From 0c35f763468942de003dd97c72e9a0013ac665aa Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Thu, 3 Sep 2026 15:35:00 +0100 Subject: [PATCH 01/20] 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> --- examples/echo_agent/client.py | 39 ++++++++++++++++++++++++++- pyproject.toml | 5 +++- slima2a/client_transport.py | 26 ++++++++++++++++++ slima2a/handler.py | 30 +++++++++++++++++++++ slima2a/types/v1/a2a_pb2_slimrpc.py | 42 ++++++++++++++++++++++++++++- uv.lock | 24 ++++------------- 6 files changed, 144 insertions(+), 22 deletions(-) diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index f540ba8..8e67100 100644 --- a/examples/echo_agent/client.py +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index 4f9d383..d443664 100644 --- a/pyproject.toml +++ b/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"] diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index 2e81224..5c62731 100644 --- a/slima2a/client_transport.py +++ b/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 diff --git a/slima2a/handler.py b/slima2a/handler.py index bc8f3f5..fbf4f62 100644 --- a/slima2a/handler.py +++ b/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, diff --git a/slima2a/types/v1/a2a_pb2_slimrpc.py b/slima2a/types/v1/a2a_pb2_slimrpc.py index cb3843a..980dafe 100644 --- a/slima2a/types/v1/a2a_pb2_slimrpc.py +++ b/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), + ) diff --git a/uv.lock b/uv.lock index b7bc27b..2e445ec 100644 --- a/uv.lock +++ b/uv.lock @@ -9,23 +9,18 @@ resolution-markers = [ [[package]] name = "a2a-sdk" -version = "1.1.0" -source = { registry = "https://pypi.org/simple" } +version = "0.0.0.post752.dev0+7346c66" +source = { git = "https://github.com/Tehsmash/a2a-python?branch=feat%2Fsend-live-message#7346c660bceb5063ad23646976f537aa217c1867" } dependencies = [ { name = "culsans", marker = "python_full_version < '3.13'" }, { name = "google-api-core" }, { name = "googleapis-common-protos" }, { name = "httpx" }, - { name = "httpx-sse" }, { name = "json-rpc" }, { name = "packaging" }, { name = "protobuf" }, { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/c7/7e/8ac10bbf8b15b16574355f39b17dbdf617a282c27b41c7ff2116e30336df/a2a_sdk-1.1.0.tar.gz", hash = "sha256:e8102dad1b36709dbdc3d19319e38e6dfa3b3a79c30416030eb2d482576be204", size = 375726, upload-time = "2026-05-29T09:34:43.015Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/d4/ea/3a5b160cfd51c67759b08748051094d9365ceff18127633d0021950c9860/a2a_sdk-1.1.0-py3-none-any.whl", hash = "sha256:d7f5846caf18033d8bf3108b11ec827dd8dd32f867c98848ede0e39474be93be", size = 241886, upload-time = "2026-05-29T09:34:41.484Z" }, -] [package.optional-dependencies] http-server = [ @@ -541,15 +536,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, ] -[[package]] -name = "httpx-sse" -version = "0.4.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/0f/4c/751061ffa58615a32c31b2d82e8482be8dd4a89154f003147acee90f2be9/httpx_sse-0.4.3.tar.gz", hash = "sha256:9b1ed0127459a66014aec3c56bebd93da3c1bc8bb6618c8082039a44889a755d", size = 15943, upload-time = "2025-10-10T21:48:22.271Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/d2/fd/6668e5aec43ab844de6fc74927e155a3b37bf40d7c3790e49fc0406b6578/httpx_sse-0.4.3-py3-none-any.whl", hash = "sha256:0ac1c9fe3c0afad2e0ebb25a934a59f4c7823b60792691f779fad2c5568830fc", size = 8960, upload-time = "2025-10-10T21:48:21.158Z" }, -] - [[package]] name = "idna" version = "3.13" @@ -1567,7 +1553,7 @@ wheels = [ [[package]] name = "slima2a" -version = "0.6.1" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "a2a-sdk", extra = ["sqlite", "telemetry"] }, @@ -1593,13 +1579,13 @@ testing = [ [package.metadata] requires-dist = [ - { name = "a2a-sdk", extras = ["telemetry", "sqlite"], specifier = "==1.1.0" }, + { name = "a2a-sdk", extras = ["telemetry", "sqlite"], git = "https://github.com/Tehsmash/a2a-python?branch=feat%2Fsend-live-message" }, { name = "slim-bindings", specifier = "~=2.0" }, ] [package.metadata.requires-dev] examples = [ - { name = "a2a-sdk", extras = ["http-server"] }, + { name = "a2a-sdk", extras = ["http-server"], git = "https://github.com/Tehsmash/a2a-python?branch=feat%2Fsend-live-message" }, { name = "langchain-core" }, { name = "langchain-openai" }, { name = "typing-extensions" }, From 1888e4fefbe1310907c1aaf3cfa75ebce0fa0259 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:01:54 +0100 Subject: [PATCH 02/20] 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> --- slima2a/client_transport.py | 44 +++++++++++++++++++++++++++++ slima2a/types/v1/a2a_pb2_slimrpc.py | 8 ++++++ 2 files changed, 52 insertions(+) diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index 5c62731..754a7c2 100644 --- a/slima2a/client_transport.py +++ b/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() diff --git a/slima2a/types/v1/a2a_pb2_slimrpc.py b/slima2a/types/v1/a2a_pb2_slimrpc.py index 980dafe..c6a9fd6 100644 --- a/slima2a/types/v1/a2a_pb2_slimrpc.py +++ b/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: From cf57eadf6fdab0e2dc20e46ff1534e0aae9c4311 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Tue, 8 Sep 2026 18:04:41 +0100 Subject: [PATCH 03/20] 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> --- examples/echo_agent/client.py | 48 +++++++++- examples/echo_agent/server.py | 9 +- pyproject.toml | 1 + slima2a/__init__.py | 2 + slima2a/client_transport.py | 26 ++++++ slima2a/handler.py | 135 ++++++++++++++++++++++++++++ slima2a/types/v1/a2a_pb2_slimrpc.py | 65 ++++++++++++++ uv.lock | 54 +++++++---- 8 files changed, 316 insertions(+), 24 deletions(-) diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index 8e67100..6e04760 100644 --- a/examples/echo_agent/client.py +++ b/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()) diff --git a/examples/echo_agent/server.py b/examples/echo_agent/server.py index bea58ff..86db50b 100644 --- a/examples/echo_agent/server.py +++ b/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() diff --git a/pyproject.toml b/pyproject.toml index d443664..8db72d0 100644 --- a/pyproject.toml +++ b/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"] diff --git a/slima2a/__init__.py b/slima2a/__init__.py index e0513a5..dd89520 100644 --- a/slima2a/__init__.py +++ b/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", ] diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index 754a7c2..c9b0aaa 100644 --- a/slima2a/client_transport.py +++ b/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) diff --git a/slima2a/handler.py b/slima2a/handler.py index fbf4f62..e3c00b7 100644 --- a/slima2a/handler.py +++ b/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 diff --git a/slima2a/types/v1/a2a_pb2_slimrpc.py b/slima2a/types/v1/a2a_pb2_slimrpc.py index c6a9fd6..bdca796 100644 --- a/slima2a/types/v1/a2a_pb2_slimrpc.py +++ b/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), + ) diff --git a/uv.lock b/uv.lock index 2e445ec..89831ec 100644 --- a/uv.lock +++ b/uv.lock @@ -40,9 +40,9 @@ name = "aiologic" version = "0.17.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "sniffio", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, - { name = "wrapt", marker = "python_full_version < '3.13'" }, + { name = "sniffio" }, + { name = "typing-extensions" }, + { name = "wrapt" }, ] sdist = { url = "https://files.pythonhosted.org/packages/f1/7a/d51f2fde1e8ae8a83431f8e97b7a71e9358cdb1d4d2ce6be387fa44d68de/aiologic-0.17.1.tar.gz", hash = "sha256:2e1b93b9e88ced318c2a63ad7b382688f40cbfe40e3d42258d49dc9c5aea179d", size = 252354, upload-time = "2026-06-27T20:41:33.25Z" } wheels = [ @@ -363,8 +363,8 @@ name = "culsans" version = "0.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "aiologic", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "aiologic" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d9/e3/49afa1bc180e0d28008ec6bcdf82a4072d1c7a41032b5b759b60814ca4b0/culsans-0.11.0.tar.gz", hash = "sha256:0b43d0d05dce6106293d114c86e3fb4bfc63088cfe8ff08ed3fe36891447fe33", size = 107546, upload-time = "2025-12-31T23:15:38.196Z" } wheels = [ @@ -385,7 +385,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1538,17 +1538,35 @@ wheels = [ [[package]] name = "slim-bindings" -version = "2.0.0" -source = { registry = "https://pypi.org/simple" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/0b/cc/0841f22d32c7b5c57a6c402287d30d7f8dc48d6abb4c6b7de54c9f1b36ff/slim_bindings-2.0.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:c242eb2ce33ed251301879145347ad4255d001a56bfbe7c4955eaa98929ae723", size = 18959817, upload-time = "2026-08-05T16:13:35.074Z" }, - { url = "https://files.pythonhosted.org/packages/6c/6b/45973eca8a94e01aad34bf5fd12a88ae0245f8632ee31c9663552c6ea233/slim_bindings-2.0.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fcb8efc80f33d41fc034704550ec600eba6d20d1be7634c7568ca725f87c2325", size = 18374062, upload-time = "2026-08-05T16:13:37.461Z" }, - { url = "https://files.pythonhosted.org/packages/b3/7a/235c9836bbd900829e3b4092ab5b8602728222bb0076ca13360e7f2cffed/slim_bindings-2.0.0-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:9d569602eb8bdb63aad16104d04f9371bf4bbfb1f6ea801c49c563e87f09bf6b", size = 19113120, upload-time = "2026-08-05T16:13:39.889Z" }, - { url = "https://files.pythonhosted.org/packages/3e/d2/56539434db59d19131760d84adb48bdfca122b78b1550845b98d57e8256f/slim_bindings-2.0.0-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:9d2b3e76063f307c0a8cb4e26d6fec0e19e5fd9312fb516101517a47909bec15", size = 19759699, upload-time = "2026-08-05T16:13:41.935Z" }, - { url = "https://files.pythonhosted.org/packages/28/35/da680723dc975059b5bec01ffa4e9da53bcda32bb59145aa465d896ee766/slim_bindings-2.0.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:218383d2066f242a5d8a9bddcfc0d67d5c5585ab4b20391915280e91dd043149", size = 19111600, upload-time = "2026-08-05T16:13:44.479Z" }, - { url = "https://files.pythonhosted.org/packages/89/c1/823731e7ebb39a25264294dfa21d142c9f3b57d4a5444f9b68006c5a71b7/slim_bindings-2.0.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:96f9453877bd4ed4cac6cdb3f1faf608800acd5e8d496b1eda4fbc72b5561671", size = 19763962, upload-time = "2026-08-05T16:13:46.872Z" }, - { url = "https://files.pythonhosted.org/packages/a8/53/fb2b38bf0d12d1b81e9ee66cd3a4a96a5e7bfc2c01555b6261aadbf21de4/slim_bindings-2.0.0-py3-none-win_amd64.whl", hash = "sha256:60182551290a1d3d0377630e743a0153c58ebccea0170a870d7be6dfabb27749", size = 17248028, upload-time = "2026-08-05T16:13:49.223Z" }, - { url = "https://files.pythonhosted.org/packages/72/56/5f2d2086f0f4f977d7fbcfe752b28bf10e1576b85a3c8603b08bfeb8637d/slim_bindings-2.0.0-py3-none-win_arm64.whl", hash = "sha256:827b1ba925b666561493cc4203139170c3669650eef7fd2d95465803b2294890", size = 16155968, upload-time = "2026-08-05T16:13:51.653Z" }, +source = { editable = "../slim-bindings/python" } + +[package.metadata] +requires-dist = [ + { name = "googleapis-common-protos", marker = "extra == 'examples'", specifier = ">=1.72.0" }, + { name = "prompt-toolkit", marker = "extra == 'examples'", specifier = ">=3.0.52" }, + { name = "protobuf", marker = "extra == 'examples'", specifier = ">=6.33.4" }, + { name = "pydantic", marker = "extra == 'examples'", specifier = ">=2.0.0" }, + { name = "pydantic-settings", marker = "extra == 'examples'", specifier = ">=2.0.0" }, + { name = "pyyaml", marker = "extra == 'examples'", specifier = ">=6.0.3" }, + { name = "tomli", marker = "extra == 'examples'", specifier = ">=2.4.0" }, +] +provides-extras = ["examples"] + +[package.metadata.requires-dev] +building = [ + { name = "maturin", specifier = ">=1.11,<1.15" }, + { name = "maturin", extras = ["patchelf"], marker = "sys_platform == 'linux'", specifier = ">=1.8.2,<1.15" }, + { name = "uniffi-bindgen", specifier = "==0.31.0" }, +] +linting = [ + { name = "griffe", specifier = ">=1.5.5" }, + { name = "mypy", specifier = ">=1.14.1" }, + { name = "pyright", specifier = ">=1.1.0" }, + { name = "ruff", specifier = ">=0.9.3" }, +] +testing = [ + { name = "pytest", specifier = ">=7.4.0" }, + { name = "pytest-asyncio", specifier = ">=0.21.0" }, ] [[package]] @@ -1580,7 +1598,7 @@ testing = [ [package.metadata] requires-dist = [ { name = "a2a-sdk", extras = ["telemetry", "sqlite"], git = "https://github.com/Tehsmash/a2a-python?branch=feat%2Fsend-live-message" }, - { name = "slim-bindings", specifier = "~=2.0" }, + { name = "slim-bindings", editable = "../slim-bindings/python" }, ] [package.metadata.requires-dev] From 311c10bc9cce1bfa08ea42052f9835bed3d36ed9 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 09:53:04 +0100 Subject: [PATCH 04/20] refactor: use Name.from_string() in channel factories Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- slima2a/client_transport.py | 43 ++++++------------------------------- 1 file changed, 7 insertions(+), 36 deletions(-) diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index c9b0aaa..79056c1 100644 --- a/slima2a/client_transport.py +++ b/slima2a/client_transport.py @@ -44,19 +44,8 @@ def slimrpc_channel_factory( conn_id: int, ) -> Callable[[str], slim_bindings.Channel]: def factory(remote: str) -> slim_bindings.Channel: - # Parse the remote name from the URL - remote_parts = remote.split("/") - if len(remote_parts) != 3: - raise ValueError( - f"Invalid remote format: '{remote}'. Expected format: 'component1/component2/component'" - ) - - remote_name = slim_bindings.Name( - remote_parts[0], remote_parts[1], remote_parts[2] - ) - return slim_bindings.Channel.new_with_connection( - local_app, remote_name, conn_id + local_app, slim_bindings.Name.from_string(remote), conn_id ) return factory @@ -67,19 +56,10 @@ def slimrpc_group_channel_factory( 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_with_connection( - local_app, members, conn_id + local_app, + [slim_bindings.Name.from_string(r) for r in remotes], + conn_id, ) return factory @@ -90,19 +70,10 @@ def slimrpc_group_shared_channel_factory( 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 + local_app, + [slim_bindings.Name.from_string(r) for r in remotes], + conn_id, ) return factory From 651aad3ef599f259570426a4acbd13a5cee02714 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 10:46:55 +0100 Subject: [PATCH 05/20] fix: update EchoAgentExecutor.execute() signature with input_queue arg Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 2de7d7a..e7ffe73 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -2,6 +2,7 @@ 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.tasks.task_updater import TaskUpdater from a2a.types import Message, Part, Role @@ -17,6 +18,7 @@ async def execute( self, context: RequestContext, event_queue: EventQueue, + input_queue: AgentInputQueue, ) -> None: if ( (not context.message) From 2d6d14c0a4977b8db1ec432cbf495bcb59f444aa Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:03:37 +0100 Subject: [PATCH 06/20] fix: read first message from input_queue instead of context Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index e7ffe73..3a0aed1 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -20,20 +20,22 @@ async def execute( event_queue: EventQueue, input_queue: AgentInputQueue, ) -> None: + turn = await input_queue.get() + if ( - (not context.message) - or (not context.message.task_id) - or (not context.message.context_id) + (not turn.message) + or (not turn.message.task_id) + or (not turn.message.context_id) ): raise Exception("invalid message") - logging.debug(f"received message: {context.message}") + logging.debug(f"received message: {turn.message}") # The V2 request handler requires an initial Task to be enqueued # before any status/artifact update events are emitted. - task = context.current_task + task = turn.current_task if task is None: - task = new_task_from_user_message(context.message) + task = new_task_from_user_message(turn.message) await event_queue.enqueue_event(task) task_updater = TaskUpdater( @@ -42,14 +44,14 @@ async def execute( context_id=task.context_id, ) - if context.message.parts[0].WhichOneof("content") != "text": + if turn.message.parts[0].WhichOneof("content") != "text": raise Exception("only text parts are supported") - result = await self.agent.invoke(context.message.parts[0].text) + result = await self.agent.invoke(turn.message.parts[0].text) response = Message( role=Role.ROLE_AGENT, - message_id=context.message.message_id, + message_id=turn.message.message_id, parts=[Part(text=result)], ) await task_updater.add_artifact( From f1ba1bf9c76a669d8af6309855abd023757be367 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:09:11 +0100 Subject: [PATCH 07/20] chore: change executor log level to info Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 3a0aed1..9e261af 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -29,7 +29,7 @@ async def execute( ): raise Exception("invalid message") - logging.debug(f"received message: {turn.message}") + logging.info(f"received message: {turn.message}") # The V2 request handler requires an initial Task to be enqueued # before any status/artifact update events are emitted. From 2eb19fb46ab00078f824099339cc8022cbc3969c Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:19:51 +0100 Subject: [PATCH 08/20] fix: use named logger in executor, default server log level to INFO Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 4 +++- examples/echo_agent/server.py | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 9e261af..e3fa264 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -1,5 +1,7 @@ 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 @@ -29,7 +31,7 @@ async def execute( ): raise Exception("invalid message") - logging.info(f"received message: {turn.message}") + logger.info(f"received message: {turn.message}") # The V2 request handler requires an initial Task to be enqueued # before any status/artifact update events are emitted. diff --git a/examples/echo_agent/server.py b/examples/echo_agent/server.py index 86db50b..a7c6bbd 100644 --- a/examples/echo_agent/server.py +++ b/examples/echo_agent/server.py @@ -126,7 +126,7 @@ def parse_arguments() -> argparse.Namespace: "--log-level", type=str, required=False, - default="ERROR", + default="INFO", ) parser.add_argument( From f20407e9b1a646db2790300b9b729cd8898c45f7 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:05:39 +0100 Subject: [PATCH 09/20] fix: SRPCSharedHandler receives decoded StreamRequest stream, not raw RequestStream Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- slima2a/handler.py | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/slima2a/handler.py b/slima2a/handler.py index e3c00b7..7f88caf 100644 --- a/slima2a/handler.py +++ b/slima2a/handler.py @@ -353,7 +353,7 @@ async def raise_error_response(self, error: A2AError) -> None: async def SendLiveMessage( self, - request_stream: slim_bindings.RequestStream, + request_stream: AsyncIterable[a2a_pb2.StreamRequest], context: slim_bindings.Context, sink: slim_bindings.ResponseSink, peer_responses: AsyncIterable, @@ -365,18 +365,10 @@ async def SendLiveMessage( 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) + async for req in request_stream: + if not server_context.tenant: + server_context.tenant = req.tenant + await queue.put(req) finally: await queue.put(None) From 541d74385faf536dbf88a0cf20aa45915fc783e4 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:19:08 +0100 Subject: [PATCH 10/20] fix: loop on input_queue for multi-turn/broadcast, complete task on QueueShutDown Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 71 ++++++++++++---------- 1 file changed, 39 insertions(+), 32 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index e3fa264..fbf803a 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -6,6 +6,7 @@ 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 @@ -22,45 +23,51 @@ async def execute( event_queue: EventQueue, input_queue: AgentInputQueue, ) -> None: - turn = await input_queue.get() + task_updater: TaskUpdater | None = None - if ( - (not turn.message) - or (not turn.message.task_id) - or (not turn.message.context_id) - ): - raise Exception("invalid message") + try: + while True: + turn = await input_queue.get() - logger.info(f"received message: {turn.message}") + if not turn.message: + continue - # The V2 request handler requires an initial Task to be enqueued - # before any status/artifact update events are emitted. - task = turn.current_task - if task is None: - task = new_task_from_user_message(turn.message) - await event_queue.enqueue_event(task) + logger.info(f"received message: {turn.message}") - task_updater = TaskUpdater( - event_queue=event_queue, - task_id=task.id, - context_id=task.context_id, - ) + 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": - raise Exception("only text parts are supported") + 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) + 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", - ) - await task_updater.complete(message=response) + 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", + ) + except QueueShutDown: + pass + + 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") From a804423eb576a4808a4be5e1de4a94feafc1a768 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:45:34 +0100 Subject: [PATCH 11/20] fix: catch QueueShutDown at get() so executor can also exit via break Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 66 +++++++++++----------- 1 file changed, 33 insertions(+), 33 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index fbf803a..fece5b2 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -25,46 +25,46 @@ async def execute( ) -> None: task_updater: TaskUpdater | None = None - try: - while True: + while True: + try: turn = await input_queue.get() + except QueueShutDown: + break - if not turn.message: - continue + if not turn.message: + continue - logger.info(f"received message: {turn.message}") + 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 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 + 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) + 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", - ) - except QueueShutDown: - pass + 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() From 82e635b51e797981b839d43855bc4ca89a679758 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:48:11 +0100 Subject: [PATCH 12/20] fix: catch QueueShutDown in SendLiveMessage handlers when task finishes early Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- slima2a/handler.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/slima2a/handler.py b/slima2a/handler.py index 7f88caf..a51c5ad 100644 --- a/slima2a/handler.py +++ b/slima2a/handler.py @@ -308,6 +308,8 @@ async def _decoded_stream() -> AsyncIterable[a2a_pb2.StreamRequest]: _decoded_stream(), server_context ): yield event + except asyncio.QueueShutDown: + pass except A2AError as e: await self.raise_error_response(e) @@ -402,6 +404,8 @@ async def _merged_stream() -> AsyncIterable[a2a_pb2.StreamRequest]: _merged_stream(), server_context ): yield event + except asyncio.QueueShutDown: + pass except A2AError as e: await self.raise_error_response(e) From 3a2011634ccb68a154e55746244098c180692fcf Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:28:11 +0100 Subject: [PATCH 13/20] fix: skip echo for peer messages; clean shutdown on KeyboardInterrupt in client Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/client.py | 36 +++++++++++++--------- examples/echo_agent/echo_agent_executor.py | 5 +++ 2 files changed, 26 insertions(+), 15 deletions(-) diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index 6e04760..c0678e8 100644 --- a/examples/echo_agent/client.py +++ b/examples/echo_agent/client.py @@ -103,21 +103,27 @@ async def main() -> None: client = client_factory.create(card=cards) - 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: - 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) - print(f"> {args.text}") - print(response_text) + try: + 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: + 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) + print(f"> {args.text}") + print(response_text) + except KeyboardInterrupt: + pass + finally: + await client.close() + await httpx_client.aclose() def parse_arguments() -> argparse.Namespace: diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index fece5b2..21f70c6 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -50,6 +50,11 @@ async def execute( context_id=task.context_id, ) + # Peer-translated messages carry slim-src metadata — don't echo them + if turn.metadata.get("slim-src"): + logger.info(f"skipping peer message from {turn.metadata['slim-src']}") + continue + if turn.message.parts[0].WhichOneof("content") != "text": logger.warning("skipping non-text message part") continue From 470975e276577b379a0daf0862d842c1d3fd70fd Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:40:04 +0100 Subject: [PATCH 14/20] fix: use slim-peer-task-id to detect peer messages, not slim-src Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 21f70c6..3e6a8e0 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -50,9 +50,9 @@ async def execute( context_id=task.context_id, ) - # Peer-translated messages carry slim-src metadata — don't echo them - if turn.metadata.get("slim-src"): - logger.info(f"skipping peer message from {turn.metadata['slim-src']}") + # Peer-translated messages carry slim-peer-task-id — don't echo them + if turn.metadata.get("slim-peer-task-id"): + logger.info(f"skipping peer message from {turn.metadata.get('slim-src')}") continue if turn.message.parts[0].WhichOneof("content") != "text": From 8ad722d5cd249ce54b0485c2d13d8435d5d7d082 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:53:51 +0100 Subject: [PATCH 15/20] fix: filter peer messages by comparing slim-src to client identity Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 3e6a8e0..3f40c30 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -24,6 +24,7 @@ async def execute( input_queue: AgentInputQueue, ) -> None: task_updater: TaskUpdater | None = None + client_slim_src: str | None = None while True: try: @@ -37,9 +38,10 @@ async def execute( logger.info(f"received message: {turn.message}") if task_updater is None: - # First turn: bootstrap the task + # First turn: bootstrap the task and capture the client's slim-src if not turn.message.task_id or not turn.message.context_id: raise Exception("invalid message") + client_slim_src = turn.metadata.get("slim-src") task = turn.current_task if task is None: task = new_task_from_user_message(turn.message) @@ -50,9 +52,11 @@ async def execute( context_id=task.context_id, ) - # Peer-translated messages carry slim-peer-task-id — don't echo them - if turn.metadata.get("slim-peer-task-id"): - logger.info(f"skipping peer message from {turn.metadata.get('slim-src')}") + # In broadcast mode slim-src identifies the sender; skip messages + # not from the original client + msg_src = turn.metadata.get("slim-src") + if client_slim_src and msg_src and msg_src != client_slim_src: + logger.info(f"skipping peer message from {msg_src}") continue if turn.message.parts[0].WhichOneof("content") != "text": From 83dc6e8187e52a405ab6222a464edae797322a11 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:13:43 +0100 Subject: [PATCH 16/20] fix: skip peer messages by presence of slim-src, not by comparison Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 3f40c30..e6cefd7 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -24,7 +24,6 @@ async def execute( input_queue: AgentInputQueue, ) -> None: task_updater: TaskUpdater | None = None - client_slim_src: str | None = None while True: try: @@ -35,13 +34,19 @@ async def execute( if not turn.message: continue + # Peer-translated messages carry slim-src in metadata; direct client + # messages never do — skip peer messages + msg_src = turn.metadata.get("slim-src") + if msg_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 and capture the client's slim-src + # First turn: bootstrap the task if not turn.message.task_id or not turn.message.context_id: raise Exception("invalid message") - client_slim_src = turn.metadata.get("slim-src") task = turn.current_task if task is None: task = new_task_from_user_message(turn.message) @@ -52,13 +57,6 @@ async def execute( context_id=task.context_id, ) - # In broadcast mode slim-src identifies the sender; skip messages - # not from the original client - msg_src = turn.metadata.get("slim-src") - if client_slim_src and msg_src and msg_src != client_slim_src: - logger.info(f"skipping peer message from {msg_src}") - continue - if turn.message.parts[0].WhichOneof("content") != "text": logger.warning("skipping non-text message part") continue From f2d0331ebabab9e362b11cdbae8e6baa381d8007 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:21:31 +0100 Subject: [PATCH 17/20] =?UTF-8?q?docs:=20clarify=20slim-src=20discriminato?= =?UTF-8?q?r=20per=20spec=20=C2=A76=20(peer-only,=20not=20all=20messages)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/echo_agent_executor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index e6cefd7..aa59be7 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -34,8 +34,8 @@ async def execute( if not turn.message: continue - # Peer-translated messages carry slim-src in metadata; direct client - # messages never do — skip peer messages + # slim-src is set only on peer-translated messages (per spec §6); + # direct client messages never carry it — use its presence to skip peers msg_src = turn.metadata.get("slim-src") if msg_src: logger.info(f"skipping peer message from {msg_src}") From e109d47ab06af181efd0835c2f682b9708dd95a2 Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:10:26 +0100 Subject: [PATCH 18/20] feat: inject slim-src from client identity on SendLiveMessage calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per spec §6 slim-src must be set on all messages in broadcast mode. Client sets slim-src in RPC call metadata; server reads it from context.metadata() and stamps each decoded StreamRequest from the client stream. Executor now captures client identity on first turn and skips messages from other senders by comparing slim-src. Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/client.py | 1 + examples/echo_agent/echo_agent_executor.py | 10 +++++++--- slima2a/client_transport.py | 15 +++++++++++---- slima2a/types/v1/a2a_pb2_slimrpc.py | 7 ++++++- 4 files changed, 25 insertions(+), 8 deletions(-) diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index c0678e8..d342357 100644 --- a/examples/echo_agent/client.py +++ b/examples/echo_agent/client.py @@ -76,6 +76,7 @@ async def main() -> None: slimrpc_group_shared_channel_factory=slimrpc_group_shared_channel_factory( slim_local_app, conn_id ), + local_name=local_name, ) client_factory = MultiAgentClientFactory(client_config) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index aa59be7..13732f2 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/examples/echo_agent/echo_agent_executor.py @@ -24,6 +24,7 @@ async def execute( input_queue: AgentInputQueue, ) -> None: task_updater: TaskUpdater | None = None + client_slim_src: str | None = None while True: try: @@ -34,10 +35,13 @@ async def execute( if not turn.message: continue - # slim-src is set only on peer-translated messages (per spec §6); - # direct client messages never carry it — use its presence to skip peers + # 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 msg_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 diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index 79056c1..496da3b 100644 --- a/slima2a/client_transport.py +++ b/slima2a/client_transport.py @@ -88,6 +88,7 @@ class ClientConfig(A2AClientConfig): slimrpc_group_shared_channel_factory: ( Callable[[list[str]], slim_bindings.Channel] | None ) = None + local_name: slim_bindings.Name | None = None @trace_class(kind=SpanKind.CLIENT) @@ -98,11 +99,13 @@ def __init__( self, channel: slim_bindings.Channel, agent_card: AgentCard | None, + local_name: slim_bindings.Name | None = None, ) -> None: """Initializes the SRPCTransport.""" self.agent_card = agent_card self.channel = channel self.stub = a2a_pb2_slimrpc.A2AServiceStub(channel) + self._local_name = local_name @classmethod def create( @@ -134,7 +137,7 @@ def create( if url is None: raise ValueError("url is required for unicast sRPC") channel = config.slimrpc_channel_factory(url) - return cls(channel, card) + return cls(channel, card, config.local_name) async def send_message( self, @@ -247,7 +250,8 @@ async def send_live_message( context: ClientCallContext | None = None, ) -> AsyncGenerator[StreamResponse, None]: """Sends a bidirectional streaming live message request to the agent.""" - bidi = self.stub.SendLiveMessage() + metadata = {"slim-src": str(self._local_name)} if self._local_name else None + bidi = self.stub.SendLiveMessage(metadata=metadata) async def _send(): async for req in request_stream: await bidi.send_async(req.SerializeToString()) @@ -349,9 +353,11 @@ class SRPCMulticastTransport: def __init__( self, channel: slim_bindings.Channel, + local_name: slim_bindings.Name | None = None, ) -> None: self.channel = channel self.stub = a2a_pb2_slimrpc.A2AServiceGroupStub(channel) + self._local_name = local_name @classmethod def create( @@ -365,7 +371,7 @@ def create( "slimrpc_group_channel_factory is required when using sRPC multicast" ) channel = config.slimrpc_group_channel_factory(agent_names) - return cls(channel) + return cls(channel, config.local_name) async def send_message( self, @@ -480,7 +486,8 @@ async def send_live_message( Yields (source, StreamResponse) tuples as events arrive from any agent. """ - bidi = self.stub.SendLiveMessage() + metadata = {"slim-src": str(self._local_name)} if self._local_name else None + bidi = self.stub.SendLiveMessage(metadata=metadata) async def _send(): async for req in request_stream: await bidi.send_async(req.SerializeToString()) diff --git a/slima2a/types/v1/a2a_pb2_slimrpc.py b/slima2a/types/v1/a2a_pb2_slimrpc.py index bdca796..474ab44 100644 --- a/slima2a/types/v1/a2a_pb2_slimrpc.py +++ b/slima2a/types/v1/a2a_pb2_slimrpc.py @@ -805,6 +805,8 @@ def __init__(self, servicer): async def handle(self, stream: slim_bindings.RequestStream, context: slim_bindings.Context, sink: slim_bindings.ResponseSink, peer_stream: slim_bindings.PeerResponseStream): try: + slim_src = context.metadata().get("slim-src") + async def request_iterator(): while True: stream_msg = await stream.next_async() @@ -813,7 +815,10 @@ async def request_iterator(): if stream_msg.is_error(): raise stream_msg[0] if stream_msg.is_data(): - yield a2a__pb2.StreamRequest.FromString(stream_msg[0]) + req = a2a__pb2.StreamRequest.FromString(stream_msg[0]) + if slim_src: + req.metadata["slim-src"] = slim_src + yield req async def peer_iterator(): while True: From da703ac4aa386fe123696c074f2e73a4acfc41fd Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Thu, 10 Sep 2026 11:54:45 +0100 Subject: [PATCH 19/20] feat: use Channel.local_name for slim-src and add interactive stdin mode Read slim-src identity directly from Channel.local_name accessor instead of threading local_name through ClientConfig and transport constructors. Add interactive stdin mode to the echo agent client: omitting --text in --live or --broadcast mode reads lines from stdin, sending each as a StreamRequest and printing responses inline. Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- examples/echo_agent/client.py | 95 ++++++++++++++++++++++++++++++++--- slima2a/client_transport.py | 19 +++---- 2 files changed, 96 insertions(+), 18 deletions(-) diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index d342357..7e89ab2 100644 --- a/examples/echo_agent/client.py +++ b/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])) @@ -55,7 +56,7 @@ 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", @@ -76,7 +77,6 @@ async def main() -> None: slimrpc_group_shared_channel_factory=slimrpc_group_shared_channel_factory( slim_local_app, conn_id ), - local_name=local_name, ) client_factory = MultiAgentClientFactory(client_config) @@ -106,15 +106,21 @@ async def main() -> None: try: if isinstance(client, MulticastClient) and args.broadcast: - print(f"> {args.text} (broadcast live to {agent_names})") - await send_live_message_broadcast(client, args.text) + 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.") - print(f"> {args.text} (live)") - await send_live_message(client, args.text) + 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) @@ -159,7 +165,9 @@ def parse_arguments() -> argparse.Namespace: 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", @@ -187,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 @@ -316,5 +328,74 @@ async def _requests(): 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()) diff --git a/slima2a/client_transport.py b/slima2a/client_transport.py index 496da3b..48e7053 100644 --- a/slima2a/client_transport.py +++ b/slima2a/client_transport.py @@ -88,7 +88,6 @@ class ClientConfig(A2AClientConfig): slimrpc_group_shared_channel_factory: ( Callable[[list[str]], slim_bindings.Channel] | None ) = None - local_name: slim_bindings.Name | None = None @trace_class(kind=SpanKind.CLIENT) @@ -99,13 +98,11 @@ def __init__( self, channel: slim_bindings.Channel, agent_card: AgentCard | None, - local_name: slim_bindings.Name | None = None, ) -> None: """Initializes the SRPCTransport.""" self.agent_card = agent_card self.channel = channel self.stub = a2a_pb2_slimrpc.A2AServiceStub(channel) - self._local_name = local_name @classmethod def create( @@ -137,7 +134,7 @@ def create( if url is None: raise ValueError("url is required for unicast sRPC") channel = config.slimrpc_channel_factory(url) - return cls(channel, card, config.local_name) + return cls(channel, card) async def send_message( self, @@ -250,8 +247,9 @@ async def send_live_message( context: ClientCallContext | None = None, ) -> AsyncGenerator[StreamResponse, None]: """Sends a bidirectional streaming live message request to the agent.""" - metadata = {"slim-src": str(self._local_name)} if self._local_name else None - bidi = self.stub.SendLiveMessage(metadata=metadata) + bidi = self.stub.SendLiveMessage( + metadata={"slim-src": str(self.channel.local_name)} + ) async def _send(): async for req in request_stream: await bidi.send_async(req.SerializeToString()) @@ -353,11 +351,9 @@ class SRPCMulticastTransport: def __init__( self, channel: slim_bindings.Channel, - local_name: slim_bindings.Name | None = None, ) -> None: self.channel = channel self.stub = a2a_pb2_slimrpc.A2AServiceGroupStub(channel) - self._local_name = local_name @classmethod def create( @@ -371,7 +367,7 @@ def create( "slimrpc_group_channel_factory is required when using sRPC multicast" ) channel = config.slimrpc_group_channel_factory(agent_names) - return cls(channel, config.local_name) + return cls(channel) async def send_message( self, @@ -486,8 +482,9 @@ async def send_live_message( Yields (source, StreamResponse) tuples as events arrive from any agent. """ - metadata = {"slim-src": str(self._local_name)} if self._local_name else None - bidi = self.stub.SendLiveMessage(metadata=metadata) + bidi = self.stub.SendLiveMessage( + metadata={"slim-src": str(self.channel.local_name)} + ) async def _send(): async for req in request_stream: await bidi.send_async(req.SerializeToString()) From 3ba3592f4726ab06aeeb5b361f5164c1ad3c3bed Mon Sep 17 00:00:00 2001 From: Sam Betts <1769706+Tehsmash@users.noreply.github.com> Date: Thu, 10 Sep 2026 12:34:37 +0100 Subject: [PATCH 20/20] chore: point slim-bindings to feat/slimrpc-shared-responses branch Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com> --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 8db72d0..ad392ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +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 } +slim-bindings = { git = "https://github.com/agntcy/slim-bindings", branch = "feat/slimrpc-shared-responses", subdirectory = "python" } [tool.ruff.format] exclude = ["*_pb2.py", "*_pb2.pyi", "*_pb2_slimrpc.py"]