使用说明

基于A2A和MCP的多Agent客户服务系统

概述

该项目实现了一个多代理客户服务自动化系统，该系统演示了：

LangGraph SDK 用于代理编排和工作流管理
代理对代理（A2A）协调 使用LangGraph的基于消息的通信
模型上下文协议（MCP） 结构化数据访问的集成
多代理任务分配、协商和多步协调
实用的客户服务自动化 具有专门的代理角色

系统使用 LangGraph SDK 在 demo/main.py 以执行多代理工作流。该工作流协调了通过A2A协议进行通信的代理，展示了专业代理如何协作分析客户查询、检索结构化数据、上报问题和生成面向用户的响应。

系统架构

第1部分：系统架构

设计一个至少有两个专门代理的多代理系统：

路由器代理（编排器）

接收客户查询
分析查询意图
通往合适专业代理的路线
协调多个代理的响应

客户数据代理（专家）

通过MCP访问客户数据库
检索客户信息
更新客户记录
处理数据验证

支持代理（专家）

处理一般客户支持查询
可能使复杂问题升级
从Data Agent请求客户上下文
提供解决方案和建议

注：所有代理人都是 真正的代理人 使用LLM后端进行智能推理和决策，而不是硬编码的聊天机器人：

路由器代理 使用LLM分析查询、检测意图并动态决定路由（没有预定义的场景匹配）
客户数据代理 使用LLM根据查询上下文推断需要哪些数据操作（没有硬编码的动作映射）
支援谍员 使用LLM根据所有可用数据生成自然、上下文感知的响应（没有基于模板的响应）

关键设计原则： 代理在不硬编码显式场景的情况下推理和解决问题。所有决策（路由、数据操作、响应生成）都是基于查询和当前状态通过LLM推理做出的。

注：如果未配置LLM API密钥，则代理可以使用基于规则的回退逻辑，但真正的代理行为需要LLM推理。

架构图

┌─────────────────────────────────────────────────────────────┐
│                    demo/main.py                             │
│              (Uses LangGraph SDK)                           │
│                                                             │
│  from agents.graph import build_workflow                    │
│  graph_app = build_workflow()                               │
│  final_state = graph_app.invoke(initial_state)              │
└───────────────────────┬─────────────────────────────────────┘
                        │
                        v
        ┌───────────────────────────────────┐
        │   LangGraph Workflow (StateGraph) │
        │   (Orchestrates agents via A2A)   │
        └───────────────┬───────────────────┘
                        │
                        v
        ┌───────────────────────────────────┐
        │   Router Agent (Orchestrator)     │
        │   - Receives customer queries     │
        │   - Analyzes query intent         │
        │   - Routes to specialist agents   │
        │   - Coordinates responses         │
        └───────────────┬───────────────────┘
                        │
        ┌───────────────┴───────────────┐
        │                               │
        v                               v
┌───────────────────────┐ ┌───────────────────────┐
│  Customer Data Agent  │ │   Support Agent       │
│     (Specialist)      │ │     (Specialist)      │
│                       │ │                       │
│ - Accesses DB via MCP │ │ - Handles support     │
│ - Retrieves customer  │ │   queries             │
│   information         │ │ - Escalates issues    │
│ - Updates records     │ │ - Requests context    │
│ - Handles validation  │ │   from Data Agent     │
│                       │ │ - Provides solutions  │
└───────────┬───────────┘ └───────────┬───────────┘
            │                         │
            │   A2A                   │   A2A
            │                         │
            │                         │
            ▼                         ▼
        ┌─────────────────────────────────┐
        │      MCP Server                 │
        │      (Database)                 │
        │                                 │
        │  - get_customer                 │
        │  - list_customers               │
        │  - update_customer              │
        │  - create_ticket                │
        │  - get_customer_history         │
        └─────────────────────────────────┘
                        │
                        │
                        v
         ┌───────────────────────────────┐
         │      Final Response           │
         └───────────────────────────────┘

协议遵从

