omop MCP

OMOP MCP服务器

用于智能OMOP通用数据模型（CDM）探索、概念发现和队列查询生成的模型上下文协议（MCP）服务器。

  

______________________________________________________________________

🎯 这个MCP能做什么？

此MCP服务器使AI助手（Claude、自定义代理）能够通过自然语言与OMOP CDM数据库无缝协作。下面是真实世界的用例，可以帮助您了解它的功能。

______________________________________________________________________

📋 用例

1. 🔍 临床概念发现

场景：您需要找到临床术语的OMOP标准概念。

你可以问：

• *“查找2型糖尿病的所有OMOP概念”*
• *“将ICD-10代码E11.9映射到OMOP标准概念”*
• *“流感的SNOMED代码是什么？”*

发生了什么：

1. MCP搜索ATHENA词汇服务
2. 返回带有ID、名称、词汇表和域的标准概念
3. 显示关系（父概念、映射、子概念）
4. 按域（条件、药物、程序）筛选（如果指定）

输出示例：

{
  "concepts": [
    {
      "concept_id": 201826,
      "concept_name": "Type 2 diabetes mellitus",
      "domain_id": "Condition",
      "vocabulary_id": "SNOMED",
      "standard_concept": "S"
    }
  ],
  "relationships": {...}
}

______________________________________________________________________

2. 📊 患者计数查询

场景：您想知道您的数据库中有多少患者患有特定疾病。

你可以问：

• *“有多少病人得了流感？”*
• *“统计2型糖尿病患者”*
• *“有多少人服用二甲双胍？”*

发生了什么：

1. MCP发现相关的OMOP概念（例如流感→ 概念ID 4171852、4171853）
2. 根据数据库生成并验证SQL查询
3. 可选地执行查询并返回结果
4. 显示估计的查询成本（对于BigQuery）

工作流程示例：

User: "How many patients with flu?"
→ Step 1: discover_concepts("flu") → [4171852, 4171853]
→ Step 2: query_omop(type="count", concept_ids=[...]) → {"patient_count": 1234}

______________________________________________________________________

3. 🧬 人口结构细分

场景：你需要对患有某种疾病的患者进行人口统计分析。

你可以问：

• *“显示糖尿病患者的年龄和性别分布”*
• *“按人口统计数据细分流感患者”*
• *“服用他汀类药物的患者的年龄分布如何？”*

发生了什么：

1. MCP找到相关概念
2. 为人口统计数据生成SQL联接人员表
3. 按性别和年龄分组
4. 返回患者计数明细

输出示例：

{
  "results": [
    {"gender_concept_id": 8507, "age_years": 65, "patient_count": 145},
    {"gender_concept_id": 8532, "age_years": 58, "patient_count": 132},
    ...
  ]
}

______________________________________________________________________

4. 🔗 概念关系探索

场景：您需要探索概念层次结构和映射。

你可以问：

• *“向我展示‘糖尿病’下的所有儿童概念”*
• *“ICD-10 E11.9在SNOMED中对应什么？”*
• *“查找二甲双胍500mg的母体概念”*

发生了什么：

1. MCP从ATHENA获取关系
2. 按关系类型筛选（映射到、子项、Is a）
3. 返回层次概念树
4. 显示词汇人行横道（ICD-10→ SNOMED等）

用途：

• 构建全面的概念集
• 理解词汇映射
• 制定纳入/排除标准

______________________________________________________________________

5. 💊 队列SQL生成

场景：你需要用时间逻辑定义一个研究队列。

你可以问：

• *“为90天内出现急性肾损伤的二甲双胍患者生成SQL”*
• *“创建一组1年内中风的糖尿病患者”*
• *“查找暴露X后为结果Y的患者”*

发生了什么：

1. MCP使用概念ID进行暴露和结果
2. 生成具有时间约束的SQL
3. 包括重复数据删除逻辑（每位患者首次接触）
4. 验证查询和估算成本
5. 返回适用于您的平台（BigQuery或Postgres）的可执行SQL

SQL输出示例：

WITH exposure AS (
  SELECT DISTINCT person_id, drug_exposure_start_date AS exposure_date
  FROM drug_exposure
  WHERE drug_concept_id IN (1503297)  -- Metformin
),
outcome AS (
  SELECT DISTINCT person_id, condition_start_date AS outcome_date
  FROM condition_occurrence
  WHERE condition_concept_id IN (46271022)  -- Acute kidney injury
),
cohort AS (
  SELECT e.person_id, e.exposure_date, o.outcome_date,
    DATE_DIFF(o.outcome_date, e.exposure_date, DAY) AS days_to_outcome
  FROM exposure e
  INNER JOIN outcome o ON e.person_id = o.person_id
  WHERE e.exposure_date <= o.outcome_date
    AND DATE_DIFF(o.outcome_date, e.exposure_date, DAY) <= 90
)
SELECT * FROM cohort
QUALIFY ROW_NUMBER() OVER (PARTITION BY person_id ORDER BY exposure_date) = 1;

