Multiagent Cs

多代理客户服务系统

使用A2A协调和MCP工具进行多代理客户支持编排。

概述

该项目实施了一个多代理客户服务自动化系统，包括：

• 分析用户意图并协调任务的路由器主机代理
• 客户数据代理是唯一允许通过MCP工具访问数据库的代理
• 处理面向客户的响应、升级和操作的支持代理
• 一个完整的MCP服务器，将SQLite客户支持数据库作为结构化工具公开
• 用于任务分配、协商和升级的多步A2A通信
• 通过以下方式用示例用户和票证初始化SQLite数据库 database_setup.py

该系统旨在反映真实的客户支持工作流程，并演示 A2A协调加MCP整合完成作业。

结论

在这个项目上的工作使A2A模型感觉更加具体。将路由器主机代理、专用客户数据代理和支持代理连接起来，展示了当代理之间相互通信而不是每个代理拥有自己的随机工具时，如何清晰地分离责任。

MCP服务器在这里特别适合，因为它迫使客户数据代理将数据库视为具有类型化工具的定义良好的外部功能，而不是让模型产生SQL幻觉或制造客户对象。让路由器将这两位专家视为远程A2A代理，也反映了真正的多服务生产系统的样子。

有一些真正的挑战。让A2A SDK、ADK和MCP服务器就端口、URL和超时达成一致需要一些迭代，Gemini速率限制和API密钥设置导致失败，直到配置被清除。调试多跳交互比调试单个代理更难，因此详细记录A2A事件并将完整的终端输出保存在 runs/main_output.txt 和 runs/mcp_output.txt 事实证明，这是至关重要的。

总体而言，该项目在设计代理边界、仔细考虑数据所有权以及使系统具有足够的可观察性以实际跟踪多代理协调错误方面是一个很好的练习。

系统架构

代理

所有代理都是使用Google ADK实现的 main.py:

• 客户数据代理

- 使用MCP工具进行读写 support.db - 电话： - get_customer(customer_id) - list_customers(status, limit) - update_customer(customer_id, data) - create_ticket(customer_id, issue, priority) - get_customer_history(customer_id) - 返回结构化JSON customer, tickets，还有一丝 db_calls

• 客服专员

- 接收原始用户查询和 customer_context JSON对象 - 决定如何应对，何时升级，以及要求哪些说明或后续行动 - 总是返回结构化JSON： - final_reply 为客户 - actions 人类或下游系统列表 - escalate 和 escalation_reason - coordination_log 描述内部推理

• 路由器主机代理

- 暴露为自己的A2A代理 - 用途 RemoteA2aAgent 客户数据和支持代理的包装器 - 对于每个传入查询： - 首先调用客户数据代理，通过MCP获取或更新上下文 - 将生成的上下文传递给Support Agent - 返回支持代理的JSON作为最终结果

A2A协调

每个代理在以下位置公开一张代理卡 /.well-known/agent-card.json 并经营自己的 本地主机上的A2A服务器：

• 客户数据代理： http://127.0.0.1:10020
• 支持代理： http://127.0.0.1:10021
• 路由器主机代理： http://127.0.0.1:10022

main.py 包括一个小 A2ASimpleClient 即：

1. 获取路由器的代理卡
2. 通过JSON RPC将用户查询发送到路由器
3. 打印支持代理返回的最终JSON回复

MCP服务器和数据库

MCP服务器 src/mcp/db_mcp_server.py 将数据库作为工具公开，使用 FastMCP. 它连接到 support.db 在项目根中。

工具：

• get_customer(customer_id)
• list_customers(status="active", limit=50)
• update_customer(customer_id, data)
• create_ticket(customer_id, issue, priority="medium")
• get_customer_history(customer_id)

数据库架构与分配匹配：

客户

• id 整数主键
• name 文本不为空
• email 文本
• phone 文本
• status 文本('active' 或 'disabled')
• created_at 时间戳
• updated_at 时间戳

票

• id 整数主键
• customer_id 英特尔（FK至 customers.id)
• issue 文本不为空
• status 文本('open', 'in_progress', 'resolved')
• priority 文本('low', 'medium', 'high')
• created_at 日期时间

src/database_setup.py 是一个辅助脚本，用于创建表、触发器和插入 示例数据，并可以运行示例查询。

项目结构

