AI导哈

多智能体编排系统

一个结合Google ADK、LangGraph、Model Context Protocol (MCP)和A2A通信的分布式AI智能体编排系统，用于构建可扩展的多智能体工作流。

  • by AuthurBoyle-03
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 AuthurBoyle-03/MCP-A2A-LangGraph-Pipelining

使用说明

MCP-A2A-LangGraph-流水线

一个复杂的多代理编排系统，结合了 谷歌ADK, LangGraph, 模型上下文协议（MCP），以及 A2A（代理人对代理人） 通信以构建可扩展的分布式AI代理管道。

🎯 概述

该项目实现了一个分层多代理架构，其中：

  • 根编排器（ADK代理）：管理任务委派和代理间沟通
  • LangGraph代理：通过工具集成处理代理工作流
  • MCP工具：通过模型上下文协议提供数学和天气功能
  • A2A服务器：允许通过HTTP/RPC进行代理间通信

架构图

┌─────────────────────────────────────────────────────────────┐
│              Root Orchestrator (ADK Agent)                   │
│  - Model: gemini-2.0-flash                                   │
│  - Tools: [A2A Remote Tool]                                  │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ├── Sub-Agent 1 (LangGraph Agent)
                       │   ├── Chatbot Node (LLM with tools)
                       │   ├── Tool Node (MCP integration)
                       │   └── State Management
                       │
                       └── A2A Server (Remote Agent Caller)
                           └── HTTP/RPC Communication

📁 项目结构

Pipelines/
├── main.py                          # Application entry point
├── requirements.txt                 # Python dependencies
├── README.md                        # This file
├── agents.py                        # Additional agent utilities
├── tools_server.py                  # MCP tools server setup
│
└── app/
    ├── __init__.py
    ├── adk/
    │   ├── adk_agent.py            # ADK orchestrator builder
    │   ├── adk_client_tool.py       # A2A client tool implementation
    │   └── __pycache__/
    │
    ├── graph/
    │   ├── langgraph_app.py         # LangGraph workflow definition
    │   ├── state.py                 # Agent state schema
    │   ├── tool_server.py           # MCP server configuration
    │   └── __pycache__/
    │
    └── server/
        └── a2a_server.py            # A2A HTTP server implementation

🚀 关键组件

1. ADK代理（app/ADK/ADK_Agent.py）

协调所有子代理并管理工具执行的根编排器。

orchestrator = ADKAgent(
    name="root_orchestrator",
    model="gemini-2.0-flash",
    instruction="Delegate tasks to LangGraph agent. Use A2A tool when needed.",
    tools=[remote_tool],
    sub_agents=[graph_agent]
)

责任：

  • 任务编排和委派
  • 模型调用（Gemini 2.0 Flash）
  • 工具管理和执行
  • 子代理生命周期管理

2. LangGraph代理（app/graph/LangGraph_app.py）

通过状态管理和工具集成实现代理工作流。

主要特点：

  • 状态图：使用消息历史记录管理对话状态
  • 聊天机器人节点：带工具绑定的LLM推理（AWS Bedrock Nova）
  • 工具节点：执行MCP工具（数学、天气）
  • 条件边：需要时提供工具路线

工作流程：

START → chatbot (LLM inference) → tools_condition → tool_execution → chatbot → END

3. MCP工具集成

连接到MCP服务器以执行工具。

可用工具：

  • 数学运算：基本算术计算
  • 天气信息：实时天气数据

配置：

client = MultiServerMCPClient({
    "math_weather": {
        "transport": "sse",
        "url": "http://localhost:8000/sse"
    }
})

4. A2A服务器（app/Server/A2A_Server.py）

将代理作为HTTP服务公开，用于远程代理通信。

终点：

  • POST /a2a：代理方法调用的RPC端点
  • GET /.well-known/agent.json：代理卡/元数据

特征：

  • 内存代理运行程序
  • JSON-RPC 2.0支持
  • FastAPI服务器
  • 通过知名端点发现代理

5. A2A客户端工具（app/adk/adk_Client_Tool.py）

使编排器能够通过A2A协议调用远程代理。

remote_tool = A2AClientTool(
    name="call_remote_agent",
    target_agent_url="http://remote-agent.com/a2a"
)

🔧 安装

先决条件

  • Python 3.10+
  • 谷歌云证书（适用于ADK和Gemini API）
  • AWS证书（适用于Bedrock）

设置步骤

  1. 克隆仓库
git clone <repository-url>
cd Pipelines
  1. 创建虚拟环境
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
  1. 安装依赖项
pip install -r requirements.txt
  1. 配置环境变量
# Set up Google Cloud credentials
export GOOGLE_APPLICATION_CREDENTIALS="path/to/credentials.json"

# Set up AWS credentials
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_REGION="us-east-1"

📦 依赖项

