【技术收藏】A2A协议实战：构建可互操作的AI代理系统，大模型开发必备技能

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?"}')