4. 智能体执行器

A2A 智能体如何处理请求并生成响应与事件，由**智能体执行器（Agent Executor）**负责。A2A Python SDK 提供抽象基类 a2a.server.agent_execution.AgentExecutor，由你来实现。

`AgentExecutor` 接口

AgentExecutor 定义了两个主要方法：

async def execute(self, context: RequestContext, event_queue: EventQueue)：处理需要响应或事件流的请求。它处理用户输入（通过 context 获得），并通过 event_queue 回送 Message、Task、TaskStatusUpdateEvent 或 TaskArtifactUpdateEvent；
async def cancel(self, context: RequestContext, event_queue: EventQueue)：处理取消正在进行的任务的请求。

RequestContext 提供传入请求的信息，例如用户消息和已有的任务详情；EventQueue 用于执行器向客户端回送事件。

Helloworld 智能体执行器

来看 agent_executor.py，它定义了 HelloWorldAgentExecutor。

1. 智能体（HelloWorldAgent）：一个封装实际“业务逻辑”的简单辅助类。

class HelloWorldAgent:
    """Hello World Agent."""

    async def invoke(self, user_request: str) -> str:
        """Invoke the Hello World agent to generate a response."""
        return f'Hello, World! I have received your request ({user_request})'

它有一个简单的 invoke 方法，返回一段问候文本。

2. 执行器（HelloWorldAgentExecutor）：实现 AgentExecutor 接口。

__init__ 中实例化了 HelloWorldAgent：

class HelloWorldAgentExecutor(AgentExecutor):
    """Test AgentProxy Implementation."""

    def __init__(self) -> None:
        self.agent = HelloWorldAgent()

execute 的实现如下：

async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:
    """Process user request."""
    # 1. Collect a task from request context
    if context.current_task:
        task = context.current_task
    else:
        # 1.1 If there is no task, create one and add it event queue
        task = new_task_from_user_message(context.message)
        await event_queue.enqueue_event(task)

    # 2. Update task status in EventQueue using TaskUpdater class object
    task_updater = TaskUpdater(
        event_queue=event_queue, task_id=task.id, context_id=task.context_id
    )
    await task_updater.update_status(
        state=TaskState.TASK_STATE_WORKING,
        message=new_text_message('Processing request...'),
    )

    # 3. Collect user request from request content and invoke LLM agent to generate content
    query = get_message_text(context.message)
    if query:
        result = await self.agent.invoke(user_request=query)
    else:
        result = 'No text input is provided!'

    # 4. Add generated response as an artifact to EventQueue
    await task_updater.add_artifact(parts=[new_text_part(text=result, media_type='text/plain')])
    print('Result: ', result)

    # 5. Update task status to completed
    await task_updater.update_status(
        state=TaskState.TASK_STATE_COMPLETED,
        message=new_text_message('Request is completed!'),
    )

当收到 SendMessage 或 SendStreamingMessage 请求时（在这个简化的执行器中两者都由 execute 处理），会发生以下步骤：

从上下文中取出当前任务；如果上下文中没有任务，就创建一个新任务并加入 EventQueue；
通过 TaskUpdater 入队一个状态为 TASK_STATE_WORKING 的状态更新，表示智能体已开始处理；
调用 self.agent.invoke() 执行实际业务逻辑；
入队一个包含智能体结果文本的产出物（TaskArtifactUpdateEvent）；
最后入队状态为 TASK_STATE_COMPLETED 的状态更新，结束任务。

AgentExecutor 是 A2A 协议（由请求处理器和服务端应用管理）与你的智能体具体逻辑之间的桥梁：它接收请求上下文，并通过事件队列回送结果或更新。任务状态的含义见任务的生命周期。

下一步

准备好 Agent Card 与执行器之后，就可以启动服务器了。

来源：A2A 官方文档 · The Agent Executor

3. 智能体技能与 Agent Card 5. 启动服务器

4. 智能体执行器

AgentExecutor 接口

Helloworld 智能体执行器

下一步

`AgentExecutor` 接口