多代理客户服务(MCP+A2A)
最小的、可运行的参考:SQLite上的FastMCP服务器加上三个协调代理(路由器、客户数据、支持),具有显式的代理到代理日志。
项目布局
mcp_server/server.py--FastMCP服务器公开所需工具(get_customer,list_customers,update_customer,create_ticket,get_customer_history)在SQLite的支持下。src/agents/router_agent.py--异步编排器/路由器代理,负责分配任务、协商升级和协调多步流。src/agents/customer_data_agent.py--客户数据专家,将MCP工具与A2A日志相结合。src/agents/support_agent.py--支持专家处理一般支持、升级和总结;可以向数据代理请求上下文。demo.py--端到端运行器,为数据库播种并执行所需的场景。database_setup.py--提供的助手(演示中未使用)仅供参考。
设置
- 创建并激活虚拟环境(可选但推荐):
python -m venv .venv
source .venv/bin/activate- 安装依赖项:
pip install -r requirements.txt运行旧版演示
python demo.py演示重置 mcp_server/support.db,种子样本数据,运行测试场景,打印面向客户的响应以及每个跃点的A2A日志,并将成绩单写入 demos/output/run.log.
A2A规范HTTP服务
三个FastAPI/uvicorn服务将代理暴露为A2A兼容的HTTP端点:
- 路由器:
python -m uvicorn services.router_server:app --host 0.0.0.0 --port 8010 - 客户数据:
python -m uvicorn services.customer_data_server:app --host 0.0.0.0 --port 8011 - 技术支持:
python -m uvicorn services.support_server:app --host 0.0.0.0 --port 8012
每项服务在以下地址提供一张代理卡 /.well-known/agent-card.json 以及JSON-RPC message/send 处理程序在 /rpc.
A2A规格演示
运行端到端A2A演示(启动所有服务、运行场景并写入 demos/output/run_a2a_spec.log):
python demos/run_a2a_spec_demo.pyMCP服务器工具
所有工具均已安装 mcp_server/server.py,通过HTTP上的FastMCP公开:
get_customer(customer_id)list_customers(status, limit)update_customer(customer_id, data)(验证允许的字段)create_ticket(customer_id, issue, priority)get_customer_history(customer_id)
代理与协调
- 路由器分配请求,协商升级,并对多步工作进行排序。
- 客户数据代理调用MCP工具进行检索/更新。
- Support Agent处理响应、工单创建和报告。
ConversationLog记录每个路由器↔ 代理跳跃;demo.py打印出来,这样你就可以追踪协调。
演示场景(按顺序)
demo.py 跑:
Get customer information for ID 5I'm customer 12345 and need help upgrading my accountShow me all active customers who have open ticketsI've been charged twice, please refund immediately!Update my email to new@email.com(换行符)and show my ticket historyWhat's the status of all high-priority tickets for premium customers?
输出显示响应和A2A日志,以实现可追溯性。
延伸
- 调整种子数据
mcp_server/server.py如果你想要不同的装置。 - 替换中的简单关键字路由
src/agents/router_agent.py使用您偏好的意图分类器。 - MCP服务器类与框架无关;如果需要,您可以用HTTP或套接字传输来包装它。
