feat: add broadcast SendLiveMessage support via slim shared-responses

Adds support for the broadcast extension to SendLiveMessage as described
in the slimrpc-broadcast-live spec. Every StreamResponse from any agent
in a group channel is delivered to all other participants via SLIM's
shared-responses transport primitive (slim-bindings PR #85).

- a2a_pb2_slimrpc.py: add A2AServiceSharedServicer base class,
  _A2AServiceServicer_SendLiveMessage_SharedHandler extending
  StreamStreamSharedHandler, and add_A2AServiceServicer_to_server_shared()
- handler.py: add SRPCSharedHandler with asyncio.Queue fan-in merge of
  client request stream and translated peer StreamResponse events;
  add _translate_peer_response() implementing spec Section 5 translation
  (task/status_update/artifact_update/message_update -> StreamRequest)
- client_transport.py: add slimrpc_group_shared_channel_factory() using
  Channel.new_group_shared_with_connection; add field to ClientConfig
- __init__.py: export slimrpc_group_shared_channel_factory
- examples/echo_agent/server.py: use Server.new_with_shared_responses_and_connection
  and register both SRPCHandler and SRPCSharedHandler so the server
  accepts both standard and broadcast sessions without any flags
- examples/echo_agent/client.py: add --broadcast flag that uses the
  shared channel factory and routes to MulticastClient.send_live_message

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

Author: Tehsmash

examples/echo_agent/client.py

@@ -18,7 +18,7 @@
     AGENT_CARD_WELL_KNOWN_PATH,
 )
 
-from slima2a import setup_slim_client
+from slima2a import setup_slim_client, slimrpc_group_shared_channel_factory
 from slima2a.client_transport import (
     ClientConfig,
     MultiAgentClientFactory,
@@ -62,12 +62,18 @@ async def main() -> None:
         secret="my_shared_secret_for_testing_purposes_only",
     )
 
+    group_factory = (
+        slimrpc_group_shared_channel_factory(slim_local_app, conn_id)
+        if args.broadcast
+        else slimrpc_group_channel_factory(slim_local_app, conn_id)
+    )
     client_config = ClientConfig(
         supported_protocol_bindings=["slimrpc"],
         streaming=args.stream,
         httpx_client=httpx_client,
         slimrpc_channel_factory=slimrpc_channel_factory(slim_local_app, conn_id),
-        slimrpc_group_channel_factory=slimrpc_group_channel_factory(
+        slimrpc_group_channel_factory=group_factory,
+        slimrpc_group_shared_channel_factory=slimrpc_group_shared_channel_factory(
             slim_local_app, conn_id
         ),
     )
@@ -97,7 +103,10 @@ async def main() -> None:
 
     client = client_factory.create(card=cards)
 
-    if isinstance(client, MulticastClient):
+    if isinstance(client, MulticastClient) and args.broadcast:
+        print(f"> {args.text} (broadcast live to {agent_names})")
+        await send_live_message_broadcast(client, args.text)
+    elif isinstance(client, MulticastClient):
         print(f"> {args.text} (multicast to {agent_names})")
         await send_message_multicast(client, args.text)
     elif args.live:
@@ -133,6 +142,13 @@ def parse_arguments() -> argparse.Namespace:
         default=False,
         help="Use SendLiveMessage bidirectional streaming",
     )
+    parser.add_argument(
+        "--broadcast",
+        action="store_true",
+        required=False,
+        default=False,
+        help="Use broadcast SendLiveMessage (shared-responses) — requires multiple --agents",
+    )
     parser.add_argument(
         "--text",
         type=str,
@@ -267,5 +283,31 @@ async def _requests():
     print(output)
 
 
+async def send_live_message_broadcast(client: MulticastClient, text: str) -> None:
+    message = new_text_message(text, role=Role.ROLE_USER)
+
+    async def _requests():
+        yield StreamRequest(message=message)
+
+    try:
+        async for source, stream_response in client.send_live_message(_requests()):
+            which = stream_response.WhichOneof("payload")
+            output = ""
+            if which == "message":
+                for part in stream_response.message.parts:
+                    if part.WhichOneof("content") == "text":
+                        output += part.text
+            elif which == "artifact_update":
+                artifact = stream_response.artifact_update.artifact
+                for part in artifact.parts:
+                    if part.WhichOneof("content") == "text":
+                        output += part.text
+            if output:
+                print(f"  [{source}] {output}")
+    except Exception as e:
+        logger.error(f"failed sending broadcast live message: {e}", exc_info=True)
+        raise RuntimeError("failed sending broadcast live message") from e
+
+
 if __name__ == "__main__":
     asyncio.run(main())

examples/echo_agent/server.py

@@ -65,8 +65,8 @@ async def main() -> None:
                 secret="my_shared_secret_for_testing_purposes_only",
             )
 
