医疗保健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.py
MCP环境变量
| 变量 |
默认值 |
描述 |
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服务器。
服务器启动于 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 |
已批准 |
架构图
渲染方式 美人鱼实时编辑器 或任何与Mermaid兼容的查看器。
MCP检查员
MCP检查员 是一个基于浏览器的工具,用于交互式调用MCP工具。
备注
- 所有数据都在内存中,并在重新启动时重置。
estimate_member_responsibility 使用模拟乘数(75%在网络中,60%在网络外)。不是付款保证。
- 没有PHI,没有外部API调用,没有持久存储。
- 官方的A2A Python SDK是
a2a-sdk>=1.0.1The a2a/ 这个项目中的模块是相同协议模式的自定义教育实现。