多代理客户服务系统

MCP供电•A2A协议•LangGraph SDK•独立代理•SQLite接地

该项目实现了一个功能齐全的 多代理客户服务平台 为任务5——多代理系统和MCP而构建。

它结合了 代理到代理（A2A）通信, 模型上下文协议（MCP）, LangGraph SDK编排, 独立代理服务，以及 数据库支持的工具 通过SQLite。

______________________________________________________________________

项目概述

该系统模拟了智能客户服务工作流程。三家专业代理商合作：

了解并处理客户查询
通过MCP检索和更新客户数据
处理支持请求和工单管理
协调多步骤工作流程
提供自然语言响应
通过A2A接口证明真正的代理独立性

管道是 基于真实的SQLite数据库 通过MCP HTTP服务器，由 LangGraph SDK 用于A2A协调，并演示 独立代理服务 具有适当的A2A接口。

______________________________________________________________________

项目结构

multiagent-mcp/
├── src/                         # Source code
│   ├── __init__.py
│   ├── agents.py                # Agent implementations and A2A coordination
│   ├── mcp_http_server.py       # MCP HTTP server implementation
│   ├── mcp_http_client.py       # MCP HTTP client for agents
│   ├── a2a_specs.py             # A2A agent cards and specifications
│   ├── langgraph_a2a.py         # LangGraph SDK integration for A2A
│   ├── agent_services.py        # Individual agent HTTP services (A2A protocol)
│   └── server.py                # HTTP server with streaming support
├── scripts/                     # Utility scripts
│   ├── setup_database.py        # Database initialization script
│   ├── start_all_services.sh    # Start all independent agent services
│   ├── start_server.sh          # Server startup helper
│   └── start_mcp_server.sh      # MCP server startup helper
├── tests/                       # Test files
│   ├── demo.py                  # End-to-end demonstration script
│   ├── test_http.py             # HTTP endpoint test suite
│   └── validate_pipeline.py     # Comprehensive pipeline validation
├── docs/                        # Documentation
│   ├── A2A_SPECIFICATIONS.md    # A2A protocol, agent cards, and LangGraph
│   └── CONCLUSION.md            # Learning outcomes and challenges
├── requirements.txt             # Python dependencies
├── README.md                    # This file
├── .gitignore                   # Git ignore rules
└── customer_service.db          # SQLite database (created after setup, gitignored)

______________________________________________________________________

系统架构

flowchart TD
    User[User Query] --> MainServer[Main HTTP Server
Port 8000]
    
    MainServer --> RouterAgent[Router Agent
Orchestrator
A2A Interface]
    
    RouterAgent -->|A2A Messages| CustomerDataAgent[Customer Data Agent
Port 8001
A2A Interface]
    RouterAgent -->|A2A Messages| SupportAgent[Support Agent
Port 8002
A2A Interface]
    
    CustomerDataAgent -->|MCP Calls| MCPServer[MCP HTTP Server
Port 8003
JSON-RPC 2.0]
    SupportAgent -->|MCP Calls| MCPServer
    
    MCPServer -->|SQL Queries| Database[(SQLite Database
customer_service.db)]
    
    RouterAgent -->|LangGraph SDK| LangGraph[LangGraph State Graph
Message Passing
Conditional Routing]
    
    MainServer -->|SSE Streaming| User

______________________________________________________________________

特性

MCP HTTP服务器

JSON-RPC 2.0协议：POST /mcp 返回JSON响应（兼容MCP检查器）
SSE流媒体：获取 /mcp 用于服务器到客户端流式传输
五种数据库工具:

- get_customer -按ID检索客户 - list_customers -按状态列出客户 - update_customer -更新客户信息 - create_ticket -创建支持票 - get_customer_history -获取客户票务历史记录

MCP检查器兼容：可与标准MCP客户端完全测试

路由器代理（编排器）

A2A接口：独立服务 /agent-card 端点
查询分析：意图检测和路由
LangGraph集成：带条件路由的状态图
多步骤协调：处理复杂的工作流程
三种协调方案:

- 任务分配 - 谈判/升级 - 多步骤协调

客户数据代理（专家）

A2A接口：8001端口独立服务
MCP客户端：通过MCP协议访问所有数据库
能力：数据检索和更新
任务：get_customer、list_customers、update_customer、get_customer_history

