深入体验 A2A SDK：一步步教你构建“服务化”的 Agent 系统

• Task（任务）：代表Agent为客户端完成的一次工作单元。其封装了整个任务过程中的多次交互消息（Message）、多个任务产出（Artifacts）等。
  
  

    注意，什么时候输出这些状态变更的Event是由服务端自行控制。

• TaskArtifactUpdateEvent（任务工件更新）：代表Agent任务执行过程中交付的多个产出。一个Artifact在内容上与Message类似，都是由多个Part（文本、文件或JSON数据）组成。

class SearchAgent:

    SYSTEM_INSTRUCTION = '''
你是一个智能助手，会智能的使用工具来完成输入任务。
如果调用工具过程中需要用户提供参数信息，你必须严格按照以下格式返回JSON字符串，不要有任何解释:
{"status": "input-required", "message": "{需要用户提供的信息}"}
'''
    ......
    async def stream(self, query: str, context_id) -> AsyncIterable[Dict[str, Any]]:
            
        agent = await self._get_agent()   #调用create_react_agent创建的graph
        config = {'configurable': {'thread_id': context_id}}

        try:
            async for chunk in  agent.astream({"messages": [("user", query)]}, config=config, stream_mode='values'):
                message = chunk["messages"][-1]
                if (isinstance(message, AIMessage) and hasattr(message, "tool_calls") and message.tool_calls):
                    yield {"status":"working","content":f"正在调用工具【{message.tool_calls[0]["name"]}】..."}
                elif isinstance(message, ToolMessage):
                    yield {"status":"working","content":"正在处理工具调用结果..."}           
                elif isinstance(message, AIMessage):                    
                    if isinstance(message.content, str) and"input-required"in message.content:
                        try: 
                            input_info = json.loads(message.content)
                            yield {"status": "input-required", "content": input_info.get("message", "需要更多信息")}
                        except Exception as e:
                            yield {"status":"failed","content":str(e)}                          
                    else:
                        yield {"status": "completed", "content": message.content}
        except Exception as e:
            yield {"status":"failed","content":str(e)}

@abstractmethod
async def execute(self, context: RequestContext, 
                        event_queue: EventQueue):

async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        
        query = context.get_user_input()
        task = context.current_task

        #如果没有task，就创建它；并通知客户端
        if not task:
            task = new_task(context.message)
            event_queue.enqueue_event(task)
        
        #用来简化发布事件的更新器
        updater = TaskUpdater(event_queue, task.id, task.contextId)
        
        try:
            response = self.agent.stream(query, task.contextId)

            #对后端agent的迭代输出做处理
            asyncfor item in response:
                
                status = item["status"]
                content = item["content"]

                #“任务状态更新”：working、input-required等
                if status != "completed":
                    updater.update_status(
                        status,
                        new_agent_text_message(
                            content,
                            task.contextId,
                            task.id,
                        ),
                    )

                else: 
                    #“任务工件更新”：Agent处理成果 
                    updater.add_artifact(
[Part(root=TextPart(text=content))],
                        name="search_result",
                    )

                    #“任务状态更新”：completed，SSE关闭
                    updater.complete(
                        new_agent_text_message(
                            '任务完成',
                            task.contextId,
                            task.id,)
                    )
                    break

        except Exception as e:
......

....#省略agent_card的准备过程......

        httpx_client = httpx.AsyncClient()
        request_handler = DefaultRequestHandler(
            agent_executor=SearchAgentExecutor(model_name=model),
            ......
        )
        server = A2AStarletteApplication(
            agent_card=agent_card, http_handler=request_handler
        )

        logger.info(f"服务地址: http://{host}:{port}/")
        logger.info(f"Agent Card: http://{host}:{port}/.well-known/agent.json")
        uvicorn.run(server.build(), host=host, port=port)