______________________________________________________________________

6. 🌐 跨词汇映射

场景：您有来自多个词汇表的代码，需要对其进行标准化。

你可以问：

• *将这些ICD-10代码映射到SNOMED:E11.9、E10.9、I10*
• *“将RxNorm代码转换为OMOP标准药物概念”*
• *“这些ICD-9代码的标准等效物是什么？”*

发生了什么：

1. MCP在源词汇表中搜索每个代码
2. 遵循标准概念的“映射到”关系
3. 返回源概念和标准概念
4. 确保所有概念都已准备好进行队列查询

为什么重要：

• 电子健康记录使用不同的编码系统
• OMOP要求查询使用标准概念
• 一个查询从所有源词汇表中捕获数据

______________________________________________________________________

7. 🔄 多后端可移植性

场景：不同的数据库平台需要相同的队列查询。

你可以问：

• *“为BigQuery和Postgres生成此队列查询”*
• *“显示在BigQuery和Postgres上运行此程序的成本差异”*

发生了什么：

1. MCP生成特定方言的SQL
2. BigQuery版本使用UNNEST、QUALIFY、回溯引用表
3. Postgres版本使用数组、子查询、schema.table格式
4. 两者都从OMOP CDM数据中返回相同的结果

支持的后端：

• ✅ BigQuery （全面支持、成本估算、模拟运行验证）
• ✅ 雪花 （全面支持，EXPLAIN验证，企业就绪）
• ✅ 鸭数据库 （完全支持，本地执行，零设置，免费！）
• ✅ 通用SQL转换 （通过SQLGlot使用10多种方言）

______________________________________________________________________

8. 💰 成本估算与验证

场景：在运行昂贵的分析之前，您需要检查查询成本。

你可以问：

• *“查询所有有心血管事件的糖尿病患者需要多少费用？”*
• *“在运行此队列查询之前估算成本”*
• *“验证此SQL而不执行它”*

发生了什么：

1. MCP运行BigQuery模拟运行验证
2. 返回扫描的估计字节数
3. 计算近似成本（BigQuery：$5/TB）
4. 如果成本超过配置的阈值（默认值：$1），则发出警告
5. 需要确认昂贵的查询

安全特性：

• 🚫 阻止超过成本限制的查询
• 📊 显示查询计划详细信息
• ⏱️ 估计执行时间
• 🔒 防止意外昂贵的跑步

______________________________________________________________________

9. 🛡️ 安全、受控的查询

场景：您需要企业级安全和审计跟踪。

你可以问：

• *“使用我团队的凭据运行此查询”*
• *“显示已执行查询的审核日志”*
• *“检查我是否有权列出患者ID”*

发生了什么：

1. MCP验证OAuth2.1承载令牌
2. 检查用户角色和权限
3. 在生产中阻止PHI返回查询（例如，患者列表）
4. 使用用户ID、时间戳、成本、结果记录所有查询
5. 强制行限制（最多1000行）
6. 阻止更改查询（DELETE、UPDATE、DROP）

安全控制：

• 🔐 OAuth2.1身份验证
• 👥 基于角色的授权
• 📝 完整的审计跟踪
• 🚫 突变阻断
• 💵 成本上限
• ⏰ 查询超时（默认30秒）

______________________________________________________________________

10. 🧪 探索性数据分析

场景：您正在探索一个新的OMOP数据集，并想了解其中的内容。

你可以问：

• *“此数据库中最常见的10种情况是什么？”*
• *“显示药物暴露分布”*
• *“数据集中患者的年龄范围是多少？”*
• *“有多少患者有登记数据？”*

发生了什么：

1. MCP生成探索性SQL查询
2. 跨核心OMOP表运行聚合
3. 返回汇总统计信息
4. 帮助您了解数据完整性

非常适合：

• 数据质量评估
• 研究可行性分析
• 了解数据覆盖范围
• 识别常见与罕见疾病

______________________________________________________________________

✨ 新功能

🤖 AI驱动的代理（PydanticAI）

OMOP查询的自然语言接口，具有智能概念发现和SQL生成功能：

from omop_mcp.agents import ConceptDiscoveryAgent, SQLGenerationAgent

# AI-powered concept discovery
agent = ConceptDiscoveryAgent()
result = await agent.run("Find all diabetes medications")
# Returns structured list of drug concepts with confidence scores

# AI-powered SQL generation
sql_agent = SQLGenerationAgent()
cohort_sql = await sql_agent.run(
    "Patients on metformin who developed kidney problems within 90 days"
)
# Returns complete, validated cohort SQL

🗄️ 多数据库支持

通过自动SQL转换跨不同数据库平台查询OMOP CDM：

DuckDB（地方发展）：

from omop_mcp.backends import DuckDBBackend

