Fhir MCP

🚀 FHIR Careplan - 通用FHIR服务器与工具集

这是一个全面的FHIR（快速医疗互操作性资源）服务器和工具包，专为医疗数据集成、患者护理规划和临床决策支持而设计。该项目提供了一个通用接口，可连接多个FHIR服务器，并具备先进的人工智能临床分析能力。

🚀 快速开始

本项目主要包含两个核心组件：

1. 通用FHIR MCP服务器 (fhir_server.py) - 一个模型上下文协议（MCP）服务器，用于为多个FHIR服务器提供标准化访问。
2. FHIR工具库 (fhir_tools.py) - 一个全面的Python库，用于FHIR数据的操作、分析以及基于人工智能的临床洞察。

✨ 主要特性

🔗 多服务器FHIR集成

• 通用FHIR接口：可连接多个FHIR服务器（如Epic、Cerner、HAPI、Firely等）。
• 与供应商无关：无论底层FHIR服务器的实现如何，都提供标准化的API。
• 实时连接测试：自动进行服务器健康检查和诊断。
• 智能故障转移：自动在服务器之间切换，以实现最佳性能。

🤖 人工智能驱动的临床分析

• 集成OpenAI：从自由文本中提取临床关键词和概念。
• 语义映射：将临床术语映射到标准化的FHIR代码。
• 相似患者匹配：查找具有相似临床特征的患者。
• 预测分析：根据历史数据生成护理建议。

📊 全面的患者数据访问

• 完整的患者记录：包括人口统计信息、病情、药物治疗、手术记录和就诊记录。
• 生命体征和实验室检查结果：分类观察数据，包含时间序列信息。
• 护理计划和团队信息：治疗计划和医疗服务提供者信息。
• 过敏史和手术史：完整的医疗历史跟踪。

🚀 性能优化

• 异步操作：采用非阻塞I/O，实现高性能数据访问。
• 智能缓存：缓存条件代码和常用数据。
• 批量处理：高效处理多个患者记录。
• 连接池管理：为多个服务器优化HTTP连接。

📦 安装指南

前提条件

• Python 3.11 或更高版本
• OpenAI API密钥（用于人工智能功能）
• 访问FHIR服务器（本地或远程）

设置MCP服务器

1. 克隆仓库

   git clone https://github.com/Kushagra-Dutta/Fhir-MCP.git
   cd FHIR-MCP
   

2. 安装依赖项

   pip install -r requirements.txt
   # 或者使用uv
   uv sync
   

3. 配置环境变量

   # 创建.env文件
   echo "OPENAI_API_KEY=your_openai_api_key" > .env
   

4. 运行FHIR MCP服务器

   python fhir_server.py
   

本地设置Firely测试数据库

1. 获取Firely服务器许可证密钥

   • 访问 Firely Server试用页面
   • 填写表格，通过电子邮件接收许可证密钥
   • 您将收到许可证密钥和下载文件（许可证有效期为7天）
   • 将许可证文件保存为 firelyserver-license.json

2. 使用Docker进行设置

   # 拉取Firely服务器镜像
   docker pull firely/server
   
   # 运行容器
   # 对于Windows CMD：
   docker run -d -p 9090:4080 --name firely.server -v %CD%/firelyserver-license.json:/app/firelyserver-license.json firely/server
   
   # 对于PowerShell或macOS/Linux：
   docker run -d -p 9090:4080 --name firely.server -v ${PWD}/firelyserver-license.json:/app/firelyserver-license.json firely/server
   
   # 验证容器是否正在运行
   docker ps
   

