feat: add SendLiveMessage support for group (multicast) channels

- Add A2AServiceGroupStub.SendLiveMessage using call_multicast_stream_stream,
  returning a MulticastBidiStreamHandler
- Add SRPCMulticastTransport.send_live_message: concurrently sends requests
  and yields (source, StreamResponse) tuples from all group members
- Add MulticastClient.send_live_message delegating to the transport

Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com>

Author: Tehsmash

slima2a/client_transport.py

@@ -473,6 +473,34 @@ async def get_extended_agent_card(
         async for source, response in self.stub.GetExtendedAgentCard(request):
             yield source, response
 
+    async def send_live_message(
+        self,
+        request_stream: "AsyncGenerator[Any, None]",
+        *,
+        context: ClientCallContext | None = None,
+    ) -> AsyncGenerator[tuple[Any, StreamResponse], None]:
+        """Sends a bidirectional streaming live message to all agents in the group.
+
+        Yields (source, StreamResponse) tuples as events arrive from any agent.
+        """
+        bidi = self.stub.SendLiveMessage()
+        async def _send():
+            async for req in request_stream:
+                await bidi.send_async(req.SerializeToString())
+            await bidi.close_send_async()
+        send_task = asyncio.ensure_future(_send())
+        try:
+            while True:
+                msg = await bidi.recv_async()
+                if msg.is_end():
+                    break
+                if msg.is_error():
+                    raise msg.error
+                if msg.is_data():
+                    yield msg.item.context, StreamResponse.FromString(msg.item.message)
+        finally:
+            await send_task
+
     async def close(self) -> None:
         """Closes the transport and releases any resources."""
         pass
@@ -648,6 +676,22 @@ async def get_extended_agent_card(
         ):
             yield source, response
 
+    async def send_live_message(
+        self,
+        request_stream: "AsyncGenerator[Any, None]",
+        *,
+        context: ClientCallContext | None = None,
+    ) -> AsyncGenerator[tuple[Any, StreamResponse], None]:
+        """Sends a bidirectional streaming live message to all agents in the group.
+
+        Yields (source, StreamResponse) tuples as events arrive from any agent.
+        Use ``source`` to demultiplex per-agent.
+        """
+        async for source, response in self._transport.send_live_message(
+            request_stream, context=context
+        ):
+            yield source, response
+
     async def close(self) -> None:
         await self._transport.close()
 

slima2a/types/v1/a2a_pb2_slimrpc.py

@@ -382,6 +382,14 @@ async def DeleteTaskPushNotificationConfig(self, request: a2a__pb2.DeleteTaskPus
             if msg.is_data():
                 yield msg.item.context, google__protobuf__empty_pb2.Empty.FromString(msg.item.message)
 
+    def SendLiveMessage(self, timeout: Optional[timedelta] = None, metadata: Optional[dict[str, str]] = None) -> slim_bindings.MulticastBidiStreamHandler:
+        """Open a bidirectional streaming SendLiveMessage call to all group members."""
+        return self._channel.call_multicast_stream_stream(
+            "lf.a2a.v1.A2AService",
+            "SendLiveMessage",
+            timeout,
+            metadata,
+        )
 
 
 class A2AServiceServicer: