Skip to main content
Glama

MCPDischarge — 跨部门 MCP 互操作性

EHR × 药房 × 计费 | RBAC | PHI 边界 | FastMCP

CitiusTech 生成式 AI 与智能体 AI 培训 — 项目 5


传统 API 无法解决的问题

患者准备出院。数据必须在三个从未共享过通用协议的部门之间流动:

Traditional workflow (45 minutes, 15 manual handoffs):
  Ward nurse    → prints discharge note
  Ward nurse    → phones pharmacy to check drug availability
  Pharmacy      → calls back 2 hours later (drug out of stock)
  Nurse         → calls doctor to re-prescribe
  Doctor        → updates chart
  Nurse         → re-contacts pharmacy
  Pharmacy      → dispenses (brand name ≠ generic name — wrong drug dispensed?)
  Nurse         → separately calls billing department
  Billing clerk → manually re-enters ICD-10 codes from printed note
  Billing clerk → can see full medication list including controlled substances (HIPAA risk)
  Patient       → waits, often 4–6 hours post-clinical-readiness

MCP(模型上下文协议)通过标准化的、类型化的、强制执行 RBAC 的工具调用层解决了这个问题:

MCP workflow (< 1 second, automated):
  DischargeAgent.EHR.get_discharge_medications()           ← structured, not free text
  DischargeAgent.Pharmacy.check_stock()                    ← semantic name matching
  DischargeAgent.Pharmacy.get_alternative()                ← out-of-stock resolution
  DischargeAgent.EHR.get_billing_safe_summary()            ← PHI stripped at source
  DischargeAgent.Billing.generate_invoice()                ← billing never sees clinical notes

Related MCP server: FHIR MCP Server

架构

┌────────────────────────────────────────────────────────────────┐
│                 Discharge Coordination Agent                    │
│                   (MCP Client — role: discharge_coordinator)   │
└────────┬───────────────────┬───────────────────┬──────────────┘
         │ MCP calls         │ MCP calls          │ MCP calls
         ▼                   ▼                    ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│  EHR MCP Server │ │ Pharmacy Server  │ │ Billing Server   │
│  (port 8001)    │ │ (port 8002)      │ │ (port 8003)      │
│                 │ │                  │ │                  │
│ Tools:          │ │ Tools:           │ │ Tools:           │
│ • discharge_meds│ │ • check_stock    │ │ • get_charges    │
│ • diagnosis_cod │ │ • get_alternative│ │ • get_insurance  │
│ • billing_safe  │ │ • get_price      │ │ • gen_invoice    │
│   _summary      │ │ • dispense_req   │ │                  │
│ [RBAC enforced] │ │ [RBAC enforced]  │ │ [RBAC enforced]  │
└─────────────────┘ └─────────────────┘ └─────────────────┘

PHI Boundary:
  EHR → Billing path uses get_billing_safe_summary()
  PHI fields blocked: name, DOB, MRN, discharge_note, attending_physician
  Billing receives: ICD-10 codes, LOS, ward — non-PHI operational data only

RBAC 策略矩阵

角色

EHR 临床记录

EHR 药物

EHR 诊断代码

药房

计费

discharge_coordinator

billing_agent

✗ 已屏蔽

✗ 已屏蔽

仅价格

pharmacy_agent

✗ 已屏蔽

clinical_agent

库存检查

✗ 已屏蔽

每次工具调用在返回数据前都会验证调用者的角色。未经授权的调用会引发 RBACError 并记录到遥测馈送中。


快速入门

第 1 步:安装依赖项

pip install -r requirements.txt

第 2 步:生成数据

cd data/
python generate_dataset.py

第 3 步:运行服务器

FastMCP HTTP 服务器(生产环境风格,异步 MCP 智能体所需):

# Terminal 1:
python src/servers/mcp_servers.py --server ehr

# Terminal 2:
python src/servers/mcp_servers.py --server pharmacy

# Terminal 3:
python src/servers/mcp_servers.py --server billing

或者在单个进程中运行所有三个(启动 3 个后台线程):

python src/servers/mcp_servers.py --all

直接使用 Python(无 HTTP,仅供培训):

from src.servers.ehr_server import EHRServer

ehr = EHRServer()
meds = ehr.get_discharge_medications("PAT-001", role="discharge_coordinator")

第 4 步:运行出院智能体