# Zero setup - works immediately!
backend = DuckDBBackend()  # In-memory by default
results = await backend.execute_query("SELECT COUNT(*) FROM person")
# Fast, free, local execution - perfect for development!

雪花（企业）：

from omop_mcp.backends import SnowflakeBackend

# Enterprise cloud data warehouse
backend = SnowflakeBackend()
parts = await backend.build_cohort_sql(
    exposure_ids=[1234],
    outcome_ids=[5678],
    pre_outcome_days=30
)
# Production-ready with EXPLAIN validation

BigQuery（云规模）：

from omop_mcp.backends import BigQueryBackend

# Google Cloud Platform
backend = BigQueryBackend()
validation = await backend.validate_sql(sql)  # Dry-run with cost estimate

🔄 通用SQL转换

自动在10多种SQL方言之间翻译查询：

from omop_mcp.backends import translate_sql

# Translate BigQuery SQL to Snowflake
bigquery_sql = "SELECT DATE_DIFF(end_date, start_date, DAY) FROM visits"
snowflake_sql = translate_sql(bigquery_sql, "bigquery", "snowflake")
# Result: "SELECT DATEDIFF(DAY, start_date, end_date) FROM visits"

# Translate to DuckDB
duckdb_sql = translate_sql(bigquery_sql, "bigquery", "duckdb")
# Result: "SELECT date_diff('day', start_date, end_date) FROM visits"

支持的方言：

• BigQuery、雪花、DuckDB
• PostgreSQL、MySQL、SQLite
• 红移、Spark、Trino、Presto

📊 数据导出工具

以标准格式导出OMOP数据：

from omop_mcp.tools.export import (
    export_concept_set,
    export_sql_query,
    export_query_results,
    export_cohort_definition
)

# Export concept set to CSV
await export_concept_set(
    concepts=concept_list,
    format="csv",
    output_path="diabetes_concepts.csv"
)

# Export SQL query with metadata
await export_sql_query(
    sql=cohort_sql,
    metadata={"description": "Diabetes cohort", "author": "researcher"},
    output_path="cohort_query.json"
)

# Export query results
await export_query_results(
    results=query_results,
    format="json",
    output_path="cohort_results.json",
    include_metadata=True
)

支持的格式： JSON、CSV，具有自动类型处理功能

______________________________________________________________________

🚀 快速开始

先决条件

• Python 3.11+ （使用现代字体提示）
• OMOP CDM数据库 （BigQuery或Postgres）
• 雅典API访问 （公开，无需密钥）

安装

使用UV Extras快速安装

# Local development (DuckDB only)
uv pip install omop-mcp[duckdb]

# Cloud analytics (BigQuery + Snowflake)
uv pip install omop-mcp[cloud]

# All backends
uv pip install omop-mcp[all-backends]

# Development with all tools
uv pip install omop-mcp[dev,all-backends]

传统安装

# Install from PyPI (when published)
pip install omop-mcp

# Or install from source
git clone https://github.com/aandresalvarez/omop-mcp.git
cd omop-mcp
uv sync

配置

复制 .env.example 到 .env 并根据您的环境进行定制：

cp .env.example .env

或者直接设置环境变量：

# Required: Database backend
BACKEND_TYPE=bigquery  # or "snowflake" or "duckdb"

# For BigQuery
BIGQUERY_PROJECT_ID=your-gcp-project
BIGQUERY_DATASET_ID=omop_cdm
BIGQUERY_CREDENTIALS_PATH=/path/to/service-account.json

# Alternative: Use Application Default Credentials (ADC)
# BIGQUERY_CREDENTIALS_PATH=  # Leave empty to use ADC
#
# ADC Authentication Methods:
# 1. User credentials (development):
#    gcloud auth application-default login
#
# 2. Service account via environment variable:
#    export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
#
# 3. Metadata service (GCP environments - Cloud Run, Compute Engine, etc.):
#    Automatically available - no additional setup needed
#
# 4. Workload Identity (Kubernetes):
#    Configured via service account annotations

# For Snowflake
SNOWFLAKE_ACCOUNT=your-account.snowflakecomputing.com
SNOWFLAKE_USER=your_username
SNOWFLAKE_PASSWORD=your_password
SNOWFLAKE_DATABASE=omop_db
SNOWFLAKE_SCHEMA=cdm
SNOWFLAKE_WAREHOUSE=compute_wh

# For DuckDB (local/embedded - no credentials needed!)
DUCKDB_DATABASE_PATH=:memory:  # or "./omop.duckdb" for persistent
DUCKDB_SCHEMA=main

# Optional: Security
MAX_COST_USD=1.0          # Cost limit for BigQuery queries
MAX_QUERY_TIMEOUT_SEC=30  # Query timeout
PHI_MODE=false            # Set true to allow patient_id queries

# Optional: OAuth (for production)
OAUTH_ISSUER=https://your-auth-provider.com
OAUTH_AUDIENCE=omop-mcp-api