此实现旨在符合这两个要求 MCP（模型上下文协议） 和 A2A（代理人对代理人） 任务要求的规格。

MCP协议合规性

MCP服务器(db_mcp_server.py)实施：

SSE（服务器发送事件） 端点(/sse)用于流媒体通信
/tools/list 端点返回带有JSON模式定义的完整工具模式
/tools/call 工具执行端点
兼容 MCP检查员 以及标准MCP客户端

MCP检验员测试：

# Start MCP server
python db_mcp_server.py

# Test tools/list
curl http://localhost:8000/tools/list

# Test tools/call
curl -X POST http://localhost:8000/tools/call \
  -H "Content-Type: application/json" \
  -d '{"tool": "get_customer", "arguments": {"customer_id": 1}}'

A2A协议合规性

每个代理都实现了完整的A2A规范：

/agent/card 代理发现端点（GET）
/agent/tasks 任务执行端点（POST）
/a2a/{assistant_id} LangGraph A2A兼容性（GET）的端点
/health 服务监控端点
通过A2A协议进行通信的独立代理服务
代理使用标准化端点发现并相互通信

代理端点：

路由器代理： http://localhost:8001/agent/card, /agent/tasks, /a2a/router-agent
数据代理： http://localhost:8002/agent/card, /agent/tasks, /a2a/customer-data-agent
支持代理： http://localhost:8003/agent/card, /agent/tasks, /a2a/support-agent

看 PROTOCOLS.md 详细的协议规范和测试程序。

MCP集成

该系统包括一个完整的MCP服务器实现，带有以下工具：

所需工具

get_customer(customer_id) -按ID检索单个客户记录
list_customers(status, limit) -列出按状态（活动/禁用）筛选的客户
update_customer(customer_id, data) -更新客户字段（姓名、电子邮件、电话、状态）
create_ticket(customer_id, issue, priority) -创建新的支持票
get_customer_history(customer_id) -检索客户的所有门票

数据库模式

客户表：

id （整数主键）
name （文本不为空）
email （文本）
phone （文本）
status （文本：“活动”或“禁用”）
created_at （时间戳）
updated_at （时间戳）

门票表：

id （整数主键）
customer_id （INTEGER，FK致客户.id）
issue （文本不为空）
status （文本：“打开”、“正在进行中”、“已解决”）
priority （文本：“低”、“中”、“高”）
created_at （日期时间）

A2A协调场景

该系统实现了三种核心协调模式：

场景1：任务分配

查询： “我需要帮助处理我的帐户，客户ID 12345。”

A2A流量：

路由器代理接收查询并提取客户ID
路由器代理→ 客户数据代理：“获取ID 12345的客户信息”
客户数据代理通过MCP获取
客户数据代理→ 路由器代理：返回客户数据
路由器代理→ 支持代理：“为客户提供支持”
支持代理生成响应
路由器代理返回最终响应

情景2：谈判/升级

查询： “我想取消订阅，但我遇到了账单问题”

A2A流量：

路由器检测到多个意图（取消+计费）
路由器→ 客服：“你能处理这个吗？”
支援谍员→ 路由器：“我需要账单上下文”
路由器在代理之间协商以制定响应
支持代理通过MCP创建高优先级票证
向客户发送协调响应

场景3：多步协调

查询： “高级客户的所有高优先级机票的状态如何？”

A2A流量：

路由器分解为子任务
路由器→ 客户数据代理：“获取所有活跃客户”
客户数据代理→ 路由器：返回客户列表
路由器→ 支持代理：“获取这些ID的高优先级票证”
支持代理通过MCP为每个客户查询工单
代理协调以格式化报告
路由器合成最终答案

安装和设置

先决条件

Python 3.8或更高版本
pip（Python包管理器）

步骤1：创建虚拟环境

# Create virtual environment
python3 -m venv venv

# Activate virtual environment
# On macOS/Linux:
source venv/bin/activate
# On Windows:
# venv\Scripts\activate

步骤2：安装依赖项

pip install -r requirements.txt

步骤3：配置LLM（代理推理所需）

