python打造Google(A2A)+MCP，Langchain生态互补

Google的Agent2Agent （A2A）协议是一种开放标准，旨在实现 AI 代理之间的无缝通信和协作。在一个代理使用不同框架和不同供应商构建的世界中，A2A 提供了一种通用语言，打破了孤岛并促进了互作性。

datian1234

470人浏览 · 2025-05-14 14:05:17

datian1234 · 2025-05-14 14:05:17 发布

A2A

Google的Agent2Agent （A2A）协议是一种开放标准，旨在实现 AI 代理之间的无缝通信和协作。
在一个代理使用不同框架和不同供应商构建的世界中，A2A 提供了一种通用语言，打破了孤岛并促进了互作性。

A2A的特性

在不同平台（LangGraph、CrewAI、Semantic Kernel、自定义解决方案）上构建的代理，以创建强大的复合 AI 系统。
能处理复杂的流程：使代理能够委派子任务、交换信息和协调作，以解决单个代理无法解决的复杂问题。
安全透明：代理无需共享内部存储器、工具或专有逻辑即可进行交互，从而确保安全并保护知识产权。

A2A 与MCP相互互补

MCP（模型上下文协议）：通过结构化输入/输出将代理连接到工具、API 和资源。将其视为代理访问其功能的方式。
A2A（Agent2Agent 协议）：促进不同代理作为对等体之间的动态、多模式通信。这是代理协作、委派和管理共享任务的方式。

python-A2A

Python A2A 是一个全面且适用于生产环境的库，用于实现 Google 的 Agent-to-Agent (A2A) 协议，并完全支持 Model Context Protocol (MCP)。
它提供了构建可互操作的 AI 代理生态系统所需的所有功能，这些代理可以无缝协作以解决复杂问题。

python-A2A的特性

完全实现google官方 A2A 规范
内置代理注册表和发现功能，用于构建代理生态系统
对 MCP(Model Context Protocol) 的一流支持，实现强大的工具使用代理
为生产环境构建，具有强大的错误处理和验证功能
与任何 Python 框架兼容（Flask、FastAPI、Django 等）
原生集成 OpenAI、Anthropic、AWS Bedrock 等
与 LangChain 的工具和代理无缝集成,在任何代理中使用 LangChain 和 MCP 工具
在 A2A 代理和 LangChain 代理之间进行转换
能即时访问数千个预构建工具

python-A2A实战

基本用法，创建一个查询天气的Agent

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
    name="Weather Agent",
    description="Provides weather information",
    version="1.0.0"
)
class WeatherAgent(A2AServer):
    
    @skill(
        name="Get Weather",
        description="Get current weather for a location",
        tags=["weather", "forecast"]
    )
    def get_weather(self, location):
        """Get weather for a location."""
        # Mock implementation
        return f"It's sunny and 75°F in {location}"
    
    def handle_task(self, task):
        # Extract location from message
        message_data = task.message or {}
        content = message_data.get("content", {})
        text = content.get("text", "") if isinstance(content, dict) else""
        
        if"weather"in text.lower() and "in"in text.lower():
            location = text.split("in", 1)[1].strip().rstrip("?.")
            
            # Get weather and create response
            weather_text = self.get_weather(location)
            task.artifacts = [{
                "parts": [{"type": "text", "text": weather_text}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        else:
            task.status = TaskStatus(
                state=TaskState.INPUT_REQUIRED,
                message={"role": "agent", "content": {"type": "text", 
                         "text": "Please ask about weather in a specific location."}}
            )
        return task

# Run the server
if __name__ == "__main__":
    agent = WeatherAgent()
    run_server(agent, port=5000)

与langchain集成

from python_a2a.mcp import FastMCP, text_response
from python_a2a.langchain import to_langchain_tool

# 创建带有工具的 MCP 服务器
mcp_server = FastMCP(name="Basic Tools", description="Simple utility tools")

@mcp_server.tool(
    name="calculator",
    description="Calculate a mathematical expression"
)
def calculator(input):
    """Simple calculator that evaluates an expression."""
    try:
        result = eval(input)
        return text_response(f"Result: {result}")
    except Exception as e:
        return text_response(f"Error: {e}")

# 启动服务器
import threading, time
def run_server(server, port):
    server.run(host="0.0.0.0", port=port)
server_thread = threading.Thread(target=run_server, args=(mcp_server, 5000), daemon=True)
server_thread.start()
time.sleep(2)  # 允许服务器启动

# 将 MCP 工具转换为 LangChain
calculator_tool = to_langchain_tool("http://localhost:5000", "calculator")

# 在 LangChain 中使用工具
result = calculator_tool.run("5 * 9 + 3")
print(f"Result: {result}")

将 LangChain 工具转换为 MCP 服务器

from langchain.tools import Tool
from langchain_core.tools import BaseTool
from python_a2a.langchain import to_mcp_server

# 创建 LangChain 工具
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression"""
    try:
        result = eval(expression)
        return f"Result: {expression} = {result}"
    except Exception as e:
        return f"Error: {e}"

calculator_tool = Tool(
    name="calculator",
    description="Evaluate a mathematical expression",
    func=calculator
)

# 转换为 MCP 服务器
mcp_server = to_mcp_server(calculator_tool)

# 运行服务器
mcp_server.run(port=5000)

将 LangChain 组件转换为 A2A 服务器

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from python_a2a import A2AClient, run_server
from python_a2a.langchain import to_a2a_server

# 创建 LangChain LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 转换为 A2A 服务器
llm_server = to_a2a_server(llm)

# 创建简单链
template = "You are a helpful travel guide.\n\nQuestion: {query}\n\nAnswer:"
prompt = PromptTemplate.from_template(template)
travel_chain = prompt | llm | StrOutputParser()

# 转换为 A2A 服务器
travel_server = to_a2a_server(travel_chain)

# 在后台线程中运行服务器
import threading
llm_thread = threading.Thread(
    target=lambda: run_server(llm_server, port=5001),
    daemon=True
)
llm_thread.start()

travel_thread = threading.Thread(
    target=lambda: run_server(travel_server, port=5002),
    daemon=True
)
travel_thread.start()

# 测试服务器
llm_client = A2AClient("http://localhost:5001")
travel_client = A2AClient("http://localhost:5002")

llm_result = llm_client.ask("What is the capital of France?")
travel_result = travel_client.ask('{"query": "What are some must-see attractions in Paris?"}')