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 ServerA2A兼容性要求我们的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:responserequests.get(headers{X-QW-Api-Key:os.getenv(QW_API_KEY)},urlurl,paramsparams)ifresponse.status_code200:returnextract_result(response)else:raiseException(f请求失败状态码{response.status_code})tooldeflook_up_location(city:str)-str:查询指定城市的地理位置 Args: city (str): 城市名称例如 北京 或 beijing returninvoke(urlos.getenv(QW_LOCATION_LOOKUP_URL,),params{location:city},extract_resultlambdaresponse:response.json()[location][0][id])tooldefget_weather(location:str)-dict:获取指定位置的实时天气信息 Args: location (str): 工具look_up_location返回的指定城市的地理位置 returninvoke(urlos.getenv(QW_WEATHER_URL,),params{location:location},extract_resultlambdaresponse:response.json()[now])graphcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[look_up_location,get_weather],system_prompt(作为一个出行助手提供指定程序实时天气和着装建议))我们执行命令行langgraph dev以dev模式部署我们定义的Agent并启动作为宿主的Web服务器。2. 获取AgentCardA2A调用需要预先得到被调用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:2024A2ACardResolver会附加上默认的路径.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_id1970a4eb-17e2-41c7-8848-894281d10bbeasyncdefcreate_new_assistant(client:AssistantsClient,assistant_id:str,graph_id:str,**kwargs):assistantsawaitclient.search()ifany(assistant[assistant_id]assistant_idforassistantinassistants):awaitclient.delete(assistant_idassistant_id)returnawaitclient.create(assistant_idassistant_id,nameclothing_assistant,graph_idgraph_id,**kwargs)asyncdefget_agent_card(httpx_client:AsyncClient,assistant_id:str)-AgentCard:resolverA2ACardResolver(httpx_clienthttpx_client,base_urlhttp://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_clientawaitClientFactory.connect(agentagent_card,client_configClientConfig(httpx_clienthttpx_client))requestMessage(message_iduuid.uuid4().hex,roleRole.user,parts[Part(rootTextPart(text苏州目前天气如何我该穿什么))])asyncforchunkina2a_client.send_message(requestrequest):task,eventcast(tuple[Task,TaskStatusUpdateEvent],chunk)if(eventandevent.statusandevent.status.stateTaskState.completedandtask.history):messagetask.history[-1]forpartinmessage.parts:ifisinstance(part.root,TextPart):returnpart.root.textreturn未能获取到有效的回复asyncdefmain():asyncwithget_client(urlhttp://127.0.0.1:2024)asclient:awaitcreate_new_assistant(client.assistants,assistant_idassistant_id,graph_idtest_agent)asyncwithAsyncClient(timeout120)ashttpx_client:agent_cardawaitget_agent_card(httpx_clienthttpx_client,assistant_idassistant_id)print(json.dumps(agent_card.model_dump(),indent2))resultawaitcall_agent(httpx_clienthttpx_client,agent_cardagent_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恤 / 薄衬衫 - **外搭**薄外套或针织开衫方便随时增减 - **下身**长裤或薄牛仔裤 - **鞋子**透气的运动鞋或休闲鞋 如果你**需要外出时间较长**或**傍晚还在外面**带一件薄外套会更合适。 需要我帮你看看**今晚或明天的天气**吗