代理使用LLM后端进行智能推理。您需要配置一个API密钥：

选项1：使用OpenAI

export OPENAI_API_KEY="your_openai_api_key_here"
export LLM_PROVIDER="openai"
export LLM_MODEL="gpt-5-nano"  # or "gpt-4"

选项2：使用Anthropic

export ANTHROPIC_API_KEY="your_anthropic_api_key_here"
export LLM_PROVIDER="anthropic"
export LLM_MODEL="claude-3-haiku-20240307"

选项3：使用.env文件（推荐）

复制示例文件：
```
cp .env.example .env
```

编辑 .env 并添加您的API密钥：

OPENAI_API_KEY=your_openai_api_key_here
LLM_PROVIDER=openai
LLM_MODEL=gpt-3.5-turbo

安装python dotenv（如果尚未安装）：
```
pip install python-dotenv
```

系统将自动从以下位置加载环境变量 .env 文件。

注：如果没有配置API密钥，系统将使用基于规则的回退逻辑（对于测试很有用，但代理不具有LLM推理功能）。

步骤4：设置数据库

python database_setup.py

出现提示时：

进入 y 插入示例数据
可选择输入 y 运行示例查询

这将创建 support.db 有样品客户和门票。

步骤5：验证安装

运行快速测试以确保一切设置正确：

python -c "from agents.graph import build_workflow; print('Setup successful!')"

用法

运行演示（使用LangGraph SDK）

演示脚本(demo/main.py) 直接使用LangGraph SDK 执行多代理工作流：

python demo/main.py

它是如何工作的：

导入LangGraph SDK： from agents.graph import build_workflow
构建工作流： graph_app = build_workflow()
使用创建初始状态 messages 键（LangGraph A2A所需）
执行工作流： graph_app.invoke(initial_state)
LangGraph工作流协调通过A2A协议进行通信的代理

先决条件：

MCP服务器必须正在运行（端口8000）
数据代理服务器必须正在运行（端口8002）
支持代理服务器必须正在运行（端口8003）
注：由于我们直接使用LangGraph SDK，因此不需要路由器代理服务器

演示显示了什么：

用户查询
检测到的场景和意图
代理间通信日志
A2A消息
最终答复

启动所需服务

在运行演示之前，请启动所需的服务：

终端1：启动MCP服务器

python db_mcp_server.py

服务器在上运行 http://localhost:8000
MCP端点： /sse, /tools/list, /tools/call
MCP检验员测试： 连接到 http://localhost:8000/sse

终端2：启动数据代理（A2A）

python data_agent_server.py

服务器在上运行 http://localhost:8002
A2A端点： /agent/card, /agent/tasks, /a2a/{assistant_id}, /health

终端3：启动支持代理（A2A）

python support_agent_server.py

服务器在上运行 http://localhost:8003
A2A端点： /agent/card, /agent/tasks, /a2a/{assistant_id}, /health

可选：路由器代理服务器（用于HTTP A2A访问）

如果你想通过HTTP A2A访问路由器代理（而不是直接使用LangGraph SDK）：

python router_agent_server.py

服务器在上运行 http://localhost:8001
A2A端点： /agent/card, /agent/tasks, /a2a/{assistant_id}, /health
在内部使用LangGraph SDK通过A2A协议协调其他代理

验证协议合规性

测试MCP服务器：

# Test tools/list (should return full tool schemas)
curl http://localhost:8000/tools/list

# Test tools/call
curl -X POST http://localhost:8000/tools/call \
  -H "Content-Type: application/json" \
  -d '{"tool": "get_customer", "arguments": {"customer_id": 1}}'

测试A2A试剂：

# Get agent cards (discovery)
curl http://localhost:8001/agent/card  # Router (if running)
curl http://localhost:8002/agent/card  # Data Agent
curl http://localhost:8003/agent/card  # Support Agent

# Get agent cards via A2A endpoint
curl http://localhost:8002/a2a/customer-data-agent
curl http://localhost:8003/a2a/support-agent