python src/agents/discharge_agent.py PAT-001
python src/agents/discharge_agent.py PAT-003

第 5 步:完整演示

python demo/demo.py               # Runs 4 scenarios
python demo/demo.py --scenario 3  # RBAC violation only

聊天 UI (React)

本仓库包含一个简单的 React 聊天前端,它调用一个轻量级的 FastAPI 网关,该网关进而调用 MCP 服务器。

1) 启动 MCP 服务器 (SSE)

python src/servers/mcp_servers.py --all

2) 启动聊天网关 API (端口 8000)

copy .env.example .env   # then fill in Azure OpenAI settings (optional)
python -m uvicorn src.gateway.chat_gateway:app --reload --port 8000

3) 启动 React 开发服务器 (端口 5173)

cd frontend
npm install
npm run dev

第 6 步:评估

cd evaluation/
python eval_dashboard.py

注意:评估需要运行 MCP 服务器(第 3 步),因为它通过 SSE 调用异步 MCP 智能体。


项目结构

mcpdischarge/
├── data/
│   ├── generate_dataset.py          ← Run this first
│   ├── ehr_patients.json            ← 6 patient records with discharge medications
│   ├── pharmacy_inventory.json      ← 17 drugs (4 out of stock, aliases table)
│   ├── billing_rate_cards.json      ← 15 charge codes
│   ├── insurance_contracts.json     ← 2 insurer contracts
│   ├── patient_insurance_map.json   ← Patient → insurer mappings
│   ├── icd10_billing_codes.json     ← ICD-10 → DRG billing mappings
│   └── rbac_policies.json           ← RBAC matrix (role → server → tools)
│
├── src/
│   ├── servers/
│   │   └── mcp_servers.py           ← EHRServer, PharmacyServer, BillingServer + FastMCP wrappers
│   └── agents/
│       └── discharge_agent.py       ← DischargeCoordinationAgent + WorkflowMetrics
│
├── evaluation/
│   ├── eval_dashboard.py
│   ├── 01_manual_vs_mcp.png
│   ├── 02_rbac_telemetry.png
│   └── 03_data_integrity.png
│
├── demo/
│   └── demo.py                      ← 4 scenarios + 2 limitations
│
├── configs/
│   ├── fastmcp_deployment.md        ← FastMCP HTTP server setup
│   ├── azure_foundry_mcp.md         ← Azure AI Foundry MCP integration
│   └── rbac_design.md               ← RBAC policy design guide
│
└── README.md

注入的挑战模式

模式

患者

药物

注入的问题

[NAME_MISMATCH]

PAT-001

Dapagliflozin/Farxiga

EHR 使用品牌名;药房存储通用名

[OUT_OF_STOCK]

PAT-001

Furosemide 40mg

库存=0;MCP 显示 Torsemide 作为替代品

[OUT_OF_STOCK]

PAT-003

Humira/Adalimumab

品牌药缺货;找到生物类似药 Exemptia

[OUT_OF_STOCK]

PAT-004

Tafamidis/Vyndamax

罕见病药物 — 无替代品;升级处理

[OUT_OF_STOCK]

PAT-005

Osimertinib/Tagrisso

特殊药物 — 中央药房订单

[DATA_DRIFT]

PAT-002

Semaglutide 0.5mg

EHR 维持剂量 vs 处方集起始剂量 0.25mg

[SCOPE_VIOLATION]

PAT-006

Modafinil Schedule H

计费系统不得查看受控物质详情

[PHI_BOUNDARY]

全部

计费发票前屏蔽 5 个 PHI 字段


三个 MCP 服务器(详细信息)

EHR 服务器

PHI 敏感工具(仅限临床角色):

get_patient_discharge_summary(patient_id, caller_role)  # full clinical note
get_discharge_medications(patient_id, caller_role)       # medication list

PHI 安全工具(所有角色,包括计费):

get_diagnosis_codes(patient_id, caller_role)             # ICD-10 only
get_admission_info(patient_id, caller_role)              # LOS, ward, dates
get_billing_safe_summary(patient_id, caller_role)        # strips PHI fields

PHI 脱敏(计费系统被屏蔽的内容):

PHI_FIELDS = {"name", "dob", "mrn", "discharge_note", "attending_physician"}
# Billing receives: patient_id, ward, admission_date, discharge_date, los_days, diagnosis_icd10

