feat: add SendLiveMessage bidirectional streaming support

examples/echo_agent/client.py

@@ -3,6 +3,7 @@
 import logging
 import sys
 from pathlib import Path
+from typing import AsyncGenerator
 
 sys.path.insert(0, str(Path(__file__).parents[2]))
 
@@ -13,12 +14,12 @@
     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,
 )
 
-from slima2a import setup_slim_client
+from slima2a import setup_slim_client, slimrpc_group_shared_channel_factory
 from slima2a.client_transport import (
     ClientConfig,
     MultiAgentClientFactory,
@@ -55,19 +56,25 @@ async def main() -> None:
     httpx_client = httpx.AsyncClient()
 
     # Initialize and connect to SLIM
-    service, slim_local_app, local_name, conn_id = await setup_slim_client(
+    service, slim_local_app, _, conn_id = await setup_slim_client(
         namespace="agntcy",
         group="demo",
         name="client",
         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,14 +104,33 @@ async def main() -> None:
 
     client = client_factory.create(card=cards)
 
-    if isinstance(client, MulticastClient):
-        print(f"> {args.text} (multicast to {agent_names})")
-        await send_message_multicast(client, args.text)
-    else:
-        logger.info("A2AClient initialized.")
-        response_text = await send_message(client, args.text)
-        print(f"> {args.text}")
-        print(response_text)
+    try:
+        if isinstance(client, MulticastClient) and args.broadcast:
+            if args.text:
+                print(f"> {args.text} (broadcast live to {agent_names})")
+                await send_live_message_broadcast(client, args.text)
+            else:
+                await interactive_live_message_broadcast(client, agent_names)
+        elif 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.")
+            if args.text:
+                print(f"> {args.text} (live)")
+                await send_live_message(client, args.text)
+            else:
+                await interactive_live_message(client)
+        else:
+            logger.info("A2AClient initialized.")
+            response_text = await send_message(client, args.text)
+            print(f"> {args.text}")
+            print(response_text)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        await client.close()
+        await httpx_client.aclose()
 
 
 def parse_arguments() -> argparse.Namespace:
@@ -122,10 +148,26 @@ 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(
+        "--broadcast",
+        action="store_true",
+        required=False,
+        default=False,
+        help="Use broadcast SendLiveMessage (shared-responses) — requires multiple --agents",
+    )
     parser.add_argument(
         "--text",
         type=str,
-        required=True,
+        required=False,
+        default=None,
+        help="Message text. If omitted in --live/--broadcast mode, reads lines from stdin interactively.",
     )
     parser.add_argument(
         "--type",
@@ -153,6 +195,10 @@ def parse_arguments() -> argparse.Namespace:
     if args.type not in ["slimrpc", "starlette"]:
         raise ValueError(f"Invalid client type: {args.type}")
 
+    interactive_mode = args.live or args.broadcast
+    if not interactive_mode and args.text is None:
+        parser.error("--text is required unless --live or --broadcast is set")
+
     return args
 
 
@@ -230,5 +276,126 @@ 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)
+
+
+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
+
+
+async def _stdin_lines() -> AsyncGenerator[str, None]:
+    loop = asyncio.get_event_loop()
+    while True:
+        try:
+            line = await loop.run_in_executor(None, sys.stdin.readline)
+        except (EOFError, KeyboardInterrupt):
+            return
+        if not line:
+            return
+        line = line.rstrip("\n")
+        if line:
+            yield line
+
+
+async def interactive_live_message(client: Client) -> None:
+    print("Interactive live session (Ctrl-D or Ctrl-C to quit)")
+
+    async def _requests() -> AsyncGenerator:
+        async for line in _stdin_lines():
+            print(f"> {line}")
+            yield StreamRequest(message=new_text_message(line, role=Role.ROLE_USER))
+
+    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":
+                        print(part.text)
+            elif which == "artifact_update":
+                artifact = stream_response.artifact_update.artifact
+                for part in artifact.parts:
+                    if part.WhichOneof("content") == "text":
+                        print(part.text)
+    except Exception as e:
+        logger.error(f"failed in interactive live session: {e}", exc_info=True)
+        raise RuntimeError("failed in interactive live session") from e
+
+
+async def interactive_live_message_broadcast(
+    client: MulticastClient, agent_names: list[str]
+) -> None:
+    print(f"Interactive broadcast session to {agent_names} (Ctrl-D or Ctrl-C to quit)")
+
+    async def _requests() -> AsyncGenerator:
+        async for line in _stdin_lines():
+            print(f"> {line}")
+            yield StreamRequest(message=new_text_message(line, role=Role.ROLE_USER))
+
+    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 in interactive broadcast session: {e}", exc_info=True)
+        raise RuntimeError("failed in interactive broadcast session") from e
+
+
 if __name__ == "__main__":
     asyncio.run(main())

examples/echo_agent/echo_agent_executor.py

@@ -1,8 +1,12 @@
 import logging
 
+logger = logging.getLogger(__name__)
+
 from a2a.helpers import new_task_from_user_message
 from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.agent_execution.agent_input_queue import AgentInputQueue
 from a2a.server.events import EventQueue
+from a2a.server.events.event_queue_v2 import QueueShutDown
 from a2a.server.tasks.task_updater import TaskUpdater
 from a2a.types import Message, Part, Role
 
@@ -17,44 +21,64 @@ async def execute(
         self,
         context: RequestContext,
         event_queue: EventQueue,
+        input_queue: AgentInputQueue,
     ) -> None:
-        if (
-            (not context.message)
-            or (not context.message.task_id)
-            or (not context.message.context_id)
-        ):
-            raise Exception("invalid message")
-
-        logging.debug(f"received message: {context.message}")
-
-        # The V2 request handler requires an initial Task to be enqueued
-        # before any status/artifact update events are emitted.
-        task = context.current_task
-        if task is None:
-            task = new_task_from_user_message(context.message)
-            await event_queue.enqueue_event(task)
-
-        task_updater = TaskUpdater(
-            event_queue=event_queue,
-            task_id=task.id,
-            context_id=task.context_id,
-        )
-
-        if context.message.parts[0].WhichOneof("content") != "text":
-            raise Exception("only text parts are supported")
-
-        result = await self.agent.invoke(context.message.parts[0].text)
-
-        response = Message(
-            role=Role.ROLE_AGENT,
-            message_id=context.message.message_id,
-            parts=[Part(text=result)],
-        )
-        await task_updater.add_artifact(
-            parts=list(response.parts),
-            name="result",
-        )
-        await task_updater.complete(message=response)
+        task_updater: TaskUpdater | None = None
+        client_slim_src: str | None = None
+
+        while True:
+            try:
+                turn = await input_queue.get()
+            except QueueShutDown:
+                break
+
+            if not turn.message:
+                continue
+
+            # slim-src is set on all messages in broadcast mode (spec §6).
+            # Capture the client's identity from the first turn, then skip messages
+            # from other senders (peer agents).
+            msg_src = turn.metadata.get("slim-src")
+            if task_updater is None:
+                client_slim_src = msg_src
+            elif msg_src and msg_src != client_slim_src:
+                logger.info(f"skipping peer message from {msg_src}")
+                continue
+
+            logger.info(f"received message: {turn.message}")
+
+            if task_updater is None:
+                # First turn: bootstrap the task
+                if not turn.message.task_id or not turn.message.context_id:
+                    raise Exception("invalid message")
+                task = turn.current_task
+                if task is None:
+                    task = new_task_from_user_message(turn.message)
+                    await event_queue.enqueue_event(task)
+                task_updater = TaskUpdater(
+                    event_queue=event_queue,
+                    task_id=task.id,
+                    context_id=task.context_id,
+                )
+
+            if turn.message.parts[0].WhichOneof("content") != "text":
+                logger.warning("skipping non-text message part")
+                continue
+
+            result = await self.agent.invoke(turn.message.parts[0].text)
+
+            response = Message(
+                role=Role.ROLE_AGENT,
+                message_id=turn.message.message_id,
+                parts=[Part(text=result)],
+            )
+            await task_updater.add_artifact(
+                parts=list(response.parts),
+                name="result",
+            )
+
+        if task_updater is not None:
+            await task_updater.complete()
 
     async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
         raise NotImplementedError("cancel not supported")