Gemini authored: update tests

Author: mikeas1

src/a2a/client/base_client.py

@@ -73,7 +73,9 @@ async def send_message(
         params = MessageSendParams(message=request, configuration=config)
 
         if not self._config.streaming or not self._card.capabilities.streaming:
-            response = await self._transport.send_message(params, context=context)
+            response = await self._transport.send_message(
+                params, context=context
+            )
             result = (
                 (response, None) if isinstance(response, Task) else response
             )
@@ -85,6 +87,9 @@ async def send_message(
         stream = self._transport.send_message_streaming(params, context=context)
 
         first_event = await anext(stream)
+        # The response from a server may be either exactly one Message or a
+        # series of Task updates. Separate out the first message for special
+        # case handling, which allows us to simplify further stream processing.
         if isinstance(first_event, Message):
             await self.consume(first_event, self._card)
             yield first_event
@@ -102,7 +107,7 @@ async def _process_response(
     ) -> ClientEvent:
         if isinstance(event, Message):
             raise A2AClientInvalidStateError(
-                "received a streamed Message from server after first response; this is not supported"
+                'received a streamed Message from server after first response; this is not supported'
             )
         await tracker.process(event)
         task = tracker.get_task_or_raise()
@@ -201,11 +206,16 @@ async def resubscribe(
         """
         if not self._config.streaming or not self._card.capabilities.streaming:
             raise NotImplementedError(
-                "client and/or server do not support resubscription."
+                'client and/or server do not support resubscription.'
             )
 
         tracker = ClientTaskManager()
-        async for event in self._transport.resubscribe(request, context=context):
+        # Note: resubscribe can only be called on an existing task. As such,
+        # we should never see Message updates, despite the typing of the service
+        # definition indicating it may be possible.
+        async for event in self._transport.resubscribe(
+            request, context=context
+        ):
             yield await self._process_response(tracker, event)
 
     async def get_card(

src/a2a/client/client.py

@@ -119,6 +119,8 @@ async def send_message(
         pairs, or a `Message`. Client will also send these values to any
         configured `Consumer`s in the client.
         """
+        return
+        yield
 
     @abstractmethod
     async def get_task(
@@ -164,6 +166,8 @@ async def resubscribe(
         context: ClientCallContext | None = None,
     ) -> AsyncIterator[ClientEvent]:
         """Resubscribes to a task's event stream."""
+        return
+        yield
 
     @abstractmethod
     async def get_card(

src/a2a/client/client_factory.py

@@ -111,9 +111,9 @@ def create(
                     transport_protocol = x
                     break
         if not transport_protocol:
-            raise ValueError("no compatible transports found.")
+            raise ValueError('no compatible transports found.')
         if transport_protocol not in self._registry:
-            raise ValueError(f"no client available for {transport_protocol}")
+            raise ValueError(f'no client available for {transport_protocol}')
 
         all_consumers = self._consumers.copy()
         if consumers:
@@ -146,8 +146,8 @@ def minimal_agent_card(
         capabilities=AgentCapabilities(),
         default_input_modes=[],
         default_output_modes=[],
-        description="",
+        description='',
         skills=[],
-        version="",
-        name="",
+        version='',
+        name='',
     )

src/a2a/client/transports/base.py

@@ -38,6 +38,7 @@ async def send_message_streaming(
         Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent
     ]:
         """Sends a streaming message request to the agent and yields responses as they arrive."""
+        return
         yield
 
     @abstractmethod
@@ -86,6 +87,7 @@ async def resubscribe(
         Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent
     ]:
         """Reconnects to get task updates."""
+        return
         yield
 
     @abstractmethod

src/a2a/client/transports/grpc.py

@@ -7,8 +7,8 @@
     import grpc
 except ImportError as e:
     raise ImportError(
-        "A2AGrpcClient requires grpcio and grpcio-tools to be installed. "
-        "Install with: "
+        'A2AGrpcClient requires grpcio and grpcio-tools to be installed. '
+        'Install with: '
         "'pip install a2a-sdk[grpc]'"
     ) from e
 
@@ -47,7 +47,9 @@ def __init__(
         self.agent_card = agent_card
         self.stub = grpc_stub
         self._needs_extended_card = (
-            agent_card.supports_authenticated_extended_card if agent_card else True
+            agent_card.supports_authenticated_extended_card
+            if agent_card
+            else True
         )
 
     async def send_message(
@@ -101,7 +103,7 @@ async def resubscribe(
     ]:
         """Reconnects to get task updates."""
         stream = self.stub.TaskSubscription(
-            a2a_pb2.TaskSubscriptionRequest(name=f"tasks/{request.id}")
+            a2a_pb2.TaskSubscriptionRequest(name=f'tasks/{request.id}')
         )
         while True:
             response = await stream.read()
@@ -117,7 +119,7 @@ async def get_task(
     ) -> Task:
         """Retrieves the current state and history of a specific task."""
         task = await self.stub.GetTask(
-            a2a_pb2.GetTaskRequest(name=f"tasks/{request.id}")
+            a2a_pb2.GetTaskRequest(name=f'tasks/{request.id}')
         )
         return proto_utils.FromProto.task(task)
 
@@ -129,7 +131,7 @@ async def cancel_task(
     ) -> Task:
         """Requests the agent to cancel a specific task."""
         task = await self.stub.CancelTask(
-            a2a_pb2.CancelTaskRequest(name=f"tasks/{request.id}")
+            a2a_pb2.CancelTaskRequest(name=f'tasks/{request.id}')
         )
         return proto_utils.FromProto.task(task)
 
@@ -142,9 +144,11 @@ async def set_task_callback(
         """Sets or updates the push notification configuration for a specific task."""
         config = await self.stub.CreateTaskPushNotificationConfig(
             a2a_pb2.CreateTaskPushNotificationConfigRequest(
-                parent="",
-                config_id="",
-                config=proto_utils.ToProto.task_push_notification_config(request),
+                parent='',
+                config_id='',
+                config=proto_utils.ToProto.task_push_notification_config(
+                    request
+                ),
             )
         )
         return proto_utils.FromProto.task_push_notification_config(config)
@@ -158,7 +162,7 @@ async def get_task_callback(
         """Retrieves the push notification configuration for a specific task."""
         config = await self.stub.GetTaskPushNotificationConfig(
             a2a_pb2.GetTaskPushNotificationConfigRequest(
-                name=f"tasks/{request.id}/pushNotification/{request.push_notification_config_id}",
+                name=f'tasks/{request.id}/pushNotification/{request.push_notification_config_id}',
             )
         )
         return proto_utils.FromProto.task_push_notification_config(config)
@@ -170,11 +174,10 @@ async def get_card(
     ) -> AgentCard:
         """Retrieves the agent's card."""
         card = self.agent_card
-        if card is None and not self._needs_extended_card:
-            raise ValueError("Agent card is not available.")
-
-        if not self._needs_extended_card:
+        if card and not self._needs_extended_card:
             return card
+        if card is None and not self._needs_extended_card:
+            raise ValueError('Agent card is not available.')
 
         card_pb = await self.stub.GetAgentCard(
             a2a_pb2.GetAgentCardRequest(),
@@ -186,5 +189,5 @@ async def get_card(
 
     async def close(self) -> None:
         """Closes the gRPC channel."""
-        if hasattr(self.stub, "close"):
+        if hasattr(self.stub, 'close'):
             await self.stub.close()