feat: add SendLiveMessage bidirectional streaming support

Implements the SendLiveMessage bidi streaming RPC from the A2A v1.1
proposal (https://github.com/Tehsmash/a2a-python/tree/feat/send-live-message).

- Add SendLiveMessage to A2AServiceStub (client, via call_stream_stream),
  A2AServiceServicer (UNIMPLEMENTED stub), _A2AServiceServicer_SendLiveMessage_Handler
  (StreamStreamHandler), and register_stream_stream in add_A2AServiceServicer_to_server
- Add SRPCHandler.SendLiveMessage: bridges RequestStream to decoded async
  iterable of StreamRequest protos, delegates to request_handler.on_live_message_send
- Add SRPCTransport.send_live_message: opens BidiStreamHandler, concurrently
  sends serialized StreamRequests and yields decoded StreamResponse events
- Point a2a-sdk dependency to Tehsmash/a2a-python@feat/send-live-message
  (the branch adding StreamRequest and on_live_message_send)
- Add --live flag to echo_agent client example demonstrating the new method

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

Author: Tehsmash

examples/echo_agent/client.py

@@ -13,7 +13,7 @@
     minimal_agent_card,
 )
 from a2a.helpers import new_text_message
-from a2a.types.a2a_pb2 import AgentCard, Role, SendMessageRequest
+from a2a.types.a2a_pb2 import AgentCard, Role, SendMessageRequest, StreamRequest
 from a2a.utils.constants import (
     AGENT_CARD_WELL_KNOWN_PATH,
 )
@@ -100,6 +100,10 @@ async def main() -> None:
     if isinstance(client, MulticastClient):
         print(f"> {args.text} (multicast to {agent_names})")
         await send_message_multicast(client, args.text)
+    elif args.live:
+        logger.info("A2AClient initialized.")
+        print(f"> {args.text} (live)")
+        await send_live_message(client, args.text)
     else:
         logger.info("A2AClient initialized.")
         response_text = await send_message(client, args.text)
@@ -122,6 +126,13 @@ def parse_arguments() -> argparse.Namespace:
         required=False,
         default=False,
     )