运行服务器

身份验证方法

OMOP MCP服务器支持BigQuery访问的多种身份验证方法：

方法1：服务帐户（推荐用于生产）

# Download service account key
gcloud iam service-accounts keys create omop-mcp-key.json \
    --iam-account=omop-mcp-server@your-project-id.iam.gserviceaccount.com

# Set environment variable
BIGQUERY_CREDENTIALS_PATH=/path/to/omop-mcp-key.json

方法2：应用程序默认凭据（ADC）

# Option 1: User credentials (development)
gcloud auth application-default login

# Option 2: Service account via environment variable
export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json

# Option 3: Metadata service (GCP environments)
# Automatically available in Cloud Run, Compute Engine, etc.

# Leave credentials path empty to use ADC
BIGQUERY_CREDENTIALS_PATH=

身份验证优先级：

1. 服务帐户JSON文件（如果 BIGQUERY_CREDENTIALS_PATH 已设置且文件存在）
2. 应用程序默认凭据（ADC）

- GOOGLE_APPLICATION_CREDENTIALS 环境变量 - 元数据服务（GCP环境） - 用户凭据(gcloud auth application-default login)

启动服务器

选项1：作为MCP服务器（用于Claude Desktop等）

添加到您的MCP客户端配置中（例如。， claude_desktop_config.json):

{
  "mcpServers": {
    "omop": {
      "command": "uv",
      "args": ["run", "omop-mcp"],
      "env": {
        "BIGQUERY_PROJECT_ID": "your-project",
        "BIGQUERY_DATASET_ID": "omop_cdm",
        "BIGQUERY_CREDENTIALS_PATH": "/path/to/credentials.json"
      }
    }
  }
}

重新启动Claude Desktop，您将看到可用的OMOP工具。

选项2：直接使用Python

import asyncio
from omop_mcp.tools.athena import discover_concepts
from omop_mcp.tools.query import query_by_concepts

async def main():
    # Step 1: Discover concepts
    result = await discover_concepts(
        query="type 2 diabetes",
        domain="Condition",
        standard_only=True
    )

    print(f"Found {len(result.concepts)} concepts:")
    for concept in result.concepts:
        print(f"  - {concept.concept_name} ({concept.concept_id})")

    # Step 2: Query database
    concept_ids = [c.concept_id for c in result.concepts]
    query_result = await query_by_concepts(
        query_type="count",
        concept_ids=concept_ids,
        domain="Condition",
        backend="bigquery",
        execute=True
    )

    print(f"\nSQL Generated:\n{query_result.sql}")
    print(f"\nPatient count: {query_result.results[0]['patient_count']}")
    print(f"Estimated cost: ${query_result.estimated_cost_usd:.4f}")

asyncio.run(main())

选项3：作为独立服务器

# Run MCP server on stdio
uv run python -m omop_mcp.server

# Or with explicit backend
BACKEND_TYPE=postgres uv run python -m omop_mcp.server

______________________________________________________________________

🛠️ 可用的MCP工具

核心工具

直接SQL工具（新！🎉)

导出工具（新增！🎉)

SQL工具（新增！🎉)

AI代理工具（新！🤖)

资源（可缓存数据）

提示（AI指导）

______________________________________________________________________

📚 文档

核心文件

• SQL验证和安全 -全面的安全功能、表满列表、PHI保护
• 执行和缓存架构 -性能优化、缓存策略、监控
• API 参考 -完整的工具文档和示例

集成指南

• Claude桌面集成 -Claude Desktop的完整设置指南
• LibreChat+Ollama集成 -通过LibreChat和Ollama进行本地部署
• 通用MCP客户端指南 -与任何兼容MCP的客户端集成

配置

• 环境配置 -带示例的完整配置参考
• 全面配置指南 -所有环境、数据库和AI客户端的详细设置
• UV附加组件安装 -后端特定安装选项

______________________________________________________________________

🔒 安全功能

OMOP MCP服务器实施了全面的安全措施来保护医疗数据：

SQL安全层

• 只允许SELECT语句 -阻止所有变异操作（DELETE、UPDATE、DROP等）
• OMOP表格列表 -仅限制访问已批准的OMOP CDM表
• PHI列阻塞 -防止访问敏感的源值列
• 自动行限制 -防止过度的数据检索
• 成本验证 -具有成本限制的BigQuery模拟运行验证

配置选项

# Enable strict table validation
STRICT_TABLE_VALIDATION=true

# Block PHI columns
OMOP_BLOCKED_COLUMNS=person_source_value,provider_source_value

# Set cost limits
MAX_COST_USD=1.0

# Disable patient ID queries in production
ALLOW_PATIENT_LIST=false

错误类型

• SecurityViolationError -检测到危险的SQL操作
• TableNotAllowedError -访问了未分配的表
• ColumnBlockedError -访问被阻止的PHI列
• CostLimitExceededError -查询成本超出限制

