Merge remote-tracking branch 'origin/1.0-dev' into ishymko/763-client-timeouts

Author: ishymko

pyproject.toml

@@ -176,6 +176,7 @@ omit = [
     "*/__init__.py",
     "src/a2a/types/a2a_pb2.py",
     "src/a2a/types/a2a_pb2_grpc.py",
+    "src/a2a/compat/*/*_pb2*.py",
 ]
 
 [tool.coverage.report]

src/a2a/client/client_factory.py

@@ -14,6 +14,7 @@
 from a2a.client.transports.base import ClientTransport
 from a2a.client.transports.jsonrpc import JsonRpcTransport
 from a2a.client.transports.rest import RestTransport
+from a2a.client.transports.tenant_decorator import TenantTransportDecorator
 from a2a.types.a2a_pb2 import (
     AgentCapabilities,
     AgentCard,
@@ -216,28 +217,27 @@ def create(
             TransportProtocol.JSONRPC
         ]
         transport_protocol = None
-        transport_url = None
+        selected_interface = None
         if self._config.use_client_preference:
             for protocol_binding in client_set:
-                supported_interface = next(
+                selected_interface = next(
                     (
                         si
                         for si in card.supported_interfaces
                         if si.protocol_binding == protocol_binding
                     ),
                     None,
                 )
-                if supported_interface:
+                if selected_interface:
                     transport_protocol = protocol_binding
-                    transport_url = supported_interface.url
                     break
         else:
             for supported_interface in card.supported_interfaces:
                 if supported_interface.protocol_binding in client_set:
                     transport_protocol = supported_interface.protocol_binding
-                    transport_url = supported_interface.url
+                    selected_interface = supported_interface
                     break
-        if not transport_protocol or not transport_url:
+        if not transport_protocol or not selected_interface:
             raise ValueError('no compatible transports found.')
         if transport_protocol not in self._registry:
             raise ValueError(f'no client available for {transport_protocol}')
@@ -252,9 +252,14 @@ def create(
             self._config.extensions = all_extensions
 
         transport = self._registry[transport_protocol](
-            card, transport_url, self._config, interceptors or []
+            card, selected_interface.url, self._config, interceptors or []
         )
 
+        if selected_interface.tenant:
+            transport = TenantTransportDecorator(
+                transport, selected_interface.tenant
+            )
+
         return BaseClient(
             card,
             self._config,

src/a2a/client/transports/grpc.py

@@ -345,6 +345,6 @@ async def _call_grpc_stream(
         )
         while True:
             response = await stream.read()
-            if response == grpc.aio.EOF:
+            if response == grpc.aio.EOF:  # pyright: ignore[reportAttributeAccessIssue]
                 break
             yield response

src/a2a/client/transports/jsonrpc.py

@@ -458,11 +458,11 @@ async def _apply_interceptors(
 
     def _get_http_args(
         self, context: ClientCallContext | None
-    ) -> dict[str, Any] | None:
+    ) -> dict[str, Any]:
         http_kwargs: dict[str, Any] = {}
         if context and context.timeout is not None:
             http_kwargs['timeout'] = httpx.Timeout(context.timeout)
-        return http_kwargs if http_kwargs else None
+        return http_kwargs
 
     def _create_jsonrpc_error(self, error_dict: dict[str, Any]) -> Exception:
         """Creates the appropriate A2AError from a JSON-RPC error dictionary."""

src/a2a/client/transports/rest.py

@@ -79,7 +79,7 @@ async def send_message(
             request, context, extensions
         )
         response_data = await self._send_post_request(
-            '/message:send', payload, modified_kwargs
+            '/message:send', request.tenant, payload, modified_kwargs
         )
         response: SendMessageResponse = ParseDict(
             response_data, SendMessageResponse()
@@ -97,10 +97,10 @@ async def send_message_streaming(
         payload, modified_kwargs = await self._prepare_send_message(
             request, context, extensions
         )
-
         async for event in self._send_stream_request(
             'POST',
             '/message:stream',
+            request.tenant,
             http_kwargs=modified_kwargs,
             json=payload,
         ):
@@ -130,6 +130,7 @@ async def get_task(
 
         response_data = await self._send_get_request(
             f'/tasks/{request.id}',
+            request.tenant,
             params,
             modified_kwargs,
         )
@@ -153,8 +154,10 @@ async def list_tasks(
             modified_kwargs,
             extensions if extensions is not None else self.extensions,
         )
+
         response_data = await self._send_get_request(
             '/tasks',
+            request.tenant,
             _model_to_query_params(request),
             modified_kwargs,
         )
@@ -181,8 +184,12 @@ async def cancel_task(
             modified_kwargs,
             context,
         )
+
         response_data = await self._send_post_request(
-            f'/tasks/{request.id}:cancel', payload, modified_kwargs
+            f'/tasks/{request.id}:cancel',
+            request.tenant,
+            payload,
+            modified_kwargs,
         )
         response: Task = ParseDict(response_data, Task())
         return response
@@ -203,8 +210,10 @@ async def create_task_push_notification_config(
         payload, modified_kwargs = await self._apply_interceptors(
             payload, modified_kwargs, context
         )
+
         response_data = await self._send_post_request(
             f'/tasks/{request.task_id}/pushNotificationConfigs',
+            request.tenant,
             payload,
             modified_kwargs,
         )
@@ -235,8 +244,10 @@ async def get_task_push_notification_config(
             del params['id']
         if 'task_id' in params:
             del params['task_id']
+
         response_data = await self._send_get_request(
             f'/tasks/{request.task_id}/pushNotificationConfigs/{request.id}',
+            request.tenant,
             params,
             modified_kwargs,
         )
@@ -265,8 +276,10 @@ async def list_task_push_notification_configs(
         )
         if 'task_id' in params:
             del params['task_id']
+
         response_data = await self._send_get_request(
             f'/tasks/{request.task_id}/pushNotificationConfigs',
+            request.tenant,
             params,
             modified_kwargs,
         )
@@ -297,8 +310,10 @@ async def delete_task_push_notification_config(
             del params['id']
         if 'task_id' in params:
             del params['task_id']
+
         await self._send_delete_request(
             f'/tasks/{request.task_id}/pushNotificationConfigs/{request.id}',
+            request.tenant,
             params,
             modified_kwargs,
         )
@@ -319,6 +334,7 @@ async def subscribe(
         async for event in self._send_stream_request(
             'GET',
             f'/tasks/{request.id}:subscribe',
+            request.tenant,
             http_kwargs=modified_kwargs,
         ):
             yield event
@@ -347,7 +363,7 @@ async def get_extended_agent_card(
             context,
         )
         response_data = await self._send_get_request(
-            '/extendedAgentCard', {}, modified_kwargs
+            '/extendedAgentCard', request.tenant, {}, modified_kwargs
         )
         response: AgentCard = ParseDict(response_data, AgentCard())
 
@@ -363,6 +379,10 @@ async def close(self) -> None:
         """Closes the httpx client."""
         await self.httpx_client.aclose()
 
+    def _get_path(self, base_path: str, tenant: str) -> str:
+        """Returns the full path, prepending the tenant if provided."""
+        return f'/{tenant}{base_path}' if tenant else base_path
+
     async def _apply_interceptors(
         self,
         request_payload: dict[str, Any],
@@ -376,7 +396,7 @@ async def _apply_interceptors(
 
     def _get_http_args(
         self, context: ClientCallContext | None
-    ) -> dict[str, Any] | None:
+    ) -> dict[str, Any]:
         http_kwargs: dict[str, Any] = {}
         if context and context.timeout is not None:
             http_kwargs['timeout'] = httpx.Timeout(context.timeout)
@@ -428,16 +448,18 @@ async def _send_stream_request(
         self,
         method: str,
         target: str,
+        tenant: str,
         http_kwargs: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> AsyncGenerator[StreamResponse]:
         final_kwargs = dict(http_kwargs or {})
         final_kwargs.update(kwargs)
+        path = self._get_path(target, tenant)
 
         async for sse_data in send_http_stream_request(
             self.httpx_client,
             method,
-            f'{self.url}{target}',
+            f'{self.url}{path}',
             self._handle_http_error,
             **final_kwargs,
         ):
@@ -452,13 +474,15 @@ async def _send_request(self, request: httpx.Request) -> dict[str, Any]:
     async def _send_post_request(
         self,
         target: str,
+        tenant: str,
         rpc_request_payload: dict[str, Any],
         http_kwargs: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
+        path = self._get_path(target, tenant)
         return await self._send_request(
             self.httpx_client.build_request(
                 'POST',
-                f'{self.url}{target}',
+                f'{self.url}{path}',
                 json=rpc_request_payload,
                 **(http_kwargs or {}),
             )
@@ -467,13 +491,15 @@ async def _send_post_request(
     async def _send_get_request(
         self,
         target: str,
+        tenant: str,
         query_params: dict[str, str],
         http_kwargs: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
+        path = self._get_path(target, tenant)
         return await self._send_request(
             self.httpx_client.build_request(
                 'GET',
-                f'{self.url}{target}',
+                f'{self.url}{path}',
                 params=query_params,
                 **(http_kwargs or {}),
             )
@@ -482,13 +508,15 @@ async def _send_get_request(
     async def _send_delete_request(
         self,
         target: str,
+        tenant: str,
         query_params: dict[str, Any],
         http_kwargs: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
+        path = self._get_path(target, tenant)
         return await self._send_request(
             self.httpx_client.build_request(
                 'DELETE',
-                f'{self.url}{target}',
+                f'{self.url}{path}',
                 params=query_params,
                 **(http_kwargs or {}),
             )