Healthcare MCP + A2A Server (Mock)

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 -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install -r requirements.txt

python mcp_server/server.py

export MCP_TRANSPORT=streamable-http
export FASTMCP_HOST=127.0.0.1
export FASTMCP_PORT=8000
python mcp_server/server.py

export MCP_TRANSPORT=sse
python mcp_server/server.py

python run_a2a.py

curl http://127.0.0.1:8001/.well-known/agent.json | python -m json.tool

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

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

npm install
npm start