diff --git a/examples/tool_call_handler.py b/examples/tool_call_handler.py index 7a56379..3b27ff2 100644 --- a/examples/tool_call_handler.py +++ b/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( diff --git a/src/anam/_streaming.py b/src/anam/_streaming.py index c4f1222..fe49030 100644 --- a/src/anam/_streaming.py +++ b/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. diff --git a/src/anam/_tool_call_manager.py b/src/anam/_tool_call_manager.py index 5b47b90..7789597 100644 --- a/src/anam/_tool_call_manager.py +++ b/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", ""), diff --git a/src/anam/client.py b/src/anam/client.py index defbb26..afccaf7 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -133,7 +133,7 @@ def __init__( event: [] for event in AnamEvent } - # Tool call manager + # Tool call manager (send_tool_result is wired once the streaming client is created) self._tool_call_manager = ToolCallManager(emit=self._emit) # Internal state @@ -262,8 +262,26 @@ async def connect_async(self, session_options: SessionOptions = SessionOptions() custom_ice_servers=self._options.ice_servers, ) + # Bind the tool call manager to this session so it can filter stale events + # and send client tool results back to the engine. + self._tool_call_manager.set_active_session(self._session_info.session_id) + self._tool_call_manager.set_send_tool_result(self._streaming_client.send_tool_result) + # Connect - await self._streaming_client.connect() + try: + await self._streaming_client.connect() + except Exception: + self._tool_call_manager.clear_session_state() + self._tool_call_manager.set_send_tool_result(None) + try: + await self._streaming_client.close() + except Exception as close_err: + logger.warning( + "Error closing streaming client after failed connect: %s", close_err + ) + self._streaming_client = None + self._session_info = None + raise self._is_streaming = True return Session(self) @@ -402,12 +420,10 @@ def register_tool_call_handler( The handler receives callbacks when the named tool is started, completed, or fails. For **client** tools, returning a string from - ``handler.on_start()`` is treated by the local tool call manager as a - successful completion result (it will emit a ``TOOL_CALL_COMPLETED`` - event with that value). Raising an exception will cause a - ``TOOL_CALL_FAILED`` event to be emitted. This behavior affects local - client-side events only and does not by itself send a completion - message to the engine over the data channel. + ``handler.on_start()`` sends the result back to the engine over the + data channel and emits a local ``TOOL_CALL_COMPLETED`` event. + Raising an exception sends the error back to the engine and emits a + local ``TOOL_CALL_FAILED`` event. Args: tool_name: The name of the tool to handle. @@ -458,7 +474,8 @@ async def close(self) -> None: """Close the connection and clean up resources.""" if self._streaming_client and self.is_streaming: self._is_streaming = False - self._tool_call_manager.clear_pending_calls() + self._tool_call_manager.clear_session_state() + self._tool_call_manager.set_send_tool_result(None) await self._handle_connection_closed(ConnectionClosedCode.NORMAL.value, None) await self._streaming_client.close() self._streaming_client = None diff --git a/src/anam/types.py b/src/anam/types.py index a3b04bc..f5a05ec 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -81,20 +81,33 @@ class ClientToolConfig: name: Tool name (must be unique within the persona). description: Description of what the tool does (shown to the LLM). parameters: JSON Schema for the tool's parameters. + await_result: If True, the return value from the registered handler's + ``on_start`` is sent back to the LLM as the tool result so it can + incorporate it into its response. If False (default), the tool + call is fire-and-forget and the LLM continues without the result. + tool_timeout_seconds: How long the engine waits for the client to + return a tool result before timing out. Only applies when + ``await_result`` is True. Range: 1–600 seconds. Defaults to 10 + seconds on the engine when omitted. """ name: str description: str parameters: ToolParametersConfig | None = None + await_result: bool = False + tool_timeout_seconds: int | None = None def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { "type": "client", "name": self.name, "description": self.description, + "awaitResult": self.await_result, } if self.parameters is not None: result["parameters"] = self.parameters.to_dict() + if self.tool_timeout_seconds is not None: + result["toolTimeoutSeconds"] = self.tool_timeout_seconds return result @@ -404,6 +417,7 @@ class ToolCallStartedPayload: Attributes: event_uid: Unique event identifier. + session_id: ID of the session this tool call belongs to. tool_call_id: Unique ID from the LLM for this tool call. tool_name: Name of the tool being called. tool_type: Type of tool ("client" or "server"). @@ -413,6 +427,7 @@ class ToolCallStartedPayload: """ event_uid: str + session_id: str tool_call_id: str tool_name: str tool_type: str @@ -427,6 +442,7 @@ class ToolCallCompletedPayload: Attributes: event_uid: Unique event identifier. + session_id: ID of the session this tool call belongs to. tool_call_id: Unique ID from the LLM for this tool call. tool_name: Name of the tool that was called. tool_type: Type of tool ("client" or "server"). @@ -438,6 +454,7 @@ class ToolCallCompletedPayload: """ event_uid: str + session_id: str tool_call_id: str tool_name: str tool_type: str @@ -454,6 +471,7 @@ class ToolCallFailedPayload: Attributes: event_uid: Unique event identifier. + session_id: ID of the session this tool call belongs to. tool_call_id: Unique ID from the LLM for this tool call. tool_name: Name of the tool that failed. tool_type: Type of tool ("client" or "server"). @@ -464,6 +482,7 @@ class ToolCallFailedPayload: """ event_uid: str + session_id: str tool_call_id: str tool_name: str tool_type: str @@ -480,10 +499,10 @@ class ToolCallHandler: respond to tool call lifecycle events for a specific tool name. For **client** tools, returning a string from ``on_start`` causes the SDK - to emit a local ``TOOL_CALL_COMPLETED`` event with that result. Raising an - exception in ``on_start`` causes a local ``TOOL_CALL_FAILED`` event to be - emitted. These events are handled within the SDK and do not themselves - send results back to the engine over the data channel. + to emit a local ``TOOL_CALL_COMPLETED`` event with that result AND send + the result back to the engine over the data channel so the LLM can + continue the conversation. Raising an exception in ``on_start`` sends the + error back to the engine and emits a local ``TOOL_CALL_FAILED`` event. For **server** tools, ``on_start`` is informational only — the engine handles execution and sends completed/failed events. @@ -492,10 +511,10 @@ class ToolCallHandler: async def on_start(self, payload: ToolCallStartedPayload) -> str | None: """Called when a tool call starts. - For client tools, return a string to auto-complete the call by - emitting a local ``TOOL_CALL_COMPLETED`` event with that result. - Return ``None`` if the handler only needs the start notification. - The SDK does not currently pass the result of client tool calls back to the engine""" + For client tools, return a string to send the result back to the + engine and emit a local ``TOOL_CALL_COMPLETED`` event. Return + ``None`` if the handler only needs the start notification and no + result should be sent back to the engine.""" return None async def on_complete(self, payload: ToolCallCompletedPayload) -> None: diff --git a/tests/test_tool_call_manager.py b/tests/test_tool_call_manager.py index 3a7b532..c6dbec3 100644 --- a/tests/test_tool_call_manager.py +++ b/tests/test_tool_call_manager.py @@ -1,7 +1,5 @@ """Tests for ToolCallManager.""" -import asyncio - import pytest from anam._tool_call_manager import ToolCallManager @@ -66,11 +64,36 @@ def emitted_events(self): return [] @pytest.fixture - def manager(self, emitted_events): + def sent_tool_results(self): + return [] + + @pytest.fixture + def manager(self, emitted_events, sent_tool_results): async def emit(event, *args, **kwargs): emitted_events.append((event, args, kwargs)) - return ToolCallManager(emit=emit) + def send_tool_result( + session_id, + tool_call_id, + user_action_correlation_id, + timestamp_user_action, + result, + error_message, + ): + sent_tool_results.append( + { + "session_id": session_id, + "tool_call_id": tool_call_id, + "user_action_correlation_id": user_action_correlation_id, + "timestamp_user_action": timestamp_user_action, + "result": result, + "error_message": error_message, + } + ) + + mgr = ToolCallManager(emit=emit, send_tool_result=send_tool_result) + mgr.set_active_session("sess-1") + return mgr @pytest.mark.asyncio async def test_started_event_emits_public_event(self, manager, emitted_events): @@ -145,7 +168,9 @@ async def test_execution_time_on_failure(self, manager, emitted_events): assert payload.execution_time == pytest.approx(2000.0, abs=1.0) @pytest.mark.asyncio - async def test_client_tool_auto_complete_on_handler_return(self, manager, emitted_events): + async def test_client_tool_auto_complete_on_handler_return( + self, manager, emitted_events, sent_tool_results + ): """Test that client tools auto-complete when handler returns a string.""" class MyHandler: @@ -168,8 +193,37 @@ async def on_fail(self, payload): completed_payload = emitted_events[1][1][0] assert completed_payload.result == "my_result" + # Should send the result back to the engine over the data channel + assert len(sent_tool_results) == 1 + sent = sent_tool_results[0] + assert sent["session_id"] == "sess-1" + assert sent["tool_call_id"] == "tc-1" + assert sent["user_action_correlation_id"] == "corr-1" + assert sent["timestamp_user_action"] == "2025-01-01T00:00:00+00:00" + assert sent["result"] == "my_result" + assert sent["error_message"] is None + @pytest.mark.asyncio - async def test_client_tool_auto_fail_on_handler_exception(self, manager, emitted_events): + async def test_client_tool_no_result_not_sent( + self, manager, emitted_events, sent_tool_results + ): + """Returning None from on_start should NOT send a tool_result or auto-complete.""" + + class MyHandler: + async def on_start(self, payload): + return None + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + + # Only STARTED is emitted, no auto-complete and no tool_result sent + assert [e[0] for e in emitted_events] == [AnamEvent.TOOL_CALL_STARTED] + assert sent_tool_results == [] + + @pytest.mark.asyncio + async def test_client_tool_auto_fail_on_handler_exception( + self, manager, emitted_events, sent_tool_results + ): """Test that client tools auto-fail when handler raises.""" class MyHandler: @@ -192,6 +246,48 @@ async def on_fail(self, payload): failed_payload = emitted_events[1][1][0] assert "handler error" in failed_payload.error_message + # Should send the error back to the engine + assert len(sent_tool_results) == 1 + sent = sent_tool_results[0] + assert sent["result"] is None + assert "handler error" in sent["error_message"] + + @pytest.mark.asyncio + async def test_failed_call_blocks_subsequent_completed( + self, manager, emitted_events + ): + """A completed event arriving after a failure should be ignored.""" + + class MyHandler: + async def on_start(self, payload): + raise ValueError("handler error") + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + # Stale completed event for the same tool_call_id should be ignored + await manager.process_completed_event(_make_completed_event()) + assert emitted_events == [] + + @pytest.mark.asyncio + async def test_events_filtered_by_active_session(self, manager, emitted_events): + """Events whose session_id doesn't match the active session are dropped.""" + await manager.process_started_event(_make_started_event(session_id="sess-other")) + await manager.process_completed_event(_make_completed_event(session_id="sess-other")) + await manager.process_failed_event(_make_failed_event(session_id="sess-other")) + + assert emitted_events == [] + + @pytest.mark.asyncio + async def test_clear_session_state_drops_subsequent_events( + self, manager, emitted_events + ): + """After clear_session_state, subsequent events are ignored until a new session is set.""" + manager.clear_session_state() + await manager.process_started_event(_make_started_event()) + assert emitted_events == [] + @pytest.mark.asyncio async def test_server_tool_does_not_auto_complete(self, manager, emitted_events): """Test that server tools don't auto-complete from handler return.""" @@ -366,6 +462,8 @@ async def test_handle_data_message_tool_call_started(self): from anam import AnamClient client = AnamClient(api_key="test-key", persona_id="test-persona") + # Simulate a connected session so the manager accepts events + client._tool_call_manager.set_active_session("sess-1") received = [] @@ -389,6 +487,8 @@ async def test_handle_data_message_tool_call_completed(self): from anam import AnamClient client = AnamClient(api_key="test-key", persona_id="test-persona") + # Simulate a connected session so the manager accepts events + client._tool_call_manager.set_active_session("sess-1") received = [] @@ -419,6 +519,8 @@ async def test_handle_data_message_tool_call_failed(self): from anam import AnamClient client = AnamClient(api_key="test-key", persona_id="test-persona") + # Simulate a connected session so the manager accepts events + client._tool_call_manager.set_active_session("sess-1") received = []