feat: add slimrpc multicast support with unified client factory (#18)

Introduce multicast transport and client classes that leverage
slim-bindings group channels to query multiple A2A agents
simultaneously.

The SRPCClientFactory auto-detects multicast from comma-separated
names in the agent card URL, providing a single create() entry point for
both unicast and multicast clients.

Fixes: #17

Signed-off-by: Mauro Sardara <msardara@cisco.com>

Author: msardara

examples/echo_agent/client.py

@@ -10,7 +10,6 @@
 from a2a.client import (
     A2ACardResolver,
     Client,
-    ClientFactory,
     create_text_message_object,
     minimal_agent_card,
 )
@@ -22,7 +21,10 @@
 from slima2a import setup_slim_client
 from slima2a.client_transport import (
     ClientConfig,
+    MultiAgentClientFactory,
+    MulticastClient,
     slimrpc_channel_factory,
+    slimrpc_group_channel_factory,
 )
 
 BASE_URL = "http://localhost:9999"
@@ -65,39 +67,44 @@ async def main() -> None:
         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(
+            slim_local_app, conn_id
+        ),
     )
-    client_factory = ClientFactory(client_config)
+    client_factory = MultiAgentClientFactory(client_config)
 
     if args.a2a_version == "v0":
         from slima2a.compat.v3_0.client_transport import SRPCCompatTransport
 
-        # mypy: the register API expects a different callable type; safe to ignore here.
-        client_factory.register("slimrpc", SRPCCompatTransport.create)  # type: ignore
-    else:
-        from slima2a.client_transport import SRPCTransport
+        client_factory.register("slimrpc", SRPCCompatTransport.create, multiagent=True)  # type: ignore
 
-        # mypy: the register API expects a different callable type; safe to ignore here.
-        client_factory.register("slimrpc", SRPCTransport.create)  # type: ignore
+    agent_names = [f"agntcy/demo/{name.strip()}" for name in args.agents.split(",")]
 
     match args.type:
         case "slimrpc":
-            agent_card = minimal_agent_card("agntcy/demo/echo_agent", ["slimrpc"])
+            cards = [minimal_agent_card(name, ["slimrpc"]) for name in agent_names]
         case "starlette":
-            agent_card = await fetch_agent_card(
-                resolver=A2ACardResolver(
-                    httpx_client=httpx_client,
-                    base_url=BASE_URL,
+            cards = [
+                await fetch_agent_card(
+                    resolver=A2ACardResolver(
+                        httpx_client=httpx_client,
+                        base_url=BASE_URL,
+                    )
                 )
-            )
+            ]
         case _:
             raise ValueError(f"Invalid client type: {args.type}")
 
-    client = client_factory.create(card=agent_card)
-    logger.info("A2AClient initialized.")
+    client = client_factory.create(card=cards)
 
-    response_text = await send_message(client, args.text)
-    print(f"> {args.text}")
-    print(response_text)
+    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)
 
 
 def parse_arguments() -> argparse.Namespace:
@@ -133,6 +140,13 @@ def parse_arguments() -> argparse.Namespace:
         default="v1",
         choices=["v0", "v1"],
     )
+    parser.add_argument(
+        "--agents",
+        type=str,
+        required=False,
+        default="echo_agent",
+        help="Comma-separated agent names (e.g. echo_agent_1,echo_agent_2). Multiple names triggers multicast.",
+    )
 
     args = parser.parse_args()
 
@@ -180,5 +194,41 @@ async def send_message(
     return output
 
 
+async def send_message_multicast(
+    client: MulticastClient,
+    text: str,
+) -> None:
+    message = create_text_message_object(content=text)
+    request = SendMessageRequest(message=message)
+
+    try:
+        async for source, response in client.send_message(request):
+            output = ""
+            which = response.WhichOneof("payload")
+            if which == "message":
+                for part in response.message.parts:
+                    if part.WhichOneof("content") == "text":
+                        output += part.text
+            elif which == "task":
+                if response.task.artifacts:
+                    for artifact in response.task.artifacts:
+                        for part in artifact.parts:
+                            if part.WhichOneof("content") == "text":
+                                output += part.text
+            elif which == "artifact_update":
+                artifact = 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 multicast message: {e}",
+            exc_info=True,
+        )
+        raise RuntimeError("failed sending multicast message") from e
+
+
 if __name__ == "__main__":
     asyncio.run(main())

examples/echo_agent/server.py

@@ -1,5 +1,6 @@
 import argparse
 import asyncio
+import contextlib
 import logging
 import sys
 from pathlib import Path
@@ -59,7 +60,7 @@ async def main() -> None:
             service, local_app, local_name, conn_id = await setup_slim_client(
                 namespace="agntcy",
                 group="demo",
-                name="echo_agent",
+                name=args.name,
                 secret="my_shared_secret_for_testing_purposes_only",
             )
 
@@ -106,6 +107,7 @@ def parse_arguments() -> argparse.Namespace:
     parser = argparse.ArgumentParser()
 
     parser.add_argument("--type", type=str, required=False, default="slimrpc")
+    parser.add_argument("--name", type=str, required=False, default="echo_agent")
 
     parser.add_argument(
         "--log-level",
@@ -131,4 +133,5 @@ def parse_arguments() -> argparse.Namespace:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    with contextlib.suppress(KeyboardInterrupt):
+        asyncio.run(main())