使用说明

多代理客户服务系统

MCP协议和A2A通信的正确实现

一个生产就绪的多代理客户服务系统，展示了：

✅ 模型上下文协议（MCP） -基于HTTP的JSON-RPC 2.0
✅ 代理到代理（A2A）通信 -显式消息传递
✅ 专业代理 -编排器、数据检索器、客户助理
✅ 整洁架构 -关注点分离和协议合规性

📋 目录

概述
建筑
主要特点
安装
用法
项目结构
MCP协议实现
A2A通信
测试
api参考
故障排除

🎯 概述

该系统实现了 符合协议 用于客户服务的多代理架构，解决了幼稚实现中常见的问题：

是什么让这个实现正确？

方面	❌ 常见错误	✅ 此实施
主控程序	带方法的Python类	带JSON-RPC 2.0的HTTP服务器
A2A	共享状态字典	显式消息传递
运输	直接函数调用	HTTP请求+Message对象
协议	无协议	标准化MCP+A2A协议

🏗️ 建筑

┌─────────────────────────────────────────────────────────────┐
│                          USER QUERY                          │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
        ┌────────────────────────────────────┐
        │    Orchestrator Agent (Router)     │
        │  • Analyzes intent                 │
        │  • Coordinates workflow            │
        │  • Sends A2A messages              │
        └────────┬───────────────────┬───────┘
                 │ A2A Message       │ A2A Message
                 ▼                   ▼
    ┌────────────────────┐   ┌──────────────────────┐
    │ Data Retriever     │   │ Customer Assistant   │
    │ • Calls MCP server │   │ • Generates response │
    │ • Returns data     │   │ • Uses context       │
    └────────┬───────────┘   └──────────────────────┘
             │ JSON-RPC
             ▼
    ┌────────────────────┐
    │   MCP Server       │
    │ • HTTP/JSON-RPC    │
    │ • Database access  │
    │ • 5 tools          │
    └────────────────────┘

✨ 主要特点

1.正确实施MCP协议

位置: protocol_server/server.py

# JSON-RPC 2.0 Request
POST /rpc
{
  "jsonrpc": "2.0",
  "method": "fetch_customer",
  "params": {"customer_id": 1},
  "id": 1
}

# JSON-RPC 2.0 Response
{
  "jsonrpc": "2.0",
  "result": {"success": true, "data": {...}},
  "id": 1
}

特性:

✅ HTTP传输（Flask服务器）
✅ JSON-RPC 2.0协议
✅ 服务器发送事件（SSE）支持
✅ 健康检查和工具发现
✅ 正确的错误处理

2.真正的A2A消息传递

位置: messaging/agent_messenger.py

# Agents communicate via messages, not shared state
message = Message(
    msg_type=MessageType.REQUEST,
    sender="orchestrator",
    receiver="data_retriever",
    content={"action": "fetch_history", "customer_id": 1}
)

response = messenger.send_message(message)

特性:

✅ 带有元数据的消息对象
✅ 异步消息传递
✅ 对话跟踪
✅ 通信记录
✅ 代理注册表

3.专业代理

代理	角色	沟通
编排器	协调工作流程	向专家发送消息
数据检索器	MCP客户端	调用JSON-RPC工具
客户助理	支持响应	通过消息接收上下文

🚀 安装

先决条件

Python 3.10或更高版本
OpenAI API密钥
Git

步骤1：克隆存储库

git clone https://github.com/kewang34k/Multi-Agent-Customer-Service-System-with-A2A-and-MCP.git
cd Multi-Agent-Customer-Service-System-with-A2A-and-MCP

第二步：创建虚拟环境

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

步骤3：安装依赖项

pip install -r requirements.txt

步骤4：设置OpenAI API密钥

export OPENAI_API_KEY='your-api-key-here'

或者创建一个 .env 文件：

echo "OPENAI_API_KEY=your-api-key-here" > .env

步骤5：初始化数据库

python database_setup.py

输出:

✓ Removed existing database: customer_service.db
✓ Created customers table
✓ Created tickets table
✓ Inserted 10 test customers
✓ Inserted 14 test tickets

💻 用法

快速启动（演示模式）

终端1-启动MCP服务器:

python start_server.py

输出:

============================================================
Starting MCP Protocol Server
============================================================
MCP Server Endpoints:
  Health Check: http://127.0.0.1:5001/health
  Tools List:   http://127.0.0.1:5001/tools
  JSON-RPC:     http://127.0.0.1:5001/rpc
  SSE Stream:   http://127.0.0.1:5001/stream
============================================================

2号航站楼-运行系统:

python run_system.py

这将运行4个演示场景，显示：

客户数据检索
根据客户情况提供支持
一般支持（无客户数据）
高优先级问题处理

交互模式

python run_system.py interactive

示例会话:

💬 Your query: I need help with billing, customer ID 2