看 SQL验证文档 了解完整的安全细节。

______________________________________________________________________

📚 详细示例

示例1：基本概念发现

from omop_mcp.tools.athena import discover_concepts

# Search for flu concepts
result = await discover_concepts(
    query="influenza",
    domain="Condition",
    standard_only=True,
    limit=10
)

print(f"Found {len(result.concepts)} concepts")
for concept in result.concepts:
    print(f"{concept.concept_id}: {concept.concept_name}")
    print(f"  Domain: {concept.domain_id}, Vocabulary: {concept.vocabulary_id}")
    print(f"  Standard: {concept.is_standard()}, Valid: {concept.is_valid()}")

输出：

Found 3 concepts
4171852: Influenza
  Domain: Condition, Vocabulary: SNOMED
  Standard: True, Valid: True
4171853: Influenza due to seasonal influenza virus
  Domain: Condition, Vocabulary: SNOMED
  Standard: True, Valid: True

示例2：患者计数查询

from omop_mcp.tools.query import query_by_concepts

# Count patients with diabetes (concept IDs from discovery)
result = await query_by_concepts(
    query_type="count",
    concept_ids=[201826, 201254],  # Type 2 diabetes concepts
    domain="Condition",
    backend="bigquery",
    execute=False  # Dry-run first
)

print(f"SQL: {result.sql}")
print(f"Estimated cost: ${result.estimated_cost_usd:.4f}")
print(f"Estimated bytes: {result.estimated_bytes:,}")

# If cost acceptable, execute
if result.estimated_cost_usd < 0.10:
    result = await query_by_concepts(
        query_type="count",
        concept_ids=[201826, 201254],
        domain="Condition",
        backend="bigquery",
        execute=True  # Actually run it
    )
    print(f"Patient count: {result.results[0]['patient_count']}")

示例3：人口统计细分

# Get age/gender breakdown for diabetes patients
result = await query_by_concepts(
    query_type="breakdown",
    concept_ids=[201826],
    domain="Condition",
    backend="bigquery",
    execute=True
)

print("Demographics:")
for row in result.results:
    gender = "Male" if row['gender_concept_id'] == 8507 else "Female"
    print(f"  {gender}, Age {row['age_years']}: {row['patient_count']} patients")

输出：

Demographics:
  Male, Age 65: 145 patients
  Female, Age 58: 132 patients
  Male, Age 72: 98 patients
  ...

示例4：多步骤工作流（发现→ 查询）

async def analyze_condition(condition_name: str):
    """Complete workflow: discover concepts and query database."""

    # Step 1: Discover concepts
    print(f"Discovering concepts for '{condition_name}'...")
    discovery = await discover_concepts(
        query=condition_name,
        domain="Condition",
        standard_only=True
    )

    if not discovery.concepts:
        print("No concepts found!")
        return

    print(f"Found {len(discovery.concepts)} concepts:")
    for c in discovery.concepts:
        print(f"  - {c.concept_name} ({c.concept_id})")

    # Step 2: Estimate query cost
    concept_ids = [c.concept_id for c in discovery.concepts]
    print("\nEstimating query cost...")
    estimate = await query_by_concepts(
        query_type="count",
        concept_ids=concept_ids,
        domain="Condition",
        backend="bigquery",
        execute=False
    )
    print(f"Estimated cost: ${estimate.estimated_cost_usd:.4f}")

    # Step 3: Execute if cost acceptable
    if estimate.estimated_cost_usd < 1.0:
        print("\nExecuting query...")
        result = await query_by_concepts(
            query_type="count",
            concept_ids=concept_ids,
            domain="Condition",
            backend="bigquery",
            execute=True
        )
        patient_count = result.results[0]['patient_count']
        print(f"✅ Found {patient_count:,} patients with {condition_name}")
    else:
        print("❌ Query too expensive, skipping execution")

# Run the workflow
await analyze_condition("type 2 diabetes")

示例5：跨域查询（药物+条件）

# Find patients on Metformin who developed acute kidney injury
from omop_mcp.tools.athena import discover_concepts

# Discover drug concept
drug_result = await discover_concepts(query="metformin", domain="Drug")
drug_ids = [c.concept_id for c in drug_result.concepts]

# Discover condition concept
condition_result = await discover_concepts(query="acute kidney injury", domain="Condition")
condition_ids = [c.concept_id for c in condition_result.concepts]

# Query drug exposures
drug_query = await query_by_concepts(
    query_type="count",
    concept_ids=drug_ids,
    domain="Drug",
    backend="bigquery",
    execute=True
)

# Query condition occurrences
condition_query = await query_by_concepts(
    query_type="count",
    concept_ids=condition_ids,
    domain="Condition",
    backend="bigquery",
    execute=True
)

print(f"Patients on Metformin: {drug_query.results[0]['patient_count']}")
print(f"Patients with AKI: {condition_query.results[0]['patient_count']}")