3. 加载测试数据

   • 使用Postman加载测试数据捆绑包
   • 在Postman中创建一个新的PUT请求
   • 将请求类型设置为原始JSON
   • 复制测试数据捆绑包中的内容
   • 将请求发送到基础URL (http://localhost:9090)
   • 对每个数据捆绑包重复此操作

完成这些步骤后，您将拥有一个可在 http://localhost:9090 访问的测试数据库。

设置MCP聊天机器人客户端

您可以通过分叉仓库或使用Docker来设置MCP聊天机器人客户端。该聊天机器人将作为与FHIR服务器交互的前端界面。

1. 获取聊天机器人设置

   # 克隆仓库
   git clone https://github.com/cgoinglove/mcp-client-chatbot.git
   cd mcp-client-chatbot
   

2. 安装PNPM（如果未安装）

   npm install -g pnpm
   

3. 选择设置方法

   Docker Compose设置 🐳

   # 安装依赖项
   pnpm i
   
   # 配置环境变量
   # 编辑.env文件，添加您的API密钥（在pnpm i后自动创建）
   
   # 启动所有服务，包括PostgreSQL
   pnpm docker-compose:up
   

   本地设置 🚀

   # 安装依赖项
   pnpm i
   
   # 设置环境变量
   pnpm initial:env
   
   # 启动PostgreSQL（如果已经运行则跳过）
   pnpm docker:pg
   
   # 运行数据库迁移
   pnpm db:migrate
   
   # 启动开发服务器
   pnpm dev
   
   # 可选：构建并启动以进行类似生产环境的测试
   pnpm build:local && pnpm start
   

4. 配置环境变量 创建或编辑 .env 文件，添加所需的API密钥：

   # LLM提供商API密钥（添加您计划使用的密钥）
   OPENAI_API_KEY=****
   GOOGLE_GENERATIVE_AI_API_KEY=****
   ANTHROPIC_API_KEY=****
   XAI_API_KEY=****
   OPENROUTER_API_KEY=****
   OLLAMA_BASE_URL=http://localhost:11434/api
   
   # 身份验证配置
   BETTER_AUTH_SECRET=**** # 使用：npx @better-auth/cli@latest secret生成
   BETTER_AUTH_URL= # 可选：您访问应用的URL
   
   # 数据库配置
   POSTGRES_URL=postgres://your_username:your_password@localhost:5432/your_database_name
   
   # MCP配置
   FILE_BASED_MCP_CONFIG=false
   
   # 可选的OAuth设置（用于Google/GitHub登录）
   GOOGLE_CLIENT_ID=
   GOOGLE_CLIENT_SECRET=
   GITHUB_CLIENT_ID=
   GITHUB_CLIENT_SECRET=
   

5. 将MCP服务器连接到聊天机器人

   • 在 http://localhost:3000 访问聊天机器人
   • 创建账户并登录
   • 转到MCP配置
   • 点击“添加服务器”
   • 复制并粘贴您的 .chatbot-config.json 配置

6. 配置系统提示

   • 导航到聊天机器人仓库中的 src/app/prompts.ts
   • 用 system_prompt.txt 中的自定义提示替换默认系统提示
   • 示例系统提示结构：

     export const defaultSystemPrompt = `您是一个强大的智能AI编码助手。
     您可以使用一组工具来回答用户的问题。
     
     <communication>
     1. 保持对话风格，但要专业。
     2. 用第二人称称呼用户，用第一人称称呼自己。
     3. 以Markdown格式格式化您的回复。
     4. 绝不撒谎或编造信息。
     </communication>
     
     <tool_calling>
     遵循以下关于工具调用的规则：
     1. 始终严格按照工具调用模式进行操作。
     2. 绝不调用未明确提供的工具。
     3. 仅在必要时调用工具。
     4. 在调用每个工具之前，先向用户解释调用原因。
     </tool_calling>
     
     <making_code_changes>
     进行代码更改时：
     1. 添加所有必要的导入语句和依赖项。
     2. 除非用户要求，否则绝不向用户输出代码。
     3. 使用适当的代码编辑工具进行实现。
     </making_code_changes>
     `;
     

   • 根据您的需求自定义提示部分：
     • <communication>：定义AI的交互方式
     • <tool_calling>：指定使用工具的规则
     • <making_code_changes>：设置代码修改的准则
   • 保存更改并重启开发服务器

完成这些步骤后，您的聊天机器人将连接到MCP服务器，并准备好与FHIR数据库进行交互。

💻 使用示例

基础用法

1. 服务器管理

from fhir_tools import UniversalFhirMcpServer

# 初始化服务器
fhir_server = UniversalFhirMcpServer()
await fhir_server.initialize()

# 测试服务器连接
result = await fhir_server.test_server_connectivity("firely_local")
print(result)

# 切换到不同的服务器
await fhir_server.switch_server("hapi_r4")

2. 患者搜索

# 搜索患者
search_criteria = {
    "family": "Smith",
    "given": "John",
    "birthdate": "1990-01-01"
}
patients = await fhir_server.find_patient(search_criteria)

3. 患者数据检索

# 获取全面的患者信息
patient_info = await fhir_server.get_comprehensive_patient_info("patient-123")

# 获取特定类型的数据
conditions = await fhir_server.get_patient_conditions("patient-123")
medications = await fhir_server.get_patient_medications("patient-123")
vital_signs = await fhir_server.get_vital_signs("patient-123")
lab_results = await fhir_server.get_lab_results("patient-123")

高级用法

1. 人工智能驱动的临床分析

# 从自由文本中提取临床关键词
clinical_note = "45岁女性，患有HER2阳性浸润性导管癌，III期A"
keywords = await fhir_server.extract_clinical_keywords(clinical_note)

# 将临床术语映射到FHIR代码
clinical_data = {
    "conditions": ["乳腺癌", "糖尿病"],
    "age": 45,
    "gender": "女性"
}
fhir_codes = await fhir_server.map_to_fhir_codes_fast(clinical_data)

# 查找具有相似临床特征的患者
criteria = {
    "age": 45,
    "gender": "女性",
    "conditions": ["乳腺癌"]
}
similar_patients = await fhir_server.find_similar_patients_simple(criteria)

2. MCP服务器集成

FHIR服务器可以用作MCP服务器，以与AI助手集成：

{
  "mcpServers": {
    "fhir-server": {
      "command": "python",
      "args": ["fhir_server.py"],
      "env": {
        "OPENAI_API_KEY": "your-key-here"
      }
    }
  }
}

📚 详细文档

架构

核心组件

graph TD
    A[FHIR Server] --> B[Universal FHIR MCP Server]
    B --> C[FHIR Tools Library]
    C --> D[Multiple FHIR Servers]
    C --> E[OpenAI Integration]
    C --> F[Caching Layer]
    
    D --> G[Firely Local]
    D --> H[HAPI R4]
    D --> I[Epic Systems]
    D --> J[Cerner]
    
    E --> K[Clinical Text Analysis]
    E --> L[Code Mapping]
    E --> M[Patient Matching]

服务器注册表

系统维护了一个全面的FHIR服务器注册表：

• Firely Local (http://localhost:9090) - 本地开发服务器
• HAPI R4 (http://hapi.fhir.org/baseR4) - 公共测试服务器
• Epic、Cerner、Azure - 企业医疗系统（可配置）

配置

服务器配置

系统支持在 fhir_tools.py 中配置多个FHIR服务器：

servers = {
    "firely_local": {
        "name": "Firely Server Local",
        "base_url": "http://localhost:9090",
        "version": "R4",
        "vendor": "Firely",
        "auth_type": "none"
    },
    "hapi_r4": {
        "name": "HAPI FHIR R4 Public",
        "base_url": "http://hapi.fhir.org/baseR4",
        "version": "R4",
        "vendor": "HAPI",
        "auth_type": "none"
    }
}

环境变量

# 必需
OPENAI_API_KEY=your_openai_api_key

# 可选
FHIR_SERVER_URL=http://localhost:9090
FHIR_SERVER_AUTH_TOKEN=your_auth_token

可用工具

核心FHIR操作

• switch_server(server_name) - 在FHIR服务器之间切换
• test_server_connectivity(server_name) - 测试服务器连接
• find_patient(search_criteria) - 搜索患者
• get_comprehensive_patient_info(patient_id) - 获取完整的患者数据

临床数据访问

• get_patient_observations(patient_id) - 获取患者观察数据
• get_patient_conditions(patient_id) - 获取患者病情信息
• get_patient_medications(patient_id) - 获取患者用药信息
• get_vital_signs(patient_id) - 获取生命体征数据
• get_lab_results(patient_id) - 获取实验室检查结果
• get_patient_encounters(patient_id) - 获取患者就诊记录
• get_patient_allergies(patient_id) - 获取患者过敏史
• get_patient_procedures(patient_id) - 获取患者手术记录

人工智能驱动的分析

• extract_clinical_keywords(text) - 从文本中提取临床信息
• map_to_fhir_codes_fast(clinical_data) - 将术语映射到FHIR代码
• find_similar_patients_simple(criteria) - 查找相似患者
• extract_condition_codes_from_fhir() - 提取所有病情代码

系统管理

• list_available_servers() - 列出所有配置的服务器
• get_server_registry() - 获取完整的服务器注册表
• diagnose_fhir_server(server_name) - 诊断服务器功能
• clear_condition_cache() - 清除病情代码缓存
• get_condition_cache_stats() - 获取缓存性能统计信息

测试工具

MCP检查器

MCP（模型上下文协议）检查器是一个强大的开发和测试工具，可帮助调试和验证FHIR服务器交互。它提供对服务器行为和API响应的实时检查。

使用MCP检查器

# 在特定文件上运行MCP检查器
mcp dev fhir_server.py

# 这将：
# 1. 以开发模式启动检查器
# 2. 监控所有FHIR服务器交互
# 3. 将详细信息记录到logs/mcp_dev_inspector.log

功能

• 实时监控：实时观察FHIR服务器交互
• 请求/响应日志记录：详细记录所有API调用和响应
• 错误检测：立即反馈API错误或配置错误
• 性能指标：跟踪响应时间和服务器性能
• 调试模式：增强日志记录，用于开发故障排除

日志文件位置

检查器将详细日志写入：

logs/mcp_dev_inspector.log

最佳实践

• 在开发过程中使用MCP检查器验证服务器行为
• 监控日志文件，以发现意外错误或性能问题
• 在实现新的FHIR端点时运行检查器
• 使用它来调试与外部FHIR服务器的连接问题

业务应用

医院参与平台

该工具包可支持自动化医院参与平台：

1. 患者护理协调

   • 自动生成护理计划
   • 管理治疗时间表
   • 多学科团队协调

2. 临床决策支持

   • 基于证据的治疗建议
   • 风险评估和早期预警
   • 根据类似病例进行结果预测

3. 资源优化

   • 资源分配的预测分析
   • 自动调度和容量管理
   • 通过数据驱动的洞察实现成本优化

4. 质量改进

   • 自动跟踪质量指标
   • 合规性监控和报告
   • 性能分析和基准测试

FHIR时间线代理

概述

FHIR时间线代理 (fhir_timeline_agent.py) 是一个专门的代理，用于根据患者查询，使用真实的FHIR数据生成详细的临床治疗时间线。它专门配置为与Firely本地FHIR服务器配合使用。

特性

• 自然语言处理：将自由文本患者查询转换为结构化临床数据。
• 真实患者数据分析：使用实际的FHIR患者记录生成时间线。
• 人工智能驱动的时间线生成：利用OpenAI GPT - 4生成准确的临床时间线。
• 交互式CLI界面：用户友好的命令行界面，具有丰富的格式。
• 全面的患者匹配：根据年龄、性别和病情查找相似患者。

使用方法

# 以交互式聊天界面模式运行
python fhir_timeline_agent.py

# 以演示模式运行，使用示例查询
python fhir_timeline_agent.py --demo

示例查询

• "45岁男性，患有胰腺癌"
• "62岁女性，患有HER2阳性乳腺癌"
• "58岁男性，患有IIIA期肺腺癌"

时间线生成过程

1. 查询处理

   • 从自然语言中提取临床关键词。
   • 识别年龄、性别、病情、阶段和生物标志物。

2. FHIR代码映射

   • 将临床术语映射到标准化的FHIR代码。
   • 使用Firely本地服务器的代码系统。

3. 相似患者搜索

   • 在FHIR数据库中查找匹配的患者。
   • 根据年龄、性别和病情对匹配结果进行评分。

4. 数据聚合

   • 收集全面的医疗历史。
   • 包括手术记录、用药记录和就诊记录。

5. 时间线生成

   • 使用人工智能创建详细的治疗时间线。
   • 按时间顺序组织事件，并提供临床背景信息。

输出格式

代理生成丰富的格式化输出，包括：

• 患者概况：人口统计信息、诊断、阶段、生物标志物。
• 治疗时间线：按日期排列的逐步临床事件。
• 临床结果：治疗反应、生存状态、毒性。
• 数据来源：服务器信息和分析指标。

配置

该代理硬编码使用：

• Firely本地FHIR服务器 (http://localhost:9090)
• OpenAI GPT - 4进行时间线生成
• 丰富的控制台输出进行格式化显示

要求

• OpenAI API密钥（在 .env 文件中设置）
• 运行中的Firely本地FHIR服务器
• Python包：

  openai>=1.86.0
  rich>=13.7.0
  python-dotenv>=1.0.0
  aiohttp>=3.8.0
  

最佳实践

• 在查询中提供完整的患者信息。
• 包括年龄、性别和主要诊断。
• 如有可用，添加阶段和生物标志物信息。
• 使用特定的临床术语以获得更好的匹配效果。

错误处理

代理包含强大的错误处理机制，用于处理：

• 患者信息缺失
• FHIR服务器连接问题
• 人工智能生成失败
• 数据解析错误

每个错误都会显示有用的解决建议。

🔧 技术细节

安全与合规

• 符合HIPAA标准：设计时考虑了医疗数据隐私。
• 安全通信：所有服务器通信均采用HTTPS/TLS加密。
• 认证支持：支持多种认证方法（OAuth、API密钥等）。
• 审计日志：全面的日志记录，用于合规性和调试。

性能特性

• 异步操作：非阻塞I/O，实现高吞吐量。
• 连接池管理：高效的HTTP连接管理。
• 智能缓存：缓存病情代码和元数据。
• 批量处理：高效处理多个记录。
• 错误处理：强大的错误处理和重试机制。

测试

运行测试

# 运行基本连接测试
python -c "
import asyncio
from fhir_tools import UniversalFhirMcpServer

async def test():
    server = UniversalFhirMcpServer()
    await server.initialize()
    result = await server.test_server_connectivity('firely_local')
    print(result)

asyncio.run(test())
"

服务器诊断

# 全面的服务器诊断
diagnostics = await fhir_server.diagnose_fhir_server("firely_local")
print(diagnostics)

日志记录

系统提供全面的日志记录：

import logging

# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 日志自动写入logs/mcp_dev_inspector.log

🤝 贡献指南

1. 分叉仓库
2. 创建功能分支
3. 进行更改
4. 为新功能添加测试
5. 提交拉取请求

📄 许可证

本项目采用MIT许可证 - 详情请参阅 LICENSE 文件。

🆘 支持

如有支持需求或问题：

• 在GitHub仓库中创建问题
• 查看日志目录以获取详细的错误信息
• 使用诊断工具进行服务器故障排除

🔄 版本历史

• v0.1.0 - 初始版本，具备核心FHIR功能
• v0.2.0 - 添加人工智能驱动的临床分析功能
• v0.3.0 - 增强多服务器支持和缓存功能

🎯 路线图

• [ ] 高级分析仪表盘
• [ ] 实时数据流
• [ ] 机器学习模型集成
• [ ] 增强安全功能
• [ ] 移动应用支持
• [ ] 云部署模板

---

怀着对医疗互操作性和患者护理改善的热爱而构建