使用说明
基于MCP和A2A的多Agent客户服务系统
本项目实施了5项要求:
- 一个功能齐全的MCP HTTP服务器,提供五个必需的数据库工具,包括SQLite存储和SSE流。
- 多个独立的A2A代理(通过LangGraph的路由器、客户数据代理、支持代理、计费代理),每个代理都运行自己的FastAPI+JSON-RPC服务器。
- LangGraph驱动的路由器代理,使用真实的A2A请求跨专家协调多步工作流。
- 本地运行堆栈的演示实用程序和设置说明。
项目布局
/mcp_server # FastAPI MCP server implementing tools/list and tools/call
/agents/router # LangGraph router agent (A2A server)
/agents/data # Customer Data Agent (calls MCP tools)
/agents/support # Support Agent (non-billing support)
/agents/billing # Billing Agent (optional specialist)
/shared # Shared helpers (SQLite + A2A helpers)
requirements.txt # Python dependencies
先决条件
- Python 3.11+
pip install -r requirements.txt
运行服务器
为每个服务打开单独的终端(所有uvicon进程都绑定到 0.0.0.0):
# 1) Start MCP server (tools/list, tools/call, SSE stream)
uvicorn mcp_server.app:app --port 8000 --reload
# 2) Customer Data Agent (calls MCP tools)
uvicorn agents.data.main:app --port 8011 --reload
# 3) Support Agent
uvicorn agents.support.main:app --port 8012 --reload
# 4) Billing Agent (optional)
uvicorn agents.billing.main:app --port 8013 --reload
# 5) LangGraph Router Agent
uvicorn agents.router.main:app --port 8010 --reload
演示脚本
在所有服务都运行的情况下,执行:
python demo.py "Need the history for customer 1"
脚本将发送JSON-RPC message/send 向路由器代理发送请求,路由器代理将请求转发给数据和支持代理,然后打印聚合的响应。
MCP协议支持
GET /tools/list返回所有工具定义和JSON模式。POST /tools/call执行所请求的工具。GET /events/stream提供了一个服务器发送事件流的工具调用,以实现可观察性。
MCP服务器在首次启动时使用SQLite和种子示例客户、票证和交互。
A2A代理互操作性
每个代理暴露:
- 代理卡在
/.well-known/agent.json - JSON-RPC端点位于
/rpc(支持message/send+流媒体) - A2A客户端所需的技能、能力和元数据(例如,MCP检查器、LangGraph流)。
路由器协调利用LangGraph进行条件路由、多步骤工作流(数据→支持),并在出现“付款”或“退款”等关键字时升级到计费。
测试场景
- 简单查询:“获取ID 5的客户信息” 单代理,直接的MCP呼叫
- 协同查询:“我是客户12345,需要帮助升级我的帐户” 多个代理协调:数据获取+支持响应
- 复杂查询:“显示所有有未结门票的活跃客户” 需要数据和支持代理之间进行协商
- 升级:“我已经被收费两次了,请立即退款!” 路由器必须识别紧急情况并适当路由
- Multi-Intent:“将我的电子邮件更新为new@email.com并显示我的门票历史记录” 并行任务执行和协调
备注
- 环境变量
MCP_SERVER_URL,DATA_AGENT_RPC,SUPPORT_AGENT_RPC,以及BILLING_AGENT_RPC让您将代理重新指向不同的主机。 - 共享SQLite数据库路径默认为
./database.sqlite并将使用种子数据自动初始化。