支持代理（专家）

A2A接口：8002端口独立服务
MCP客户端：通过MCP协议进行票务管理
能力：票务管理和支持响应
任务：handle_support、create_ticket、get_tickets_by_priority、check_can_handle

LangGraph SDK集成

状态图：定义为状态机的代理工作流
消息传递：用于结构化A2A通信的LangChain消息
条件路由：基于查询分析的动态代理选择
MCP集成：LangGraph节点使用MCP HTTP客户端进行数据访问

代理人独立性

独立服务：每个代理都作为单独的HTTP服务运行
A2A协议：代理通信的结构化消息格式
代理卡：每个代理通过以下方式公开功能和任务 /agent-card
服务发现：代理可以发现彼此的能力

______________________________________________________________________

安装

1.克隆存储库

git clone 
cd multiagent-mcp

2.创建虚拟环境

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

3.安装依赖项

pip install -r requirements.txt

4.初始化数据库

python scripts/setup_database.py

这将创建SQLite数据库(customer_service.db)带有客户和票务数据样本。

______________________________________________________________________

如何跑步

🔹 快速入门（所有独立服务）

使用一个命令启动所有服务：

./scripts/start_all_services.sh

这将开始：

MCP服务器（端口8003）
客户数据代理（端口8001）
支持代理（端口8002）
路由器代理（端口8004）
主HTTP服务器（端口8000）

🔹 手动设置（独立服务）

要完全控制，请手动启动每个服务：

终端1-MCP服务器：

python -m src.mcp_http_server

终端2-客户数据代理：

python -m src.agent_services customer_data 8001

终端3-支持代理：

python -m src.agent_services support 8002

终端4-路由器代理：

python -m src.agent_services router 8004

终端5-主服务器：

export A2A_USE_HTTP=true
export A2A_CUSTOMER_DATA_URL=http://localhost:8001
export A2A_SUPPORT_URL=http://localhost:8002
python -m src.server

🔹 单流程模式（开发）

用于测试或开发：

# Start MCP Server
python -m src.mcp_http_server

# In another terminal, start main server
python -m src.server

🔹 运行验证测试

全面的合规性验证：

python tests/validate_pipeline.py

这验证了：

✅ MCP服务器可访问性和协议合规性
✅ 主控程序 tools/list 和 tools/call 方法
✅ MCP协议端点（JSON-RPC 2.0）
✅ A2A代理卡和规格
✅ 端到端查询处理

预期产量： 所有7项测试均应通过✅

🔹 运行端到端演示

python tests/demo.py

🔹 测试HTTP端点

python tests/test_http.py

______________________________________________________________________

端到端演示（A2A协调）

以下是运行的输出 python tests/demo.py:

所有场景都展示了多步代理之间的协调 路由器代理, 客户数据代理，以及 支援谍员 通过MCP服务器和适当的A2A协议通信使用真实的数据库接地。

______________________________________________________________________

⭐ 场景1——任务分配

用户： *“获取ID 1的客户信息”*

Router Agent → Analyzing query intent
Router → Data Agent: Get customer 1
Data Agent → Router: Customer data retrieved via MCP
Router → Final response synthesized

Final Answer → "Customer Information:
  ID: 1
  Name: Alice Johnson
  Email: alice@example.com
  Phone: 555-0101
  Status: active"

协调步骤： 2

______________________________________________________________________

⭐ 情景2——谈判/升级

用户： *“我已经被收费两次了，请立即退款！”*

Router Agent → Analyzing query intent
Router → Support Agent: Can you handle this?
Support Agent → Router: Response with context check
Router → Data Agent: Get customer context
Data Agent → Router: Customer data retrieved
Router → Support Agent: Generate response with context
Support Agent → Router: Coordinated response ready

Final Answer → "I understand you've been charged twice. 
Let me help you resolve this billing issue immediately..."

协调步骤： 5

______________________________________________________________________

⭐ 场景3——多步协调

用户： *“显示所有有未结门票的活跃客户”*

Router Agent → Analyzing query intent
Router → Data Agent: Get all active customers
Data Agent → Router: Found 3 active customers
Router → Support Agent: Get open tickets for these customers
Support Agent → Router: Found 2 open tickets
Router → Synthesizes report from both responses

