feat: pass client tool results back to engine

examples/tool_call_handler.py

@@ -1,13 +1,14 @@
-"""Tool call handler example — client-side logging for tool events.
+"""Tool call handler example — client tool with result passback.
 
 This example shows how to:
 1. Define a client tool (get_weather) on an ephemeral persona
 2. Register a ToolCallHandler to observe tool call lifecycle events
-3. Log when a tool starts, completes, or fails on the client side
+3. Return a result from ``on_start`` — the SDK sends it back to the engine
+   over the data channel so the LLM can use it in its response
 
 When the user asks about the weather, the LLM invokes the get_weather
-tool. The handler logs the event and returns a dummy result so the
-LLM can incorporate it into its response.
+tool. The handler returns a JSON string, which the SDK forwards to the
+engine; the LLM then incorporates the result into its spoken reply.
 
 Requirements:
     uv sync --extra display
@@ -68,7 +69,9 @@ class WeatherToolHandler(ToolCallHandler):
     async def on_start(self, payload: ToolCallStartedPayload) -> str | None:
         city = payload.arguments.get("city", "unknown")
         print(f"\n[get_weather] Started — city={city!r}")
-        # Return a dummy result. The LLM wont use this in its response.
+        # Returning a string sends the result back to the engine over the data
+        # channel so the LLM can incorporate it into its response. Returning
+        # None would acknowledge the call locally without sending a result.
         return json.dumps({
             "city": city,
             "temperature_celsius": 18,
@@ -217,6 +220,8 @@ def main() -> None:
     api_base_url = os.environ.get("ANAM_API_BASE_URL", "https://api.anam.ai").strip().strip('"')
 
     # Define the get_weather client tool inline on the ephemeral persona.
+    # await_result=True tells the engine to wait for the handler's return value
+    # and pass it back to the LLM as the tool result.
     get_weather_tool = ClientToolConfig(
         name="get_weather",
         description="Get the current weather for a city.",
@@ -229,6 +234,8 @@ def main() -> None:
             },
             required=["city"],
         ),
+        await_result=True,
+        tool_timeout_seconds=10,
     )
 
     persona_config = PersonaConfig(

src/anam/_streaming.py

@@ -559,6 +559,38 @@ def send_interrupt(self) -> None:
         }
         self.send_data_message(json.dumps(message))
 
+    def send_tool_result(
+        self,
+        session_id: str,
+        tool_call_id: str,
+        user_action_correlation_id: str,
+        timestamp_user_action: str,
+        result: str | None = None,
+        error_message: str | None = None,
+    ) -> None:
+        """Send a client tool result back to the engine over the data channel.
+
+        Args:
+            session_id: The session ID the tool call belongs to.
+            tool_call_id: The tool_call_id from the original toolCallStarted event.
+            user_action_correlation_id: Correlation ID from the original event.
+            timestamp_user_action: User-action timestamp from the original event.
+            result: The tool result string (optional).
+            error_message: Error message if the tool failed (optional).
+        """
+        message: dict[str, str] = {
+            "session_id": session_id,
+            "message_type": "tool_result",
+            "tool_call_id": tool_call_id,
+            "user_action_correlation_id": user_action_correlation_id,
+            "timestamp_user_action": timestamp_user_action,
+        }
+        if result is not None:
+            message["result"] = result
+        if error_message:
+            message["error"] = error_message
+        self.send_data_message(json.dumps(message))
+
     async def send_talk(self, content: str) -> None:
         """Send a single text message directly to TTS via REST API.
 

src/anam/_tool_call_manager.py

@@ -25,6 +25,10 @@
 # Type for the emit callback
 EmitCallback = Callable[..., Awaitable[None]]
 
+# Type for the send_tool_result callback (session_id, tool_call_id,
+# user_action_correlation_id, timestamp_user_action, result, error_message)
+SendToolResultCallback = Callable[[str, str, str, str, str | None, str | None], None]
+
 
 @dataclass
 class _PendingToolCall:
@@ -42,13 +46,38 @@ class ToolCallManager:
     - Maintains a registry of per-tool-name handlers.
     - Tracks pending (in-flight) tool calls for execution time calculation.
     - Converts wire-format (snake_case) events to public payload dataclasses.
-    - For client tools, auto-completes or auto-fails based on handler result.
+    - For client tools, auto-completes or auto-fails based on handler result,
+      and sends the result/error back to the engine over the data channel.
+    - Filters events by active session to avoid processing stale events.
     """
 
-    def __init__(self, emit: EmitCallback) -> None:
+    def __init__(
+        self,
+        emit: EmitCallback,
+        send_tool_result: SendToolResultCallback | None = None,
+    ) -> None:
         self._emit = emit
+        self._send_tool_result = send_tool_result
         self._handlers: dict[str, ToolCallHandler] = {}
         self._pending_calls: dict[str, _PendingToolCall] = {}