# Check health
curl http://localhost:8001/health  # Router (if running)
curl http://localhost:8002/health
curl http://localhost:8003/health

通过HTTP A2A执行查询（可选）

如果路由器代理服务器正在运行，您也可以通过HTTP调用它：

curl -X POST http://localhost:8001/agent/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "user_query": "Get customer information for ID 5"
    }
  }'

路由器代理将：

在内部使用LangGraph SDK来编排工作流
通过A2A呼叫数据代理(/agent/tasks)
数据代理调用MCP服务器(/tools/call)
路由器通过A2A呼叫支持代理
返回最终响应

测试场景

系统处理以下测试场景：

1.简单查询

查询： “获取ID 5的客户信息”
预期： 单代理，直接的MCP呼叫

2.协同查询

查询： “我是客户12345，需要帮助升级我的帐户”
预期： 多个代理协调：数据获取+支持响应

3.复杂查询

查询： “显示所有有未结门票的活跃客户”
预期：

Data Agent通过MCP获取活跃客户
支持代理为这些客户获取未结工单
响应列出了有未结票证的客户，然后列出了他们所有未结票证及其详细信息

4.升级

查询： “我已经被收费两次了，请立即退款！”
预期： 路由器识别紧急情况并适当路由，创建高优先级票证

5.多意图

查询： “更新我的电子邮件至new@email.com并显示我的门票历史记录”
预期： 并行任务执行和协调

项目结构

.
├── agents/                  # Agent implementations
│   ├── __init__.py
│   ├── router_agent.py     # Router/orchestrator agent
│   ├── data_agent.py       # Customer data specialist
│   ├── support_agent.py    # Support specialist
│   ├── graph.py            # LangGraph workflow definition
│   ├── state.py            # Shared state structure
│   ├── llm_config.py       # LLM configuration and initialization
│   └── mcp_client.py       # MCP tool client wrapper
├── demo/                   # Demo scripts
│   └── main.py            # End-to-end demonstration
├── .env.example            # Environment variables template
├── database_setup.py       # Database initialization script
├── db_mcp_server.py        # MCP server (FastAPI)
├── router_agent_server.py  # Router agent A2A server
├── data_agent_server.py    # Data agent A2A server
├── support_agent_server.py # Support agent A2A server
├── config.py               # Configuration (URLs, paths)
├── requirements.txt        # Python dependencies
├── README.md              # This file
├── QUICKSTART.md          # Quick start guide
├── PROTOCOLS.md           # Protocol documentation
└── support.db             # SQLite database (created after setup)

技术栈

LangGraph SDK：用于代理编排和工作流管理
- StateGraph 从 langgraph.graph 用于定义多代理工作流
- A2A兼容性的基于消息的状态（必须具有 messages 按键）
- 条件路由和边缘管理
- 本地工作流编译和执行
- 直接用于 demo/main.py 在路由器代理服务器内部
LangChain：用于LLM集成（OpenAI、Anthropic）
- ChatOpenAI 和 ChatAnthropic LLM后端
- 提示模板和输出解析
快速API：用于MCP和A2A服务器端点
- MCP服务器： /sse, /tools/list, /tools/call
- 代理服务器： /agent/card, /agent/tasks, /a2a/{assistant_id}, /health
SQLite：客户和票务数据的数据库后端

关键实施细节

LangGraph SDK使用方法

此实现使用 LangGraph SDK 对于代理编排：

工作流定义：用途 StateGraph 定义多代理工作流
节点注册：每个代理都注册为LangGraph节点
条件路由：用途 add_conditional_edges() 用于基于场景的路由
状态管理：共享状态(CSState)在代理人之间传递
工作流编译：工作流是使用编译的 workflow.compile()
执行：使用以下命令调用工作流 app.invoke(initial_state)

关键实施：

demo/main.py 直接使用LangGraph SDK： from agents.graph import build_workflow
路由器代理服务器内部也使用LangGraph SDK进行编排
两者都使用来自的相同工作流定义 agents/graph.py