-            # Create server
-            server = slim_bindings.Server.new_with_connection(
+            # Create server — always use shared-responses so broadcast clients work
+            server = slim_bindings.Server.new_with_shared_responses_and_connection(
                 local_app, local_name, conn_id
             )
 
@@ -80,13 +80,16 @@ async def main() -> None:
                 add_v0(compat_handler, server)
 
             if args.a2a_version in ("v1", "both"):
-                from slima2a.handler import SRPCHandler
+                from slima2a.handler import SRPCHandler, SRPCSharedHandler
                 from slima2a.types.v1.a2a_pb2_slimrpc import (
                     add_A2AServiceServicer_to_server as add_v1,
+                    add_A2AServiceServicer_to_server_shared as add_v1_shared,
                 )
 
                 handler = SRPCHandler(agent_card, default_request_handler)
                 add_v1(handler, server)
+                shared_handler = SRPCSharedHandler(agent_card, default_request_handler)
+                add_v1_shared(shared_handler, server)
 
             # Run server
             await server.serve_async()

pyproject.toml

@@ -47,6 +47,7 @@ default-groups = ["linting", "testing", "examples"]
 
 [tool.uv.sources]
 a2a-sdk = { git = "https://github.com/Tehsmash/a2a-python", branch = "feat/send-live-message" }
+slim-bindings = { path = "../slim-bindings/python", editable = true }
 
 [tool.ruff.format]
 exclude = ["*_pb2.py", "*_pb2.pyi", "*_pb2_slimrpc.py"]

slima2a/__init__.py

@@ -3,6 +3,7 @@
 
 """SLIMA2A - A2A protocol over slimrpc."""
 
+from slima2a.client_transport import slimrpc_group_shared_channel_factory
 from slima2a.slim_helper import (
     connect_and_subscribe,
     initialize_slim_service,
@@ -13,4 +14,5 @@
     "connect_and_subscribe",
     "initialize_slim_service",
     "setup_slim_client",
+    "slimrpc_group_shared_channel_factory",
 ]

slima2a/client_transport.py

@@ -85,12 +85,38 @@ def factory(remotes: list[str]) -> slim_bindings.Channel:
     return factory
 
 
+def slimrpc_group_shared_channel_factory(
+    local_app: slim_bindings.App,
+    conn_id: int,
+) -> Callable[[list[str]], slim_bindings.Channel]:
+    def factory(remotes: list[str]) -> slim_bindings.Channel:
+        members = []
+        for remote in remotes:
+            remote_parts = remote.split("/")
+            if len(remote_parts) != 3:
+                raise ValueError(
+                    f"Invalid remote format: '{remote}'. Expected format: 'component1/component2/component'"
+                )
+            members.append(
+                slim_bindings.Name(remote_parts[0], remote_parts[1], remote_parts[2])
+            )
+
+        return slim_bindings.Channel.new_group_shared_with_connection(
+            local_app, members, conn_id
+        )
+
+    return factory
+
+
 @dataclass
 class ClientConfig(A2AClientConfig):
     slimrpc_channel_factory: Callable[[str], slim_bindings.Channel] | None = None
     slimrpc_group_channel_factory: (
         Callable[[list[str]], slim_bindings.Channel] | None
     ) = None
+    slimrpc_group_shared_channel_factory: (
+        Callable[[list[str]], slim_bindings.Channel] | None
+    ) = None
 
 
 @trace_class(kind=SpanKind.CLIENT)

slima2a/handler.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 # ruff: noqa: N802
+import asyncio
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Callable
 
@@ -320,3 +321,137 @@ async def GetExtendedAgentCard(
         if self.card_modifier:
             card_to_serve = self.card_modifier(card_to_serve)
         return card_to_serve
+
+
+class SRPCSharedHandler(a2a_pb2_slimrpc.A2AServiceSharedServicer):
+    """Maps incoming broadcast-live SlimRPC SendLiveMessage calls to the request handler.
+
+    Peer agent StreamResponse events are translated to StreamRequest items and
+    merged into the unified inbound stream before on_live_message_send is called,
+    so application code sees a single mixed stream of client and peer messages.
+    """
+
+    def __init__(
+        self,
+        agent_card: AgentCard,
+        request_handler: RequestHandler,
+        context_builder: CallContextBuilder | None = None,
+        card_modifier: Callable[[AgentCard], AgentCard] | None = None,
+    ) -> None:
+        self.agent_card = agent_card
+        self.request_handler = request_handler
+        self.context_builder = context_builder or DefaultCallContextBuilder()
+        self.card_modifier = card_modifier
+
+    async def raise_error_response(self, error: A2AError) -> None:
+        code = _SLIM_ERROR_CODE_MAP.get(type(error), code_pb2.UNKNOWN)
+        raise slim_bindings.RpcError.Rpc(
+            code=code,
+            message=f"{type(error).__name__}: {error.message}",
+            details=None,
+        )
+
+    async def SendLiveMessage(
+        self,
+        request_stream: slim_bindings.RequestStream,
+        context: slim_bindings.Context,
+        sink: slim_bindings.ResponseSink,
+        peer_responses: AsyncIterable,
+    ) -> AsyncIterable[a2a_pb2.StreamResponse]:
+        """Handles broadcast SendLiveMessage: merges client stream with translated peer events."""
+        server_context = self.context_builder.build(context)
+
+        queue: asyncio.Queue[a2a_pb2.StreamRequest | None] = asyncio.Queue()
+
+        async def _feed_client() -> None:
+            try:
+                while True:
+                    msg = await request_stream.next_async()
+                    if msg.is_end():
+                        break
+                    if msg.is_error():
+                        await queue.put(None)
+                        return
+                    if msg.is_data():
+                        req = a2a_pb2.StreamRequest.FromString(msg[0])
+                        if not server_context.tenant:
+                            server_context.tenant = req.tenant
+                        await queue.put(req)
+            finally:
+                await queue.put(None)
+
+        async def _feed_peers() -> None:
+            try:
+                async for source, stream_response in peer_responses:
+                    src_str = str(source)
+                    translated = _translate_peer_response(src_str, stream_response)
+                    if translated is not None:
+                        await queue.put(translated)
+            finally:
+                await queue.put(None)
+
+        async def _merged_stream() -> AsyncIterable[a2a_pb2.StreamRequest]:
+            client_task = asyncio.ensure_future(_feed_client())
+            peer_task = asyncio.ensure_future(_feed_peers())
+            pending = 2
+            try:
+                while pending > 0:
+                    item = await queue.get()
+                    if item is None:
+                        pending -= 1
+                    else:
+                        yield item
+            finally:
+                client_task.cancel()
+                peer_task.cancel()
+
+        try:
+            async for event in self.request_handler.on_live_message_send(
+                _merged_stream(), server_context
+            ):
+                yield event
+        except A2AError as e:
+            await self.raise_error_response(e)
+
+
+def _translate_peer_response(
+    source: str,
+    response: a2a_pb2.StreamResponse,
+) -> a2a_pb2.StreamRequest | None:
+    """Translate a peer StreamResponse to a StreamRequest per spec Section 5."""
+    which = response.WhichOneof("payload")
+    meta = {"slim-src": source}
+
+    if which == "task":
+        task = response.task
+        meta["slim-peer-task-id"] = task.id
+        msg = a2a_pb2.Message(
+            role=a2a_pb2.Role.ROLE_USER,
+            parts=[a2a_pb2.Part(text=f"peer task started: {task.id}")],
+        )
+        return a2a_pb2.StreamRequest(message=msg, metadata=meta)
+
+    if which == "status_update":
+        update = response.status_update
+        meta["slim-peer-task-id"] = update.task_id
+        if update.status.HasField("message"):
+            return a2a_pb2.StreamRequest(message=update.status.message, metadata=meta)
+        state_name = a2a_pb2.TaskState.Name(update.status.state)
+        meta["slim-peer-state"] = state_name
+        msg = a2a_pb2.Message(
+            role=a2a_pb2.Role.ROLE_USER,
+            parts=[a2a_pb2.Part(text=f"peer task state: {state_name}")],
+        )
+        return a2a_pb2.StreamRequest(message=msg, metadata=meta)
+
+    if which == "artifact_update":
+        update = response.artifact_update
+        meta["slim-peer-task-id"] = update.task_id
+        return a2a_pb2.StreamRequest(artifact_update=update, metadata=meta)
+
+    if which == "message_update":
+        update = response.message_update
+        meta["slim-peer-task-id"] = update.task_id
+        return a2a_pb2.StreamRequest(message=update.message, metadata=meta)
+
+    return None

slima2a/types/v1/a2a_pb2_slimrpc.py

@@ -779,3 +779,68 @@ def add_A2AServiceServicer_to_server(servicer, server: slim_bindings.Server):
         method_name="SendLiveMessage",
         handler=_A2AServiceServicer_SendLiveMessage_Handler(servicer),
     )
+
+
+class A2AServiceSharedServicer:
+    """Shared-responses server servicer for A2AService.
+
+    Register with ``add_A2AServiceServicer_to_server_shared`` on a server created
+    with ``Server.new_with_shared_responses_and_connection``. Each method receives
+    an extra ``peer_responses`` argument — an async iterable of
+    ``(source, StreamResponse)`` tuples yielded by peer agents in the GROUP.
+    """
+
+    def SendLiveMessage(self, request_stream, context, sink, peer_responses):
+        """Method for SendLiveMessage (shared-responses). Implement your service logic here."""
+        raise slim_bindings.RpcError.Rpc(
+            code=slim_bindings.RpcCode.UNIMPLEMENTED,
+            message="Method not implemented!",
+            details=None
+        )
+
+
+class _A2AServiceServicer_SendLiveMessage_SharedHandler(slim_bindings.StreamStreamSharedHandler):
+    def __init__(self, servicer):
+        self.servicer = servicer
+
+    async def handle(self, stream: slim_bindings.RequestStream, context: slim_bindings.Context, sink: slim_bindings.ResponseSink, peer_stream: slim_bindings.PeerResponseStream):
+        try:
+            async def request_iterator():
+                while True:
+                    stream_msg = await stream.next_async()
+                    if stream_msg.is_end():
+                        break
+                    if stream_msg.is_error():
+                        raise stream_msg[0]
+                    if stream_msg.is_data():
+                        yield a2a__pb2.StreamRequest.FromString(stream_msg[0])
+
+            async def peer_iterator():
+                while True:
+                    msg = await peer_stream.next_async()
+                    if msg.is_end():
+                        break
+                    if msg.is_data():
+                        yield msg.source, a2a__pb2.StreamResponse.FromString(msg.payload)
+
+            response_iter = self.servicer.SendLiveMessage(request_iterator(), context, sink, peer_iterator())
+            async for response in response_iter:
+                await sink.send_async(a2a__pb2.StreamResponse.SerializeToString(response))
+            await sink.close_async()
+        except slim_bindings.RpcError as e:
+            await sink.send_error_async(e)
+        except Exception as e:
+            rpc_error = slim_bindings.RpcError.Rpc(
+                code=slim_bindings.RpcCode.INTERNAL,
+                message=str(e),
+                details=None
+            )
+            await sink.send_error_async(rpc_error)
+
+
+def add_A2AServiceServicer_to_server_shared(servicer, server: slim_bindings.Server):
+    server.register_stream_stream_shared(
+        service_name="lf.a2a.v1.A2AService",
+        method_name="SendLiveMessage",
+        handler=_A2AServiceServicer_SendLiveMessage_SharedHandler(servicer),
+    )