示例6：使用MCP资源（缓存）

from omop_mcp.resources import get_concept_resource, search_concepts_resource

# Get single concept (cacheable by MCP client)
concept_resource = await get_concept_resource(concept_id=201826)
print(concept_resource)  # Returns concept with URI omop://concept/201826

# Search with pagination (cacheable)
page1 = await search_concepts_resource(
    query="diabetes",
    domain="Condition",
    cursor=None,  # First page
    page_size=50
)

print(f"Found {len(page1['concepts'])} concepts")
print(f"Next cursor: {page1['next_cursor']}")

# Get next page using cursor
page2 = await search_concepts_resource(
    query="diabetes",
    domain="Condition",
    cursor=page1['next_cursor'],
    page_size=50
)

示例7：使用MCP提示（AI引导）

from omop_mcp.prompts import get_prompt

# Get SQL generation guidance
prompt = await get_prompt(
    prompt_id="cohort/sql",
    arguments={
        "exposure": "Metformin",
        "outcome": "Acute Kidney Injury",
        "time_window": "90 days",
        "dialect": "bigquery"
    }
)

print(prompt["messages"][0]["content"]["text"])
# Returns detailed prompt with SQL template, best practices, and examples

示例8：使用DuckDB进行本地开发（新增！🎉)

from omop_mcp.backends import DuckDBBackend, translate_query

# Step 1: Develop and test locally with DuckDB (FREE!)
duckdb_backend = DuckDBBackend()  # Zero setup required!

# Build cohort SQL
parts = await duckdb_backend.build_cohort_sql(
    exposure_ids=[1503297],  # Metformin
    outcome_ids=[46271022],  # Acute kidney injury
    pre_outcome_days=90
)

# Test locally (instant, free)
local_results = await duckdb_backend.execute_query(
    parts.to_sql(),
    limit=10
)
print(f"✅ Found {len(local_results)} matching records locally")

# Step 2: Translate to production database
bigquery_sql = translate_query(parts.to_sql(), "duckdb", "bigquery")
snowflake_sql = translate_query(parts.to_sql(), "duckdb", "snowflake")

# Step 3: Run on production (after local validation)
from omop_mcp.backends import BigQueryBackend
bigquery_backend = BigQueryBackend()
validation = await bigquery_backend.validate_sql(bigquery_sql)
print(f"💰 Estimated cost: ${validation.estimated_cost_usd:.2f}")

if validation.estimated_cost_usd < 1.0:
    prod_results = await bigquery_backend.execute_query(bigquery_sql)
    print(f"🚀 Production results: {len(prod_results)} records")

为什么是这个工作流程？

• 🆓 免费本地测试 -开发期间无云成本
• ⚡ 即时迭代 -测试更改（毫秒）
• ✅ 部署前验证 -在本地捕获错误
• 💰 注重成本 -仅为生产查询付费
• 🔄 跨平台 -同样的SQL适用于BigQuery、Snowflake、DuckDB

例9:AI驱动的概念发现（新！🤖)

from omop_mcp.agents import ConceptDiscoveryAgent

# Initialize AI agent
agent = ConceptDiscoveryAgent()

# Natural language concept search
result = await agent.run(
    "Find all concepts related to type 2 diabetes and its complications"
)

print(f"Found {len(result.concepts)} concepts:")
for concept in result.concepts[:5]:
    print(f"  - {concept.concept_name} ({concept.concept_id})")
    print(f"    Domain: {concept.domain_id}, Confidence: {concept.confidence}")

# Agent automatically:
# - Understands medical context
# - Searches multiple domains
# - Filters for relevance
# - Returns structured results

示例10：基于AI的SQL生成（新增！🤖)

from omop_mcp.agents import SQLGenerationAgent

# Initialize SQL agent
agent = SQLGenerationAgent()

# Generate cohort SQL from natural language
result = await agent.run(
    description="Patients on metformin who developed acute kidney injury within 90 days",
    exposure="metformin",
    outcome="acute kidney injury"
)

print("Generated SQL:")
print(result.sql)
print(f"\nExposure concepts: {result.exposure_ids}")
print(f"Outcome concepts: {result.outcome_ids}")
print(f"Validation: {'✅ Valid' if result.validation.valid else '❌ Invalid'}")

# Agent automatically:
# - Discovers relevant concepts
# - Generates cohort SQL
# - Validates syntax
# - Returns complete, executable query

______________________________________________________________________

📊 工作流示例

研究问题： *“有多少患者在服用二甲双胍后出现急性肾损伤？”*

1. User asks the question
   ↓
2. MCP discovers concepts:
   - Metformin → 1503297 (Drug)
   - Acute kidney injury → 46271022 (Condition)
   ↓
3. MCP generates cohort SQL with 90-day temporal window
   ↓
4. MCP validates query (estimated cost: $0.08)
   ↓
