菜鸟教程 – 学的不仅是技术，更是梦想！

技术网站，更是交流网站

MCP

AI Agent 开发实战：MCP+A2A+LangGraph 驱动的智能体全流程开发

likuolei

AI Agent 开发实战：MCP + A2A + LangGraph 驱动的智能体全流程开发

引言

在 AI Agent 开发领域，MCP（Model Context Protocol，模型上下文协议）用于标准化工具和资源的访问，让 LLM 能无缝调用外部数据源；A2A（Agent2Agent，代理间协议）则实现代理间的标准化通信，支持多代理协作；LangGraph 是 LangChain 的扩展库，用于构建状态ful、多演员的代理工作流，支持复杂决策图谱。这些技术结合，能打造高效、可扩展的智能体系统。本指南基于实际案例（如天气代理、货币转换代理和采购管家），提供全流程开发实战，从环境搭建到部署集成。目标：构建一个多代理系统，其中一个 LangGraph 代理通过 MCP 调用工具，并经 A2A 与远程代理协作。

适用人群：有 Python 和 LLM 基础的开发者。预计耗时：2-4 小时。

前置条件

  • 环境：Python 3.10+，Node.js（若集成 Chrome DevTools MCP）。
  • 库安装：
  pip install langgraph langchain-openai fastmcp a2a-sdk httpx pydantic requests asyncio
  • API 密钥：OpenAI 或 Google Gemini API key；可选：OpenWeatherMap API key（天气示例）、Frankfurter API（货币示例）。
  • 部署：Google Cloud 项目（免费额度内），gcloud CLI。
  • GitHub 资源：克隆示例仓库，如 MCP-A2A-step-by-step。

步骤 1: 构建 MCP Server（工具暴露层）

MCP Server 作为工具后端，提供标准化 API（如天气查询）。我们用 FastMCP 实现一个天气工具服务器。

代码实现（mcp_server.py）

import asyncio
import os
import requests
from fastmcp import FastMCP
from pydantic import BaseModel

mcp = FastMCP("Weather MCP Server")

class CurrentWeather(BaseModel):
    location: str
    temperature: float
    description: str

@mcp.tool()
def get_current_weather(location: str) -> str:
    """获取指定地点的当前天气"""
    url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={os.getenv('OPENWEATHERMAP_API_KEY')}&units=metric"
    response = requests.get(url).json()
    temp = response['main']['temp']
    desc = response['weather'][0]['description']
    return f"{location} 当前天气：{temp}°C，{desc}"

@mcp.prompt()
def weather_analysis_prompt(location: str, data: str) -> str:
    """天气分析提示模板"""
    return f"基于以下数据分析 {location} 的穿衣建议：{data}"

if __name__ == "__main__":
    asyncio.run(mcp.run_async(transport="streamable-http", host="0.0.0.0", port=8080))

运行与测试

  • 运行：python mcp_server.py。
  • 测试：用 curl 发送 JSON-RPC 请求：
  curl -X POST http://localhost:8080/jsonrpc -H "Content-Type: application/json" -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}'
  • 部署到 Cloud Run：构建 Docker 镜像并 gcloud run deploy mcp-weather --source .。

此服务器暴露工具，供代理发现和调用。

步骤 2: 使用 LangGraph 构建核心代理（MCP 集成）

LangGraph 用于编排代理逻辑：LLM + MCP 工具 + 状态管理。我们构建一个“天气顾问”代理，查询 MCP 工具并生成建议。

代码实现（langgraph_agent.py）

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from fastmcp import Client
import asyncio
import json
import os

llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
mcp_url = "http://localhost:8080"  # 或 Cloud Run URL

async def discover_mcp_tools():
    async with Client(mcp_url) as client:
        tools = await client.list_tools()
        return tools

def create_langchain_tool(mcp_tool):
    def tool_func(input_str: str) -> str:
        args = json.loads(input_str) if input_str.startswith('{') else {"location": input_str}
        # 同步调用 MCP
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(call_mcp_tool(mcp_tool.name, args))
        loop.close()
        return result
    return tool(name=mcp_tool.name, description=mcp_tool.description, func=tool_func)

async def call_mcp_tool(tool_name: str, args: dict):
    async with Client(mcp_url) as client:
        result = await client.call_tool(tool_name, args)
        return result

# 发现工具并创建代理
tools_data = asyncio.run(discover_mcp_tools())
tools = [create_langchain_tool(t) for t in tools_data]

agent = create_react_agent(llm, tools, checkpointer=MemorySaver())  # 需 pip install langgraph-checkpoint

def run_agent(query: str):
    return agent.invoke({"messages": [("user", query)]})

测试

  • 查询：run_agent("纽约明天穿什么？") → 代理调用 MCP 获取天气，生成建议。
  • 扩展：集成 Chrome DevTools MCP（npx chrome-devtools-mcp@latest），添加浏览器自动化工具，如 @mcp.tool() def navigate_page(url: str)。

此步实现单代理的工具调用与推理。

步骤 3: 暴露代理为 A2A Server（代理间通信）

A2A 让 LangGraph 代理成为远程服务，支持发现（Agent Card）和任务执行。我们构建货币转换代理作为示例。

Agent Card（agent.json）

