Healthcare MCP + A2A Server (Mock)

Ein Beispielprojekt zur Demonstration des Model Context Protocol (MCP) und der Agent-to-Agent (A2A)-Kommunikation für den Bereich der Krankenversicherungsabrechnungen.

mcp_server/ — FastMCP-Server, der 8 Healthcare-Tools bereitstellt
a2a/ — Benutzerdefiniertes A2A-Protokoll mit 4 spezialisierten Agenten, die über JSON-RPC 2.0 koordinieren
run_a2a.py — Eigenständiger HTTP-Server, der alle Agenten zusammen ausführt

Alle Daten sind synthetisch — keine echten PHI (geschützte Gesundheitsdaten), keine externen Aufrufe.

Offizielles A2A SDK: Das a2a-sdk (v1.0.1+) ist jetzt die offizielle Python-Bibliothek für das A2A-Protokoll. Dieses Projekt enthält eine benutzerdefinierte Implementierung, die zu Schulungszwecken erstellt wurde.

Projektstruktur

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

Einrichtung

Voraussetzungen: Python 3.10+

python -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Ausführen des MCP-Servers

Der MCP-Server stellt Healthcare-Tools für jeden MCP-kompatiblen Client bereit.

stdio (Standard — für MCP-Clients)

python mcp_server/server.py

Streamable HTTP

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

Endpunkt: http://127.0.0.1:8000/mcp

SSE

export MCP_TRANSPORT=sse
python mcp_server/server.py

MCP-Umgebungsvariablen

Variable	Standard	Beschreibung
`MCP_TRANSPORT`	`stdio`	Transport: `stdio`, `streamable-http` oder `sse`
`FASTMCP_HOST`	`127.0.0.1`	Bind-Host für HTTP-Transporte
`FASTMCP_PORT`	`8000`	Bind-Port für HTTP-Transporte
`FASTMCP_STREAMABLE_HTTP_PATH`	`/mcp`	URL-Pfad für den streamable-http-Endpunkt
`MCP_MOUNT_PATH`	(keiner)	Optionaler Mount-Pfad-Präfix

Ausführen des A2A-Servers

Der A2A-Server startet alle vier Healthcare-Agenten und stellt sie über HTTP bereit. Er verwendet einen lokalen MCP-Client, der die MCP-Tool-Funktionen in-process aufruft — kein separater MCP-Server erforderlich.

python run_a2a.py

Der Server startet standardmäßig unter http://127.0.0.1:8001.

A2A-Umgebungsvariablen

Variable	Standard	Beschreibung
`A2A_HOST`	`127.0.0.1`	Bind-Host
`A2A_PORT`	`8001`	Bind-Port

A2A-Endpunkte

Methode	Pfad	Beschreibung
`GET`	`/.well-known/agent.json`	Agenten-Karte — Fähigkeiten und Metadaten zur Erkennung
`GET`	`/agents`	Liste aller registrierten Agenten
`POST`	`/`	Empfangen einer A2A JSON-RPC 2.0 Nachricht

Beispiel: Agenten entdecken

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

Beispiel: Eine A2A-Nachricht senden

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-Tools

Abrechnungen (Claims)

Tool	Parameter	Beschreibung
`list_member_claims`	`member_id`, `status?`	Abrechnungen auflisten, optional gefiltert nach `paid`/`pending`/`denied`
`get_claim_detail`	`claim_id`	Vollständige Abrechnung mit Mitglied, Anbieter und Abrechnungsbeträgen
`submit_claim_inquiry`	`claim_id`, `inquiry_type`, `note`	Ein Anfrageticket einreichen

Leistungen (Benefits)

Tool	Parameter	Beschreibung
`get_member_benefits`	`member_id`	Selbstbehalt und OOP-Salden für den Plan des Mitglieds
`estimate_member_responsibility`	`member_id`, `procedure_code`, `billed_amount`, `network?`	Kostenteilungsschätzung

Anbieter & Genehmigungen

Tool	Parameter	Beschreibung
`search_providers`	`specialty`, `zip_code?`, `network?`	Anbieter nach Fachgebiet suchen
`create_prior_authorization`	`member_id`, `provider_id`, `procedure_codes`, `service_date`, `diagnosis_codes?`	Vorabgenehmigungsanfrage einreichen
`get_prior_authorization_status`	`auth_id`	Aktuellen Status einer Vorabgenehmigung abrufen

A2A-Agentensystem

Agenten

Agent	ID	Rolle
`MemberAssistAgent`	`member-assist-agent`	Patientenkoordinator — prüft Anspruchsberechtigung, findet Anbieter
`ClaimsAgent`	`claims-agent`	Abrechnungsbearbeiter — Anspruchsberechtigung, Abrechnungshistorie, Kostenschätzungen
`ProviderAdvocateAgent`	`provider-advocate-agent`	Netzwerkmanager — Anbietersuche
`BenefitsAgent`	`benefits-agent`	Leistungsspezialist — Kostenberechnungen

Kommunikationsfluss

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

Verwendung von Agenten im Code

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())

Tests

pytest -v                          # all 40 tests
pytest test_a2a_communication.py   # A2A protocol tests only

Siehe TEST_GUIDE.md für Testdokumentation und QUICKSTART.md für Filterbefehle.

Referenz der Mock-Daten

Mitglieder

ID	Name	Plan
`M-1001`	Jordan Lee	`P-100` Optum Choice PPO
`M-1002`	Casey Patel	`P-200` Optum Select HMO

Pläne

ID	Name	Selbstbehalt	Verbleibend	OOP Max	Verbleibend	In-Network Mitversicherung
`P-100`	Optum Choice PPO	$1.500	$420	$5.000	$2.100	20%
`P-200`	Optum Select HMO	$500	$120	$3.000	$980	10%

Anbieter

ID	Name	Fachgebiet	Netzwerk	PLZ
`PR-2001`	Northside Primary Care	Hausarzt	in-network	55401
`PR-2002`	Lakeview Ortho Clinic	Orthopädie	in-network	55111
`PR-2003`	Metro Imaging Center	Radiologie	out-of-network	55415

Abrechnungen

ID	Mitglied	Anbieter	Status	Abgerechnet
`C-10001`	M-1001	PR-2001	bezahlt	$250,00
`C-10002`	M-1001	PR-2003	ausstehend	$980,00
`C-10003`	M-1002	PR-2002	abgelehnt (Vorabgenehmigung erforderlich)	$1.350,00

Vorabgenehmigungen

ID	Mitglied	Status
`PA-9001`	M-1002	genehmigt

Architekturdiagramme

Datei	Beschreibung
diagrams/mcp.mmd	MCP-Gateway-Routing zu Domänenservern
diagrams/a2a.mmd	A2A-Messaging zwischen Agenten

Mit dem Mermaid Live Editor oder einem beliebigen Mermaid-kompatiblen Viewer rendern.

MCP Inspector

MCP Inspector ist ein browserbasiertes Tool zum interaktiven Aufrufen von MCP-Tools.

npm install
npm start

Hinweise

Alle Daten sind im Arbeitsspeicher und werden beim Neustart zurückgesetzt.
estimate_member_responsibility verwendet Mock-Multiplikatoren (75% in-network, 60% out-of-network). Dies ist keine Zahlungsgarantie.
Keine PHI, keine externen API-Aufrufe, kein persistenter Speicher.
Das offizielle A2A Python SDK ist a2a-sdk>=1.0.1. Das a2a/-Modul in diesem Projekt ist eine benutzerdefinierte, zu Bildungszwecken erstellte Implementierung derselben Protokollmuster.

Health Claims MCP Server