+        self._failed_calls: dict[str, ToolCallFailedPayload] = {}
+        self._active_session_id: str | None = None
+
+    def set_send_tool_result(self, send_tool_result: SendToolResultCallback | None) -> None:
+        """Set the callback used to send tool results back to the engine."""
+        self._send_tool_result = send_tool_result
+
+    def set_active_session(self, session_id: str) -> None:
+        """Set the active session ID. Events with a different session_id are ignored."""
+        self._active_session_id = session_id
+        self.clear_pending_calls()
+        self.clear_failed_calls()
+
+    def clear_session_state(self) -> None:
+        """Clear all session-scoped state (active session, pending + failed calls)."""
+        self._active_session_id = None
+        self.clear_pending_calls()
+        self.clear_failed_calls()
 
     def register_handler(self, tool_name: str, handler: ToolCallHandler) -> Callable[[], None]:
         """Register a handler for a specific tool name.
@@ -72,9 +101,21 @@ async def process_started_event(self, event: dict[str, Any]) -> None:
         """Process a toolCallStarted data channel message.
 
         For client tools with a registered handler:
-        - If on_start returns a string, auto-complete the call.
-        - If on_start raises, auto-fail the call.
+        - If on_start returns, send the result back to the engine
+          and auto-complete the call.
+        - If await_result is set to true the engine will wait for this returned response
+          else it will consider the tool call fire-and-forget
+        - If on_start raises, send the error back to the engine and auto-fail.
         """
+        session_id = event.get("session_id", "")
+        if self._active_session_id is None or self._active_session_id != session_id:
+            logger.debug(
+                "Ignoring toolCallStarted for inactive session (event=%s, active=%s)",
+                session_id,
+                self._active_session_id,
+            )
+            return
+
         tool_call_id = event.get("tool_call_id", "")
         tool_name = event.get("tool_name", "")
         tool_type = event.get("tool_type", "")
@@ -108,31 +149,36 @@ async def process_started_event(self, event: dict[str, Any]) -> None:
         handler = self._handlers.get(tool_name)
         if handler and hasattr(handler, "on_start") and handler.on_start is not None:
             if tool_type == "client":
-                # For client tools, auto-complete/fail based on handler result
+                # For client tools, send result/error back to engine and auto-complete/fail
                 try:
                     result = await handler.on_start(payload)
-                    if result is not None:
-                        # Auto-complete with the returned result
-                        completed_payload = self._build_completed_payload(event, result=result)
-                        await self._emit(AnamEvent.TOOL_CALL_COMPLETED, completed_payload)
-                        if tool_call_id:
-                            self._pending_calls.pop(tool_call_id, None)
-                        # Dispatch to handler's on_complete callback
-                        if hasattr(handler, "on_complete") and handler.on_complete is not None:
-                            try:
-                                await handler.on_complete(completed_payload)
-                            except Exception as cb_err:
-                                logger.error(
-                                    "Error in on_complete handler for tool '%s': %s",
-                                    tool_name,
-                                    cb_err,
-                                )
+                    self._send_client_tool_result(event, result=result, error_message=None)
+                    # Auto-complete with the returned result
+                    completed_payload = self._build_completed_payload(event, result=result)
+                    await self._emit(AnamEvent.TOOL_CALL_COMPLETED, completed_payload)
+                    if tool_call_id:
+                        self._pending_calls.pop(tool_call_id, None)
+                    # Dispatch to handler's on_complete callback
+                    if hasattr(handler, "on_complete") and handler.on_complete is not None:
+                        try:
+                            await handler.on_complete(completed_payload)
+                        except Exception as cb_err:
+                            logger.error(
+                                "Error in on_complete handler for tool '%s': %s",
+                                tool_name,
+                                cb_err,
+                            )
                 except Exception as e:
-                    # Auto-fail
+                    # Auto-fail: send error back to engine
+                    error_message = f"Error in handler: {e}"
+                    self._send_client_tool_result(event, result=None, error_message=error_message)
                     failed_payload = self._build_failed_payload(event, error_message=str(e))
                     await self._emit(AnamEvent.TOOL_CALL_FAILED, failed_payload)
                     if tool_call_id:
                         self._pending_calls.pop(tool_call_id, None)
+                    # Track the failed call so a stale completed event doesn't double-process it
+                    if tool_call_id:
+                        self._failed_calls[tool_call_id] = failed_payload
                     # Dispatch to handler's on_fail callback
                     if hasattr(handler, "on_fail") and handler.on_fail is not None:
                         try:
@@ -156,9 +202,23 @@ async def process_started_event(self, event: dict[str, Any]) -> None:
 
     async def process_completed_event(self, event: dict[str, Any]) -> None:
         """Process a toolCallCompleted data channel message."""
+        session_id = event.get("session_id", "")
+        if self._active_session_id is None or self._active_session_id != session_id:
+            logger.debug(
+                "Ignoring toolCallCompleted for inactive session (event=%s, active=%s)",
+                session_id,
+                self._active_session_id,
+            )
+            return
+
         tool_call_id = event.get("tool_call_id", "")
         tool_name = event.get("tool_name", "")
 
+        # If this call was previously marked as failed, do not process it as completed
+        if tool_call_id and tool_call_id in self._failed_calls:
+            del self._failed_calls[tool_call_id]
+            return
+
         # Build payload (calculates execution time from pending call)
         payload = self._to_completed_payload(event)
 
@@ -182,12 +242,25 @@ async def process_completed_event(self, event: dict[str, Any]) -> None:
 
     async def process_failed_event(self, event: dict[str, Any]) -> None:
         """Process a toolCallFailed data channel message."""
+        session_id = event.get("session_id", "")
+        if self._active_session_id is None or self._active_session_id != session_id:
+            logger.debug(
+                "Ignoring toolCallFailed for inactive session (event=%s, active=%s)",
+                session_id,
+                self._active_session_id,
+            )
+            return
+
         tool_call_id = event.get("tool_call_id", "")
         tool_name = event.get("tool_name", "")
 
         # Build payload (calculates execution time from pending call)
         payload = self._to_failed_payload(event)
 
+        # Mark the call as failed so a late completed event is ignored
+        if tool_call_id:
+            self._failed_calls[tool_call_id] = payload
+
         # Clean up pending call
         self._pending_calls.pop(tool_call_id, None)
 
@@ -210,6 +283,42 @@ def clear_pending_calls(self) -> None:
         """Clear all pending tool calls (e.g., on session close)."""
         self._pending_calls.clear()
 
+    def clear_failed_calls(self) -> None:
+        """Clear all tracked failed calls (e.g., on session close)."""
+        self._failed_calls.clear()
+
+    # --- Internal helpers ---
+
+    def _send_client_tool_result(
+        self,
+        started_event: dict[str, Any],
+        result: str | None,
+        error_message: str | None,
+    ) -> None:
+        """Send a tool_result data channel message for a client tool."""
+        if self._send_tool_result is None:
+            logger.debug(
+                "No send_tool_result callback configured; skipping tool_result send."
+            )
+            return
+
+        session_id = started_event.get("session_id", "")
+        tool_call_id = started_event.get("tool_call_id", "")
+        user_action_correlation_id = started_event.get("user_action_correlation_id", "")
+        timestamp_user_action = started_event.get("timestamp_user_action", "")
+
+        try:
+            self._send_tool_result(
+                session_id,
+                tool_call_id,
+                user_action_correlation_id,
+                timestamp_user_action,
+                result,
+                error_message,
+            )
+        except Exception as e:
+            logger.error("Failed to send tool_result for tool_call_id=%s: %s", tool_call_id, e)
+
     # --- Payload conversion helpers ---
 
     def _calculate_execution_time(self, tool_call_id: str, timestamp_str: str) -> float:
@@ -234,6 +343,7 @@ def _calculate_execution_time(self, tool_call_id: str, timestamp_str: str) -> fl
     def _to_started_payload(event: dict[str, Any]) -> ToolCallStartedPayload:
         return ToolCallStartedPayload(
             event_uid=event.get("event_uid", ""),
+            session_id=event.get("session_id", ""),
             tool_call_id=event.get("tool_call_id", ""),
             tool_name=event.get("tool_name", ""),
             tool_type=event.get("tool_type", ""),
@@ -247,6 +357,7 @@ def _to_completed_payload(self, event: dict[str, Any]) -> ToolCallCompletedPaylo
         timestamp = event.get("timestamp", "")
         return ToolCallCompletedPayload(
             event_uid=event.get("event_uid", ""),
+            session_id=event.get("session_id", ""),
             tool_call_id=tool_call_id,
             tool_name=event.get("tool_name", ""),
             tool_type=event.get("tool_type", ""),
@@ -262,6 +373,7 @@ def _to_failed_payload(self, event: dict[str, Any]) -> ToolCallFailedPayload:
         timestamp = event.get("timestamp", "")
         return ToolCallFailedPayload(
             event_uid=event.get("event_uid", ""),
+            session_id=event.get("session_id", ""),
             tool_call_id=tool_call_id,
             tool_name=event.get("tool_name", ""),
             tool_type=event.get("tool_type", ""),
@@ -279,6 +391,7 @@ def _build_completed_payload(
         completion_time = datetime.now(timezone.utc).isoformat()
         return ToolCallCompletedPayload(
             event_uid=started_event.get("event_uid", ""),
+            session_id=started_event.get("session_id", ""),
             tool_call_id=tool_call_id,
             tool_name=started_event.get("tool_name", ""),
             tool_type=started_event.get("tool_type", ""),
@@ -296,6 +409,7 @@ def _build_failed_payload(
         completion_time = datetime.now(timezone.utc).isoformat()
         return ToolCallFailedPayload(
             event_uid=started_event.get("event_uid", ""),
+            session_id=started_event.get("session_id", ""),
             tool_call_id=tool_call_id,
             tool_name=started_event.get("tool_name", ""),
             tool_type=started_event.get("tool_type", ""),