multiagent-cs/
├── README.md
├── LICENSE
├── main.py                  # Entry point that starts all A2A agents and runs scenarios
├── pyproject.toml           # uv / PEP 621 project configuration
├── requirements.txt         # Package list for non uv installs
├── uv.lock                  # Lockfile managed by uv
├── .env.example             # Example env file with GOOGLE_API_KEY
├── .pre-commit-config.yaml  # Pre-commit hooks
├── .python-version
├── runs/
│   ├── main_output.txt      # Saved terminal output from running main.py
│   └── mcp_output.txt       # Saved terminal output from running the MCP server
└── src/
    ├── database_setup.py    # Creates and populates support.db
    └── mcp/
        └── db_mcp_server.py # FastMCP server exposing DB tools

support.db 运行时在项目根目录中创建 src/database_setup.py.

设置和安装

您可以使用 uv （推荐）或普通 venv 随着 requirements.txt.

1.克隆仓库

git clone git@github.com:sohammandal/multiagent-cs.git
cd multiagent-cs

2.安装依赖项 uv

# Install uv if needed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create the virtual environment and install dependencies
uv sync

# Activate the virtual environment
source .venv/bin/activate

如果你喜欢素色 venv:

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

3.环境变量

复制 .env.example 到 .env 并设置您的Gemini密钥：

cp .env.example .env
# then edit .env and put:
# GOOGLE_API_KEY=your_real_key

代码使用 GOOGLE_API_KEY 和Gemini API，模型：

gemini-2.0-flash-lite

4.初始化数据库

从项目根目录运行数据库安装脚本。这将创建 support.db 并且可选地插入样本数据。

uv run python src/database_setup.py

出现提示时：

• 回答 y 如果你想插入客户和门票样本
• 如果要检查数据，可以选择运行交互式示例查询

5.运行MCP服务器

在一个终端中（venv已激活）：

uv run python src/mcp/db_mcp_server.py

您应该看到显示正在侦听的服务器的日志：

http://127.0.0.1:8001

并回应 /sse 和 ListToolsRequest此步骤的样本输出保存在：

runs/mcp_output.txt

6.运行多代理A2A系统

打开第二个终端，激活相同的venv，并从项目根运行：

uv run python main.py

这将：

1. 在后台启动三个A2A代理：

- 端口10020上的客户数据代理 - 端口10021上的支持代理 - 端口10022上的路由器主机代理

1. 使用 A2ASimpleClient 向路由器发送一系列测试查询

1. 打印来自支持代理的结构化JSON结果

完整捕获的跑步记录保存在：

runs/main_output.txt

它显示了每个场景的日志输出和JSON。

测试场景

main.py 自动对Router Host Agent运行五种方案：

1. 简单查询

- 查询： "Get customer information for ID 5" - 流程：路由器通过A2A呼叫客户数据代理，A2A呼叫 get_customer 通过MCP。

1. 协同查询（任务分配）

- 查询： "I'm customer 12345 and need help upgrading my account" - 流程：路由器向客户数据代理询问客户5，然后要求支持代理 为该客户处理升级。

1. 复杂查询（多步骤和协商）

- 查询： "Show me all active customers who have open tickets" - 流程：路由器询问活跃客户和票务信息。支持代理决定 当可用工具无法满足确切的报告请求时，升级。

1. 升级

- 查询： "I've been charged twice, please refund immediately!" - 流程：路由器通过客户数据代理获取上下文，支持代理识别 紧急情况，创建高优先级票证，并提供清晰的升级日志。

1. 多意图

- 查询： "Update my email to new@email.com and show my ticket history" - 流程：路由器使用客户数据代理通过以下方式更新电子邮件 update_customer 并通过以下方式获取门票历史记录 get_customer_history，然后传递该上下文 支持最终答复。

这些流程表明：

• 基于A2A的任务分配和路由
• 谈判和升级决定
• 跨代理和MCP工具的多步协调

端到端演示

作业允许Colab笔记本或端到端运行的python程序。 此存储库使用python程序方法：

• main.py 是端到端的入口点。

• runs/main_output.txt 包含一个完整的终端截图，显示：

- 代理启动和A2A日志 - MCP呼叫 - 所有五个必需场景的JSON响应

开发说明

预提交挂钩

可选，但包括：

uv tool install pre-commit
pre-commit install
pre-commit run --all-files

MCP检查员

因为MCP服务器在上使用SSE 127.0.0.1:8001，可以使用MCP工具进行检查 例如MCP检查器，通过将它们指向该URL。