5. User approves execution
   ↓
6. MCP returns results:
   - 1,234 patients
   - Median time to event: 45 days
   - SQL available for reproduction

______________________________________________________________________

🎓 谁应该使用这个？

• 临床研究人员：使用自然语言更快地建立队列
• 数据科学家：生成经过验证的SQL，而无需记忆OMOP模式
• 医疗保健分析师：交互式浏览OMOP数据集
• 情报学家：自动映射临床术语
• 研究协调员：通过快速患者计数评估可行性
• AI开发人员：将OMOP功能集成到健康AI应用程序中

______________________________________________________________________

🧪 测试

运行综合测试套件：

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=omop_mcp --cov-report=html

# Run specific test categories
uv run pytest tests/test_integration.py  # E2E workflows
uv run pytest tests/test_athena.py       # ATHENA API
uv run pytest tests/test_query_security.py  # Security guards

# Run with verbose output
uv run pytest -v

测试覆盖范围： 173次测试，100%通过✅

• 单元测试：161（模型、后端、工具、资源、提示、代理、导出、sqlgen、方言）
• 集成测试：12（E2E发现→查询工作流、多后端、跨方言翻译）
• 覆盖区域：

- ✅ 核心OMOP工具（athena、query、sqlgen） - ✅ AI代理（概念发现、SQL生成） - ✅ 导出工具（JSON、CSV、所有数据类型） - ✅ 多后端（BigQuery、Snowflake、DuckDB） - ✅ SQL翻译（10+方言） - ✅ 安全与验证

______________________________________________________________________

👨‍💻 发展

快速开始

# Clone and setup
git clone https://github.com/aandresalvarez/omop-mcp.git
cd omop-mcp

# Install with all dev dependencies
uv sync --extra dev

# Run quality checks
make check           # Format, lint, typecheck, test
make check-all       # All checks + pylint + pyright + security

质量工具

该项目使用全面的质量基础设施进行医疗保健等级代码：

代码质量

make format          # Black + ruff auto-formatting
make lint            # Ruff linting
make pylint          # Strict linting
make typecheck       # mypy type checking
make pyright         # pyright type checking

安全扫描 🔒

make security        # bandit (Python security) + pip-audit (vulnerabilities)
make audit           # Comprehensive: security + safety dependency scan

SQL质量 （对OMOP查询至关重要）

make sql-lint        # Lint SQL files with sqlfluff
make sql-fix         # Auto-fix SQL formatting

测试和覆盖范围

make test            # Run pytest
make coverage        # Detailed coverage report (HTML + JSON + terminal)

可用生成目标

make help            # Show all available commands

# Setup
make dev             # Full development setup (venv + deps + tools)

# Quality checks
make check           # Standard checks (format, lint, typecheck, test)
make check-all       # ALL checks including security
make pre-commit      # Run pre-commit hooks

# Security
make security        # Security scans (bandit + pip-audit)
make audit           # Full security audit (+ safety)

# SQL
make sql-lint        # Lint SQL with sqlfluff
make sql-fix         # Auto-format SQL

# Testing
make test            # Run tests
make coverage        # Generate coverage reports

# Server
make http            # Run MCP server (HTTP mode)
make stdio           # Run MCP server (stdio mode)

CI/CD

GitHub Actions会在每次推送时自动运行：

• ✅ 测试 （Python 3.11、3.12）
• ✅ 代码质量 （格式，lint）
• ✅ 类型检查 （mypy，pyright）
• ✅ 安全扫描 （土匪、pip审计、安全）
• ✅ SQL质量 （sqlfluff）
• ✅ 覆盖率报告

看 了解详情。

预提交钩子

安装git挂钩进行自动质量检查：

make pre-commit-install

这将在每次提交之前运行格式化、linting和类型检查。

______________________________________________________________________

🔧 故障排除

常见问题

问题： “找不到后端：bigquery”

# Solution: Install backend dependencies
uv pip install google-cloud-bigquery
# Or use DuckDB for local development (no setup required!)
export BACKEND_TYPE=duckdb

问题： “雅典API超时”

# Solution: The public ATHENA API can be slow. Increase timeout:
export ATHENA_TIMEOUT_SEC=60
# Or use cached results from MCP resources

问题： “查询超出成本限制”

# Solution: Increase cost cap or optimize query
export MAX_COST_USD=5.0
# Or run with execute=False to see SQL first

问题： “concept_ids不能为空”

# Solution: Discovery returned no results. Try broader search:
result = await discover_concepts(
    query="diabetes",  # Broader term
    standard_only=False  # Include non-standard
)

问题： “OAuth令牌无效”

# Solution: Check token format and issuer
# Token must be Bearer JWT with correct audience
export OAUTH_AUDIENCE=omop-mcp-api

调试提示

启用调试日志记录：

import structlog
structlog.configure(
    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG)
)

检查后端连接：

