feat: handle tenant in Client

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/rest.py

@@ -79,7 +79,7 @@
             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 @@
         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 @@
 
         response_data = await self._send_get_request(
             f'/tasks/{request.id}',
+            request.tenant,
             params,
             modified_kwargs,
         )
@@ -153,8 +154,10 @@
             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 @@
             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 @@
         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,
         )
@@ -221,22 +230,24 @@
         extensions: list[str] | None = None,
     ) -> TaskPushNotificationConfig:
         """Retrieves the push notification configuration for a specific task."""
         params = MessageToDict(request)
         modified_kwargs = update_extension_header(
             self._get_http_args(context),
             extensions if extensions is not None else self.extensions,
         )
         params, modified_kwargs = await self._apply_interceptors(
             params,
             modified_kwargs,
             context,
         )
         if 'id' in params:
             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 @@
         )
         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,
         )
@@ -283,22 +296,24 @@
         extensions: list[str] | None = None,
     ) -> None:
         """Deletes the push notification configuration for a specific task."""
         params = MessageToDict(request)
         modified_kwargs = update_extension_header(
             self._get_http_args(context),
             extensions if extensions is not None else self.extensions,
         )
         params, modified_kwargs = await self._apply_interceptors(
             params,
             modified_kwargs,
             context,
         )
         if 'id' in params:
             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,20 +334,21 @@
         async for event in self._send_stream_request(
             'GET',
             f'/tasks/{request.id}:subscribe',
+            request.tenant,
             http_kwargs=modified_kwargs,
         ):
             yield event
 
     async def get_extended_agent_card(
         self,
         request: GetExtendedAgentCardRequest,
         *,
         context: ClientCallContext | None = None,
         extensions: list[str] | None = None,
         signature_verifier: Callable[[AgentCard], None] | None = None,
     ) -> AgentCard:
         """Retrieves the Extended AgentCard."""
         modified_kwargs = update_extension_header(
             self._get_http_args(context),
             extensions if extensions is not None else self.extensions,
         )
@@ -347,7 +363,7 @@
             context,
         )
         response_data = await self._send_get_request(
-            '/extendedAgentCard', {}, modified_kwargs
+            '/extendedAgentCard', request.tenant, {}, modified_kwargs
         )
         response: AgentCard = ParseDict(response_data, AgentCard())
 
@@ -363,15 +379,19 @@
         """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],
         http_kwargs: dict[str, Any] | None,
         context: ClientCallContext | None,
     ) -> tuple[dict[str, Any], dict[str, Any]]:
         final_http_kwargs = http_kwargs or {}
         final_request_payload = request_payload
         # TODO: Implement interceptors for other transports
         return final_request_payload, final_http_kwargs
 
     def _get_http_args(
@@ -425,16 +445,18 @@
         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,
         ):
@@ -449,13 +471,15 @@
     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 {}),
             )
@@ -464,13 +488,15 @@
     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 {}),
             )
@@ -479,13 +505,15 @@
     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 {}),
             )