feat(server): add async context manager support to EventQueue (#743)

# Description

Thank you for opening a Pull Request!
Before submitting your PR, there are a few things you can do to make
sure it goes smoothly:

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [x] Ensure the tests and linter pass (Run `bash scripts/format.sh`
from the repository root to format)
- [x] Appropriate docs were updated (if necessary)

Fixes #720 🦕

## Problem

`EventQueue` has a sophisticated `close()` with graceful/immediate
modes, child propagation, and cross-version handling — but doesn't
support `async with`. Server-side code must use explicit `try/finally`
or risk leaking resources on exceptions:

```python
queue = EventQueue()
try:
    await queue.enqueue_event(event)
    ...
finally:
    await queue.close()
```

## Fix

Add `__aenter__` and `__aexit__` as concrete methods on `EventQueue`:

* `__aenter__` returns `Self` (via `typing_extensions`).
* `__aexit__` delegates to `close()` with default `immediate=False`
(graceful). Code needing immediate shutdown can still call `await
queue.close(immediate=True)` explicitly.

This enables the idiomatic pattern:

```python
async with EventQueue() as queue:
    await queue.enqueue_event(event)
    ...
# close() called automatically, even on exceptions
```

Unlike the client-side hierarchy where `__aenter__`/`__aexit__` were
lifted to the abstract `Client` (#719), `EventQueue` is a concrete class
with no abstract base above it — `QueueManager` manages queue lifecycles
by task ID but does not wrap or extend `EventQueue`. This is the correct
and only place for these methods.

Non-breaking, additive change. Manual `close()` and `try/finally`
continue to work as before.

Follows the pattern established in `ClientTransport` (#682),
`BaseClient` (#688), and `Client` (#719).

## Tests

Two tests added to `tests/server/events/test_event_queue.py`, following
the same approach as `ClientTransport` and `BaseClient`.

Author: cchinchilla-dev

src/a2a/server/events/event_queue.py

@@ -2,6 +2,10 @@
 import logging
 import sys
 
+from types import TracebackType
+
+from typing_extensions import Self
+
 from a2a.types.a2a_pb2 import (
     Message,
     Task,
@@ -43,6 +47,19 @@ def __init__(self, max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE) -> None:
         self._lock = asyncio.Lock()
         logger.debug('EventQueue initialized.')
 
+    async def __aenter__(self) -> Self:
+        """Enters the async context manager, returning the queue itself."""
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Exits the async context manager, ensuring close() is called."""
+        await self.close()
+
     async def enqueue_event(self, event: Event) -> None:
         """Enqueues an event to this queue and all its children.
 

tests/server/events/test_event_queue.py

@@ -78,6 +78,28 @@ def test_constructor_invalid_max_queue_size() -> None:
         EventQueue(max_queue_size=-10)
 
 
+@pytest.mark.asyncio
+async def test_event_queue_async_context_manager(
+    event_queue: EventQueue,
+) -> None:
+    """Test that EventQueue can be used as an async context manager."""
+    async with event_queue as q:
+        assert q is event_queue
+        assert event_queue.is_closed() is False
+    assert event_queue.is_closed() is True
+
+
+@pytest.mark.asyncio
+async def test_event_queue_async_context_manager_on_exception(
+    event_queue: EventQueue,
+) -> None:
+    """Test that close() is called even when an exception occurs inside the context."""
+    with pytest.raises(RuntimeError, match='boom'):
+        async with event_queue:
+            raise RuntimeError('boom')
+    assert event_queue.is_closed() is True
+
+
 @pytest.mark.asyncio
 async def test_enqueue_and_dequeue_event(event_queue: EventQueue) -> None:
     """Test that an event can be enqueued and dequeued."""