药房服务器

语义名称解析:

# EHR says "Dapagliflozin" → Pharmacy stores as "Farxiga"
# MCP alias table: {"farxiga": "PH-001", "dapa": "PH-001", "sglt2 inhibitor": "PH-001"}
drug = _find_drug_by_name("Dapagliflozin")  # → PH-001 (Dapagliflozin)
drug = _find_drug_by_name("Humira")          # → PH-008 (Adalimumab, branded)

剂量冲突检测:

# EHR prescribes Semaglutide 0.5mg, formulary standard is 0.25mg starter
if queried_dose not in formulary_dose:
    dose_conflict = True  # triggers clinical review alert

语义匹配评分:

# score = word overlap / max(len(ehr_words), len(pharm_words))
# score < 0.85 → NAME_MISMATCH alert even if drug found
semantic_drug_match_score("Humira", "Adalimumab")  # → 0.0 (no word overlap)
semantic_drug_match_score("Furosemide", "Furosemide")  # → 1.0 (exact)

计费服务器

发票生成(PHI 防护):

def generate_invoice(patient_id, billing_safe_ehr, drug_costs, ...):
    # Verify PHI is stripped
    for phi_field in PHI_FIELDS:
        if phi_field in billing_safe_ehr:
            raise PermissionError(f"PHI field '{phi_field}' in billing payload")
    # Process invoice using only: ICD-10 + LOS + ward + drug prices

MCP 与传统 API 对比

能力

传统 REST API

MCP 协议

模式发现

静态 Swagger 文档

动态工具清单

跨部门调用

脆弱的点对点

标准化工具调用

RBAC 强制执行

应用层(不一致)

协议层(有保证)

PHI 边界

手动策略

每个工具强制执行

药物名称解析

硬编码映射

语义别名表

缺货处理

手动药房回拨

自动替代品查找

遥测

自定义日志记录

内置工具调用追踪

新部门入驻

新 API 集成

注册新 MCP 服务器


评估结果(6 例患者出院)

患者

MCP 调用

成功

警报

PHI 已屏蔽

PAT-001 HFrEF

16

100%

1

5 个字段

PAT-002 AKI

11

100%

1

5 个字段

PAT-003 RA

13

100%

2

5 个字段

PAT-004 ATTR

14

100%

2

5 个字段

PAT-005 NSCLC

9

100%

1

5 个字段

PAT-006 MS

9

100%

1

5 个字段

总计:72 次 MCP 工具调用 | 100% 成功 | 每次出院减少 15 次手动交接 | 每个案例节省约 45 分钟


FastMCP HTTP 部署

参见 configs/fastmcp_deployment.md。关键模式:

from fastmcp import FastMCP

ehr_mcp = FastMCP("EHR-Server")

@ehr_mcp.tool()
def get_discharge_medications(patient_id: str, caller_role: str) -> dict:
    """Get discharge medication list from EHR."""
    return EHRServer().get_discharge_medications(patient_id, caller_role)

# Run as HTTP SSE server
ehr_mcp.run(transport="sse", host="0.0.0.0", port=8001)

智能体作为 MCP 客户端连接:

from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

async with sse_client("http://localhost:8001/sse") as (read, write):
    async with ClientSession(read, write) as session:
        result = await session.call_tool(
            "get_discharge_medications",
            {"patient_id": "PAT-001", "caller_role": "discharge_coordinator"}
        )

Azure AI Foundry 集成

参见 configs/azure_foundry_mcp.md。MCP 服务器注册为 Foundry 工具:

from azure.ai.projects.models import McpToolDefinition

mcp_tools = [
    McpToolDefinition(server_url="http://ehr-server:8001/sse", name="ehr-server"),
    McpToolDefinition(server_url="http://pharmacy-server:8002/sse", name="pharmacy-server"),
    McpToolDefinition(server_url="http://billing-server:8003/sse", name="billing-server"),
]

agent = client.agents.create_agent(
    model="gpt-4o",
    name="DischargeCoordinationAgent",
    instructions=DISCHARGE_AGENT_SYSTEM_PROMPT,
    tools=[t.as_tool_definition() for t in mcp_tools],
)

CitiusTech 生成式 AI 与智能体 AI 培训计划 — 5 个项目中的第 5 个

Related MCP Connectors

Related MCP Servers