news 2026/5/15 19:32:05

[利用LangGraph SDK调用部署的Agent-07]以A2A的形式调用Agent

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
[利用LangGraph SDK调用部署的Agent-07]以A2A的形式调用Agent

Agent2Agent (A2A) 是Google用于实现对话式AI Agent之间通信的协议。LangSmith实现了A2A支持,使您的Agent能够通过标准化协议与其他兼容A2A的Agent进行通信。A2A对应Endpoint的路径为/a2a/{assistant_id}。Agent Server支持以下A2A RPC方法:

  • message/send:向Assistant发送消息并接收完整响应;
  • message/stream:发送消息并使用服务器发送事件 (SSE) 实时传输响应流;
  • tasks/get:检索先前创建的任务的状态和结果;

1. 部署兼容A2A要求的Agent Server

A2A兼容性要求我们的Graph的状态必需有一个messages字段来存储对话消息,并且这个字段需要满足A2A协议的规范。为了演示A2A调用,我们将部署的Graph切换到之前定义的用于提供指定城市实时天气信息。 回顾一下Agent Server根目录的结构:

root/ ├── .env # 存放环境变量 ├── langgraph.json # 配置Graph ├── agent.py # 定义Graph └── pyproject.toml # 项目配置文件

langgraph.json文件中,设置了一个名为test_agent的Graph,并指定了它的定义文件为./agent.py:graph

{"dependencies":["."],"graphs":{"test_agent":"./agent.py:graph"},"env":".env"}

我们在agent.py中定义了一个Graph用来提供指定城市实时的天气信息,并在此基础上提供着装建议。我们调用的是“和风天气”提供的API,相应的URL和API Key等信息我们都放在了.env文件中,该文件中也存放了与模型和LangSmith相关的环境变量。agent.py文件的内容如下所示:利用create_agent函数创建了一个Agent注册了两个工具,get_weather根据指定的位置编码查询实时天气信息,指定城市的地理位置编码则由look_up_location工具查询获得。

fromtypingimportCallable,Anyfromlangchain.agentsimportcreate_agentfromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromrequestsimportResponseimportrequests,osdefinvoke(url:str,params:dict,extract_result:Callable[[Response],Any])->Any:response=requests.get(headers={"X-QW-Api-Key":os.getenv("QW_API_KEY")},url=url,params=params)ifresponse.status_code==200:returnextract_result(response)else:raiseException(f"请求失败,状态码:{response.status_code}")@tooldeflook_up_location(city:str)->str:"""查询指定城市的地理位置 Args: city (str): 城市名称,例如 "北京" 或 "beijing" """returninvoke(url=os.getenv("QW_LOCATION_LOOKUP_URL",""),params={"location":city},extract_result=lambdaresponse:response.json()["location"][0]["id"])@tooldefget_weather(location:str)->dict:"""获取指定位置的实时天气信息 Args: location (str): 工具look_up_location返回的指定城市的地理位置 """returninvoke(url=os.getenv("QW_WEATHER_URL",""),params={"location":location},extract_result=lambdaresponse:response.json()["now"])graph=create_agent(model=ChatOpenAI(model="gpt-5.2-chat"),tools=[look_up_location,get_weather],system_prompt=("作为一个出行助手,提供指定程序实时天气和着装建议"))

我们执行命令行langgraph dev以dev模式部署我们定义的Agent,并启动作为宿主的Web服务器。

2. 获取AgentCard

A2A调用需要预先得到被调用Agent的AgentCard。AgentCard是一个 JSON文档,相当于Agent的名片,提供有关Agent的基本元数据。客户端解析这些信息,以确定Agent是否适合特定任务、如何构建请求以及如何进行安全通信。关键信息包括身份、服务端点(URL)、A2A 功能、身份验证要求和技能列表。由于针对Agent调用本质上是针对某个Assistant的调用,所以我们得先得到被调用Assitant的ID。如果确定了assistant_id,我们就可以按照如下的形式请求路径.well-known/agent-card.json?assistant_id={assistant_id}来获取AgentCard。

3. 编写A2A客户端程序远程调用Agent