🤖 Response:
Hello Bob Martinez,

I see you're inquiring about billing. I've reviewed your account and
noticed you have 2 open tickets related to system integration and API
rate limits.

How can I specifically help with billing today?

📌 Intent: billing_question
🔧 Workflow: Orchestrator → Data Retriever → Customer Assistant

📁 项目结构

Multi-Agent-Customer-Service-System-with-A2A-and-MCP/
│
├── protocol_server/              # MCP Protocol Implementation
│   ├── __init__.py
│   ├── database.py               # Database operations
│   └── server.py                 # Flask HTTP server with JSON-RPC
│
├── service_agents/               # A2A Agents
│   ├── __init__.py
│   ├── orchestrator.py           # Workflow coordinator
│   ├── data_retriever.py         # MCP client
│   └── customer_assistant.py     # Support response generator
│
├── messaging/                    # A2A Message Passing
│   ├── __init__.py
│   └── agent_messenger.py        # Message protocol implementation
│
├── helpers/                      # Utilities
│   ├── __init__.py
│   └── parsers.py                # JSON parsing, logging
│
├── database_setup.py             # Database initialization
├── start_server.py               # MCP server launcher
├── run_system.py                 # System launcher
├── requirements.txt              # Dependencies
├── README.md                     # This file
└── customer_service.db           # SQLite database (created)

🔌 MCP协议实现

MCP服务器端点

1.健康检查

GET http://127.0.0.1:5001/health

回应:

{
  "status": "healthy",
  "protocol": "MCP",
  "version": "1.0"
}

2.列出工具

GET http://127.0.0.1:5001/tools

回应:

{
  "tools": [
    {
      "name": "fetch_customer",
      "description": "Retrieve customer by ID",
      "parameters": {"customer_id": "integer"}
    },
    // ... 4 more tools
  ]
}

3.JSON-RPC工具调用

POST http://127.0.0.1:5001/rpc
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "method": "get_complete_history",
  "params": {"customer_id": 1},
  "id": 1
}

可用的MCP工具

工具	说明	参数
`fetch_customer`	通过ID获取客户	`customer_id: int`
`list_customers`	列出客户	`account_status: str, max_results: int`
`modify_customer`	更新客户	`customer_id: int, updates: dict`
`create_ticket`	创建工单	`customer_id: int, issue_description: str, priority_level: str`
`get_complete_history`	完整历史	`customer_id: int`

📡 A2A通信

消息结构

@dataclass
class Message:
    msg_type: MessageType       # REQUEST, RESPONSE, INFORM, etc.
    sender: str                 # Agent identifier
    receiver: str               # Target agent
    content: Dict[str, Any]     # Message payload
    conversation_id: str        # Links related messages
    message_id: str             # Unique message ID
    timestamp: str              # ISO format
    reply_to: Optional[str]     # For responses

通信流示例

User Query: "Help with account, ID 1"
    │
    ▼
[Message 1] User → Orchestrator
    type: REQUEST
    content: {"query": "Help with account, ID 1", "customer_id": 1}
    │
    ▼
[Message 2] Orchestrator → Data Retriever
    type: REQUEST
    content: {"action": "fetch_history", "customer_id": 1}
    │
    ▼
[Message 3] Data Retriever → MCP Server
    JSON-RPC: get_complete_history(customer_id=1)
    │
    ▼
[Message 4] Data Retriever → Orchestrator
    type: RESPONSE
    content: {"success": true, "data": {...}}
    │
    ▼
[Message 5] Orchestrator → Customer Assistant
    type: REQUEST
    content: {"query": "...", "customer_context": {...}}
    │
    ▼
[Message 6] Customer Assistant → Orchestrator
    type: RESPONSE
    content: {"response": "Hello Alice, ..."}
    │
    ▼
[Message 7] Orchestrator → User
    type: RESPONSE
    content: {"response": "...", "workflow": "..."}

查看通信日志

# Get formatted log
comm_log = messenger.get_communication_log()

# Output:
# [2024-11-30T10:30:45] orchestrator → data_retriever: request
# [2024-11-30T10:30:46] data_retriever → orchestrator: response
# [2024-11-30T10:30:47] orchestrator → customer_assistant: request
# [2024-11-30T10:30:48] customer_assistant → orchestrator: response

🧪 测试

手动测试

测试MCP服务器:

# Terminal 1
python start_server.py

# Terminal 2
curl http://127.0.0.1:5001/health

curl -X POST http://127.0.0.1:5001/rpc \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "fetch_customer",
    "params": {"customer_id": 1},
    "id": 1
  }'

测试A2A系统:

python run_system.py

验证检查表

[]MCP服务器启动时没有错误
[]健康检查返回“健康”
[]工具端点列出了5个工具
[]JSON-RPC调用返回有效响应
[]代理注册成功
[]信息记录在A2A系统中
[]演示场景已成功完成

