医疗保健 MCP + A2A 服务器（模拟）

一个演示健康保险理赔领域模型上下文协议 (MCP) 和 代理间 (Agent-to-Agent, A2A) 通信的示例项目。

mcp_server/ — 公开 8 个医疗保健工具的 FastMCP 服务器
a2a/ — 自定义 A2A 协议，包含 4 个通过 JSON-RPC 2.0 进行协调的专业代理
run_a2a.py — 同时运行所有代理的独立 HTTP 服务器

所有数据均为合成数据 — 无真实个人健康信息 (PHI)，无外部调用。

官方 A2A SDK：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

SSE

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 服务器。

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`	成员计划的免赔额和自付额余额
`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

请参阅 TEST_GUIDE.md 获取测试文档，参阅 QUICKSTART.md 获取过滤命令。

模拟数据参考

成员

ID	姓名	计划
`M-1001`	Jordan Lee	`P-100` Optum Choice PPO
`M-1002`	Casey Patel	`P-200` Optum Select HMO

计划

ID	姓名	免赔额	剩余	OOP 最大值	剩余	网络内共同保险
`P-100`	Optum Choice PPO	$1,500	$420	$5,000	$2,100	20%
`P-200`	Optum Select HMO	$500	$120	$3,000	$980	10%

提供者

ID	姓名	专业	网络	邮编
`PR-2001`	Northside Primary Care	初级保健	网络内	55401
`PR-2002`	Lakeview Ortho Clinic	骨科	网络内	55111
`PR-2003`	Metro Imaging Center	放射科	网络外	55415

理赔

ID	成员	提供者	状态	账单金额
`C-10001`	M-1001	PR-2001	已支付	$250.00
`C-10002`	M-1001	PR-2003	待处理	$980.00
`C-10003`	M-1002	PR-2002	已拒绝（需预授权）	$1,350.00

预授权

ID	成员	状态
`PA-9001`	M-1002	已批准

架构图

文件	描述
diagrams/mcp.mmd	路由到领域服务器的 MCP 网关
diagrams/a2a.mmd	代理间的 A2A 消息传递

使用 Mermaid Live Editor 或任何兼容 Mermaid 的查看器进行渲染。

MCP 检查器

MCP Inspector 是一个基于浏览器的工具，用于交互式调用 MCP 工具。

npm install
npm start

注意事项

所有数据均在内存中，重启后重置。
estimate_member_responsibility 使用模拟乘数（网络内 75%，网络外 60%）。不作为支付保证。
无 PHI，无外部 API 调用，无持久化存储。
官方 A2A Python SDK 为 a2a-sdk>=1.0.1。本项目中的 a2a/ 模块是相同协议模式的自定义教学实现。

Health Claims MCP Server