30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度如果你正在构建多智能体系统或者想让不同的 AI 代理能够像人类团队一样协作那么 Google 的 Agent-to-Agent (A2A) 协议是你必须了解的技术。它定义了智能体之间标准化的通信格式是构建可互操作、可扩展的 AI 生态系统的基石。而 Python A2A 库正是将这一协议从理论变为现实的关键工具。Python A2A 是一个由 Manoj Desai 维护的开源项目它完整实现了 Google 的 A2A 协议并深度集成了 Model Context Protocol (MCP)。这意味着你可以用它来创建能够相互发现、通信、并使用外部工具如 GitHub API、浏览器自动化、文件系统的智能体网络。它的核心价值在于“标准化”和“互操作性”——无论你的智能体底层是基于 OpenAI、Anthropic Claude还是本地部署的 Ollama它们都能通过 A2A 协议无缝对话。这篇文章将带你快速上手 Python A2A。我们不空谈概念而是聚焦于实战如何安装、如何创建一个具备收发邮件能力的智能体、如何为智能体赋予专属的“邮箱”技能、以及如何构建一个能自动进行邮件采编和数据清洗的工作流。整个过程你只需要基础的 Python 知识无需复杂的硬件配置重点在于理解其架构和 API 的使用。1. 核心能力速览在深入代码之前我们先通过一个表格快速了解 Python A2A 的核心特性这能帮你判断它是否适合你的项目。能力项说明项目类型Python 库用于实现 Google A2A 协议和 MCP 协议。开源团队个人开发者 Manoj Desai 维护。主要功能1. 创建符合 A2A 协议的智能体 (Agent)。2. 构建智能体网络 (Agent Network) 并实现智能路由。3. 集成 MCP 协议让智能体使用外部工具如 GitHub、浏览器、文件系统。4. 提供可视化工作流编辑器 (Agent Flow UI)。5. 与 LangChain 生态无缝集成。硬件/环境门槛极低。纯 Python 库无特殊硬件要求。核心依赖仅requests库可在任何能运行 Python 的环境包括 CPU中使用。性能瓶颈主要取决于你集成的 LLM 后端如 OpenAI API 调用速度。显存/内存占用库本身不直接消耗大量显存。内存占用取决于你运行的智能体数量和工作流复杂度通常很小。主要资源消耗来自你集成的 LLM 模型如本地部署的 Ollama。支持平台跨平台Windows/macOS/Linux。启动方式1.命令行启动通过a2aCLI 工具快速启动服务器或 UI。2.代码启动通过run_server函数在 Python 脚本中启动。3.可视化 UI通过a2a ui命令启动拖拽式工作流编辑器。是否支持 API是。每个智能体本质上都是一个 HTTP 服务器提供标准的 A2A API 端点可直接通过A2AClient或requests库调用。是否支持批量任务是。通过Flow工作流引擎和parallel()方法可以轻松定义并行执行的批量任务。适合场景1. 研究多智能体协作机制。2. 构建企业级 AI 业务流程自动化系统。3. 为现有 LLM 应用增加工具调用和智能体协作能力。4. 快速原型验证复杂的、多步骤的 AI 任务。2. 适用场景与使用边界Python A2A 是一个强大的框架但理解其边界能让你更有效地使用它。它非常适合构建专用智能体例如创建一个专门处理邮件收发、数据清洗、日程安排或代码审查的智能体。编排复杂工作流将多个单一功能的智能体串联起来完成如“监控新闻 - 摘要 - 生成报告 - 邮件发送”这样的复杂链条。集成现有工具通过 MCP 协议让智能体安全地操作 GitHub、数据库、浏览器或文件系统极大扩展 AI 的能力边界。教学与实验其清晰的协议实现和丰富的示例是学习 A2A/MCP 协议和智能体架构的绝佳材料。它可能不适合超低延迟场景智能体间通过 HTTP 通信存在网络开销。对于需要微秒级响应的场景可能需要更底层的通信机制。完全离线的封闭环境虽然智能体可以本地部署但部分功能如 Agent Flow UI 的某些特性、与云端 MCP 服务器的连接可能需要网络。替代简单的单次 LLM 调用如果你的需求只是向 ChatGPT 问一个问题那么直接调用 API 更简单。重要合规与安全边界工具使用授权当通过 MCP 连接 GitHub、邮箱等外部服务时务必确保你拥有合法的 API 令牌或账户权限并遵守相关服务的使用条款。数据隐私智能体处理的数据尤其是通过 MCP 工具访问的可能包含敏感信息。需确保数据传输和存储的加密并在生产环境中实施严格的访问控制。操作风险赋予智能体文件系统、浏览器自动化等能力后其操作具有实际影响如删除文件、发送邮件。在赋予高权限前应在沙箱环境中充分测试。内容合规智能体生成的内容需符合法律法规和平台政策避免产生有害或侵权信息。3. 环境准备与前置条件开始实战前确保你的开发环境就绪。基础环境要求操作系统Windows 10/11 macOS 10.15 或主流 Linux 发行版如 Ubuntu 20.04。Python 版本Python 3.8 或更高版本。这是运行大多数现代 AI 库的基本要求。包管理工具pip或更推荐的uv一个更快的 Python 包安装器。网络能正常访问 PyPI 和 GitHub用于安装依赖和克隆示例。可选但推荐的组件虚拟环境强烈建议使用venv、conda或uv创建独立的 Python 环境避免依赖冲突。代码编辑器VS Code、PyCharm 等具备良好的 Python 支持。LLM 后端你需要一个或多个 LLM 服务来为智能体提供“大脑”。可以选择云端 APIOpenAI, Anthropic Claude, Google Gemini 等。你需要相应的 API Key。本地模型通过 Ollama、LM Studio 或 vLLM 等本地部署模型。这需要一定的 GPU 显存或 CPU 内存。Python A2A 内置库也支持直接启动一个简单的、基于配置的 LLM 服务器。环境检查清单在终端中执行以下命令确认基础环境正常。# 检查 Python 版本 python --version # 或 python3 --version # 检查 pip 是否可用 pip --version # 可选安装 uv curl -LsSf https://astral.sh/uv/install.sh | sh # 安装后重启终端或运行 source $HOME/.local/bin/env 来激活 uv uv --version如果上述命令都能正确返回版本号说明基础环境已就绪。4. 安装部署与启动方式Python A2A 的安装非常灵活你可以根据需求选择不同的安装方式。方式一使用 pip 安装传统方式这是最直接的方法适合大多数用户。# 安装基础包包含核心功能 pip install python-a2a # 或者根据你的需求安装特定组件 pip install python-a2a[server] # 包含 Flask 服务器支持 pip install python-a2a[openai] # 包含 OpenAI 集成 pip install python-a2a[anthropic] # 包含 Anthropic Claude 集成 pip install python-a2a[ollama] # 包含 Ollama 集成 pip install python-a2a[mcp] # 包含 MCP 支持 pip install python-a2a[all] # 安装所有可选依赖方式二使用 uv 安装推荐速度更快uv是新一代的 Python 包管理工具安装和解决依赖的速度远超pip。# 如果尚未安装 uv先安装 curl -LsSf https://astral.sh/uv/install.sh | sh # 使用 uv 安装 python-a2a uv install python-a2a # uv 会自动处理依赖和虚拟环境方式三从源码安装用于开发或体验最新特性如果你想贡献代码或体验尚未发布的特性可以克隆仓库安装。# 克隆仓库 git clone https://github.com/themanojdesai/python-a2a.git cd python-a2a # 使用 uv 创建虚拟环境并安装开发依赖推荐 uv venv source .venv/bin/activate # Linux/macOS # 在 Windows 上使用.venv\Scripts\activate uv pip install -e .[dev] # -e 表示可编辑模式安装 # 或者使用 pip pip install -e .[dev]验证安装安装完成后可以通过命令行工具快速验证。# 检查 a2a 命令行工具是否可用 a2a --help # 应该能看到类似如下的输出列出可用的命令 # Usage: a2a [OPTIONS] COMMAND [ARGS]... # # Options: # --help Show this message and exit. # # Commands: # agent Manage A2A agents # anthropic Start an Anthropic Claude-powered A2A server # mcp-agent Start an MCP-enabled A2A agent # mcp-call Call an MCP tool directly # mcp-serve Start an MCP server # network Manage agent networks # openai Start an OpenAI-powered A2A server # send Send a message to an A2A agent # stream Stream a response from an A2A agent # ui Start the Agent Flow UI # workflow Run a workflow from a script看到命令列表说明安装成功。5. 功能测试与效果验证创建你的第一个邮件智能体理论说再多不如动手跑一遍。我们现在就创建一个具备“收发邮件”技能的智能体并测试其基本功能。为了简化我们先模拟邮件发送功能后续再接入真实的邮件服务。5.1 创建基础的邮件智能体创建一个名为mail_agent.py的文件内容如下from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState from typing import Dict, Any, List import json agent( nameMail Agent, descriptionA smart agent capable of sending and receiving emails, with content analysis skills., version1.0.0 ) class MailAgent(A2AServer): 一个模拟的邮件智能体。 在实际应用中这里的 send_mail 和 check_inbox 方法应该调用真实的邮件 API如 SMTP、IMAP 或第三方服务如 SendGrid。 # 模拟的“发件箱”和“收件箱”用于演示 _outbox: List[Dict] [] _inbox: List[Dict] [ {from: newstechsite.com, subject: Weekly AI Digest, body: New breakthroughs in multimodal models...}, {from: alicecompany.com, subject: Meeting Notes, body: Here are the notes from our project sync.}, {from: spamexample.com, subject: You won a prize!, body: Click this link to claim your reward.} ] skill( namesend_email, descriptionSend an email to a recipient with a subject and body., tags[communication, email] ) def send_email(self, to: str, subject: str, body: str) - str: 模拟发送邮件。实际应集成 SMTP 或邮件服务 API。 email_record { to: to, subject: subject, body: body, status: sent } self._outbox.append(email_record) print(f[Mail Agent] Simulated sending email to {to}: {subject}) return json.dumps({status: success, message_id: fmock_{len(self._outbox)}, detail: email_record}) skill( namecheck_inbox, descriptionCheck the inbox for new emails, optionally with filters., tags[communication, email, inbox] ) def check_inbox(self, filter_unread: bool False, sender: str None) - str: 模拟检查收件箱。实际应集成 IMAP/POP3 或邮件服务 API。 results self._inbox if sender: results [mail for mail in results if mail[from] sender] # 简单模拟“未读”过滤 if filter_unread: results results[:1] # 假设只有第一封是“未读” print(f[Mail Agent] Inbox checked. Found {len(results)} email(s).) return json.dumps({count: len(results), emails: results}) skill( namesummarize_inbox, descriptionAnalyze the inbox and provide a summary (e.g., count by sender, key topics)., tags[analysis, email, summarization] ) def summarize_inbox(self) - str: 分析收件箱并提供摘要。 from collections import Counter senders Counter([mail[from] for mail in self._inbox]) # 这里可以集成一个真正的 LLM 调用来分析邮件主题和正文 # 为了演示我们做一个简单的统计 summary { total_emails: len(self._inbox), senders: dict(senders), sample_subjects: [mail[subject] for mail in self._inbox[:2]] } print(f[Mail Agent] Inbox summarized: {summary}) return json.dumps(summary) def handle_task(self, task): 核心任务处理函数。根据用户消息调用相应的技能。 message_data task.message or {} content message_data.get(content, {}) # 处理文本输入 text content.get(text, ) if isinstance(content, dict) else str(content) # 简单的意图识别在实际应用中这里应该用一个更强大的 NLU 或 LLM 来解析意图 text_lower text.lower() if send email in text_lower or send mail in text_lower: # 这是一个非常简单的解析。理想情况下应从 task 中提取结构化参数。 # 例如假设消息格式是 “send email to aliceexample.com with subject ‘Hello’ and body ‘World’” # 这里我们使用一个预设的响应来演示。 task.artifacts [{ parts: [{ type: text, text: I can send emails. Please provide details in a structured format or use the send_email skill directly. }] }] task.status TaskStatus(stateTaskState.COMPLETED) elif check inbox in text_lower or new emails in text_lower: # 调用 check_inbox 技能 result self.check_inbox() task.artifacts [{ parts: [{type: text, text: fInbox check result: {result}}] }] task.status TaskStatus(stateTaskState.COMPLETED) elif summarize in text_lower and inbox in text_lower: # 调用 summarize_inbox 技能 result self.summarize_inbox() task.artifacts [{ parts: [{type: text, text: fInbox summary: {result}}] }] task.status TaskState.COMPLETED) else: # 如果无法识别意图请求更多输入 task.status TaskStatus( stateTaskState.INPUT_REQUIRED, message{ role: agent, content: { type: text, text: I am your Mail Agent. I can help you send emails, check your inbox, or summarize your emails. What would you like to do? } } ) return task if __name__ __main__: # 实例化并运行智能体服务器监听 5000 端口 agent MailAgent() print(Starting Mail Agent on http://localhost:5000) run_server(agent, port5000, host0.0.0.0) # host0.0.0.0 允许外部访问5.2 启动并测试智能体启动智能体服务器 在终端中进入mail_agent.py所在目录运行python mail_agent.py如果一切正常你会看到类似Starting Mail Agent on http://localhost:5000的输出并且服务器开始运行。使用 cURL 测试基础 API 打开另一个终端使用curl命令与智能体交互。# 1. 发送一个简单的问候触发 INPUT_REQUIRED curl -X POST http://localhost:5000/tasks \ -H Content-Type: application/json \ -d { message: { role: user, content: {type: text, text: Hello} } } # 预期响应会提示你它可以做什么INPUT_REQUIRED状态。 # 2. 请求检查收件箱 curl -X POST http://localhost:5000/tasks \ -H Content-Type: application/json \ -d { message: { role: user, content: {type: text, text: Can you check my inbox?} } } # 预期响应会包含模拟的收件箱内容状态为 COMPLETED。观察第一个终端的日志你会看到[Mail Agent] Inbox checked...的输出。直接调用技能Skill A2A 协议支持直接调用智能体暴露的技能。这是更结构化的交互方式。# 直接调用 check_inbox 技能并传递参数 curl -X POST http://localhost:5000/skills/check_inbox/execute \ -H Content-Type: application/json \ -d { parameters: { filter_unread: true } }这将直接执行check_inbox方法并返回 JSON 格式的结果绕过了handle_task中的自然语言解析。测试成功的关键指标服务器正常启动无报错。curl命令能收到格式正确的 JSON 响应。服务器终端打印出相应的技能执行日志如[Mail Agent] Inbox checked...。响应中的state字段根据任务情况变为COMPLETED或INPUT_REQUIRED。如果遇到Address already in use错误说明 5000 端口被占用。修改run_server(agent, port5000)中的端口号即可。6. 接口 API 与批量任务智能体作为 HTTP 服务器其核心价值在于提供了标准化的 API 接口便于集成和自动化。Python A2A 的客户端库让调用变得非常简单。6.1 使用 A2AClient 进行编程调用创建一个test_client.py文件使用官方客户端进行交互import asyncio from python_a2a import A2AClient async def test_mail_agent(): # 创建客户端指向我们刚刚启动的邮件智能体 client A2AClient(http://localhost:5000) print( 测试1: 发送自然语言请求 ) # 方法1: 使用 ask 方法发送自然语言消息 response await client.ask(Can you summarize my inbox?) print(fResponse: {response}) print(\n 测试2: 直接调用技能 ) # 方法2: 直接调用特定的技能更精确 skill_response await client.execute_skill( skill_namesummarize_inbox, parameters{} # 这个技能不需要参数 ) print(fSkill Response: {skill_response}) print(\n 测试3: 发送邮件模拟) send_response await client.execute_skill( skill_namesend_email, parameters{ to: colleaguecompany.com, subject: Project Update, body: The A2A agent integration is working perfectly. } ) print(fSend Email Response: {send_response}) if __name__ __main__: # 运行异步测试 asyncio.run(test_mail_agent())运行这个脚本前确保mail_agent.py仍在运行。执行python test_client.py你将看到客户端成功调用智能体并获取结果。6.2 构建邮件采编与清洗工作流批量任务单一智能体能力有限A2A 的强大之处在于构建多智能体工作流。我们来模拟一个“邮件采编与数据清洗”的自动化场景。假设我们有三个智能体Fetcher Agent负责从某个源如 RSS获取原始数据。Cleaner Agent负责清洗和格式化数据。Mail Agent我们刚创建的负责发送整理好的摘要。创建workflow_orchestration.pyimport asyncio from python_a2a import AgentNetwork, Flow import json import random import time # 模拟的智能体类在实际中它们应该是独立运行的服务 class FetcherAgent: 模拟一个数据抓取智能体 async def fetch_news(self, topic: str) - list: await asyncio.sleep(0.5) # 模拟网络延迟 mock_news [ {title: fBreakthrough in {topic} announced, source: TechNews, url: http://example.com/1}, {title: fNew research paper on {topic} published, source: arXiv, url: http://example.com/2}, {title: fIndustry leaders discuss the future of {topic}, source: Blog, url: http://example.com/3}, ] return mock_news class CleanerAgent: 模拟一个数据清洗智能体 async def clean_and_summarize(self, items: list) - str: await asyncio.sleep(0.3) # 模拟清洗提取标题生成摘要 titles [item[title] for item in items] summary fFound {len(titles)} items on topic. Key headlines: ; .join(titles[:2]) return summary async def main(): print(Starting automated email curation workflow...) # 1. 初始化模拟的智能体网络实际应用中这些是远程服务 # 假设我们已经有三个运行在不同端口的智能体 network AgentNetwork(nameCuration Pipeline) # 这里我们模拟网络实际应使用 network.add(fetcher, http://localhost:5001) 等 fetcher FetcherAgent() cleaner CleanerAgent() # 2. 创建并运行工作流 flow Flow(agent_networknetwork, nameDaily Digest Workflow) # 注意为了演示我们直接调用对象方法。真实 Flow 会通过 HTTP 调用远程智能体。 # 模拟工作流步骤 topic AI Agents print(fStep 1: Fetching news about {topic}...) raw_news await fetcher.fetch_news(topic) print(f Fetched {len(raw_news)} items.) print(Step 2: Cleaning and summarizing data...) digest_content await cleaner.clean_and_summarize(raw_news) print(f Generated summary: {digest_content[:100]}...) print(Step 3: Preparing email...) # 这里本应调用 Mail Agent 的 send_email 技能 # 我们使用之前创建的客户端来演示 from python_a2a import A2AClient mail_client A2AClient(http://localhost:5000) try: # 调用邮件智能体的发送技能 await mail_client.execute_skill( skill_namesend_email, parameters{ to: subscriberexample.com, subject: fYour Daily Digest on {topic}, body: fHere is your curated digest:\n\n{digest_content}\n\nBest,\nAI Curation Bot } ) print( Email sent successfully (simulated).) except Exception as e: print(f Failed to send email: {e}. Make sure Mail Agent is running on port 5000.) print(\nWorkflow completed!) # 3. 展示如何使用 Flow 的并行处理真实场景 print(\n--- Demonstrating Parallel Flow ---) # 假设我们有两个数据源需要同时抓取 flow_parallel Flow(agent_networknetwork, nameParallel Fetch) # 伪代码展示并行执行的概念 # parallel_results (flow_parallel.parallel() # .ask(fetcher_tech, Fetch news about Technology) # .branch() # .ask(fetcher_science, Fetch news about Science) # .end_parallel(max_concurrency2)) # result await flow_parallel.run({}) print(In a real flow, you could fetch from multiple sources in parallel.) if __name__ __main__: asyncio.run(main())这个脚本模拟了一个完整的工作流获取数据 - 清洗摘要 - 邮件发送。关键在于Flow类和parallel()等方法它们允许你以声明式的方式定义复杂、可能并行执行的智能体任务链。运行批量任务的关键定义清晰的接口确保每个智能体的技能Skill有明确的输入输出。错误处理在工作流中为每个步骤添加重试和超时机制。状态管理对于长时间运行的批量任务需要将任务状态持久化例如存入数据库以便中断后恢复。队列与调度对于生产环境可以使用 Celery、RQ 或 Dramatiq 等任务队列来管理大批量的工作流执行。7. 资源占用与性能观察Python A2A 库本身非常轻量其资源消耗主要来自以下几个方面智能体服务进程每个通过run_server启动的智能体都是一个独立的 Python 进程通常是单线程的 WSGI 服务器如 Flask。内存占用通常在几十 MB 到一两百 MB 之间取决于你导入的库和模型。LLM 后端这是最大的变数。调用云端 API如 OpenAI几乎不占用本地计算资源性能取决于网络延迟和 API 速率限制。本地推理如 Ollama会占用显著的 GPU 显存或 CPU/内存。一个 7B 参数的模型在 GPU 上可能需要 4-8GB 显存在 CPU 上可能需要 8GB 内存。工作流引擎Flow执行器本身开销很小但它会并发调用多个智能体。如果并发数很高需要注意本地端口的占用和网络连接数。性能观察建议使用系统监控工具在 Linux/macOS 上使用htop或top在 Windows 上使用任务管理器。观察 Python 进程的 CPU 和内存使用情况。网络延迟如果智能体部署在不同机器上使用ping或curl -w time_total: %{time_total}\n来测量网络延迟这可能成为工作流的主要瓶颈。日志与追踪为你的智能体和工作流添加详细的日志记录每个步骤的开始和结束时间便于定位性能瓶颈。降低资源占用的技巧智能体复用不要让每个智能体都加载一个独立的 LLM 大模型。可以设计一个中心的LLM Gateway智能体其他智能体通过它来调用 LLM。使用轻量级服务器生产环境可以考虑使用uvicornasgi替代默认的 Flask 开发服务器以获得更好的并发性能。批量处理对于数据处理类任务尽量在智能体内部进行批量操作减少 HTTP 请求次数。8. 常见问题与排查方法在开发和部署过程中你可能会遇到以下问题。这里提供快速的排查思路。问题现象可能原因排查方式解决方案ModuleNotFoundError: No module named python_a2a1. 未安装python-a2a包。2. 在错误的 Python 环境或虚拟环境中运行。1. 运行pip list | grep python-a2a。2. 检查终端激活的 Python 解释器路径which python。1. 在正确的环境中执行pip install python-a2a。2. 确认并激活你的虚拟环境。Address already in use错误指定的端口如 5000已被其他程序占用。运行lsof -i :5000(macOS/Linux) 或netstat -ano | findstr :5000(Windows) 查看占用进程。1. 终止占用端口的进程。2. 在run_server()中更换一个端口如port5001。智能体服务启动后客户端连接超时或无响应1. 服务未成功启动。2. 防火墙或安全组阻止了端口访问。3. 服务绑定到了127.0.0.1而非0.0.0.0。1. 检查服务启动日志是否有错误。2. 在本机使用curl http://localhost:PORT测试。3. 检查run_server的host参数。1. 修复启动错误。2. 确保host0.0.0.0以允许外部连接。3. 配置防火墙规则开放端口。调用技能时返回404 Not Found或Skill not found1. 技能名称拼写错误。2. 智能体类中的skill装饰器未正确定义。3. 请求的 URL 路径不正确。1. 检查智能体源代码中的skill(name...)。2. 访问http://localhost:PORT/skills查看所有已注册技能。1. 确保客户端调用的技能名称与skill装饰器中的name完全一致。2. 使用/skills端点列出技能进行核对。工作流Flow执行卡住或报错1. 某个下游智能体服务宕机或无响应。2. 工作流逻辑有循环依赖或死锁。3. 异步await调用未正确处理。1. 检查每个智能体服务的健康状态。2. 在 Flow 中增加步骤超时设置。3. 查看详细的异常堆栈信息。1. 为ask()或execute_skill()添加超时参数。2. 简化工作流分步调试。3. 确保在异步函数 (async def) 中调用异步方法。集成真实邮箱服务如 SMTP失败1. 邮箱服务商 SMTP/IMAP 配置错误服务器、端口、加密方式。2. 账户密码或授权码错误。3. 服务器开启了二次验证但未使用应用专用密码。1. 先用标准的 Pythonsmtplib库写一个简单的测试脚本。2. 查看邮箱服务商如 Gmail, QQ Mail的官方文档获取正确的服务器地址和端口。1. 在智能体代码外单独测试邮件发送功能。2. 对于 Gmail 等可能需要开启“允许不够安全的应用”或使用 OAuth2.0。3. 使用环境变量存储敏感的 API 密钥和密码不要硬编码在代码中。a2aCLI 命令未找到python-a2a未正确安装或安装路径不在系统的 PATH 环境变量中。运行pip show -f python-a2a查看安装位置检查bin或Scripts目录是否在 PATH 中。1. 使用python -m python_a2a.cli代替a2a。2. 重新在虚拟环境中安装并确保虚拟环境的bin目录在 PATH 中。9. 最佳实践与使用建议基于项目经验遵循以下实践能让你的 A2A 智能体系统更健壮、更易维护。技能设计要单一职责每个skill装饰的方法应该只做一件事并且做好。例如send_email只负责发送validate_email_address负责验证而不是混在一起。这有利于复用和测试。为智能体编写清晰的描述agent装饰器中的name和description非常重要。当智能体注册到网络时这些信息会被用于发现和路由。一个好的描述能让 AI 路由器如AIAgentRouter更准确地将任务分配给它。使用配置管理不要将服务器地址、API 密钥、邮箱密码等硬编码在代码中。使用环境变量、配置文件或密钥管理服务。import os MAIL_SERVER os.getenv(MAIL_SERVER, smtp.gmail.com) MAIL_PORT int(os.getenv(MAIL_PORT, 587))实现健康检查端点为你的智能体服务添加一个/health端点用于监控系统是否存活。这在与容器编排系统如 Kubernetes集成时尤其有用。错误处理与重试网络调用和外部服务如 LLM API、邮件服务器可能失败。在智能体技能和工作流中实现优雅的错误处理和指数退避重试机制。日志记录结构化使用如structlog或logging模块的 JSON 格式化输出方便后续用 ELK 或 Loki 等工具进行日志聚合和分析。记录请求 ID、用户 ID、技能执行时间等关键上下文。安全第一认证与授权如果智能体服务暴露在公网必须实施 API 密钥、JWT Token 或 OAuth 等认证机制。Python A2A 可以与 FastAPI/Flask 的中间件轻松集成。输入验证与清理对所有传入智能体的参数进行严格的验证和清理防止注入攻击。权限最小化通过 MCP 连接的工具如文件系统、数据库应遵循最小权限原则使用沙箱或受限的访问令牌。从模拟到真实的渐进式开发就像本文先模拟邮件发送一样先构建一个能跑通核心逻辑的“模拟版”智能体。然后逐步替换模拟部分为真实的第三方服务 API 调用。这能让你快速验证架构降低初期开发风险。10. 总结与下一步Python A2A 库将一个前沿的协议Google A2A变得触手可及。通过今天的实践你已经掌握了其核心如何创建一个具备特定技能如邮件处理的智能体、如何将其作为 HTTP 服务启动、以及如何通过客户端和工作流来编排它们。这个项目最值得尝试的点在于它的“协议先行”和“互操作性”。你创建的智能体不仅能被自己的系统调用任何遵循 A2A 协议的其他系统也能与之对话。这为构建开放的智能体市场或企业内异构 AI 系统协作奠定了基础。你接下来可以深入探索的方向接入真实服务将MailAgent中的模拟函数替换为真实的smtplib(发送) 和imaplib(接收) 调用或者使用yagmail、sendgrid等第三方库。记得处理好认证和安全。探索 MCP 集成这才是 Python A2A 的“超级技能”所在。尝试为你的智能体集成 GitHub MCP Provider让它能自动创建 Issue或集成 Browserbase MCP Provider让它能自动浏览网页抓取信息。这将极大扩展智能体的能力边界。使用可视化编辑器运行a2a ui命令启动 Agent Flow UI。通过拖拽的方式将你的邮件智能体、数据清洗智能体、报告生成智能体连接起来构建一个可视化的自动化流水线。部署与扩展学习如何使用 Docker 容器化你的智能体并使用 Kubernetes 或 Docker Compose 来编排一整套多智能体系统。考虑如何为智能体网络添加服务发现如 Consul和负载均衡。最容易踩的坑往往在初期端口冲突、依赖版本问题、以及异步编程的理解。按照本文的步骤从最简单的模拟智能体开始每步都进行验证就能平稳上路。建议将本文的示例代码作为你的起点收藏备用。当你需要为 AI 智能体添加“手”和“脚”即操作外部工具的能力时Python A2A 及其 MCP 集成会是一个非常得力的框架。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度
30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度如果你正在构建多智能体系统或者想让不同的 AI 代理能够像人类团队一样协作那么 Google 的 Agent-to-Agent (A2A) 协议是你必须了解的技术。它定义了智能体之间标准化的通信格式是构建可互操作、可扩展的 AI 生态系统的基石。而 Python A2A 库正是将这一协议从理论变为现实的关键工具。Python A2A 是一个由 Manoj Desai 维护的开源项目它完整实现了 Google 的 A2A 协议并深度集成了 Model Context Protocol (MCP)。这意味着你可以用它来创建能够相互发现、通信、并使用外部工具如 GitHub API、浏览器自动化、文件系统的智能体网络。它的核心价值在于“标准化”和“互操作性”——无论你的智能体底层是基于 OpenAI、Anthropic Claude还是本地部署的 Ollama它们都能通过 A2A 协议无缝对话。这篇文章将带你快速上手 Python A2A。我们不空谈概念而是聚焦于实战如何安装、如何创建一个具备收发邮件能力的智能体、如何为智能体赋予专属的“邮箱”技能、以及如何构建一个能自动进行邮件采编和数据清洗的工作流。整个过程你只需要基础的 Python 知识无需复杂的硬件配置重点在于理解其架构和 API 的使用。1. 核心能力速览在深入代码之前我们先通过一个表格快速了解 Python A2A 的核心特性这能帮你判断它是否适合你的项目。能力项说明项目类型Python 库用于实现 Google A2A 协议和 MCP 协议。开源团队个人开发者 Manoj Desai 维护。主要功能1. 创建符合 A2A 协议的智能体 (Agent)。2. 构建智能体网络 (Agent Network) 并实现智能路由。3. 集成 MCP 协议让智能体使用外部工具如 GitHub、浏览器、文件系统。4. 提供可视化工作流编辑器 (Agent Flow UI)。5. 与 LangChain 生态无缝集成。硬件/环境门槛极低。纯 Python 库无特殊硬件要求。核心依赖仅requests库可在任何能运行 Python 的环境包括 CPU中使用。性能瓶颈主要取决于你集成的 LLM 后端如 OpenAI API 调用速度。显存/内存占用库本身不直接消耗大量显存。内存占用取决于你运行的智能体数量和工作流复杂度通常很小。主要资源消耗来自你集成的 LLM 模型如本地部署的 Ollama。支持平台跨平台Windows/macOS/Linux。启动方式1.命令行启动通过a2aCLI 工具快速启动服务器或 UI。2.代码启动通过run_server函数在 Python 脚本中启动。3.可视化 UI通过a2a ui命令启动拖拽式工作流编辑器。是否支持 API是。每个智能体本质上都是一个 HTTP 服务器提供标准的 A2A API 端点可直接通过A2AClient或requests库调用。是否支持批量任务是。通过Flow工作流引擎和parallel()方法可以轻松定义并行执行的批量任务。适合场景1. 研究多智能体协作机制。2. 构建企业级 AI 业务流程自动化系统。3. 为现有 LLM 应用增加工具调用和智能体协作能力。4. 快速原型验证复杂的、多步骤的 AI 任务。2. 适用场景与使用边界Python A2A 是一个强大的框架但理解其边界能让你更有效地使用它。它非常适合构建专用智能体例如创建一个专门处理邮件收发、数据清洗、日程安排或代码审查的智能体。编排复杂工作流将多个单一功能的智能体串联起来完成如“监控新闻 - 摘要 - 生成报告 - 邮件发送”这样的复杂链条。集成现有工具通过 MCP 协议让智能体安全地操作 GitHub、数据库、浏览器或文件系统极大扩展 AI 的能力边界。教学与实验其清晰的协议实现和丰富的示例是学习 A2A/MCP 协议和智能体架构的绝佳材料。它可能不适合超低延迟场景智能体间通过 HTTP 通信存在网络开销。对于需要微秒级响应的场景可能需要更底层的通信机制。完全离线的封闭环境虽然智能体可以本地部署但部分功能如 Agent Flow UI 的某些特性、与云端 MCP 服务器的连接可能需要网络。替代简单的单次 LLM 调用如果你的需求只是向 ChatGPT 问一个问题那么直接调用 API 更简单。重要合规与安全边界工具使用授权当通过 MCP 连接 GitHub、邮箱等外部服务时务必确保你拥有合法的 API 令牌或账户权限并遵守相关服务的使用条款。数据隐私智能体处理的数据尤其是通过 MCP 工具访问的可能包含敏感信息。需确保数据传输和存储的加密并在生产环境中实施严格的访问控制。操作风险赋予智能体文件系统、浏览器自动化等能力后其操作具有实际影响如删除文件、发送邮件。在赋予高权限前应在沙箱环境中充分测试。内容合规智能体生成的内容需符合法律法规和平台政策避免产生有害或侵权信息。3. 环境准备与前置条件开始实战前确保你的开发环境就绪。基础环境要求操作系统Windows 10/11 macOS 10.15 或主流 Linux 发行版如 Ubuntu 20.04。Python 版本Python 3.8 或更高版本。这是运行大多数现代 AI 库的基本要求。包管理工具pip或更推荐的uv一个更快的 Python 包安装器。网络能正常访问 PyPI 和 GitHub用于安装依赖和克隆示例。可选但推荐的组件虚拟环境强烈建议使用venv、conda或uv创建独立的 Python 环境避免依赖冲突。代码编辑器VS Code、PyCharm 等具备良好的 Python 支持。LLM 后端你需要一个或多个 LLM 服务来为智能体提供“大脑”。可以选择云端 APIOpenAI, Anthropic Claude, Google Gemini 等。你需要相应的 API Key。本地模型通过 Ollama、LM Studio 或 vLLM 等本地部署模型。这需要一定的 GPU 显存或 CPU 内存。Python A2A 内置库也支持直接启动一个简单的、基于配置的 LLM 服务器。环境检查清单在终端中执行以下命令确认基础环境正常。# 检查 Python 版本 python --version # 或 python3 --version # 检查 pip 是否可用 pip --version # 可选安装 uv curl -LsSf https://astral.sh/uv/install.sh | sh # 安装后重启终端或运行 source $HOME/.local/bin/env 来激活 uv uv --version如果上述命令都能正确返回版本号说明基础环境已就绪。4. 安装部署与启动方式Python A2A 的安装非常灵活你可以根据需求选择不同的安装方式。方式一使用 pip 安装传统方式这是最直接的方法适合大多数用户。# 安装基础包包含核心功能 pip install python-a2a # 或者根据你的需求安装特定组件 pip install python-a2a[server] # 包含 Flask 服务器支持 pip install python-a2a[openai] # 包含 OpenAI 集成 pip install python-a2a[anthropic] # 包含 Anthropic Claude 集成 pip install python-a2a[ollama] # 包含 Ollama 集成 pip install python-a2a[mcp] # 包含 MCP 支持 pip install python-a2a[all] # 安装所有可选依赖方式二使用 uv 安装推荐速度更快uv是新一代的 Python 包管理工具安装和解决依赖的速度远超pip。# 如果尚未安装 uv先安装 curl -LsSf https://astral.sh/uv/install.sh | sh # 使用 uv 安装 python-a2a uv install python-a2a # uv 会自动处理依赖和虚拟环境方式三从源码安装用于开发或体验最新特性如果你想贡献代码或体验尚未发布的特性可以克隆仓库安装。# 克隆仓库 git clone https://github.com/themanojdesai/python-a2a.git cd python-a2a # 使用 uv 创建虚拟环境并安装开发依赖推荐 uv venv source .venv/bin/activate # Linux/macOS # 在 Windows 上使用.venv\Scripts\activate uv pip install -e .[dev] # -e 表示可编辑模式安装 # 或者使用 pip pip install -e .[dev]验证安装安装完成后可以通过命令行工具快速验证。# 检查 a2a 命令行工具是否可用 a2a --help # 应该能看到类似如下的输出列出可用的命令 # Usage: a2a [OPTIONS] COMMAND [ARGS]... # # Options: # --help Show this message and exit. # # Commands: # agent Manage A2A agents # anthropic Start an Anthropic Claude-powered A2A server # mcp-agent Start an MCP-enabled A2A agent # mcp-call Call an MCP tool directly # mcp-serve Start an MCP server # network Manage agent networks # openai Start an OpenAI-powered A2A server # send Send a message to an A2A agent # stream Stream a response from an A2A agent # ui Start the Agent Flow UI # workflow Run a workflow from a script看到命令列表说明安装成功。5. 功能测试与效果验证创建你的第一个邮件智能体理论说再多不如动手跑一遍。我们现在就创建一个具备“收发邮件”技能的智能体并测试其基本功能。为了简化我们先模拟邮件发送功能后续再接入真实的邮件服务。5.1 创建基础的邮件智能体创建一个名为mail_agent.py的文件内容如下from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState from typing import Dict, Any, List import json agent( nameMail Agent, descriptionA smart agent capable of sending and receiving emails, with content analysis skills., version1.0.0 ) class MailAgent(A2AServer): 一个模拟的邮件智能体。 在实际应用中这里的 send_mail 和 check_inbox 方法应该调用真实的邮件 API如 SMTP、IMAP 或第三方服务如 SendGrid。 # 模拟的“发件箱”和“收件箱”用于演示 _outbox: List[Dict] [] _inbox: List[Dict] [ {from: newstechsite.com, subject: Weekly AI Digest, body: New breakthroughs in multimodal models...}, {from: alicecompany.com, subject: Meeting Notes, body: Here are the notes from our project sync.}, {from: spamexample.com, subject: You won a prize!, body: Click this link to claim your reward.} ] skill( namesend_email, descriptionSend an email to a recipient with a subject and body., tags[communication, email] ) def send_email(self, to: str, subject: str, body: str) - str: 模拟发送邮件。实际应集成 SMTP 或邮件服务 API。 email_record { to: to, subject: subject, body: body, status: sent } self._outbox.append(email_record) print(f[Mail Agent] Simulated sending email to {to}: {subject}) return json.dumps({status: success, message_id: fmock_{len(self._outbox)}, detail: email_record}) skill( namecheck_inbox, descriptionCheck the inbox for new emails, optionally with filters., tags[communication, email, inbox] ) def check_inbox(self, filter_unread: bool False, sender: str None) - str: 模拟检查收件箱。实际应集成 IMAP/POP3 或邮件服务 API。 results self._inbox if sender: results [mail for mail in results if mail[from] sender] # 简单模拟“未读”过滤 if filter_unread: results results[:1] # 假设只有第一封是“未读” print(f[Mail Agent] Inbox checked. Found {len(results)} email(s).) return json.dumps({count: len(results), emails: results}) skill( namesummarize_inbox, descriptionAnalyze the inbox and provide a summary (e.g., count by sender, key topics)., tags[analysis, email, summarization] ) def summarize_inbox(self) - str: 分析收件箱并提供摘要。 from collections import Counter senders Counter([mail[from] for mail in self._inbox]) # 这里可以集成一个真正的 LLM 调用来分析邮件主题和正文 # 为了演示我们做一个简单的统计 summary { total_emails: len(self._inbox), senders: dict(senders), sample_subjects: [mail[subject] for mail in self._inbox[:2]] } print(f[Mail Agent] Inbox summarized: {summary}) return json.dumps(summary) def handle_task(self, task): 核心任务处理函数。根据用户消息调用相应的技能。 message_data task.message or {} content message_data.get(content, {}) # 处理文本输入 text content.get(text, ) if isinstance(content, dict) else str(content) # 简单的意图识别在实际应用中这里应该用一个更强大的 NLU 或 LLM 来解析意图 text_lower text.lower() if send email in text_lower or send mail in text_lower: # 这是一个非常简单的解析。理想情况下应从 task 中提取结构化参数。 # 例如假设消息格式是 “send email to aliceexample.com with subject ‘Hello’ and body ‘World’” # 这里我们使用一个预设的响应来演示。 task.artifacts [{ parts: [{ type: text, text: I can send emails. Please provide details in a structured format or use the send_email skill directly. }] }] task.status TaskStatus(stateTaskState.COMPLETED) elif check inbox in text_lower or new emails in text_lower: # 调用 check_inbox 技能 result self.check_inbox() task.artifacts [{ parts: [{type: text, text: fInbox check result: {result}}] }] task.status TaskStatus(stateTaskState.COMPLETED) elif summarize in text_lower and inbox in text_lower: # 调用 summarize_inbox 技能 result self.summarize_inbox() task.artifacts [{ parts: [{type: text, text: fInbox summary: {result}}] }] task.status TaskState.COMPLETED) else: # 如果无法识别意图请求更多输入 task.status TaskStatus( stateTaskState.INPUT_REQUIRED, message{ role: agent, content: { type: text, text: I am your Mail Agent. I can help you send emails, check your inbox, or summarize your emails. What would you like to do? } } ) return task if __name__ __main__: # 实例化并运行智能体服务器监听 5000 端口 agent MailAgent() print(Starting Mail Agent on http://localhost:5000) run_server(agent, port5000, host0.0.0.0) # host0.0.0.0 允许外部访问5.2 启动并测试智能体启动智能体服务器 在终端中进入mail_agent.py所在目录运行python mail_agent.py如果一切正常你会看到类似Starting Mail Agent on http://localhost:5000的输出并且服务器开始运行。使用 cURL 测试基础 API 打开另一个终端使用curl命令与智能体交互。# 1. 发送一个简单的问候触发 INPUT_REQUIRED curl -X POST http://localhost:5000/tasks \ -H Content-Type: application/json \ -d { message: { role: user, content: {type: text, text: Hello} } } # 预期响应会提示你它可以做什么INPUT_REQUIRED状态。 # 2. 请求检查收件箱 curl -X POST http://localhost:5000/tasks \ -H Content-Type: application/json \ -d { message: { role: user, content: {type: text, text: Can you check my inbox?} } } # 预期响应会包含模拟的收件箱内容状态为 COMPLETED。观察第一个终端的日志你会看到[Mail Agent] Inbox checked...的输出。直接调用技能Skill A2A 协议支持直接调用智能体暴露的技能。这是更结构化的交互方式。# 直接调用 check_inbox 技能并传递参数 curl -X POST http://localhost:5000/skills/check_inbox/execute \ -H Content-Type: application/json \ -d { parameters: { filter_unread: true } }这将直接执行check_inbox方法并返回 JSON 格式的结果绕过了handle_task中的自然语言解析。测试成功的关键指标服务器正常启动无报错。curl命令能收到格式正确的 JSON 响应。服务器终端打印出相应的技能执行日志如[Mail Agent] Inbox checked...。响应中的state字段根据任务情况变为COMPLETED或INPUT_REQUIRED。如果遇到Address already in use错误说明 5000 端口被占用。修改run_server(agent, port5000)中的端口号即可。6. 接口 API 与批量任务智能体作为 HTTP 服务器其核心价值在于提供了标准化的 API 接口便于集成和自动化。Python A2A 的客户端库让调用变得非常简单。6.1 使用 A2AClient 进行编程调用创建一个test_client.py文件使用官方客户端进行交互import asyncio from python_a2a import A2AClient async def test_mail_agent(): # 创建客户端指向我们刚刚启动的邮件智能体 client A2AClient(http://localhost:5000) print( 测试1: 发送自然语言请求 ) # 方法1: 使用 ask 方法发送自然语言消息 response await client.ask(Can you summarize my inbox?) print(fResponse: {response}) print(\n 测试2: 直接调用技能 ) # 方法2: 直接调用特定的技能更精确 skill_response await client.execute_skill( skill_namesummarize_inbox, parameters{} # 这个技能不需要参数 ) print(fSkill Response: {skill_response}) print(\n 测试3: 发送邮件模拟) send_response await client.execute_skill( skill_namesend_email, parameters{ to: colleaguecompany.com, subject: Project Update, body: The A2A agent integration is working perfectly. } ) print(fSend Email Response: {send_response}) if __name__ __main__: # 运行异步测试 asyncio.run(test_mail_agent())运行这个脚本前确保mail_agent.py仍在运行。执行python test_client.py你将看到客户端成功调用智能体并获取结果。6.2 构建邮件采编与清洗工作流批量任务单一智能体能力有限A2A 的强大之处在于构建多智能体工作流。我们来模拟一个“邮件采编与数据清洗”的自动化场景。假设我们有三个智能体Fetcher Agent负责从某个源如 RSS获取原始数据。Cleaner Agent负责清洗和格式化数据。Mail Agent我们刚创建的负责发送整理好的摘要。创建workflow_orchestration.pyimport asyncio from python_a2a import AgentNetwork, Flow import json import random import time # 模拟的智能体类在实际中它们应该是独立运行的服务 class FetcherAgent: 模拟一个数据抓取智能体 async def fetch_news(self, topic: str) - list: await asyncio.sleep(0.5) # 模拟网络延迟 mock_news [ {title: fBreakthrough in {topic} announced, source: TechNews, url: http://example.com/1}, {title: fNew research paper on {topic} published, source: arXiv, url: http://example.com/2}, {title: fIndustry leaders discuss the future of {topic}, source: Blog, url: http://example.com/3}, ] return mock_news class CleanerAgent: 模拟一个数据清洗智能体 async def clean_and_summarize(self, items: list) - str: await asyncio.sleep(0.3) # 模拟清洗提取标题生成摘要 titles [item[title] for item in items] summary fFound {len(titles)} items on topic. Key headlines: ; .join(titles[:2]) return summary async def main(): print(Starting automated email curation workflow...) # 1. 初始化模拟的智能体网络实际应用中这些是远程服务 # 假设我们已经有三个运行在不同端口的智能体 network AgentNetwork(nameCuration Pipeline) # 这里我们模拟网络实际应使用 network.add(fetcher, http://localhost:5001) 等 fetcher FetcherAgent() cleaner CleanerAgent() # 2. 创建并运行工作流 flow Flow(agent_networknetwork, nameDaily Digest Workflow) # 注意为了演示我们直接调用对象方法。真实 Flow 会通过 HTTP 调用远程智能体。 # 模拟工作流步骤 topic AI Agents print(fStep 1: Fetching news about {topic}...) raw_news await fetcher.fetch_news(topic) print(f Fetched {len(raw_news)} items.) print(Step 2: Cleaning and summarizing data...) digest_content await cleaner.clean_and_summarize(raw_news) print(f Generated summary: {digest_content[:100]}...) print(Step 3: Preparing email...) # 这里本应调用 Mail Agent 的 send_email 技能 # 我们使用之前创建的客户端来演示 from python_a2a import A2AClient mail_client A2AClient(http://localhost:5000) try: # 调用邮件智能体的发送技能 await mail_client.execute_skill( skill_namesend_email, parameters{ to: subscriberexample.com, subject: fYour Daily Digest on {topic}, body: fHere is your curated digest:\n\n{digest_content}\n\nBest,\nAI Curation Bot } ) print( Email sent successfully (simulated).) except Exception as e: print(f Failed to send email: {e}. Make sure Mail Agent is running on port 5000.) print(\nWorkflow completed!) # 3. 展示如何使用 Flow 的并行处理真实场景 print(\n--- Demonstrating Parallel Flow ---) # 假设我们有两个数据源需要同时抓取 flow_parallel Flow(agent_networknetwork, nameParallel Fetch) # 伪代码展示并行执行的概念 # parallel_results (flow_parallel.parallel() # .ask(fetcher_tech, Fetch news about Technology) # .branch() # .ask(fetcher_science, Fetch news about Science) # .end_parallel(max_concurrency2)) # result await flow_parallel.run({}) print(In a real flow, you could fetch from multiple sources in parallel.) if __name__ __main__: asyncio.run(main())这个脚本模拟了一个完整的工作流获取数据 - 清洗摘要 - 邮件发送。关键在于Flow类和parallel()等方法它们允许你以声明式的方式定义复杂、可能并行执行的智能体任务链。运行批量任务的关键定义清晰的接口确保每个智能体的技能Skill有明确的输入输出。错误处理在工作流中为每个步骤添加重试和超时机制。状态管理对于长时间运行的批量任务需要将任务状态持久化例如存入数据库以便中断后恢复。队列与调度对于生产环境可以使用 Celery、RQ 或 Dramatiq 等任务队列来管理大批量的工作流执行。7. 资源占用与性能观察Python A2A 库本身非常轻量其资源消耗主要来自以下几个方面智能体服务进程每个通过run_server启动的智能体都是一个独立的 Python 进程通常是单线程的 WSGI 服务器如 Flask。内存占用通常在几十 MB 到一两百 MB 之间取决于你导入的库和模型。LLM 后端这是最大的变数。调用云端 API如 OpenAI几乎不占用本地计算资源性能取决于网络延迟和 API 速率限制。本地推理如 Ollama会占用显著的 GPU 显存或 CPU/内存。一个 7B 参数的模型在 GPU 上可能需要 4-8GB 显存在 CPU 上可能需要 8GB 内存。工作流引擎Flow执行器本身开销很小但它会并发调用多个智能体。如果并发数很高需要注意本地端口的占用和网络连接数。性能观察建议使用系统监控工具在 Linux/macOS 上使用htop或top在 Windows 上使用任务管理器。观察 Python 进程的 CPU 和内存使用情况。网络延迟如果智能体部署在不同机器上使用ping或curl -w time_total: %{time_total}\n来测量网络延迟这可能成为工作流的主要瓶颈。日志与追踪为你的智能体和工作流添加详细的日志记录每个步骤的开始和结束时间便于定位性能瓶颈。降低资源占用的技巧智能体复用不要让每个智能体都加载一个独立的 LLM 大模型。可以设计一个中心的LLM Gateway智能体其他智能体通过它来调用 LLM。使用轻量级服务器生产环境可以考虑使用uvicornasgi替代默认的 Flask 开发服务器以获得更好的并发性能。批量处理对于数据处理类任务尽量在智能体内部进行批量操作减少 HTTP 请求次数。8. 常见问题与排查方法在开发和部署过程中你可能会遇到以下问题。这里提供快速的排查思路。问题现象可能原因排查方式解决方案ModuleNotFoundError: No module named python_a2a1. 未安装python-a2a包。2. 在错误的 Python 环境或虚拟环境中运行。1. 运行pip list | grep python-a2a。2. 检查终端激活的 Python 解释器路径which python。1. 在正确的环境中执行pip install python-a2a。2. 确认并激活你的虚拟环境。Address already in use错误指定的端口如 5000已被其他程序占用。运行lsof -i :5000(macOS/Linux) 或netstat -ano | findstr :5000(Windows) 查看占用进程。1. 终止占用端口的进程。2. 在run_server()中更换一个端口如port5001。智能体服务启动后客户端连接超时或无响应1. 服务未成功启动。2. 防火墙或安全组阻止了端口访问。3. 服务绑定到了127.0.0.1而非0.0.0.0。1. 检查服务启动日志是否有错误。2. 在本机使用curl http://localhost:PORT测试。3. 检查run_server的host参数。1. 修复启动错误。2. 确保host0.0.0.0以允许外部连接。3. 配置防火墙规则开放端口。调用技能时返回404 Not Found或Skill not found1. 技能名称拼写错误。2. 智能体类中的skill装饰器未正确定义。3. 请求的 URL 路径不正确。1. 检查智能体源代码中的skill(name...)。2. 访问http://localhost:PORT/skills查看所有已注册技能。1. 确保客户端调用的技能名称与skill装饰器中的name完全一致。2. 使用/skills端点列出技能进行核对。工作流Flow执行卡住或报错1. 某个下游智能体服务宕机或无响应。2. 工作流逻辑有循环依赖或死锁。3. 异步await调用未正确处理。1. 检查每个智能体服务的健康状态。2. 在 Flow 中增加步骤超时设置。3. 查看详细的异常堆栈信息。1. 为ask()或execute_skill()添加超时参数。2. 简化工作流分步调试。3. 确保在异步函数 (async def) 中调用异步方法。集成真实邮箱服务如 SMTP失败1. 邮箱服务商 SMTP/IMAP 配置错误服务器、端口、加密方式。2. 账户密码或授权码错误。3. 服务器开启了二次验证但未使用应用专用密码。1. 先用标准的 Pythonsmtplib库写一个简单的测试脚本。2. 查看邮箱服务商如 Gmail, QQ Mail的官方文档获取正确的服务器地址和端口。1. 在智能体代码外单独测试邮件发送功能。2. 对于 Gmail 等可能需要开启“允许不够安全的应用”或使用 OAuth2.0。3. 使用环境变量存储敏感的 API 密钥和密码不要硬编码在代码中。a2aCLI 命令未找到python-a2a未正确安装或安装路径不在系统的 PATH 环境变量中。运行pip show -f python-a2a查看安装位置检查bin或Scripts目录是否在 PATH 中。1. 使用python -m python_a2a.cli代替a2a。2. 重新在虚拟环境中安装并确保虚拟环境的bin目录在 PATH 中。9. 最佳实践与使用建议基于项目经验遵循以下实践能让你的 A2A 智能体系统更健壮、更易维护。技能设计要单一职责每个skill装饰的方法应该只做一件事并且做好。例如send_email只负责发送validate_email_address负责验证而不是混在一起。这有利于复用和测试。为智能体编写清晰的描述agent装饰器中的name和description非常重要。当智能体注册到网络时这些信息会被用于发现和路由。一个好的描述能让 AI 路由器如AIAgentRouter更准确地将任务分配给它。使用配置管理不要将服务器地址、API 密钥、邮箱密码等硬编码在代码中。使用环境变量、配置文件或密钥管理服务。import os MAIL_SERVER os.getenv(MAIL_SERVER, smtp.gmail.com) MAIL_PORT int(os.getenv(MAIL_PORT, 587))实现健康检查端点为你的智能体服务添加一个/health端点用于监控系统是否存活。这在与容器编排系统如 Kubernetes集成时尤其有用。错误处理与重试网络调用和外部服务如 LLM API、邮件服务器可能失败。在智能体技能和工作流中实现优雅的错误处理和指数退避重试机制。日志记录结构化使用如structlog或logging模块的 JSON 格式化输出方便后续用 ELK 或 Loki 等工具进行日志聚合和分析。记录请求 ID、用户 ID、技能执行时间等关键上下文。安全第一认证与授权如果智能体服务暴露在公网必须实施 API 密钥、JWT Token 或 OAuth 等认证机制。Python A2A 可以与 FastAPI/Flask 的中间件轻松集成。输入验证与清理对所有传入智能体的参数进行严格的验证和清理防止注入攻击。权限最小化通过 MCP 连接的工具如文件系统、数据库应遵循最小权限原则使用沙箱或受限的访问令牌。从模拟到真实的渐进式开发就像本文先模拟邮件发送一样先构建一个能跑通核心逻辑的“模拟版”智能体。然后逐步替换模拟部分为真实的第三方服务 API 调用。这能让你快速验证架构降低初期开发风险。10. 总结与下一步Python A2A 库将一个前沿的协议Google A2A变得触手可及。通过今天的实践你已经掌握了其核心如何创建一个具备特定技能如邮件处理的智能体、如何将其作为 HTTP 服务启动、以及如何通过客户端和工作流来编排它们。这个项目最值得尝试的点在于它的“协议先行”和“互操作性”。你创建的智能体不仅能被自己的系统调用任何遵循 A2A 协议的其他系统也能与之对话。这为构建开放的智能体市场或企业内异构 AI 系统协作奠定了基础。你接下来可以深入探索的方向接入真实服务将MailAgent中的模拟函数替换为真实的smtplib(发送) 和imaplib(接收) 调用或者使用yagmail、sendgrid等第三方库。记得处理好认证和安全。探索 MCP 集成这才是 Python A2A 的“超级技能”所在。尝试为你的智能体集成 GitHub MCP Provider让它能自动创建 Issue或集成 Browserbase MCP Provider让它能自动浏览网页抓取信息。这将极大扩展智能体的能力边界。使用可视化编辑器运行a2a ui命令启动 Agent Flow UI。通过拖拽的方式将你的邮件智能体、数据清洗智能体、报告生成智能体连接起来构建一个可视化的自动化流水线。部署与扩展学习如何使用 Docker 容器化你的智能体并使用 Kubernetes 或 Docker Compose 来编排一整套多智能体系统。考虑如何为智能体网络添加服务发现如 Consul和负载均衡。最容易踩的坑往往在初期端口冲突、依赖版本问题、以及异步编程的理解。按照本文的步骤从最简单的模拟智能体开始每步都进行验证就能平稳上路。建议将本文的示例代码作为你的起点收藏备用。当你需要为 AI 智能体添加“手”和“脚”即操作外部工具的能力时Python A2A 及其 MCP 集成会是一个非常得力的框架。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
