鸿蒙智能体开发实战：6.A2A 模式开发接口-北京尧图网络

前言在上一篇文章中我们完成了项目的初始化和基础架构搭建。本文将深入介绍 A2A 模式的核心接口开发包括 JSON-RPC 2.0 协议的路由设计和实现。一、协议架构概览在开始接口开发之前先回顾A2A模式的整体协议架构A2A协议采用三层架构设计传输层基于HTTP/HTTPS统一使用POST方法协议层JSON-RPC 2.0格式标准化请求/响应业务层各RPC方法实现具体的业务逻辑核心设计原则无状态传输服务器不维护长连接通过Session机制管理状态流式优先所有耗时操作支持SSE流式输出统一入口所有方法通过单一端点路由二、JSON-RPC 2.0 协议回顾2.1 请求格式JSON-RPC 是一种轻量级的远程过程调用协议其请求格式如下{jsonrpc:2.0,id:request-123,method:message/stream,params:{key:value}}2.2 响应格式响应格式{jsonrpc:2.0,id:request-123,result:{data:response data}}2.3 JSON-RPC 错误码规范JSON-RPC 2.0 定义了标准的错误码范围错误码含义说明-32700解析错误服务端收到无效的JSON-32600无效请求发送的JSON不是有效的请求对象-32601方法不存在请求的方法不存在-32602无效参数方法参数无效-32603内部错误服务端内部错误-32000~-32099服务端错误服务端自定义错误三、统一端点设计A2A 协议采用统一端点设计所有 RPC 方法都通过/agent/message端点处理POST /agent/message Headers: Content-Type: application/json agent-session-id: session-id四、完整的路由实现4.1 消息分发器# routes/agent_routes.pyfromfastapiimportAPIRouter,Request,Headerfromfastapi.responsesimportJSONResponse,StreamingResponseimportuuidimportjsonfromtypingimportOptional,Dict,Any routerAPIRouter(prefix/agent,tags[agent])# 共享状态存储agent_sessions:Dict[str,Dict[str,Any]]{}conversation_contexts:Dict[str,list]{}task_states:Dict[str,Dict[str,Any]]{}router.post(/message)asyncdefhandle_agent_message(request:Request,agent_session_id:Optional[str]Header(None,aliasagent-session-id),x_request_id:Optional[str]Header(None,aliasX-Request-ID),):统一的 Agent 消息处理接口request_idx_request_idorstr(uuid.uuid4())# 解析请求体try:bodyawaitrequest.json()exceptjson.JSONDecodeError:returnJSONResponse(status_code400,content{jsonrpc:2.0,id:unknown,error:{code:400,message:Invalid JSON}})jsonrpc_requestJsonRpcRequest(**body)methodjsonrpc_request.method# 路由分发ifmethodinitialize:returnawaithandle_initialize(jsonrpc_request,request_id)elifmethodnotifications/initialized:returnawaithandle_initialized(jsonrpc_request,agent_session_id,request_id)elifmethodmessage/stream:returnawaithandle_message_stream(jsonrpc_request,agent_session_id,request_id)elifmethodtasks/cancel:returnawaithandle_tasks_cancel(jsonrpc_request,agent_session_id,request_id)elifmethodclearContext:returnawaithandle_clear_context(jsonrpc_request,agent_session_id,request_id)else:returnJSONResponse(status_code400,contentJsonRpcResponse(jsonrpc2.0,idjsonrpc_request.id,error{code:-32601,message:fMethod not found:{method}}).model_dump())4.2 Initialize 方法初始化会话获取agentSessionIdasyncdefhandle_initialize(request:JsonRpcRequest,request_id:str,)-JSONResponse:处理 initialize 方法fromdatetimeimportdatetime,timedelta session_iduuid.uuid4().hexsession_ttl7*24*60*60# 7 天# 存储会话信息agent_sessions[session_id]{created_at:datetime.utcnow(),expires_at:datetime.utcnow()timedelta(secondssession_ttl),status:initialized,}responseJsonRpcResponse(jsonrpc2.0,idrequest.id,result{version:1.0,agentSessionId:session_id,agentSessionTtl:session_ttl,})returnJSONResponse(contentresponse.model_dump())4.3 Notifications/Initialized 方法通知服务器初始化完成asyncdefhandle_initialized(request:JsonRpcRequest,agent_session_id:Optional[str],request_id:str,)-JSONResponse:处理 notifications/initialized 方法ifnotagent_session_idoragent_session_idnotinagent_sessions:returnJSONResponse(status_code401,content{error:Invalid session ID})# 更新会话状态为 activeagent_sessions[agent_session_id][status]activereturnJSONResponse(content{})4.4 Message/Stream 方法核心的流式消息处理方法支持 SSE 输出fromfastapi.responsesimportStreamingResponseimportasyncioasyncdefhandle_message_stream(request:JsonRpcRequest,agent_session_id:Optional[str],request_id:str,)-StreamingResponse:处理 message/stream 方法支持 SSE 流式输出paramsrequest.paramsor{}task_idparams.get(id,str(uuid.uuid4()))session_idparams.get(sessionId,str(uuid.uuid4()))messageparams.get(message,{})# 提取用户消息文本user_textget_text_from_parts(message.get(parts,[]))# 存储任务状态task_states[task_id]{sessionId:session_id,status:working,}asyncdefgenerate_sse():SSE 流式生成器try:# 步骤 1: 发送 submitted 状态yieldcreate_sse_event({taskId:task_id,kind:status-update,status:{state:submitted}})# 步骤 2: 处理用户请求调用大模型asyncforchunkinprocess_user_request(user_text,session_id):yieldcreate_sse_event(chunk)# 步骤 3: 发送完成状态yieldcreate_sse_event({taskId:task_id,kind:status-update,final:True,status:{state:completed}})exceptExceptionase:yieldcreate_sse_event({taskId:task_id,kind:status-update,final:True,status:{state:failed},error:str(e)})returnStreamingResponse(generate_sse(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,X-Accel-Buffering:no,})defget_text_from_parts(parts:list)-str:从 Message Parts 中提取文本内容texts[]forpartinparts:ifpart.get(kind)textandpart.get(text):texts.append(part.get(text))return\n.join(texts)defcreate_sse_event(data:dict)-str:创建 SSE 事件字符串event{jsonrpc:2.0,id:task_id,result:data,error:{code:0,message:success}}returnfdata:{json.dumps(event,ensure_asciiFalse)}\n\n4.5 Tasks/Cancel 方法取消正在执行的任务asyncdefhandle_tasks_cancel(request:JsonRpcRequest,agent_session_id:Optional[str],request_id:str,)-JSONResponse:处理 tasks/cancel 方法paramsrequest.paramsor{}task_idparams.get(id,request.id)iftask_idintask_states:task_states[task_id][status]canceledreturnJSONResponse(contentJsonRpcResponse(jsonrpc2.0,idrequest.id,result{id:task_id,status:{state:canceled}}).model_dump())4.6 ClearContext 方法清理对话上下文asyncdefhandle_clear_context(request:JsonRpcRequest,agent_session_id:Optional[str],request_id:str,)-JSONResponse:处理 clearContext 方法paramsrequest.paramsor{}session_idparams.get(sessionId,request.sessionId)ifsession_idandsession_idinconversation_contexts:conversation_contexts[session_id][]returnJSONResponse(contentJsonRpcResponse(jsonrpc2.0,idrequest.id,result{status:{state:cleared}}).model_dump())五、接口调用示例5.1 初始化会话curl-XPOST http://localhost:8080/agent/message\-HContent-Type: application/json\-d{ jsonrpc: 2.0, id: 1, method: initialize }响应{jsonrpc:2.0,id:1,result:{agentSessionId:8f01f3d172cd4396a0e535ae8aec6687,agentSessionTtl:604800}}重要字段说明agentSessionId会话唯一标识后续请求必须在Header中携带agentSessionTtl会话有效期秒默认为7天提示建议在客户端实现Session续期逻辑在过期前重新调用initialize获取新ID。5.2 发送流式消息curl-XPOST http://localhost:8080/agent/message\-HContent-Type: application/json\-Hagent-session-id: 8f01f3d172cd4396a0e535ae8aec6687\-d{ jsonrpc: 2.0, id: 2, method: message/stream, params: { id: task-001, sessionId: session-001, message: { role: user, parts: [{ kind: text, text: 你好请帮我介绍一下你自己 }] } } }六、接口认证与鉴权实现6.1 认证方式对比认证方式适用场景安全等级API Key内部测试中AK/SK生产环境高OAuth 2.0需要用户身份高6.2 API Key 校验中间件fromfastapiimportHTTPException,Headerasyncdefapi_key_auth(x_api_key:strHeader(...)):ifx_api_key!API_KEY:raiseHTTPException(status_code401,detailInvalid API Key)七、SSE 流式输出优化7.1 心跳保持asyncdefheartbeat():whileTrue:awaitasyncio.sleep(15)yield: ping\n\n7.2 流式输出控制字段推荐值说明Cache-Controlno-cache禁用缓存Connectionkeep-alive保持连接X-Accel-Bufferingno禁用代理缓冲八、接口测试用例8.1 测试 initializecurl-XPOST http://localhost:8080/agent/message\-HContent-Type: application/json\-d{jsonrpc:2.0,id:1,method:initialize}8.2 测试 message/streamcurl-XPOST http://localhost:8080/agent/message\-HContent-Type: application/json\-Hagent-session-id:$SESSION_ID\-d{ jsonrpc: 2.0, id: 2, method: message/stream, params: {...} }九、接口性能优化建议使用异步数据库连接池管理会话对大模型调用增加超时与重试对频繁访问的上下文使用 Redis 缓存9.1 超时配置示例importhttpx timeouthttpx.Timeout(10.0,connect2.0)clienthttpx.AsyncClient(timeouttimeout)小结本文详细介绍了A2A模式的核心接口开发协议架构三层架构设计传输层/协议层/业务层解耦JSON-RPC 2.0标准化请求响应格式和错误码规范统一端点设计所有RPC方法通过/agent/message路由SSE流式输出实时推送任务状态和结果会话管理基于agentSessionId的完整状态跟踪六个RPC方法消息分发器、初始化、初始化完成、流式消息、取消任务、清理上下文如果这篇文章对你有帮助欢迎点赞、收藏⭐、关注你的支持是我持续创作的动力相关资源鸿蒙Agent通信协议技术规范总览 - 华为开发者联盟鸿蒙Agent通信协议消息指令定义 - 华为开发者联盟JSON-RPC 2.0 规范FastAPI官方文档a2a-sdk Python SDK - PyPISSE (Server-Sent Events) 规范Google A2A协议规范小艺开放平台

鸿蒙智能体开发实战：6.A2A 模式开发接口

相关推荐

数据中台 ODS/DWD/DWS/DWT 4层数仓设计：从命名规范到分层建模实战解析

记一次多 Agent 架构在单节点 K8s 触发的 PID 耗尽与 Pod 驱逐（Evicted）大摸排

麦克斯韦方程组 4 大方程实战：从静电场到电磁波传播的 3 步推导

更多资讯

EasyU优启通PE维护盘实战指南：制作、启动与核心操作全解析

Rust原来很简单二：所有权系统

基于Hadoop+Spark的民宿智能推荐系统设计与实现

ADS122U04IPWR+STM32F427ZI工业级温度测量信号链设计

UART接口24位ADC搭配RS485，解决远距离温度监控难题

UART接口24位ADC搭配PIC18F46K22的RTD与热电偶测温实践

今日更新

多智能体协作系统架构设计与实践优化

大模型系统性入门实战指南：从原理到部署微调的全链路路线

鸿蒙文件系统沙箱机制解析与开发实践

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？