from omop_mcp.backends.registry import list_backends, get_backend, get_supported_dialects

# List all registered backends
backends = list_backends()
print(f"Available backends: {backends}")

# Check supported SQL dialects
dialects = get_supported_dialects()
print(f"Supported dialects: {dialects}")

# Get specific backend
backend = get_backend("duckdb")  # or "bigquery" or "snowflake"
print(f"Connected to: {backend.name} (dialect: {backend.dialect})")

测试SQL转换：

from omop_mcp.backends import translate_sql, validate_sql

# Translate SQL between dialects
bigquery_sql = "SELECT DATE_DIFF(end_date, start_date, DAY) FROM visits"
snowflake_sql = translate_sql(bigquery_sql, "bigquery", "snowflake")
print(f"Translated SQL: {snowflake_sql}")

# Validate SQL for specific dialect
is_valid, error = validate_sql(snowflake_sql, "snowflake")
print(f"Valid: {is_valid}, Error: {error}")

在不执行的情况下验证SQL：

result = await query_by_concepts(
    query_type="count",
    concept_ids=[201826],
    domain="Condition",
    backend="bigquery",
    execute=False  # SQL only, no execution
)
print(result.sql)
print(f"Cost: ${result.estimated_cost_usd}")

______________________________________________________________________

🚀 性能提示

一般提示

1. 使用MCP资源进行缓存：资源由MCP客户端缓存
2. 批量概念查找：搜索一次，查询多次
3. 以execute=False开头：运行前验证SQL和成本
4. 仅使用标准=True：减小搜索结果大小
5. 设定适当的限制：默认值为50个概念，必要时可增加

后端特定提示

DuckDB（地方发展）：

• ✅ 即时启动：内存模式(:memory:)最快的
• ✅ 免费测试：无云成本，迭代速度快
• ✅ 基于文件的持久性：使用 ./omop.duckdb 用于持久存储
• ✅ 进口拼花地板：DuckDB可以直接查询Parquet文件
• ⚡ 演出：~1-10GB数据集在几秒钟内运行

BigQuery（云规模）：

• 💰 启用查询结果缓存：结果缓存24小时（免费！）
• 📊 分区表：使用分区OMOP表降低成本
• 🔍 先进行试运行：执行前检查成本
• 💵 监控成本：设置 MAX_COST_USD 防止昂贵的查询
• ⚡ 演出：可扩展到PB

雪花（企业）：

• ❄️ 适当使用仓库大小：从X-Small for dev开始
• 🔄 启用结果缓存：Snowflake缓存相同的查询
• 📈 规模计算：根据查询复杂性调整仓库大小
• 💰 暂停仓库：5分钟不活动后自动暂停
• ⚡ 演出：非常适合复杂分析

本地→ 生产工作流

# 1. Develop locally with DuckDB (free, fast)
duckdb_backend = DuckDBBackend()
local_results = await duckdb_backend.execute_query(sql, limit=10)

# 2. Translate to production dialect
prod_sql = translate_sql(sql, "duckdb", "bigquery")

# 3. Validate cost before production run
bigquery_backend = BigQueryBackend()
validation = await bigquery_backend.validate_sql(prod_sql)
print(f"Cost: ${validation.estimated_cost_usd:.2f}")

# 4. Execute on production if cost acceptable
if validation.estimated_cost_usd < 1.0:
    prod_results = await bigquery_backend.execute_query(prod_sql)

此工作流程通过先在本地验证来节省资金和时间！

______________________________________________________________________

🔐 安全最佳实践

1. 在生产环境中使用OAuth：启用 OAUTH_ISSUER 和 OAUTH_AUDIENCE
2. 设定成本限制：默认值为1美元，根据您的预算进行调整
3. 禁用PHI模式：设置 PHI_MODE=false 阻止患者ID查询
4. 使用服务帐户：对于BigQuery，使用具有只读访问权限的专用服务帐户
5. 更喜欢云环境中的ADC：为云运行、计算引擎等使用应用程序默认凭据。
6. 启用审核日志记录：所有查询都使用structlog记录
7. 设置查询超时：默认30秒，根据需要调整

______________________________________________________________________

🔗 资源

官方文件

• OMOP通用数据模型
• ATHENA词汇浏览器
• 模型上下文协议

项目文件

• 实施计划
• 开发进度
• 集成测试

功能文档

• SQL生成模块
• PydanticAI试剂
• 导出工具
• 其他后端
• 可选增强功能摘要

代理文件

• 概念发现代理
• 查询生成器代理

______________________________________________________________________

📄 许可证

MIT许可证-请参阅 许可证 了解详情。

______________________________________________________________________

🤝 贡献

欢迎投稿！请看 贡献.md 作为指导方针。

______________________________________________________________________

💡 需要帮助？

• 📚 文档
• 💬 讨论
• 🐛 问题

______________________________________________________________________

建于❤️ OHDSI社区