医疗保健MCP+A2A服务器(模拟)
一个示例项目展示了 模型上下文协议(MCP) 和 代理到代理(A2A)通信 在健康保险索赔领域。
mcp_server/--FastMCP服务器公开8个医疗保健工具a2a/--自定义A2A协议,带有4个通过JSON-RPC 2.0进行协调的专用代理run_a2a.py--运行所有代理的独立HTTP服务器
所有数据都是合成的,没有真正的PHI,也没有外部调用。
官方A2A SDK:The a2a-sdk (v1.0.1+)现在是A2A协议的官方Python库。此项目包含为培训目的而构建的自定义实现。______________________________________________________________________
项目结构
healthcare-mcp/
├── mcp_server/ # MCP server
│ ├── __init__.py
│ └── server.py # FastMCP with 8 healthcare tools
├── a2a/ # A2A protocol implementation
│ ├── __init__.py
│ ├── message.py # JSON-RPC 2.0 message types
│ ├── a2a_protocol.py # Protocol engine, registry, routing
│ └── agents.py # 4 healthcare domain agents
├── run_a2a.py # Standalone A2A HTTP server
├── test_a2a_communication.py # A2A protocol tests (40 tests)
├── diagrams/
│ ├── mcp.mmd # MCP Gateway architecture
│ └── a2a.mmd # A2A messaging pattern
├── requirements.txt
└── pytest.ini______________________________________________________________________
设置
先决条件: Python 3.10+
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt______________________________________________________________________
运行MCP服务器
MCP服务器将医疗保健工具暴露给任何兼容MCP的客户端。
stdio(默认值--适用于MCP客户端)
python mcp_server/server.py可流式传输的HTTP
export MCP_TRANSPORT=streamable-http
export FASTMCP_HOST=127.0.0.1
export FASTMCP_PORT=8000
python mcp_server/server.py端点: http://127.0.0.1:8000/mcp
上海证券交易所
export MCP_TRANSPORT=sse
python mcp_server/server.pyMCP环境变量
| 变量 | 默认值 | 描述 |
|---|---|---|
MCP_TRANSPORT | stdio | 运输: stdio, streamable-http,或 sse |
FASTMCP_HOST | 127.0.0.1 | 为HTTP传输绑定主机 |
FASTMCP_PORT | 8000 | 绑定HTTP传输端口 |
FASTMCP_STREAMABLE_HTTP_PATH | /mcp | 可流式传输的http端点的URL路径 |
MCP_MOUNT_PATH | _(无)_ | 可选装载路径前缀 |
______________________________________________________________________
运行A2A服务器
A2A服务器启动所有四个医疗保健代理,并通过HTTP公开它们。它使用一个本地MCP客户端,在进程中调用MCP工具函数——不需要单独的MCP服务器。
python run_a2a.py服务器启动于 http://127.0.0.1:8001 默认情况下。
A2A环境变量
| 变量 | 默认值 | 描述 |
|---|---|---|
A2A_HOST | 127.0.0.1 | 绑定主机 |
A2A_PORT | 8001 | 绑定端口 |
A2A端点
| 方法 | 路径 | 描述 |
|---|---|---|
GET | /.well-known/agent.json | 代理卡——发现功能和元数据 |
GET | /agents | 列出所有注册代理人 |
POST | / | 接收A2A JSON-RPC 2.0消息 |
示例:发现代理
curl http://127.0.0.1:8001/.well-known/agent.json | python -m json.tool示例:发送A2A消息
curl -s -X POST http://127.0.0.1:8001/ \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "check_member_eligibility",
"params": {"member_id": "M-1001"},
"id": "req-001",
"sender": "external-client",
"recipient": "claims-agent",
"type": "request"
}' | python -m json.tool______________________________________________________________________
MCP工具
索赔
| 工具 | 参数 | 说明 |
|---|---|---|
list_member_claims | member_id, status? | 列出索赔,可选择按以下方式筛选 paid/pending/denied |
get_claim_detail | claim_id | 包含会员、提供者和裁决金额的全额索赔 |
submit_claim_inquiry | claim_id, inquiry_type, note | 提交查询单 |
好处
| 工具 | 参数 | 说明 |
|---|---|---|
get_member_benefits | member_id | 会员计划的免赔额和OOP余额 |
estimate_member_responsibility | member_id, procedure_code, billed_amount, network? | 成本分摊估算 |
供应商和授权
| 工具 | 参数 | 说明 |
|---|---|---|
search_providers | specialty, zip_code?, network? | 按专业搜索提供商 |
create_prior_authorization | member_id, provider_id, procedure_codes, service_date, diagnosis_codes? | 提交先前的身份验证请求 |
get_prior_authorization_status | auth_id | 获取先前身份验证的当前状态 |
______________________________________________________________________
A2A代理系统
代理
| 代理 | ID | 角色 |
|---|---|---|
MemberAssistAgent | member-assist-agent | 患者协调员-检查资格,找到提供者 |
ClaimsAgent | claims-agent | 索赔处理人----资格、索赔历史、费用估计 |
ProviderAdvocateAgent | provider-advocate-agent | 网络管理器--提供商搜索 |
BenefitsAgent | benefits-agent | 福利专家——成本计算 |
通信流
MemberAssistAgent
├─ A2A → ClaimsAgent: "check_member_eligibility"
│ ├─ MCP: list_member_claims, get_member_benefits
│ └─ MCP: get_claim_detail
│
└─ A2A → ProviderAdvocateAgent: "search_network_providers"
└─ MCP: search_providers
ClaimsAgent
└─ A2A → BenefitsAgent: "calculate_member_responsibility"
└─ MCP: estimate_member_responsibility在代码中使用代理
import asyncio
from a2a import A2AProtocol, MemberAssistAgent, ClaimsAgent, ProviderAdvocateAgent, BenefitsAgent
async def main():
protocol = A2AProtocol()
member_assist = MemberAssistAgent(protocol, mcp_client)
claims = ClaimsAgent(protocol, mcp_client)
provider_advocate = ProviderAdvocateAgent(protocol, mcp_client)
benefits = BenefitsAgent(protocol, mcp_client)
for agent in [member_assist, claims, provider_advocate, benefits]:
await agent.register()
protocol.register_handler("check_member_eligibility", claims.handle_check_member_eligibility)
protocol.register_handler("search_network_providers", provider_advocate.handle_search_network_providers)
protocol.register_handler("calculate_member_responsibility", benefits.handle_calculate_member_responsibility)
eligibility = await member_assist.check_eligibility("M-1001")
providers = await member_assist.find_providers("primary care", "55401")
asyncio.run(main())______________________________________________________________________
测试
pytest -v # all 40 tests
pytest test_a2a_communication.py # A2A protocol tests only看 测试_用户.md 用于测试文件和 QUICKSTART.md 用于过滤命令。
______________________________________________________________________
模拟数据参考
成员
| ID | 姓名 | 计划 |
|---|---|---|
M-1001 | 李 | P-100 Optum精选PPO |
M-1002 | 凯西·帕特尔 | P-200 Optum选择HMO |
计划
| ID | 姓名 | 免赔额 | 剩余 | OOP最大值 | 剩余 | 网络内共同保险 |
|---|---|---|---|---|---|---|
P-100 | Optum Choice PPO | 1500美元 | 420美元 | 5000美元 | 2100美元 | 20% |
P-200 | Optum Select HMO | 500美元 | 120美元 | 3000美元 | 980美元 | 10% |
提供商
| ID | 名称 | 专业 | 网络 | ZIP |
|---|---|---|---|---|
PR-2001 | 北侧初级保健 | 初级保健 | 网络中 | 55401 |
PR-2002 | 湖景骨科诊所 | 骨科 | 网络中 | 55111 |
PR-2003 | Metro影像中心 | 放射科 | 网络外 | 55415 |
索赔
| ID | 成员 | 提供商 | 状态 | 已计费 |
|---|---|---|---|---|
C-10001 | M-1001 | PR-2001 | 已支付 | 250.00美元 |
C-10002 | M-1001 | PR-2003 | 待定 | 980.00美元 |
C-10003 | M-1002 | PR-2002 | 拒绝(需要事先授权) | 1350.00美元 |
事先授权
| ID | 成员 | 状态 |
|---|---|---|
PA-9001 | M-1002 | 已批准 |
______________________________________________________________________
架构图
| 文件 | 描述 |
|---|---|
| 图表/mcp.mmd | MCP网关路由到域服务器 |
| 图表/a2a.mmd | A2A代理之间的消息传递 |
渲染方式 美人鱼实时编辑器 或任何与Mermaid兼容的查看器。
______________________________________________________________________
MCP检查员
MCP检查员 是一个基于浏览器的工具,用于交互式调用MCP工具。
npm install
npm start______________________________________________________________________
备注
- 所有数据都在内存中,并在重新启动时重置。
estimate_member_responsibility使用模拟乘数(75%在网络中,60%在网络外)。不是付款保证。- 没有PHI,没有外部API调用,没有持久存储。
- 官方的A2A Python SDK是
a2a-sdk>=1.0.1Thea2a/这个项目中的模块是相同协议模式的自定义教育实现。
