【Agent专题】构建一个 A2A 天气代理系统：从服务器到客户端的完整指南

作者

“任何足够先进的技术，都与魔法无异。” —— 阿瑟·克拉克

代理到代理（A2A）协议正在彻底改变自主代理相互发现和通信的方式。A2A 并不是复杂的集成设置，而是为代理提供了一种标准化的方式来找到彼此并无缝交换信息。

今天，我们将通过从零开始构建一个天气代理系统（包括 A2A 服务器和客户端）来深入理解 A2A 协议。重点介绍 A2A 协议的核心部分——发现流程。在这个过程中，客户端通过 A2A 服务器公开的代理卡（Agent Card）来查找并理解代理的功能。

接着，我们会演练客户端如何使用 /tasks/send 端点向代理发送任务，服务器如何处理该任务，以及如何返回响应并由客户端解析和展示。

在过程中，我们还会拆解任务的结构（Task）：它由消息（Message）组成，而每条消息又由若干部分（Part）构成。在这个示例中，我们保持简单，只使用 TextPart。

为了让流程清晰，我们采用一个最基础的 A2A 设置：一个服务器端代理和一个客户端。服务器代理利用大型语言模型结合 Tavily API 获取并格式化天气信息，客户端则通过标准 A2A 流程与之交互。

我们的目标不是堆砌复杂性，而是理解 A2A 协议的基本机制：发现、任务启动与响应处理。

项目设置

项目文件结构如下：

a2a-weather/├── server/│   └── weather_server.py├── client/│   └── weather_client.py├── requirements.txt└── .env

1. 创建项目环境

# 创建项目目录mkdir a2a-weathercd a2a-weather# 使用 uv 初始化环境uv init# 创建并激活虚拟环境uv venvsource .venv/bin/activate  # Windows: .venv\Scripts\activate

2. 创建 requirements.txt

flask==3.0.0requests==2.31.0tavily-python>=0.3.0python-dotenv==1.0.0tiktoken>=0.7.0langchain-google-genai==1.0.10

安装依赖：

uv pip install -r requirements.txt

3. 获取 API Key

  • Gemini API Key：前往 Google AI Studio，登录账户，点击「获取 API Key」，创建并复制。

  • Tavily API Key：前往 tavily.com 注册并获取（支持免费套餐）。

将两者配置到 .env 文件中：

TAVILY_API_KEY=your_tavily_api_key_hereGOOGLE_API_KEY=your_google_gemini_api_key_here

构建 A2A 天气服务器

在 server/weather_server.py 中实现服务器逻辑：

from flask import Flask, request, jsonifyfrom tavily import TavilyClientfrom langchain_google_genai import ChatGoogleGenerativeAIimport osfrom dotenv import load_dotenvload_dotenv()# Initialize Flask app, Tavily client, and Gemini LLMapp = Flask(__name__)tavily = TavilyClient(api_key=os.getenv('TAVILY_API_KEY'))llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash-exp", google_api_key=os.getenv('GOOGLE_API_KEY'))@app.route("/.well-known/agent.json", methods=["GET"])def agent_card():    """    A2A Discovery endpoint - tells other agents what we can do.    This is the first endpoint any A2A client will call.    """    return jsonify({        "name": "WeatherBot",        "description": "Get current weather information for any city using real-time data with AI-enhanced summaries",        "url": "http://127.0.0.1:5000",        "version": "1.0",        "capabilities": {            "streaming": False,            "pushNotifications": False        }    })@app.route("/tasks/send", methods=["POST"])def handle_task():    """    Main A2A endpoint where clients send tasks.    This is where the actual work happens.    """    try:        # Get the task data from the A2A client        task_data = request.get_json()        task_id = task_data.get("id")        # Extract the user's message text        user_message = task_data["message"]["parts"][0]["text"]    except (KeyError, IndexError, TypeError):        return jsonify({"error": "Invalid A2A task format"}), 400    try:        # Use Tavily to search for weather information        search_query = f"current weather {user_message}"        search_results = tavily.search(            query=search_query,            search_depth="basic",            max_results=3        )        # Combine search results for processing        raw_weather_data = []        for result in search_results.get('results', []):            raw_weather_data.append(f"{result['title']}: {result['content']}")        if raw_weather_data:            # Use Gemini to create a concise, well-formatted weather summary            combined_data = "\n\n".join(raw_weather_data)            prompt = f"""            Based on the following weather search results, provide a concise and well-formatted weather summary for {user_message}.            Raw weather data:            {combined_data}            Please provide:            1. Current temperature and conditions            2. Key weather details (humidity, wind, etc.)            3. Any relevant weather alerts or notable information            Keep the response under 150 words and make it clear and easy to read.            """            response = llm.invoke(prompt)            response_text = response.content        else:            response_text = "Sorry, I couldn't find current weather information for that location."    except Exception as e:        response_text = f"I encountered an error getting weather data: {str(e)}"    # Return properly formatted A2A response    return jsonify({        "id": task_id,        "status": {"state": "completed"},        "messages": [            task_data["message"],  # Echo back original message            {                "role": "agent",                "parts": [{"text": response_text}]            }        ]    })if __name__ == "__main__":    print("Starting WeatherBot A2A Agent on http://127.0.0.1:5000")    app.run(host="127.0.0.1", port=5000, debug=True)

