Enable SLIM group chat pattern (#40)

* feat: enable group chat mode for broadcasting message

Signed-off-by: samuyang <samuyang@cisco.com>

* feat: refactoring A2A broadcast to enable groupchat. Adding groupchat test and refactoring test A2A agent-executor for groupchat.

* feat: adding client-initiated group session cleanup protocol using message header

* feat: refactor end-message-signal message to be uuid with closing handshake trigger

* feat: move session close propagation to session_manager.close_session

* feat: bump package version and add CHANGELOG.md

---------

Signed-off-by: samuyang <samuyang@cisco.com>
Co-authored-by: samuyang <samuyang@cisco.com>

Author: codyhartsook

CHANGELOG.md

@@ -1,3 +1,18 @@
 # Changelog
 
-## 0.0.1
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## 0.2.3
+
+### Added
+- SLIM multi-session lifecycle management
+- SLIM groupchat sessions, initiated with A2AClient.broadcast_message(group_chat=True)
+
+### Changed
+- AgntcyFactory.create_transport requires a name field when the type is SLIM, in the form /org/namespace/service
+- A2AClient.broadcast_message, when created from factory, requires list of recipients to fulfill SLIM requirements
+
+### Fixed

README.md

@@ -88,7 +88,7 @@ from agntcy_app_sdk.factory import AgntcyFactory
 # Create factory and transport
 factory = AgntcyFactory()
 transport_instance = factory.create_transport(
-    transport="SLIM", endpoint="http://localhost:46357"
+    transport="SLIM", endpoint="http://localhost:46357", name="org/namespace/agent-foo"
 )
 
 # Create MCP client

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "agntcy-app-sdk"
-version = "0.2.2"
+version = "0.2.3"
 description = "Agntcy Application SDK for Python"
 authors = [{ name = "Cody Hartsook", email = "codyhartsook@gmail.com" }]
 requires-python = ">=3.13,<4.0"

src/agntcy_app_sdk/bridge.py

@@ -66,22 +66,17 @@ async def loop_forever(self):
     async def _process_message(self, message: Message):
         """Process an incoming message through the handler and send response."""
 
-        try:
-            # Handle the request - check if handler is async or sync
-            if inspect.iscoroutinefunction(self.handler):
-                response = await self.handler(message)
+        if inspect.iscoroutinefunction(self.handler):
+            response = await self.handler(message)
+        else:
+            result = self.handler(message)
+            # If the result is a coroutine, await it
+            if inspect.iscoroutine(result):
+                response = await result
             else:
-                result = self.handler(message)
-                # If the result is a coroutine, await it
-                if inspect.iscoroutine(result):
-                    response = await result
-                else:
-                    response = result
-
-            if not response:
-                logger.warning("Handler returned no response for message.")
+                response = result
 
-            return response
+        if not response:
+            logger.warning("Handler returned no response for message.")
 
-        except Exception as e:
-            logger.error(f"Error processing message: {e}")
+        return response

src/agntcy_app_sdk/protocols/a2a/protocol.py

@@ -11,8 +11,13 @@
 from a2a.client import A2AClient, A2ACardResolver
 from a2a.utils import AGENT_CARD_WELL_KNOWN_PATH, PREV_AGENT_CARD_WELL_KNOWN_PATH
 from a2a.server.apps import A2AStarletteApplication
-from a2a.types import AgentCard, SendMessageRequest, SendMessageResponse
-
+from a2a.types import (
+    AgentCard,
+    SendMessageRequest,
+    SendMessageResponse,
+    JSONRPCSuccessResponse,
+    MessageSendParams,
+)
 from agntcy_app_sdk.protocols.protocol import BaseAgentProtocol
 from agntcy_app_sdk.transports.transport import BaseTransport, ResponseMode
 from agntcy_app_sdk.protocols.message import Message
@@ -24,7 +29,9 @@
 logger = get_logger(__name__)
 
 
-async def get_client_from_agent_card_url(httpx_client: httpx.AsyncClient, base_url: str,
+async def get_client_from_agent_card_url(
+    httpx_client: httpx.AsyncClient,
+    base_url: str,
     http_kwargs: dict[str, Any] | None = None,
 ) -> A2AClient:
     """
@@ -33,20 +40,25 @@ async def get_client_from_agent_card_url(httpx_client: httpx.AsyncClient, base_u
     """
     try:
         agent_card: AgentCard = await A2ACardResolver(
-            httpx_client, base_url=base_url,
+            httpx_client,
+            base_url=base_url,
             agent_card_path=AGENT_CARD_WELL_KNOWN_PATH,
         ).get_agent_card(http_kwargs=http_kwargs)
     except Exception as e:
-        logger.info(f"Failed to get client from agent card url with v3 path, "
-                    f"falling back to v2 path: {e}")
+        logger.info(
+            f"Failed to get client from agent card url with v3 path, "
+            f"falling back to v2 path: {e}"
+        )
         try:
-            agent_card: AgentCard = await (A2ACardResolver(
-                httpx_client, base_url=base_url,
+            agent_card: AgentCard = await A2ACardResolver(
+                httpx_client,
+                base_url=base_url,
                 agent_card_path=PREV_AGENT_CARD_WELL_KNOWN_PATH,
-            ).get_agent_card(http_kwargs=http_kwargs))
+            ).get_agent_card(http_kwargs=http_kwargs)
         except Exception as e:
-            logger.error(f"Failed to get client from agent card url with v2 "
-                         f"path: {e}")
+            logger.error(
+                f"Failed to get client from agent card url with v2 " f"path: {e}"
+            )
             raise e
 
     return A2AClient(httpx_client=httpx_client, agent_card=agent_card)
@@ -77,22 +89,28 @@ async def get_client_from_agent_card_topic(
         try:
             request = Message(
                 type="A2ARequest",
-                payload=json.dumps({"path": AGENT_CARD_WELL_KNOWN_PATH, "method": method}),
+                payload=json.dumps(
+                    {"path": AGENT_CARD_WELL_KNOWN_PATH, "method": method}
+                ),
                 route_path=AGENT_CARD_WELL_KNOWN_PATH,
-                method=method
+                method=method,
             )
             response = await transport.request(topic, request, ResponseMode.FIRST)
 
             response.payload = json.loads(response.payload.decode("utf-8"))
             card = AgentCard.model_validate(response.payload)
         except Exception as e:
-            logger.info(f"A2A v3 path failed or invalid payload, falling back to v2: {e}")
+            logger.info(
+                f"A2A v3 path failed or invalid payload, falling back to v2: {e}"
+            )
 
             request = Message(
                 type="A2ARequest",
-                payload=json.dumps({"path": PREV_AGENT_CARD_WELL_KNOWN_PATH, "method": method}),
+                payload=json.dumps(
+                    {"path": PREV_AGENT_CARD_WELL_KNOWN_PATH, "method": method}
+                ),
                 route_path=PREV_AGENT_CARD_WELL_KNOWN_PATH,
-                method=method
+                method=method,
             )
             response = await transport.request(topic, request, ResponseMode.FIRST)
 
@@ -203,6 +221,8 @@ async def broadcast_message(
             recipients: List[str] | None = None,
             broadcast_topic: str = None,
             timeout: float = 30.0,
+            group_chat: bool = False,
+            end_message: str = "work-done",
         ) -> List[SendMessageResponse]:
             """
             Broadcast a request using the provided transport.
@@ -217,12 +237,16 @@ async def broadcast_message(
             if not broadcast_topic:
                 broadcast_topic = topic
 
+            # determine response mode, either collect len(recipients) or group chat
+            resp_mode = ResponseMode.GROUP if group_chat else ResponseMode.COLLECT_ALL
+
             try:
                 responses = await transport.request(
                     broadcast_topic,
                     msg,
-                    response_mode=ResponseMode.COLLECT_ALL,
+                    response_mode=resp_mode,
                     recipients=recipients,
+                    end_message=end_message,
                     timeout=timeout,
                 )
             except Exception as e:
@@ -305,6 +329,19 @@ async def handle_message(self, message: Message) -> Message:
         )
         method = message.method
 
+        # check if the body is a JSONRPCSuccessResponse, and if so, convert it to a SendMessageRequest
+        try:
+            inner = JSONRPCSuccessResponse.model_validate_json(body)
+            msg_params = {"message": inner.result}
+            request = SendMessageRequest(
+                id=str(uuid4()), params=MessageSendParams(**msg_params)
+            )
+            body = json.dumps(
+                request.model_dump(mode="json", exclude_none=True)
+            ).encode("utf-8")
+        except Exception:
+            pass
+
         headers = []
         for key, value in message.headers.items():
             if isinstance(value, str):

src/agntcy_app_sdk/protocols/message.py

@@ -16,8 +16,8 @@ def __init__(
         type: str,
         payload: bytes,
         reply_to: Optional[str] = None,
-        route_path: Optional[str] = None,
-        method: Optional[str] = None,
+        route_path: Optional[str] = "/",
+        method: Optional[str] = "POST",
         headers: Optional[dict] = None,
         status_code: Optional[int] = None,
     ):

src/agntcy_app_sdk/transports/nats/transport.py

@@ -103,9 +103,12 @@ async def _connect(self):
     async def close(self) -> None:
         """Close the NATS connection."""
         if self._nc:
-            await self._nc.drain()
-            await self._nc.close()
-            logger.info("NATS connection closed")
+            try:
+                await self._nc.drain()
+                await self._nc.close()
+                logger.info("NATS connection closed")
+            except Exception as e:
+                logger.error(f"Error closing NATS connection: {e}")
         else:
             logger.warning("No NATS connection to close")
 
@@ -191,6 +194,7 @@ async def _request_all(
         message: Message,
         recipients: List[str] = None,
         timeout: float = 30.0,
+        **kwargs,
     ) -> List[Message]:
         """
         Send a message to topic and wait for a response from all recipients

src/agntcy_app_sdk/transports/slim/common.py

@@ -17,8 +17,6 @@ def split_id(id):
     except ValueError as e:
         raise e
 
-    print(f"split_id: {id} -> {organization}, {namespace}, {app}")
-
     return slim_bindings.PyName(organization, namespace, app)
 
 

src/agntcy_app_sdk/transports/slim/session_manager.py

@@ -1,6 +1,7 @@
 # Copyright AGNTCY Contributors (https://github.com/agntcy)
 # SPDX-License-Identifier: Apache-2.0
 
+import asyncio
 from typing import Dict
 import datetime
 from agntcy_app_sdk.common.logging_config import configure_logging, get_logger
@@ -11,6 +12,7 @@
     PySessionConfiguration,
     PySessionDirection,
 )
+from agntcy_app_sdk.transports.transport import Message
 from threading import Lock
 
 configure_logging()
@@ -42,12 +44,17 @@ async def request_reply_session(
             raise ValueError("SLIM client is not set")
 
         # check if we already have a request-reply session
-        session_key = "RequestReply"
-
-        with self._lock:
-            if session_key in self._sessions:
-                logger.info(f"Reusing existing session: {session_key}")
-                return session_key, self._sessions[session_key]
+        for session_id, (session, q) in self._slim.sessions.items():
+            try:
+                conf = await self._slim.get_session_config(session_id)
+                # compare the type of conf to PySessionConfiguration.FireAndForget
+                if isinstance(conf, PySessionConfiguration.FireAndForget):
+                    return session_id, session
+            except Exception as e:
+                logger.warning(
+                    f"could not retrieve SLIM session config for {session_id}: {e}"
+                )
+                continue
 
         with self._lock:
             session = await self._slim.create_session(
@@ -58,9 +65,7 @@ async def request_reply_session(
                     mls_enabled=mls_enabled,
                 )
             )
-            session_key = "RequestReply"
-            self._sessions[session_key] = session
-            return session_key, session
+            return session.id, session
 
     async def group_broadcast_session(
         self,
@@ -77,15 +82,15 @@ async def group_broadcast_session(
             raise ValueError("SLIM client is not set")
 
         # check if we already have a group broadcast session for this channel and invitees
-        session_key = f"GroupChannel:{channel}:" + ",".join(
+        session_key = f"PySessionConfiguration.Streaming:{channel}:" + ",".join(
             [str(invitee) for invitee in invitees]
         )
         with self._lock:
             if session_key in self._sessions:
                 logger.info(f"Reusing existing group broadcast session: {session_key}")
                 return session_key, self._sessions[session_key]
 
-        logger.info(f"Creating new group broadcast session: {session_key}")
+        logger.debug(f"Creating new group broadcast session: {session_key}")
         with self._lock:
             session_info = await self._slim.create_session(
                 PySessionConfiguration.Streaming(
@@ -99,20 +104,61 @@ async def group_broadcast_session(
             )
 
             for invitee in invitees:
+                logger.debug(f"Inviting {invitee} to session {session_info.id}")
                 await self._slim.set_route(invitee)
                 await self._slim.invite(session_info, invitee)
 
             # store the session info
             self._sessions[session_key] = session_info
             return session_key, session_info
 
-    def close_session(self, session_key: str):
+    async def close_session(
+        self, session: PySessionInfo, remote: PyName = None, end_signal: str = None
+    ):
         """
         Close and remove a session by its key.
         """
-        session = self._sessions.pop(session_key, None)
-        if session:
-            logger.info(f"Closing session: {session_key}")
+        if not self._slim:
+            raise ValueError("SLIM client is not set")
+
+        try:
+            # send the end signal to the remote if provided
+            if remote is not None and end_signal is not None:
+                logger.info(f"Sending end signal '{end_signal}' to remote {remote}")
+
+                end_msg = Message(
+                    type="text/plain",
+                    headers={"x-session-end-message": end_signal},
+                    payload=end_signal,
+                )
+                await self._slim.publish(session, end_msg.serialize(), remote)
+                await asyncio.sleep(
+                    0.25
+                )  # give some time for the message to be sent before closing our side
+
+            await self._slim.delete_session(session.id)
+            logger.debug(f"Closed session: {session.id}")
+
+            # remove from local store
+            self._local_cache_cleanup(session.id)
+        except Exception as e:
+            logger.warning(f"Error closing SLIM session {session.id}: {e}")
+            return
+
+    def _local_cache_cleanup(self, session_id: int):
+        """
+        Perform local cleanup of a session without attempting to close it on the SLIM client.
+        """
+        with self._lock:
+            session_key = None
+            for key, sess in self._sessions.items():
+                if sess.id == session_id:
+                    session_key = key
+                    break
+
+            if session_key:
+                del self._sessions[session_key]
+                logger.debug(f"Locally cleaned up session: {session_id}")
 
     def session_details(self, session_key: str):
         """