LangGraph A2A兼容性

该系统实现了LangGraph的A2A要求：

基于消息的状态：所有州都包括a messages 列表（LangGraph A2A需要）

initial_state: CSState = {
    "messages": [],  # Required for LangGraph A2A compatibility
    "user_query": query,
    "logs": [],
}

代理卡端点：每个代理都暴露 /a2a/{assistant_id} 端点
- 路由器代理： /a2a/router-agent
- 数据代理： /a2a/customer-data-agent
- 支持代理： /a2a/support-agent
A2A通信：代理通过HTTP A2A协议进行发现和通信
- 客服可以通过拨打电话来发现彼此 /a2a/{assistant_id}
- 代理通过以下方式进行通信 /agent/tasks 使用HTTP+JSON的端点

A2A消息传递

该系统使用LangGraph的基于消息的状态结构来实现A2A兼容性：

所有州都包括 messages 列表（A2A所需）
记录消息以进行调试和透明度
代理通过结构化消息与发送方/接收方信息进行通信

MCP集成

MCP工具可通过HTTP端点访问(/tools/call)
代理可以直接调用工具（本地演示）或通过HTTP（分布式设置）调用工具
所有数据库操作都经过MCP以确保一致性

错误处理

妥善处理丢失的客户ID
验证客户状态和工单优先级
清除无效输入的错误消息

故障排除

数据库问题

问题： support.db 未找到
解决方案： 跑 python database_setup.py 创建数据库

问题： 数据库状态已损坏
解决方案： 删除 support.db 然后跑 database_setup.py 再次

端口冲突

问题： 端口已在使用中
解决方案： 更新URL config.py 使用不同的端口

导入错误

问题： 未找到模块错误
解决方案： 确保虚拟环境已激活并安装了依赖项：

source venv/bin/activate
pip install -r requirements.txt

学习成果

该项目展示了：

多代理协调： 专业代理如何协作解决复杂任务
A2A通信： 使用显式日志实现代理到代理协议
MCP集成： 结构化访问外部工具和数据源
任务分解： 将复杂的查询分解为可管理的子任务
状态管理： 通过结构化状态在代理之间共享上下文

未来的增强功能

潜在改进：

增强的LLM提示，可实现更好的意图检测和响应质量
跨会话实现持久代理内存/上下文
为生产使用添加身份验证和速率限制
支持其他MCP工具（电子邮件、通知等）
用于客户交互的Web界面
基于客户等级/高级状态的高级路由
通过SSE实时流式响应

许可证

该项目是为了教育目的而创建的，是应用生成人工智能课程作业的一部分。

使用说明

基于A2A和MCP的多Agent客户服务系统

概述

系统架构

第1部分：系统架构

路由器代理（编排器）

客户数据代理（专家）

支持代理（专家）

架构图

协议遵从

MCP协议合规性

A2A协议合规性

MCP集成

所需工具

数据库模式

A2A协调场景

场景1：任务分配

情景2：谈判/升级

场景3：多步协调

安装和设置

先决条件

步骤1：创建虚拟环境

步骤2：安装依赖项

步骤3：配置LLM（代理推理所需）

步骤4：设置数据库

步骤5：验证安装

用法

运行演示（使用LangGraph SDK）

启动所需服务

终端1：启动MCP服务器

终端2：启动数据代理（A2A）

终端3：启动支持代理（A2A）

可选：路由器代理服务器（用于HTTP A2A访问）

验证协议合规性

通过HTTP A2A执行查询（可选）

测试场景

1.简单查询

2.协同查询

3.复杂查询

4.升级

5.多意图

项目结构

技术栈

关键实施细节

LangGraph SDK使用方法

LangGraph A2A兼容性

A2A消息传递

MCP集成

错误处理

故障排除

数据库问题

端口冲突

导入错误

学习成果

未来的增强功能

许可证

相关 MCP

Git MCP 服务器

AI助手协议服务器

AI任务管理系统

AstrBot 智能体聊天机器人平台

智能代理编排平台

媒体管理协议服务器