MCP multiagent

MCP医疗分诊系统

基于模型上下文协议（MCP）和A2A（代理到代理）通信的综合多代理架构。该系统通过路由器代理协调专业代理（患者数据、分诊、保险）来处理医疗分诊场景。

目的和商业意识

该项目模拟了一个医疗分流工作流程，在该流程中，对传入的患者请求进行筛选，丰富临床背景，并将其路由到正确的专家响应。目标是提高摄入效率，同时确保患者获得与他们的症状和覆盖问题相一致的及时、一致的指导。

为什么这对医疗保健运营很重要：

• 更快的录取决定：路由器和分类代理有助于对紧急情况进行分类，并在没有手动切换的情况下提供首通指导。
• 单一数据来源：患者数据代理显示图表和遭遇历史，以便响应基于上下文。
• 覆盖清晰度：保险代理人回答经常阻碍护理决定的福利和共付额问题。
• 可扩展的协调：代理可以独立部署，使诊所能够随着需求的增长扩展分诊和覆盖工作流程。

架构概述

┌─────────────────────────────────────────────────────────────┐
│                     MCP Server (Port 8000)                  │
│              SQLite Database + Tool Registry                │
└─────────────────────────────────────────────────────────────┘
                            ▲
                            │ HTTP/REST
        ┌───────────────────┼───────────────────┐
        │                   │                   │
        ▼                   ▼                   ▼
┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│ Data Agent   │    │Triage Agent  │    │Insurance Agent │
│ (Port 8011)  │    │ (Port 8012)  │    │ (Port 8013)  │
└──────────────┘    └──────────────┘    └──────────────┘
        ▲                   ▲                   ▲
        │                   │                   │
        └───────────────────┼───────────────────┘
                            │ A2A RPC
                            │
                    ┌───────▼────────┐
                    │ Router Agent   │
                    │  (Port 8010)   │
                    └────────────────┘

项目结构

.
├── README.md                      # This file
├── requirements.txt               # Python dependencies
├── database_setup.py              # SQLite initialization
├── demo.py                        # Example client
├── common/                        # Shared utilities
│   ├── a2a.py                    # A2A runtime & JSON-RPC
│   ├── db.py                     # Async database helpers
│   ├── message_utils.py          # Message construction
│   └── __init__.py
├── sdk/                           # Core data types
│   ├── types.py                  # Pydantic models
│   ├── agent.py                  # Agent types
│   ├── task.py                   # Task management
│   └── __init__.py
├── shared/                        # Backwards compatibility
│   ├── a2a_handler.py            # Re-exports from common
│   ├── message_utils.py          # Re-exports from common
│   └── __init__.py
├── mcp_server/                    # MCP Server (Port 8000)
│   ├── app.py                    # FastAPI application
│   ├── database.py               # Database interface
│   └── __init__.py
└── agents/                        # Specialized agents
    ├── router/                   # Router Agent (Port 8010)
    │   ├── main.py              # LangGraph workflow
    │   └── __init__.py
    ├── data/                     # Data Agent (Port 8011)
    │   ├── main.py              # MCP client for records
    │   └── __init__.py
    ├── support/                  # Triage Agent (Port 8012)
    │   ├── main.py              # Customer guidance
    │   └── __init__.py
    ├── payments/                 # Insurance Agent (Port 8013)
    │   ├── main.py              # Billing responses
    │   └── __init__.py
    └── __init__.py

安装说明

先决条件

• Python 3.10+
• pip 虚拟环境支持

1.克隆并导航到项目

cd /path/to/MCP_multiagent

2.创建虚拟环境

# Create isolated Python environment
python3 -m venv venv

# Activate it
source venv/bin/activate  # macOS/Linux
# OR
venv\Scripts\activate     # Windows

3.安装依赖项

pip install --upgrade pip
pip install -r requirements.txt

里面有什么 requirements.txt:

• FastAPI/Uvicorn：代理和MCP服务器的Web框架
• LangGraph：路由器的工作流编排
• HTTPx：用于A2A通信的异步HTTP客户端
• aiosqlite：异步SQLite驱动程序
• 派丹蒂克：数据验证
• python dotenv：环境配置
• 小星星：服务器发送实时更新事件
• 其他实用程序（orjson、fastmcp等）

4.初始化数据库

python database_setup.py

创造 triage.db 与样本患者和遭遇。

5.启动所有服务

终端1-MCP服务器：

python -m mcp_server.app
# Listens on http://localhost:8000

终端2——数据代理：

python -m agents.data.main
# Listens on http://localhost:8011

3号航站楼——分检人员：

python -m agents.support.main
# Listens on http://localhost:8012

4号航站楼——保险代理人：

python -m agents.payments.main
# Listens on http://localhost:8013

终端5——路由器代理：

python -m agents.router.main
# Listens on http://localhost:8010

6.测试系统

在一个新的终端（与 venv 已激活）：

python demo.py

或者运行Jupyter场景：

jupyter notebook Assignment5_notebook.ipynb

配置

环境变量

创建一个 .env 项目根目录中的文件：

# Database path (default: ./database.sqlite)
A2A_DB_PATH=./triage.db

# MCP Server URL (for agents to reach it)
MCP_SERVER_URL=http://localhost:8000