Final Answer → "Found 2 active customer(s) with open tickets:
- Alice Johnson (ID: 1, Email: alice@example.com)
  Open Tickets: 1
    • Ticket #1: Login issue (Priority: high)
- Bob Smith (ID: 2, Email: bob@example.com)
  Open Tickets: 1
    • Ticket #2: Billing question (Priority: medium)"

协调步骤： 4

______________________________________________________________________

⭐ 场景4——简单查询

用户： *“我需要帮助处理我的帐户，客户ID 12345”*

Router Agent → Analyzing query intent
Router → Data Agent: Get customer 12345
Data Agent → Router: Customer data retrieved
Router → Support Agent: Handle support query
Support Agent → Router: Support response generated

Final Answer → "I can help you with your account! 
What specific issue are you experiencing?"

协调步骤： 3

______________________________________________________________________

⭐ 场景5——多意图查询

用户： *“更新我的电子邮件至new@email.com并显示我的门票历史记录”*

Router Agent → Analyzing query intent
Router → Data Agent: Update customer email
Data Agent → Router: Update successful
Router → Data Agent: Get customer info
Data Agent → Router: Customer data retrieved
Router → Data Agent: Get ticket history
Data Agent → Router: Found 3 tickets

Final Answer → "Updates completed:
  ✓ Updated customer 1: {'email': 'new@email.com'}

Customer Information:
  Name: Alice Johnson
  Email: new@email.com
  Status: active

Ticket History (3 tickets):
  • Ticket #1: Login issue
    Status: open, Priority: high
  • Ticket #2: Feature request
    Status: resolved, Priority: low
  • Ticket #3: Billing question
    Status: open, Priority: medium"

协调步骤： 4

______________________________________________________________________

✔ 总结

这些场景展示了：

多步推理 跨多个代理
代理间协调 通过A2A协议
工具接地 通过MCP HTTP协议
工单创建工作流程 具有数据库持久性
依赖上下文的响应 基于客户数据
SQLite支持的状态持久性 对于所有操作
独立代理服务 具有适当的A2A接口
LangGraph SDK集成 用于国家管理

______________________________________________________________________

MCP检验员测试

MCP服务器与MCP Inspector完全兼容：

启动MCP服务器：

   python -m src.mcp_http_server

连接MCP检查器 到 http://localhost:8003/mcp

试验方法：

- initialize -初始化MCP会话 - tools/list -列出所有可用工具 - tools/call -调用特定工具

MCP协议端点：

POST /mcp -客户端到服务器消息（返回JSON-RPC 2.0）
GET /mcp -服务器到客户端流媒体（SSE）
GET /tools/list -直接工具清单（用于测试）
POST /tools/call -直接工具调用（用于测试）

MCP检查器兼容性：

✅ 发布 /mcp 返回JSON响应（标准MCP协议）
✅ 所有响应均采用JSON-RPC 2.0格式
✅ 为基于web的客户端启用CORS
✅ 会话管理通过 Mcp-Session-Id 头球

______________________________________________________________________

A2A代理卡

每个代理都公开了一个与代理卡兼容的A2A接口：

访问代理卡

# List all agents
curl http://localhost:8000/agents

# Get specific agent card
curl http://localhost:8001/agent-card  # Customer Data Agent
curl http://localhost:8002/agent-card  # Support Agent
curl http://localhost:8004/agent-card  # Router Agent

代理卡结构

每张代理卡包括：

代理ID：唯一标识符
名字：人类可读名称
描述代理人做什么
能力：功能列表（例如，数据检索、票务管理）
任务：具有输入/输出模式的可用任务
端点：A2A通信的HTTP端点

看 A2A_SPECIFICATIONS.md 详细文档。

______________________________________________________________________

API示例

流式查询（服务器发送的事件）

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"query": "Get customer information for ID 5"}' \
  --no-buffer

同步查询

curl -X POST http://localhost:8000/query/sync \
  -H "Content-Type: application/json" \
  -d '{"query": "I need help with my account, customer ID 12345"}'

健康检查

curl http://localhost:8000/health

______________________________________________________________________

结论

该项目展示了一个完整的多代理系统，具有适当的A2A（代理到代理）通信协议和MCP（模型上下文协议）集成。构建MCP HTTP服务器教会了我如何实现与MCP Inspector等工具兼容的标准协议，确保互操作性和可测试性。A2A接口实现向我展示了如何设计真正独立的代理，这些代理可以发现彼此的能力，并通过结构化协议进行通信。

