鸿蒙智能体开发实战：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协议规范小艺开放平台