这个服务器提供了：

  • Agent Card：在 /.well-known/agent.json 暴露能力信息，供客户端发现。

  • 任务处理：在 /tasks/send 端点接收并处理任务。

  • 数据获取：通过 Tavily 拉取实时天气数据。

  • 智能总结：用 Gemini 将数据总结为简明天气报告。

  • 标准化响应：返回符合 A2A 格式的 JSON，包含原始请求和 AI 回复。

构建客户端代理

在 client/weather_client.py 中实现客户端：

import requestsimport uuidclass WeatherClient:    def __init__(self, server_url="http://127.0.0.1:5000"):        self.server_url = server_url        self.agent_info = None    def discover_agent(self):        """        Step 1: Discover the weather agent using A2A discovery flow.        """        discovery_url = f"{self.server_url}/.well-known/agent.json"        try:            response = requests.get(discovery_url)            response.raise_for_status()            self.agent_info = response.json()            return self.agent_info        except requests.exceptions.RequestException as e:            raise Exception(f"Could not discover agent: {str(e)}")    def ask_weather(self, location):        """        Step 2: Send a weather query to the agent using A2A task format.        """        if not self.agent_info:            raise Exception("Agent not discovered. Call discover_agent() first.")        # Create unique task ID        task_id = str(uuid.uuid4())        # Build A2A task payload        task_payload = {            "id": task_id,            "message": {                "role": "user",                "parts": [{"text": f"weather in {location}"}]            }        }        # Send task to server        task_url = f"{self.server_url}/tasks/send"        try:            response = requests.post(task_url, json=task_payload)            response.raise_for_status()            result = response.json()            # Extract agent's response from A2A message structure            messages = result.get("messages", [])            if messages and len(messages) > 1:                return messages[-1]["parts"][0]["text"]            else:                return "No response received"        except requests.exceptions.RequestException as e:            raise Exception(f"Weather request failed: {str(e)}")def main():    """    Demo function showing how to use the A2A weather client.    """    client = WeatherClient()    print("A2A Weather Client Demo")    print("=" * 40)    try:        # Step 1: Discover the weather agent        print("Discovering weather agent...")        agent_info = client.discover_agent()        print(f"Found: {agent_info['name']} - {agent_info['description']}")        # Step 2: Ask for weather in different cities        cities = ["London", "Tokyo", "New York"]        for city in cities:            print(f"\nGetting weather for {city}...")            weather_response = client.ask_weather(city)            print(f"Response:\n{weather_response}\n")            print("-" * 40)    except Exception as e:        print(f"Error: {str(e)}")if __name__ == "__main__":    main()

客户端遵循标准 A2A 流程：

  1. 发现代理：调用 /.well-known/agent.json 获取代理信息。

  2. 构建任务：生成唯一 ID，并打包为标准 A2A 消息结构。

  3. 发送任务：POST 请求到 /tasks/send。

  4. 解析响应：从标准消息格式中提取代理回复。

测试

启动服务器：

cd a2a-weathersource .venv/bin/activateuv run python server/weather_server.py

另开终端运行客户端：

cd a2a-weathersource .venv/bin/activateuv run python client/weather_client.py

你将看到 Gemini 格式化后的天气摘要，结果更加清晰、易读。

A2A 协议把原本复杂的“代理间通信”问题，变成了一个简洁的标准化流程。仅仅依靠两个端点和统一的消息格式，任何 A2A 客户端都可以与任何 A2A 服务器对话，而无需自定义适配。

在本示例中，我们通过 Tavily 获取实时天气数据，再用 Gemini LLM 对数据进行智能化总结，实现了一个 具备发现、任务执行、响应处理的完整代理系统。

  • A2A 的核心价值在于“发现”和“标准化通信”，这使得代理能够像互联网中的服务一样自由组合。

  • AI 的价值在于提升交互体验，A2A 保证了“能连通”，而大模型保证了“能用好”。

  • 对开发者来说，A2A 协议降低了代理生态系统的“摩擦成本”，未来任何新代理都能像“即插即用组件”一样快速接入。


一句话总结：A2A 是智能体世界的 TCP/IP，而 LLM 是它的应用层大脑。

（文：AI技术研习社）

发表评论