实现LangGraph SDK集成帮助我了解了基于状态的编排以及如何管理复杂的多步骤工作流。代理独立架构展示了代理如何作为单独的服务运行，同时保持适当的通信协议。这段经历使简单函数调用和真正的代理间通信之间的区别变得清晰。

最大的挑战是确保MCP协议合规性（特别是JSON-RPC 2.0格式），在所有代码路径上实现适当的错误处理，以及协调多个独立的服务。调试分布式代理通信并确保所有代理正确使用MCP客户端（而不是直接访问数据库）需要仔细注意架构和协议遵守情况。

总的来说，这个项目让我获得了构建具有适当协议、独立服务和工具基础的生产风格多代理架构的实践经验。它增强了我对代理编排、协议设计和系统级AI架构的理解。

______________________________________________________________________

技能展示

多智能体系统：具有A2A接口的独立代理
模型上下文协议（MCP）：基于HTTP的MCP服务器实现
A2A协议：使用代理卡进行代理间通信
LangGraph SDK：用于编排的状态图和消息传递
JSON-RPC 2.0：标准协议实施
快速API：支持流媒体的HTTP服务器
异步Python：异步代理通信
SQLite：数据库设计和管理
工具接地LLM：基于MCP的工具访问
代理人独立性：具有适当接口的独立服务
MCP检查器兼容性：符合标准协议
服务器发送事件（SSE）：流式响应
RESTful API：代理服务终结点

______________________________________________________________________

未来的增强功能

WebSocket支持双向通信
使用NLP进行更复杂的意图分析
跨会话的代理状态持久性
用于并行处理的多线程代理执行
用于查询提交和监控的Web界面
高级错误处理和重试逻辑
代理服务的身份验证和授权
代理性能指标和监控
支持其他MCP传输协议
与外部API和服务集成

______________________________________________________________________

故障排除

HTTP服务器问题

端口已在使用中：更改端口 src/server.py 或停止其他服务
连接被拒绝：确保服务器正在运行： python -m src.server 或 ./scripts/start_server.sh
流媒体不工作：检查客户端是否支持服务器发送事件

数据库问题

如果遇到数据库错误：

删除 customer_service.db 然后跑 python scripts/setup_database.py 再次
确保SQLite在Python环境中可用

导入错误

如果您看到导入错误：

确保安装了所有依赖项： pip install -r requirements.txt
检查Python版本： python --version （应为3.8+）

代理协调问题

检查控制台日志以了解详细的A2A消息流
验证数据库是否包含示例数据
查看输出中的coordination_log
对于HTTP A2A模式，确保代理服务在正确的端口上运行

______________________________________________________________________

文档

其他文件可在 docs/ 目录：

A2A_SPECIFICATIONS.md -A2A协议、代理卡和LangGraph集成
结论.md -学习成果和挑战

______________________________________________________________________

致谢

课程：多代理系统\ 作业：作业5-多代理系统和MCP\ 年: 2025

作为Multi-Agent系统课程作业的一部分创建。

Multiagent MCP

快速接入

详细介绍

多代理客户服务系统

MCP供电•A2A协议•LangGraph SDK•独立代理•SQLite接地

项目概述

项目结构

系统架构

特性

MCP HTTP服务器

路由器代理（编排器）

客户数据代理（专家）

支持代理（专家）

LangGraph SDK集成

代理人独立性

安装

1.克隆存储库

2.创建虚拟环境

3.安装依赖项

4.初始化数据库

如何跑步

🔹 快速入门（所有独立服务）

🔹 手动设置（独立服务）

🔹 单流程模式（开发）

🔹 运行验证测试

🔹 运行端到端演示

🔹 测试HTTP端点

端到端演示（A2A协调）

⭐ 场景1——任务分配

⭐ 情景2——谈判/升级

⭐ 场景3——多步协调

⭐ 场景4——简单查询

⭐ 场景5——多意图查询

✔ 总结

MCP检验员测试

A2A代理卡

访问代理卡

代理卡结构

API示例

流式查询（服务器发送的事件）

同步查询

健康检查

结论

技能展示

未来的增强功能

故障排除

HTTP服务器问题

数据库问题

导入错误

代理协调问题

文档

致谢

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP