a2a.server.events.event_queue module

a2a.server.events.event_queue.Event = a2a_pb2.Message | a2a_pb2.Task | a2a_pb2.TaskStatusUpdateEvent | a2a_pb2.TaskArtifactUpdateEvent

Type alias for events that can be enqueued.

class a2a.server.events.event_queue.EventQueue(*args: Any, **kwargs: Any)

Bases: ABC

Base class and factory for EventQueueSource.

EventQueue provides an abstraction for a queue of events that can be tapped by multiple consumers. EventQueue maintain main queue and source and maintain child queues in sync. GUARANTEE: All sinks (including the default one) will receive events in the exact same order.

WARNING (Concurrency): All events from all sinks (both the default queue and any tapped child queues) must be regularly consumed and marked as done. If any single consumer stops processing and its queue reaches capacity, it can block the event dispatcher and stall the entire system, causing a widespread deadlock.

WARNING (Memory Leak): Event queues spawn background tasks. To prevent memory and task leaks, all queue objects (both source and sinks) MUST be explicitly closed via await queue.close() or by using the async context manager (async with queue:). Child queues are automatically closed when parent queue is closed, but you should still close them explicitly to prevent queues from reaching capacity by unconsumed events.

Typical usage: queue = EventQueue() child_queue1 = await queue.tap() child_queue2 = await queue.tap()

async for event in child_queue1:

do_some_work(event) child_queue1.task_done()

abstractmethod async enqueue_event(event: Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent) → None

Pushes an event into the queue.

Only main queue can enqueue events. Child queues can only dequeue events.

class a2a.server.events.event_queue.EventQueueLegacy(*args: Any, **kwargs: Any)

Bases: EventQueue

Event queue for A2A responses from agent.

Acts as a buffer between the agent’s asynchronous execution and the server’s response handling (e.g., streaming via SSE). Supports tapping to create child queues that receive the same events.

async close(immediate: bool = False) → None

Closes the queue for future push events and also closes all child queues.

Parameters:

immediate – If True, immediately flushes the queue, discarding all pending events, and causes any currently blocked dequeue_event calls to raise QueueShutDown. If False (default), the queue is marked as closed to new events, but existing events can still be dequeued and processed until the queue is fully drained.

async dequeue_event() → Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent

Dequeues an event from the queue.

This implementation expects that dequeue to raise an exception when the queue has been closed. In python 3.13+ this is naturally provided by the QueueShutDown exception generated when the queue has closed and the user is awaiting the queue.get method. Python<=3.12 this needs to manage this lifecycle itself. The current implementation can lead to blocking if the dequeue_event is called before the EventQueue has been closed but when there are no events on the queue. One way to avoid this is to use an async Task management solution to cancel the get task if the queue has closed or some other condition is met. The implementation of the EventConsumer uses an async.wait with a timeout to abort the dequeue_event call and retry, when it will return with a closed error.

Returns:

The next event from the queue.

Raises:

asyncio.QueueShutDown – If the queue has been closed and is empty.

async enqueue_event(event: Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent) → None

Enqueues an event to this queue and all its children.

Parameters:

event – The event object to enqueue.

is_closed() → bool

Checks if the queue is closed.

property queue: Queue[Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent]

[DEPRECATED] Returns the underlying asyncio.Queue.

async tap(max_queue_size: int = 1024) → EventQueueLegacy

Taps the event queue to create a new child queue that receives future events.

Returns:

A new EventQueue instance that will receive all events enqueued to this parent queue from this point forward.

task_done() → None

Signals that a formerly enqueued task is complete.

Used in conjunction with dequeue_event to track processed items.