我们编写了如下的客户端程序。我们使用a2a库提供的客户端组件来调用Agent Server提供的A2A接口。函数get_agent_card用于获取AgentCard,它利用A2ACardResolver请求Agent Server的获取AgentCard。我们指定了基地址"http://127.0.0.1:2024",A2ACardResolver会附加上默认的路径.well-known/agent-card.json来格式化最终的地址。为了迎合Agent Server的实现,我们还得添加针对assistant_id的查询字符串。

fromtypingimportcastfromlanggraph_sdkimportget_clientfromlanggraph_sdk.clientimportAssistantsClientfroma2a.clientimportClientFactory,A2ACardResolver,ClientConfigfroma2a.typesimport(Message,Part,Role,TextPart,AgentCard,Task,TaskStatusUpdateEvent,TaskState)fromhttpximportAsyncClientimportasyncio,json,uuid assistant_id="1970a4eb-17e2-41c7-8848-894281d10bbe"asyncdefcreate_new_assistant(client:AssistantsClient,assistant_id:str,graph_id:str,**kwargs):assistants=awaitclient.search()ifany(assistant["assistant_id"]==assistant_idforassistantinassistants):awaitclient.delete(assistant_id=assistant_id)returnawaitclient.create(assistant_id=assistant_id,name="clothing_assistant",graph_id=graph_id,**kwargs)asyncdefget_agent_card(httpx_client:AsyncClient,assistant_id:str)->AgentCard:resolver=A2ACardResolver(httpx_client=httpx_client,base_url="http://127.0.0.1:2024")returnawaitresolver.get_agent_card(http_kwargs={"params":{"assistant_id":assistant_id}})asyncdefcall_agent(httpx_client:AsyncClient,agent_card:AgentCard)->str:a2a_client=awaitClientFactory.connect(agent=agent_card,client_config=ClientConfig(httpx_client=httpx_client))request=Message(message_id=uuid.uuid4().hex,role=Role.user,parts=[Part(root=TextPart(text="苏州目前天气如何?我该穿什么?"))])asyncforchunkina2a_client.send_message(request=request):task,event=cast(tuple[Task,TaskStatusUpdateEvent],chunk)if(eventandevent.statusandevent.status.state==TaskState.completedandtask.history):message=task.history[-1]forpartinmessage.parts:ifisinstance(part.root,TextPart):returnpart.root.textreturn"未能获取到有效的回复"asyncdefmain():asyncwithget_client(url="http://127.0.0.1:2024")asclient:awaitcreate_new_assistant(client.assistants,assistant_id=assistant_id,graph_id="test_agent")asyncwithAsyncClient(timeout=120)ashttpx_client:agent_card=awaitget_agent_card(httpx_client=httpx_client,assistant_id=assistant_id)print(json.dumps(agent_card.model_dump(),indent=2))result=awaitcall_agent(httpx_client=httpx_client,agent_card=agent_card)print(result)asyncio.run(main())

针对调用实现call_agent函数中。我们将得到的AgentCard传递给A2A客户端组件的ClientFactory来创建一个A2A客户端对象。然后我们构造一个Message对象作为请求消息,并调用A2A客户端的send_message方法来发送这个消息。send_message方法会返回一个异步迭代器来迭代响应流中的事件,我们在迭代过程中检查任务状态更新事件,当任务完成时,我们从任务历史中提取出最后一条消息,并从中获取文本部分作为最终的回复结果。

main函数中,我们首先调用create_new_assistant函数来创建一个新的Assistant,并将其与之前定义的Graph进行绑定。接着,我们使用get_agent_card函数来获取这个Assistant的AgentCard,并打印出来。最后,我们调用call_agent函数来发送一个消息给Agent,并打印出它的回复。针对AgentCard和调用结果的输出如下所示:

{"additionalInterfaces":null,"capabilities":{"extensions":null,"pushNotifications":false,"stateTransitionHistory":false,"streaming":true},"defaultInputModes":["application/json","text/plain"],"defaultOutputModes":["application/json","text/plain"],"description":"clothing_assistant assistant","documentationUrl":null,"iconUrl":null,"name":"clothing_assistant","preferredTransport":"JSONRPC","protocolVersion":"0.3.0","provider":null,"security":null,"securitySchemes":null,"signatures":null,"skills":[{"description":"clothing_assistant assistant","examples":[],"id":"1970a4eb-17e2-41c7-8848-894281d10bbe-main","inputModes":["application/json","text/plain"],"name":"clothing_assistant Capabilities","outputModes":["application/json","text/plain"],"security":null,"tags":["assistant","langgraph"]}],"supportsAuthenticatedExtendedCard":null,"url":"http://127.0.0.1:2024/a2a/1970a4eb-17e2-41c7-8848-894281d10bbe","version":"0.6.15"}
目前**苏州**的实时天气情况如下(更新时间:**13:34**): 🌤 **天气**:多云 🌡 **气温**:23℃(体感约 20℃) 💨 **风向风力**:西北风 2 级,风不大 💧 **湿度**:35%,比较干爽 🌧 **降水**:无 👀 **能见度**:良好 ### 👕 穿衣建议 现在的体感比较舒适,但早晚可能稍凉,建议: - **上身**:长袖T恤 / 薄衬衫 - **外搭**:薄外套或针织开衫(方便随时增减) - **下身**:长裤或薄牛仔裤 - **鞋子**:透气的运动鞋或休闲鞋 如果你**需要外出时间较长**或**傍晚还在外面**,带一件薄外套会更合适。 需要我帮你看看**今晚或明天的天气**吗?
版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/5/15 19:32:04

保姆级避坑指南:在Ubuntu 20.04上搞定ROS Noetic + PX4 + Gazebo仿真环境

保姆级避坑指南:在Ubuntu 20.04上搞定ROS Noetic PX4 Gazebo仿真环境 刚接触机器人仿真的开发者,往往会在环境搭建阶段耗费大量时间解决依赖冲突和版本适配问题。Ubuntu 20.04作为长期支持版本,搭配ROS Noetic和PX4生态时,既有新…

作者头像 李华
网站建设 2026/5/15 19:27:36

STM32存储性能对决:SDIO总线三种传输模式读写SD/TF卡实战评测

1. SDIO总线与存储卡基础认知 在嵌入式系统开发中,外置存储扩展是常见需求。SD/MicroSD/TF卡凭借其体积小、容量大、价格实惠的优势,成为许多项目的首选方案。这些卡片本质上都是基于NAND Flash的存储设备,内部包含闪存芯片和控制器电路。虽然…

作者头像 李华
网站建设 2026/5/15 19:25:14

C#中使用MiniExcel 快速入门:读写 .xlsx 文件

背景介绍 报表绕不开 Excel。传统方案用 Microsoft.Office.Interop&#xff0c;需要安装 Office&#xff0c;且进程管理复杂。MiniExcel 是一个轻量级库&#xff08;< 1MB&#xff09;&#xff0c;通过直接操作 ZIP 压缩包&#xff08;.xlsx 本质是 ZIP&#xff09;实现读写…

作者头像 李华
网站建设 2026/5/15 19:24:06

Hyper-V导入win11时,出现出现“win11无法启动”、“win11无法初始化、”“hyper-v 无法解包密钥保护程序”

解决方法&#xff1a;从旧主机导出证书并导入到新主机 在旧 Hyper-V 主机上&#xff1a; 按下WinR&#xff0c;输入mmc点击 “文件” → “添加 / 删除管理单元”选择 “证书”&#xff0c;点击 “添加”选择 “计算机账户” → “下一步” → “本地计算机” → “完成” → “…

作者头像 李华
网站建设 2026/5/15 19:20:04

4. 大型场馆大空间挡烟垂壁选型与布设

大型场馆、商业综合体、中庭展厅这类大空间建筑&#xff0c;空间跨度大、层高较高&#xff0c;传统隔断无法满足排烟分区要求&#xff0c;合理选用与布设挡烟垂壁&#xff0c;是解决大空间防排烟难题的核心途径。大空间场景在挡烟垂壁选型上&#xff0c;需优先适配大跨度、高空…

作者头像 李华