+    parser.add_argument(
+        "--live",
+        action="store_true",
+        required=False,
+        default=False,
+        help="Use SendLiveMessage bidirectional streaming",
+    )
     parser.add_argument(
         "--text",
         type=str,
@@ -230,5 +241,31 @@ async def send_message_multicast(
         raise RuntimeError("failed sending multicast message") from e
 
 
+async def send_live_message(client: Client, text: str) -> None:
+    message = new_text_message(text, role=Role.ROLE_USER)
+
+    async def _requests():
+        yield StreamRequest(message=message)
+
+    output = ""
+    try:
+        async for stream_response in client.send_live_message(_requests()):
+            which = stream_response.WhichOneof("payload")
+            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
+    except Exception as e:
+        logger.error(f"failed sending live message: {e}", exc_info=True)
+        raise RuntimeError("failed sending live message") from e
+
+    print(output)
+
+
 if __name__ == "__main__":
     asyncio.run(main())

pyproject.toml

@@ -8,7 +8,7 @@ description = "A2A protocol over slimrpc"
 readme = "README.md"
 requires-python = ">=3.10, <4.0"
 dependencies = [
-    "a2a-sdk[telemetry,sqlite]==1.1.0",
+    "a2a-sdk[telemetry,sqlite]",
     "slim-bindings~=2.0",
 ]
 classifiers = [
@@ -45,6 +45,9 @@ examples = [
 [tool.uv]
 default-groups = ["linting", "testing", "examples"]
 
+[tool.uv.sources]
+a2a-sdk = { git = "https://github.com/Tehsmash/a2a-python", branch = "feat/send-live-message" }
+
 [tool.ruff.format]
 exclude = ["*_pb2.py", "*_pb2.pyi", "*_pb2_slimrpc.py"]
 

slima2a/client_transport.py

@@ -1,6 +1,7 @@
 # Copyright AGNTCY Contributors (https://github.com/agntcy)
 # SPDX-License-Identifier: Apache-2.0
 
+import asyncio
 import logging
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
@@ -242,6 +243,31 @@ async def get_extended_agent_card(
             return card
         return await self.stub.GetExtendedAgentCard(request)
 
+    async def send_live_message(
+        self,
+        request_stream: "AsyncGenerator[Any, None]",
+        *,
+        context: ClientCallContext | None = None,
+    ) -> AsyncGenerator[StreamResponse, None]:
+        """Sends a bidirectional streaming live message request to the 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[0]
+                if msg.is_data():
+                    yield StreamResponse.FromString(msg[0])
+        finally:
+            await send_task
+
     async def close(self) -> None:
         """Closes the transport and releases any resources."""
         pass

slima2a/handler.py

@@ -280,6 +280,36 @@ async def DeleteTaskPushNotificationConfig(
             await self.raise_error_response(e)
         return empty_pb2.Empty()
 
+    async def SendLiveMessage(
+        self,
+        request_stream: slim_bindings.RequestStream,
+        context: slim_bindings.Context,
+        sink: slim_bindings.ResponseSink,
+    ) -> AsyncIterable[a2a_pb2.StreamResponse]:
+        """Handles the 'SendLiveMessage' SlimRPC bidi streaming method."""
+        server_context = self.context_builder.build(context)
+
+        async def _decoded_stream() -> AsyncIterable[a2a_pb2.StreamRequest]:
+            while True:
+                msg = await request_stream.next_async()
+                if msg.is_end():
+                    break
+                if msg.is_error():
+                    raise msg[0]
+                if msg.is_data():
+                    req = a2a_pb2.StreamRequest.FromString(msg[0])
+                    if not server_context.tenant:
+                        server_context.tenant = req.tenant
+                    yield req
+
+        try:
+            async for event in self.request_handler.on_live_message_send(
+                _decoded_stream(), server_context
+            ):
+                yield event
+        except A2AError as e:
+            await self.raise_error_response(e)
+
     async def GetExtendedAgentCard(
         self,
         request: a2a_pb2.GetExtendedAgentCardRequest,

slima2a/types/v1/a2a_pb2_slimrpc.py

@@ -158,6 +158,14 @@ async def DeleteTaskPushNotificationConfig(self, request: a2a__pb2.DeleteTaskPus
         )
         return google__protobuf__empty_pb2.Empty.FromString(response_bytes)
 
+    def SendLiveMessage(self, timeout: Optional[timedelta] = None, metadata: Optional[dict[str, str]] = None) -> slim_bindings.BidiStreamHandler:
+        """Open a bidirectional streaming SendLiveMessage call."""
+        return self._channel.call_stream_stream(
+            "lf.a2a.v1.A2AService",
+            "SendLiveMessage",
+            timeout,
+            metadata,
+        )
 
 
 class A2AServiceGroupStub:
@@ -467,7 +475,13 @@ def DeleteTaskPushNotificationConfig(self, request, context):
             details=None
         )
 
-
+    def SendLiveMessage(self, request_stream, context, sink):
+        """Method for SendLiveMessage. Implement your service logic here."""
+        raise slim_bindings.RpcError.Rpc(
+            code=slim_bindings.RpcCode.UNIMPLEMENTED,
+            message="Method not implemented!",
+            details=None
+        )
 
 
 class _A2AServiceServicer_SendMessage_Handler(slim_bindings.UnaryUnaryHandler):
@@ -675,6 +689,27 @@ async def handle(self, request: bytes, context: slim_bindings.Context) -> bytes:
             )
 
 
+class _A2AServiceServicer_SendLiveMessage_Handler(slim_bindings.StreamStreamHandler):
+    def __init__(self, servicer):
+        self.servicer = servicer
+
+    async def handle(self, stream: slim_bindings.RequestStream, context: slim_bindings.Context, sink: slim_bindings.ResponseSink):
+        try:
+            response_iter = self.servicer.SendLiveMessage(stream, context, sink)
+            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(servicer, server: slim_bindings.Server):
     server.register_unary_unary(
         service_name="lf.a2a.v1.A2AService",
@@ -731,3 +766,8 @@ def add_A2AServiceServicer_to_server(servicer, server: slim_bindings.Server):
         method_name="DeleteTaskPushNotificationConfig",
         handler=_A2AServiceServicer_DeleteTaskPushNotificationConfig_Handler(servicer),
     )
+    server.register_stream_stream(
+        service_name="lf.a2a.v1.A2AService",
+        method_name="SendLiveMessage",
+        handler=_A2AServiceServicer_SendLiveMessage_Handler(servicer),
+    )