# Agent RPC endpoints (used by router)
DATA_AGENT_RPC=http://localhost:8011/rpc
TRIAGE_AGENT_RPC=http://localhost:8012/rpc
INSURANCE_AGENT_RPC=http://localhost:8013/rpc

# Router endpoint (used by clients)
ROUTER_RPC=http://localhost:8010/rpc

加载环境文件：

export $(cat .env | xargs)

API概述

MCP服务器端点

可用工具：

• get_patient --按ID获取患者记录
• list_patients --列出患者（可选择按状态筛选）
• update_patient --修改患者字段
• create_case --打开分诊病例
• get_patient_history --检索遭遇历史记录

A2A代理RPC方法

所有代理都支持JSON-RPC 2.0：

{
  "jsonrpc": "2.0",
  "id": "unique-id",
  "method": "message/send",
  "params": {
    "message": {
      "messageId": "uuid",
      "role": "user",
      "parts": [{"text": "Your query here"}]
    }
  }
}

方法：

• message/send --同步请求
• message/send_stream --流媒体响应
• task/get --按ID检索任务
• task/cancel --取消正在运行的任务

代理元数据

每个代理在以下位置公开元数据 /.well-known/agent-card.json:

curl http://localhost:8010/.well-known/agent-card.json

返回代理名称、功能、技能和文档URL。

使用示例

1.查询客户数据

import asyncio
import httpx
from sdk.types import Message, MessageSendParams, Role
from common.message_utils import create_text_message

async def query_router():
    msg = create_text_message("Get patient information for ID 5", role=Role.user)
    params = {"message": msg.model_dump()}
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://localhost:8010/rpc",
            json={
                "jsonrpc": "2.0",
                "id": "demo",
                "method": "message/send",
                "params": params
            }
        )
    
    result = response.json()
    print(result["result"]["status"]["message"]["parts"][0]["text"])

asyncio.run(query_router())

2.多意图场景

query = "Update my date of birth to 1980-12-01 and show my encounter history"
# Router automatically routes to both Data and Triage agents
# Combines responses into a single coherent answer

3.升级流程

query = "Do I need a referral and what is my copay?"
# Router detects coverage keywords → routes to Insurance agent
# Insurance agent provides specialized response

工作流：请求流

1. 客户端请求 → 路由器代理(:8010/rpc)
2. 路由器意图分类 → 确定要呼叫的专家
3. 专家调用 → 数据/支持/支付代理通过A2A RPC进行响应
4. 响应聚合 → 路由器结合了专业输出
5. 最终响应 → 以统一消息返回给客户端

路由逻辑

数据库模式

患者

CREATE TABLE patients (
  id INTEGER PRIMARY KEY,
  name TEXT,
  date_of_birth TEXT,
  status TEXT,  -- 'stable', 'monitoring', 'urgent'
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);

案例

CREATE TABLE cases (
  id INTEGER PRIMARY KEY,
  patient_id INTEGER,
  complaint TEXT,
  urgency TEXT,
  status TEXT DEFAULT 'open',
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);

遭遇

CREATE TABLE encounters (
  id INTEGER PRIMARY KEY,
  patient_id INTEGER,
  channel TEXT,  -- 'email', 'phone', 'chat'
  notes TEXT,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);

停用虚拟环境

完成后：

deactivate

故障排除

端口已在使用中

# Kill existing process on port (e.g., 8010)
lsof -i :8010 | grep LISTEN | awk '{print $2}' | xargs kill -9

数据库锁定

# Remove stale database
rm -f triage.db database.sqlite
python database_setup.py

导入错误

# Ensure you're in venv and installed dependencies
source venv/bin/activate
pip install -r requirements.txt

代理无法连接

• 验证所有服务是否在预期端口上运行
• 检查 .env 包含正确的端点URL
• 测试连接性： curl http://localhost:8000/health

主要特点

✅ 模块化架构 --每个代理都可以独立部署\ ✅ LangGraph编排 --路由器使用状态图处理复杂的工作流程\ ✅ A2A协议 --用于代理间通信的标准JSON-RPC\ ✅ MCP集成 --代理将数据操作委托给集中式MCP服务器\ ✅ 异步/等待 --对并发操作的完全异步支持\ ✅ Pydantic验证 --类型安全消息传递\ ✅ 实时事件 --SSE流式传输审计日志和更新

测试

运行附带的Jupyter笔记本：

jupyter notebook Assignment5_notebook.ipynb

或者执行演示场景：

python demo.py

该测试：

• 简单的患者查询
• 多智能体协调
• 复杂的搜索查询
• 升级工作流程
• 多意图请求

业绩说明

• 数据库：SQLite aiosqlite 用于异步访问
• 并发：代理请求通过httpx无阻塞
• 记忆：存储在内存中的任务；考虑用于生产的持久存储
• 可扩展性：目前的设计适合中小型部署

许可证

作业5----教育项目

支持

有关问题或疑问，请参阅：

• 任务5_notebook.ipynb --完整场景演练
• demo.py --工作示例
• 个人代理人 main.py 实施细节文件

______________________________________________________________________

最后更新：2025年12月\ Python版本: 3.10+\ 状态: ✅ 生产就绪