Merge remote-tracking branch 'upstream/main' into fix/keep-follow-up-message-on-current-task

Author: marcinbelczewski

.github/actions/spelling/allow.txt

@@ -115,6 +115,7 @@ opensource
 otherurl
 outerjoin
 pb2
+PLW
 podman
 Podman
 poolclass

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-  ".": "1.2.3"
+  ".": "1.3.0"
 }

CHANGELOG.md

@@ -1,5 +1,39 @@
 # Changelog
 
+## [1.3.0](https://github.com/a2aproject/a2a-python/compare/v1.2.3...v1.3.0) (2026-10-27)
+
+
+### Features
+
+* **server:** release parked interrupted tasks after a configurable timeout ([#1338](https://github.com/a2aproject/a2a-python/issues/1338)) ([b22a68d](https://github.com/a2aproject/a2a-python/commit/b22a68d242c32edddaa8953d0040ca82897e0b9b))
+
+
+### Bug Fixes
+
+* **client:** Create separate httpx per client when none configured ([#1358](https://github.com/a2aproject/a2a-python/issues/1358)) ([d316d0e](https://github.com/a2aproject/a2a-python/commit/d316d0e85e4d229df261d0cee0e9d9c0e5263207))
+* **client:** raise A2A errors for malformed JSON-RPC responses ([#1375](https://github.com/a2aproject/a2a-python/issues/1375)) ([7e55413](https://github.com/a2aproject/a2a-python/commit/7e554137514b693cc952e6544a9eaf36c7d56643))
+* **client:** rejections from unrecognized fields ([#1354](https://github.com/a2aproject/a2a-python/issues/1354)) ([91929cd](https://github.com/a2aproject/a2a-python/commit/91929cda989caba089584ceafa7796583f25bec4))
+* **compat:** keep an unset v0.3 status message and provider as None ([#1362](https://github.com/a2aproject/a2a-python/issues/1362)) ([cdb4799](https://github.com/a2aproject/a2a-python/commit/cdb479957e81600f68f946f8ff0122667673cb35)), closes [#1361](https://github.com/a2aproject/a2a-python/issues/1361)
+* **compat:** return A2A error codes from the v0.3 JSON-RPC adapter ([#1369](https://github.com/a2aproject/a2a-python/issues/1369)) ([8e32d53](https://github.com/a2aproject/a2a-python/commit/8e32d53173f8924beb7df0d3210e28a9fe9020f9))
+* **compat:** set last_updated when storing a task in v0.3 format ([#1372](https://github.com/a2aproject/a2a-python/issues/1372)) ([30b6d50](https://github.com/a2aproject/a2a-python/commit/30b6d5067a9256447e3d5324163b091fc4bbad74)), closes [#1371](https://github.com/a2aproject/a2a-python/issues/1371)
+* **core:** restore structured __repr__ on A2A exceptions ([#1348](https://github.com/a2aproject/a2a-python/issues/1348)) ([79a6f45](https://github.com/a2aproject/a2a-python/commit/79a6f451f49fdeaec099d723542363b5bc6dead7))
+* **server:** allow recovery after status event validation fails ([#1365](https://github.com/a2aproject/a2a-python/issues/1365)) ([ca3ef0d](https://github.com/a2aproject/a2a-python/commit/ca3ef0d6c63384228db88d4984536a8189831d08))
+* **server:** apply history_length to legacy streaming sends ([#1364](https://github.com/a2aproject/a2a-python/issues/1364)) ([261c074](https://github.com/a2aproject/a2a-python/commit/261c074aa4de522dfa609882666c98c5bd1a52d0))
+* **server:** include legacy task states in database list filters ([#1337](https://github.com/a2aproject/a2a-python/issues/1337)) ([e12d494](https://github.com/a2aproject/a2a-python/commit/e12d494b4136351d5b5d07b5c6beae95a500fc4e))
+* **server:** record executor failures through the event consumer ([#1366](https://github.com/a2aproject/a2a-python/issues/1366)) ([ad85806](https://github.com/a2aproject/a2a-python/commit/ad85806d9e43249eef5e3ad9724b0a5540cf99fb))
+* **server:** reject boolean JSON-RPC request IDs ([#1350](https://github.com/a2aproject/a2a-python/issues/1350)) ([05392b3](https://github.com/a2aproject/a2a-python/commit/05392b373f70fea18349b8c66bfffb669b3b1137))
+* skip malformed agent card signature headers ([#1335](https://github.com/a2aproject/a2a-python/issues/1335)) ([b21a144](https://github.com/a2aproject/a2a-python/commit/b21a1445c8e3a4dcd9edfa1d23e7e6af8382d8b7))
+
+
+### Documentation
+
+* **helpers:** note that whole numbers in a data Part come back as floats ([#1373](https://github.com/a2aproject/a2a-python/issues/1373)) ([654774f](https://github.com/a2aproject/a2a-python/commit/654774f3d43b471e17fc2c6901998c71015ec450))
+
+
+### Miscellaneous Chores
+
+* release 1.3.0 ([#1389](https://github.com/a2aproject/a2a-python/issues/1389)) ([fb6373f](https://github.com/a2aproject/a2a-python/commit/fb6373f2e4bad3c9e6130325dd2ba092749f8998))
+
 ## [1.2.3](https://github.com/a2aproject/a2a-python/compare/v1.2.2...v1.2.3) (2026-10-17)
 
 

src/a2a/client/base_client.py

@@ -1,4 +1,5 @@
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable
+from contextlib import aclosing
 from typing import Any
 
 from a2a.client.client import (
@@ -12,6 +13,8 @@
     ClientCallInterceptor,
 )
 from a2a.client.transports.base import ClientTransport
+from a2a.plugins.manager import PluginManager, get_global
+from a2a.plugins.types import ClientRequest, ClientResponse
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
@@ -32,20 +35,37 @@
 
 
 class BaseClient(Client):
-    """Base implementation of the A2A client, containing transport-independent logic."""
+    """Base implementation of the A2A client, containing transport-independent logic.
+
+    Client plugin hooks run around every call: ``before_client_request`` before
+    the interceptors, ``after_client_request`` after them.
+    """
 
     def __init__(
         self,
         card: AgentCard,
         config: ClientConfig,
         transport: ClientTransport,
         interceptors: list[ClientCallInterceptor],
+        *,
+        plugins: PluginManager | None = None,
     ):
+        """Initializes the client.
+
+        Args:
+            card: The card of the agent to call.
+            config: The client configuration.
+            transport: The transport used for calls.
+            interceptors: Interceptors applied to every call.
+            plugins: Plugin manager whose client hooks run for every call.
+                Defaults to the global manager at construction time.
+        """
         super().__init__(interceptors)
         self._card = card
         self._config = config
         self._transport = transport
         self._interceptors = interceptors
+        self._plugins = plugins if plugins is not None else get_global()
 
     async def send_message(
         self,
@@ -90,15 +110,20 @@ async def send_message(
             yield stream_response
             return
 
-        async for event in self._execute_stream_with_interceptors(
-            input_data=request,
-            method='send_message_streaming',
-            context=context,
-            transport_call=lambda req, ctx: (
-                self._transport.send_message_streaming(req, context=ctx)
-            ),
-        ):
-            yield event
+        # aclosing() closes the inner stream (and the transport stream) as
+        # soon as the caller stops iterating.
+        async with aclosing(
+            self._execute_stream_with_interceptors(
+                input_data=request,
+                method='send_message_streaming',
+                context=context,
+                transport_call=lambda req, ctx: (
+                    self._transport.send_message_streaming(req, context=ctx)
+                ),
+            )
+        ) as events:
+            async for event in events:
+                yield event
 
     def _apply_client_config(self, request: SendMessageRequest) -> None:
         request.configuration.return_immediately |= self._config.polling
@@ -324,15 +349,18 @@ async def subscribe(
                 'client and/or server do not support resubscription.'
             )
 
-        async for event in self._execute_stream_with_interceptors(
-            input_data=request,
-            method='subscribe',
-            context=context,
-            transport_call=lambda req, ctx: self._transport.subscribe(
-                req, context=ctx
-            ),
-        ):
-            yield event
+        async with aclosing(
+            self._execute_stream_with_interceptors(
+                input_data=request,
+                method='subscribe',
+                context=context,
+                transport_call=lambda req, ctx: self._transport.subscribe(
+                    req, context=ctx
+                ),
+            )
+        ) as events:
+            async for event in events:
+                yield event
 
     async def get_extended_agent_card(
         self,
@@ -372,6 +400,19 @@ async def close(self) -> None:
         """Closes the underlying transport."""
         await self._transport.close()
 
+    def _client_request(
+        self,
+        input_data: Any,
+        method: str,
+        context: ClientCallContext | None,
+    ) -> ClientRequest:
+        return ClientRequest(
+            method=method,
+            payload=input_data,
+            agent_card=self._card,
+            context=context if context is not None else ClientCallContext(),
+        )
+
     async def _execute_with_interceptors(
         self,
         input_data: Any,
@@ -380,6 +421,41 @@ async def _execute_with_interceptors(
         transport_call: Callable[
             [Any, ClientCallContext | None], Awaitable[Any]
         ],
+    ) -> Any:
+        """Runs a unary call through plugins, interceptors and the transport.
+
+        ``method`` is reported to both plugins and interceptors.
+        """
+        plugins = self._plugins
+        if not plugins.has_client_hooks:
+            return await self._call_through_interceptors(
+                input_data, method, context, transport_call
+            )
+
+        request = self._client_request(input_data, method, context)
+        await plugins.before_client_request(request)
+        try:
+            result = await self._call_through_interceptors(
+                request.payload, method, request.context, transport_call
+            )
+        except Exception as e:
+            await plugins.after_client_request(
+                ClientResponse(request.method, None, request.context, error=e)
+            )
+            raise
+        await plugins.after_client_request(
+            ClientResponse(request.method, result, request.context)
+        )
+        return result
+
+    async def _call_through_interceptors(
+        self,
+        input_data: Any,
+        method: str,
+        context: ClientCallContext | None,
+        transport_call: Callable[
+            [Any, ClientCallContext | None], Awaitable[Any]
+        ],
     ) -> Any:
         before_args = BeforeArgs(
             input=input_data,
@@ -422,8 +498,62 @@ async def _execute_stream_with_interceptors(
         transport_call: Callable[
             [Any, ClientCallContext | None], AsyncIterator[StreamResponse]
         ],
-    ) -> AsyncIterator[StreamResponse]:
+    ) -> AsyncGenerator[StreamResponse, None]:
+        """Runs a streaming call through plugins, interceptors and the transport.
 
+        ``method`` is reported to both plugins and interceptors.
+        """
+        plugins = self._plugins
+        if not plugins.has_client_hooks:
+            async with aclosing(
+                self._stream_through_interceptors(
+                    input_data, method, context, transport_call
+                )
+            ) as events:
+                async for event in events:
+                    yield event
+            return
+
+        request = self._client_request(input_data, method, context)
+        await plugins.before_client_request(request)
+        error: BaseException | None = None
+        try:
+            async with aclosing(
+                self._stream_through_interceptors(
+                    request.payload, method, request.context, transport_call
+                )
+            ) as events:
+                async for event in events:
+                    await plugins.after_client_request(
+                        ClientResponse(
+                            request.method,
+                            event,
+                            request.context,
+                            is_stream_event=True,
+                            is_final=False,
+                        )
+                    )
+                    yield event
+        except Exception as e:
+            error = e
+            raise
+        finally:
+            # Also runs when the caller stops iterating early.
+            await plugins.after_client_request(
+                ClientResponse(
+                    request.method, None, request.context, error=error
+                )
+            )
+
+    async def _stream_through_interceptors(
+        self,
+        input_data: Any,
+        method: str,
+        context: ClientCallContext | None,
+        transport_call: Callable[
+            [Any, ClientCallContext | None], AsyncIterator[StreamResponse]
+        ],
+    ) -> AsyncGenerator[StreamResponse, None]:
         before_args = BeforeArgs(
             input=input_data,
             method=method,