包装 用途
a2a-sdk[http-server] 带有HTTP服务器的Google ADK代理框架
python-a2a A2A协议实现
langgraph 代理工作流编排
langchain-mcp-adapters 用于工具集成的MCP客户端
langchain-aws AWS Bedrock LLM集成
langchain-core 核心LangChain抽象
fastapi HTTP服务器框架
uvicorn ASGI服务器
httpx 异步HTTP客户端
rich 终端输出格式

▶️ 运行应用程序

选项1：启动编排器

python main.py

这将：

  • 初始化ADK代理编排器
  • 构建LangGraph工作流
  • 连接到MCP工具服务器
  • 日志初始化状态
  • 让代理为请求做好准备

选项2：作为A2A服务器运行

# Terminal 1: Start MCP tools server
python tools_server.py

# Terminal 2: Start A2A server
uvicorn app.server.a2a_server:app --port 9000 --reload

选项3：运行自检

python app/graph/langgraph_app.py

这将使用示例查询测试LangGraph工作流：

"What is (3 + 5) * 12 and weather in Tokyo?"

🔄 工作流示例

示例1：简单计算

# The LLM detects a math query and delegates to MCP math tool
User: "Calculate (3 + 5) * 12"
→ Chatbot detects tool call
→ Tool Node executes via MCP
→ Result: 96

示例2：多工具请求

# Multiple tools are called in sequence
User: "What is (10 + 5) * 2 and weather in Tokyo?"
→ Math tool: (10 + 5) * 2 = 30
→ Weather tool: Returns Tokyo weather
→ Chatbot synthesizes response

示例3：远程代理委派

# Orchestrator delegates to remote agent
User: "Process this complex request"
→ Orchestrator evaluates task
→ If specialized handler needed: call A2A remote agent
→ Remote agent processes and returns result
→ Orchestrator synthesizes final response

📊 状态管理

代理状态模式（app/graph/State.py）

class State(TypedDict):
    messages: Annotated[list, add_messages]

特征：

  • 消息历史记录：保持对话上下文
  • add_message：用于消息附加的缩减功能
  • 类型安全：带TypedDict的完整类型提示

🛠️ 定制

添加新的MCP工具

  1. 在MCP服务器中创建一个工具
  2. 更新 langgraph_app.py MCP客户端配置：
client = MultiServerMCPClient({
    "your_tools": {
        "transport": "sse",
        "url": "http://localhost:8000/sse"
    }
})

更改LLM模型

编辑 app/graph/langgraph_app.py:

# Option 1: Use different Bedrock model
llm = ChatBedrockConverse(model="amazon.nova-pro-v1:0")

# Option 2: Use Gemini directly in orchestrator
# (Already configured in adk_agent.py)

添加子代理

在 app/adk/adk_agent.py:

sub_agents = [
    graph_agent,
    your_new_agent_1,
    your_new_agent_2,
]

orchestrator = ADKAgent(
    ...
    sub_agents=sub_agents
)

🧪 测试

单元测试

pytest tests/

手动测试

# Test LangGraph workflow
python app/graph/langgraph_app.py

# Test A2A server
curl -X POST http://localhost:9000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": {"prompt": "What is 5 + 3?"}
  }'

🔐 安全注意事项

  1. API密钥：使用环境变量，从不提交凭据
  2. 跨域资源共享：如果从外部源访问，请配置FastAPI CORS
  3. 认证：将身份验证中间件添加到A2A服务器端点
  4. 速率限制：对公共端点实施速率限制
  5. 输入验证：验证所有传入的提示和参数

🐛 故障排除

问题：MCP工具未加载

Solution: Ensure MCP server is running on http://localhost:8000/sse

问题：未找到基岩证书

Solution: Verify AWS credentials are properly configured
export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...

问题：远程代理无法访问

Solution: Check A2A server is running and URL is correct
curl http://remote-agent.com/a2a -X OPTIONS

📚 学习资源

  • 谷歌ADK文档: https://cloud.google.com/docs/adk
  • LangGraph: https://langchain-ai.github.io/langgraph/
  • 模型上下文协议: https://modelcontextprotocol.io/
  • A2A协议: https://github.com/CloudWeaver-AI/a2a-spec
  • AWS基岩: https://aws.amazon.com/bedrock/

🤝 贡献

  1. 创建要素分支
  2. 进行更改
  3. 彻底测试
  4. 提交拉取请求

📝 许可证

该项目是Google ADK生态系统的一部分，并遵循其许可条款。

👤 作者

赫曼斯·卡尔蒂克
附加培训项目-多代理编排

🎓 学习目标

通过研究这个代码库，你会学到：

  • ✅ 多代理架构设计模式
  • ✅ LangGraph工作流编排
  • ✅ 模型上下文协议集成
  • ✅ 代理间通信（A2A）
  • ✅ 工具增强LLM代理
  • ✅ 异步Python模式
  • ✅ 分布式系统中的状态管理
  • ✅ FastAPI服务器开发

📞 支持

对于问题和疑问：

  • 检查故障排除部分
  • 查看应用程序中的示例代码/
  • 使用自检脚本进行测试
  • 启用调试日志记录以进行详细跟踪

相关 MCP