📚 api参考

代理信使API

# Initialize
messenger = AgentMessenger()

# Register agent
messenger.register_agent("agent_name", handler_function)

# Create message
msg = messenger.create_message(
    msg_type=MessageType.REQUEST,
    sender="sender_id",
    receiver="receiver_id",
    content={"key": "value"}
)

# Send message (synchronous)
response = messenger.send_message(msg)

# Get statistics
stats = messenger.get_statistics()

数据库处理程序API

# Initialize
db = DatabaseHandler("customer_service.db")

# Fetch customer
customer = db.fetch_customer_by_id(1)

# List customers
customers = db.fetch_customers_list(account_status="active", max_results=10)

# Modify customer
success = db.modify_customer_record(1, {"email": "new@email.com"})

# Create ticket
ticket_id = db.create_support_ticket(1, "Issue description", "high")

# Get complete history
history = db.fetch_customer_complete_history(1)

🐛 故障排除

MCP服务器未启动

错误: Address already in use

解决方案:

# Find process using port 5001
lsof -i :5001

# Kill process
kill -9 <PID>

# Or change port in start_server.py
server = ProtocolServer(db_handler, host="127.0.0.1", port=5002)

无法连接到MCP服务器

错误: MCP server unavailable

检查:

服务器正在运行吗？检查1号航站楼
正确的港口？默认值为5001
防火墙阻止？尝试 curl http://127.0.0.1:5001/health

OpenAI API错误

错误: No API key provided

解决方案:

# Set environment variable
export OPENAI_API_KEY='sk-...'

# Or use .env file
echo "OPENAI_API_KEY=sk-..." > .env
pip install python-dotenv

数据库错误

错误: Database validation failed

解决方案:

# Reinitialize database
python database_setup.py

🎓 关键概念

为什么这个实现是正确的

❌ 错误：共享状态

# Naive approach - all agents modify same dictionary
state = {"query": "...", "data": None, "response": None}
router_agent(state)
data_agent(state)  # Modifies state["data"]
support_agent(state)  # Reads state["data"]

✅ 正确的：消息传递

# Proper A2A - agents send messages
msg = Message(sender="orchestrator", receiver="data_retriever", ...)
response = messenger.send_message(msg)

❌ 错误：直接函数调用

# Naive approach - Python methods
class MCPServer:
    def get_customer(self, id):
        return {...}

result = mcp.get_customer(1)  # Direct call

✅ 正确的：HTTP+JSON-RPC

# Proper MCP - HTTP protocol
POST /rpc
{"jsonrpc": "2.0", "method": "fetch_customer", "params": {...}}

📄 许可证

MIT许可证-有关详细信息，请参阅许可证文件

👥 贡献

这个项目演示了正确的协议实现。欢迎投稿！

需要改进的领域:

向MCP服务器添加身份验证
实现WebSocket传输
添加对话持久性
创建web UI
添加监控仪表板

📞 支持

对于问题或疑问：

检查故障排除章节
查看通信日志： messenger.get_communication_log()
检查MCP服务器运行状况： curl http://127.0.0.1:5001/health
启用调试日志记录： setup_logging("DEBUG")

🌟 致谢

此实现演示 适当的 MCP和A2A协议，解决了多智能体系统中的常见错误。

学习要点:

MCP需要HTTP传输，而不是Python方法
A2A要求消息传递，而非共享状态
协议实现了互操作性和可扩展性
关注点分离提高了可维护性

为生产使用构建了适当的协议合规性 🚀

使用说明

多代理客户服务系统

📋 目录

🎯 概述

是什么让这个实现正确？

🏗️ 建筑

✨ 主要特点

1.正确实施MCP协议

2.真正的A2A消息传递

3.专业代理

🚀 安装

先决条件

步骤1：克隆存储库

第二步：创建虚拟环境

步骤3：安装依赖项

步骤4：设置OpenAI API密钥

步骤5：初始化数据库

💻 用法

快速启动（演示模式）

交互模式

📁 项目结构

🔌 MCP协议实现

MCP服务器端点

1.健康检查

2.列出工具

3.JSON-RPC工具调用

可用的MCP工具

📡 A2A通信

消息结构

通信流示例

查看通信日志

🧪 测试

手动测试

验证检查表

📚 api参考

代理信使API

数据库处理程序API

🐛 故障排除

MCP服务器未启动

无法连接到MCP服务器

OpenAI API错误

数据库错误

🎓 关键概念

为什么这个实现是正确的

❌ 错误：共享状态

✅ 正确的：消息传递

❌ 错误：直接函数调用

✅ 正确的：HTTP+JSON-RPC

📄 许可证

👥 贡献

📞 支持

🌟 致谢

相关 MCP

MCP测试服务器集合

AstrBot 智能体聊天机器人平台

模型上下文协议服务器

MCP开发框架

智能客服系统

MCP协议服务器