Fhir MCP

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编码助手。
      您可以使用一组工具来回答用户的问题。
      
      
      1. 保持对话风格,但要专业。
      2. 用第二人称称呼用户,用第一人称称呼自己。
      3. 以Markdown格式格式化您的回复。
      4. 绝不撒谎或编造信息。
      
      
      
      遵循以下关于工具调用的规则:
      1. 始终严格按照工具调用模式进行操作。
      2. 绝不调用未明确提供的工具。
      3. 仅在必要时调用工具。
      4. 在调用每个工具之前,先向用户解释调用原因。
      
      
      
      进行代码更改时:
      1. 添加所有必要的导入语句和依赖项。
      2. 除非用户要求,否则绝不向用户输出代码。
      3. 使用适当的代码编辑工具进行实现。
      
      `;
      
    • 根据您的需求自定义提示部分:
      • :定义AI的交互方式
      • :指定使用工具的规则
      • :设置代码修改的准则
    • 保存更改并重启开发服务器

完成这些步骤后,您的聊天机器人将连接到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 - 增强多服务器支持和缓存功能

🎯 路线图

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

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

  • 0 关注
  • 0 收藏,29 浏览
  • system 提出于 2025-10-03 19:36

相似服务问题

相关AI产品