{
  "name": "currency_agent",
  "description": "货币转换代理",
  "protocolVersion": "0.2.6",
  "capabilities": {"streaming": true},
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"],
  "skills": [{"id": "convert_currency", "name": "Convert Currency", "description": "转换货币汇率"}],
  "url": "https://your-cloud-run-url.com"
}

放置在 /static/.well-known/agent.json。

代码实现（a2a_server.py）

from langgraph.prebuilt import create_react_agent
from langchain_google_genai import ChatGoogleGenerativeAI  # 或 OpenAI
from a2a import A2AStarletteApplication, InMemoryTaskStore
from a2a.events import TaskStatusUpdateEvent, TaskArtifactUpdateEvent
import requests
from pydantic import BaseModel

llm = ChatGoogleGenerativeAI(model="gemini-1.0-pro")
store = InMemoryTaskStore()

@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    url = f"https://api.frankfurter.app/latest?from={from_currency}&to={to_currency}"
    data = requests.get(url).json()
    return f"1 {from_currency} = {data['rates'][to_currency]} {to_currency}"

tools = [get_exchange_rate]
graph = create_react_agent(llm, tools)

class CurrencyAgent:
    def __init__(self):
        self.graph = graph

    def invoke(self, query: str, context_id: str):
        result = self.graph.invoke({"messages": [("user", query)]}, {"configurable": {"thread_id": context_id}})
        return result["messages"][-1].content

    async def stream(self, query: str, context_id: str):
        for chunk in self.graph.stream({"messages": [("user", query)]}, {"configurable": {"thread_id": context_id}}):
            yield chunk

class CurrencyExecutor:
    def __init__(self):
        self.agent = CurrencyAgent()
        self.store = store

    async def on_message_send(self, request):  # A2A 入口
        task = self.store.create_task(request.context_id, request.message_id)
        task.status = "working"
        self.store.enqueue_event(task.id, TaskStatusUpdateEvent(status="working"))
        result = self.agent.invoke(request.params["message"]["parts"][0]["text"], request.context_id)
        artifact = TaskArtifactUpdateEvent(artifact={"text": result})
        self.store.enqueue_event(task.id, artifact)
        task.status = "completed"
        return task

app = A2AStarletteApplication(executor=CurrencyExecutor(), task_store=store)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

部署

  • Docker 化并 gcloud run deploy currency-a2a --source . --allow-unauthenticated。
  • 测试：客户端 GET / .well-known/agent.json 发现卡片。

步骤 4: 构建 A2A Client 代理（多代理协作）

客户端代理调用远程 A2A Server，实现协作。如采购管家调用汉堡/披萨卖家代理。

代码实现（a2a_client.py）

import httpx
from a2a import A2AClient, SendMessageRequest
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import uuid

llm = ChatOpenAI(model="gpt-4o-mini")
remote_url = "https://currency-a2a-xxxx.run.app"  # A2A Server URL

@tool
def call_remote_agent(agent_name: str, task: str) -> str:
    async def _call():
        async with httpx.AsyncClient() as client:
            resolver = A2ACardResolver(httpx_client=client, base_url=remote_url)
            card = await resolver.get_agent_card()
            a2a_client = A2AClient(httpx_client=client, agent_card=card)
            req = SendMessageRequest(
                id=str(uuid.uuid4()),
                params={"message": {"role": "user", "parts": [{"kind": "text", "text": task}]}}
            )
            resp = await a2a_client.send_message(req)
            return resp.root.result.artifacts[0].parts[0].root.text
    return asyncio.run(_call())

tools = [call_remote_agent]
client_agent = create_react_agent(llm, tools)

def run_client(query: str):
    return client_agent.invoke({"messages": [("user", query)]})

# 示例：run_client("将100美元转换为欧元")

测试多代理

  • 查询：”帮我订购一个汉堡和披萨” → 客户端发现远程代理，发送任务，聚合响应。

步骤 5: 全流程集成与优化

  • Chrome DevTools MCP 集成：在 LangGraph 中添加浏览器工具：
  # 运行 npx chrome-devtools-mcp@latest
  # 在工具中调用：@tool def debug_page(url: str): ... 使用 MCP Client

用于前端调试，如代理生成代码后自动测试。

  • 状态管理：用 LangGraph 的 Checkpointer 持久化会话。
  • 部署全栈：MCP → LangGraph A2A Server → Client on Agent Engine。
  • 性能优化：添加流式响应（A2A streaming），错误处理（重试 MCP 调用）。

项目案例

  1. 天气顾问：MCP 天气工具 + LangGraph 分析 → A2A 分享给旅行代理。
  2. 货币转换：如上，处理多轮澄清（如缺失目标货币）。
  3. 采购管家：LangGraph 协调远程汉堡/披萨 A2A 代理，处理订单。

注意事项

  • 安全：A2A 用认证（如 OAuth），MCP 限本地访问。
  • 调试：用 LangSmith 追踪图执行。
  • 扩展：结合 RAG 或多模态（图像工具）。
  • 资源：参考 Google Codelab A2A 采购示例，完整代码见 GitHub。

通过此流程，你能从零构建生产级多代理系统。实践时从天气示例起步，逐步添加 A2A。遇到问题？提供具体错误，我可进一步指导！

分类： MCP
网站 http://yingjuxia.com
文章已创建 3183

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

相关文章

开始在上面输入您的搜索词，然后按回车进行搜索。按ESC取消。

返回顶部