4. The Agent Executor

How an A2A agent processes requests and generates responses and events is handled by an Agent Executor. The A2A Python SDK provides the abstract base class a2a.server.agent_execution.AgentExecutor, which you implement.

The AgentExecutor interface

The AgentExecutor class defines two primary methods:

• async def execute(self, context: RequestContext, event_queue: EventQueue): handles incoming requests that expect a response or a stream of events. It processes the user’s input (available through context) and uses the event_queue to send back Message, Task, TaskStatusUpdateEvent, or TaskArtifactUpdateEvent objects.
• async def cancel(self, context: RequestContext, event_queue: EventQueue): handles requests to cancel an ongoing task.

The RequestContext provides information about the incoming request, such as the user’s message and any existing task details. The EventQueue is how the executor sends events back to the client.

The Helloworld agent executor

Look at agent_executor.py, which defines HelloWorldAgentExecutor.

1. The agent (HelloWorldAgent): a simple helper class that encapsulates the actual “business logic”.

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})'

It has a simple invoke method that returns a greeting.

2. The executor (HelloWorldAgentExecutor): implements the AgentExecutor interface.

__init__ instantiates the HelloWorldAgent:

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

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

execute is implemented as follows:

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!'),
    )

When a SendMessage or SendStreamingMessage request comes in (both are handled by execute in this simplified executor), these steps occur:

1. The current task is taken from the context. If there is no task in the context, a new task is created and added to the EventQueue.
2. A status update with state TASK_STATE_WORKING is enqueued through the TaskUpdater, indicating the agent has begun processing.
3. self.agent.invoke() runs the actual business logic.
4. An artifact (TaskArtifactUpdateEvent) containing the agent’s result text is enqueued.
5. Finally, a status update with state TASK_STATE_COMPLETED is enqueued to conclude the task.

The AgentExecutor acts as the bridge between the A2A protocol (managed by the request handler and server application) and your agent’s specific logic. It receives context about the request and uses an event queue to send results or updates back. For what the task states mean, see Life of a Task .

Next step

With an Agent Card and an executor in place, you can start the server .

Source: A2A official documentation · The Agent Executor