feat: add async context manager support to ClientTransport (#682)

# 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 #674 🦕

## Problem

`ClientTransport` defines an abstract `close()` method but does not
implement `__aenter__`/`__aexit__`. This means transports cannot be used
with `async with`, which is the idiomatic Python pattern for managing
async resources. If an exception occurs between creating a transport and
calling `close()`, the underlying connection (e.g., gRPC channel, httpx
client) is never cleaned up:

```python
transport = GrpcTransport(channel=channel, agent_card=agent_card)
result = await transport.send_message(params)  # if this raises, close() is never called
await transport.close()
```

## Fix

Added `__aenter__` and `__aexit__` methods to `ClientTransport` in
`src/a2a/client/transports/base.py`:

`__aenter__` returns `self`.
`__aexit__` awaits `close()`.

This enables the standard async context manager pattern on all transport
implementations (`GrpcTransport`, `RestTransport`, `JsonRpcTransport`):

```python
async with GrpcTransport(channel=channel, agent_card=agent_card) as transport:
    result = await transport.send_message(params)
    # close() called automatically, even on exceptions
```

This is a non-breaking, additive change. Calling `close()` manually or
via `try/finally` continues to work exactly as before.

## Test

Tests were added to `tests/client/test_base_client.py` since it already
imports and mocks `ClientTransport`. Happy to move them to a dedicated
file if maintainers prefer.

## Note

As mentioned in #674, the same pattern could also be applied to
`BaseClient`, which wraps `ClientTransport` and also exposes a `close()`
method. I've kept this PR scoped to `ClientTransport` only. Happy to
extend it to `BaseClient` in this same PR or a follow-up if maintainers
prefer.

Release-As: 0.3.23

---------

Co-authored-by: Ivan Shymko <vana.shimko@gmail.com>

Author: cchinchilla-dev

src/a2a/client/transports/base.py

@@ -1,5 +1,8 @@
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Callable
+from types import TracebackType
+
+from typing_extensions import Self
 
 from a2a.client.middleware import ClientCallContext
 from a2a.types import (
@@ -19,6 +22,19 @@
 class ClientTransport(ABC):
     """Abstract base class for a client transport."""
 
+    async def __aenter__(self) -> Self:
+        """Enters the async context manager, returning the transport 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()
+
     @abstractmethod
     async def send_message(
         self,

tests/client/test_base_client.py

@@ -1,4 +1,4 @@
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -61,6 +61,32 @@ def base_client(
     )
 
 
+@pytest.mark.asyncio
+async def test_transport_async_context_manager() -> None:
+    with (
+        patch.object(ClientTransport, '__abstractmethods__', set()),
+        patch.object(ClientTransport, 'close', new_callable=AsyncMock),
+    ):
+        transport = ClientTransport()
+        async with transport as t:
+            assert t is transport
+            transport.close.assert_not_awaited()
+        transport.close.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_transport_async_context_manager_on_exception() -> None:
+    with (
+        patch.object(ClientTransport, '__abstractmethods__', set()),
+        patch.object(ClientTransport, 'close', new_callable=AsyncMock),
+    ):
+        transport = ClientTransport()
+        with pytest.raises(RuntimeError, match='boom'):
+            async with transport:
+                raise RuntimeError('boom')
+        transport.close.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_send_message_streaming(
     base_client: BaseClient, mock_transport: MagicMock, sample_message: Message