diff --git a/examples/echo_agent/client.py b/examples/echo_agent/client.py index f540ba8..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])) @@ -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()) diff --git a/examples/echo_agent/echo_agent_executor.py b/examples/echo_agent/echo_agent_executor.py index 2de7d7a..13732f2 100644 --- a/examples/echo_agent/echo_agent_executor.py +++ b/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") diff --git a/examples/echo_agent/server.py b/examples/echo_agent/server.py index bea58ff..a7c6bbd 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() @@ -123,7 +126,7 @@ def parse_arguments() -> argparse.Namespace: "--log-level", type=str, required=False, - default="ERROR", + default="INFO", ) parser.add_argument( diff --git a/pyproject.toml b/pyproject.toml index 4f9d383..ad392ac 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,10 @@ 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" } +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"] 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 2e81224..48e7053 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 @@ -43,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 @@ -66,19 +56,24 @@ 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 + + +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: + return slim_bindings.Channel.new_group_shared_with_connection( + local_app, + [slim_bindings.Name.from_string(r) for r in remotes], + conn_id, ) return factory @@ -90,6 +85,9 @@ class ClientConfig(A2AClientConfig): 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) @@ -242,6 +240,33 @@ 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( + metadata={"slim-src": str(self.channel.local_name)} + ) + 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 @@ -447,6 +472,36 @@ 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( + metadata={"slim-src": str(self.channel.local_name)} + ) + 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 @@ -622,6 +677,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/handler.py b/slima2a/handler.py index bc8f3f5..a51c5ad 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 @@ -280,6 +281,38 @@ 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 asyncio.QueueShutDown: + pass + except A2AError as e: + await self.raise_error_response(e) + async def GetExtendedAgentCard( self, request: a2a_pb2.GetExtendedAgentCardRequest, @@ -290,3 +323,131 @@ 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: AsyncIterable[a2a_pb2.StreamRequest], + 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: + 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) + + 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 asyncio.QueueShutDown: + pass + 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 cb3843a..474ab44 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: @@ -374,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: @@ -467,7 +483,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 +697,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 +774,78 @@ 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), + ) + + +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: + slim_src = context.metadata().get("slim-src") + + 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(): + 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: + 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 b7bc27b..89831ec 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 = [ @@ -45,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 = [ @@ -368,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 = [ @@ -390,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 = [ @@ -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" @@ -1552,22 +1538,40 @@ 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]] name = "slima2a" -version = "0.6.1" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "a2a-sdk", extra = ["sqlite", "telemetry"] }, @@ -1593,13 +1597,13 @@ testing = [ [package.metadata] requires-dist = [ - { name = "a2a-sdk", extras = ["telemetry", "sqlite"], specifier = "==1.1.0" }, - { name = "slim-bindings", specifier = "~=2.0" }, + { name = "a2a-sdk", extras = ["telemetry", "sqlite"], git = "https://github.com/Tehsmash/a2a-python?branch=feat%2Fsend-live-message" }, + { name = "slim-bindings", editable = "../slim-bindings/python" }, ] [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" },