From 038c8482af71a4838adea19bcbd4a5b0adf852f4 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 2 Mar 2026 15:30:28 +0000 Subject: [PATCH 01/10] feat: add tool call lifecycle events and handler registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the tool call events refactor from the JavaScript SDK. Introduces three-phase tool call lifecycle (started → completed/failed) with per-tool-name handler registration and execution time tracking. Co-Authored-By: Claude Opus 4.6 --- src/anam/__init__.py | 8 + src/anam/_tool_call_manager.py | 276 ++++++++++++++++++++ src/anam/client.py | 55 ++++ src/anam/types.py | 119 ++++++++- tests/test_tool_call_manager.py | 449 ++++++++++++++++++++++++++++++++ 5 files changed, 906 insertions(+), 1 deletion(-) create mode 100644 src/anam/_tool_call_manager.py create mode 100644 tests/test_tool_call_manager.py diff --git a/src/anam/__init__.py b/src/anam/__init__.py index 686804e..126c274 100644 --- a/src/anam/__init__.py +++ b/src/anam/__init__.py @@ -65,6 +65,10 @@ async def consume_audio(): PersonaConfig, SessionOptions, SessionReplayOptions, + ToolCallCompletedPayload, + ToolCallFailedPayload, + ToolCallHandler, + ToolCallStartedPayload, ) __all__ = [ @@ -86,6 +90,10 @@ async def consume_audio(): "PersonaConfig", "SessionOptions", "SessionReplayOptions", + "ToolCallCompletedPayload", + "ToolCallFailedPayload", + "ToolCallHandler", + "ToolCallStartedPayload", "VideoFrame", # Errors "AnamError", diff --git a/src/anam/_tool_call_manager.py b/src/anam/_tool_call_manager.py new file mode 100644 index 0000000..f9ca3e0 --- /dev/null +++ b/src/anam/_tool_call_manager.py @@ -0,0 +1,276 @@ +"""Tool call lifecycle manager. + +Handles the started → completed/failed lifecycle for tool calls received +over the WebRTC data channel, including handler dispatch and execution +time tracking. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Awaitable, Callable + +from .types import ( + AnamEvent, + ToolCallCompletedPayload, + ToolCallFailedPayload, + ToolCallHandler, + ToolCallStartedPayload, +) + +logger = logging.getLogger(__name__) + +# Type for the emit callback +EmitCallback = Callable[..., Awaitable[None]] + + +@dataclass +class _PendingToolCall: + """Tracks an in-flight tool call for execution time calculation.""" + + tool_call_id: str + tool_name: str + started_at: datetime + + +class ToolCallManager: + """Manages tool call lifecycle events. + + Responsibilities: + - 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. + """ + + def __init__(self, emit: EmitCallback) -> None: + self._emit = emit + self._handlers: dict[str, ToolCallHandler] = {} + self._pending_calls: dict[str, _PendingToolCall] = {} + + def register_handler(self, tool_name: str, handler: ToolCallHandler) -> Callable[[], None]: + """Register a handler for a specific tool name. + + Args: + tool_name: The name of the tool to handle. + handler: An object implementing the ToolCallHandler protocol. + + Returns: + An unsubscribe function that removes the handler when called. + """ + self._handlers[tool_name] = handler + + def unsubscribe() -> None: + if self._handlers.get(tool_name) is handler: + del self._handlers[tool_name] + + return unsubscribe + + 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. + """ + tool_call_id = event.get("tool_call_id", "") + tool_name = event.get("tool_name", "") + tool_type = event.get("tool_type", "") + timestamp_str = event.get("timestamp", "") + + # Track the pending call + try: + started_at = datetime.fromisoformat(timestamp_str) + except (ValueError, TypeError): + started_at = datetime.now(timezone.utc) + + self._pending_calls[tool_call_id] = _PendingToolCall( + tool_call_id=tool_call_id, + tool_name=tool_name, + started_at=started_at, + ) + + # Build public payload + payload = self._to_started_payload(event) + + # Emit public event + await self._emit(AnamEvent.TOOL_CALL_STARTED, payload) + + # Dispatch to handler + 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 + 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) + self._pending_calls.pop(tool_call_id, None) + except Exception as e: + # Auto-fail + failed_payload = self._build_failed_payload( + event, error_message=str(e) + ) + await self._emit(AnamEvent.TOOL_CALL_FAILED, failed_payload) + self._pending_calls.pop(tool_call_id, None) + else: + # For server tools, just call on_start informally + try: + await handler.on_start(payload) + except Exception as e: + logger.error( + "Error in on_start handler for server tool '%s': %s", + tool_name, + e, + ) + + async def process_completed_event(self, event: dict[str, Any]) -> None: + """Process a toolCallCompleted data channel message.""" + 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_completed_payload(event) + + # Clean up pending call + self._pending_calls.pop(tool_call_id, None) + + # Emit public event + await self._emit(AnamEvent.TOOL_CALL_COMPLETED, payload) + + # Dispatch to handler + handler = self._handlers.get(tool_name) + if handler and hasattr(handler, "on_complete") and handler.on_complete is not None: + try: + await handler.on_complete(payload) + except Exception as e: + logger.error( + "Error in on_complete handler for tool '%s': %s", + tool_name, + e, + ) + + async def process_failed_event(self, event: dict[str, Any]) -> None: + """Process a toolCallFailed data channel message.""" + 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) + + # Clean up pending call + self._pending_calls.pop(tool_call_id, None) + + # Emit public event + await self._emit(AnamEvent.TOOL_CALL_FAILED, payload) + + # Dispatch to handler + handler = self._handlers.get(tool_name) + if handler and hasattr(handler, "on_fail") and handler.on_fail is not None: + try: + await handler.on_fail(payload) + except Exception as e: + logger.error( + "Error in on_fail handler for tool '%s': %s", + tool_name, + e, + ) + + def clear_pending_calls(self) -> None: + """Clear all pending tool calls (e.g., on session close).""" + self._pending_calls.clear() + + # --- Payload conversion helpers --- + + def _calculate_execution_time(self, tool_call_id: str, timestamp_str: str) -> float: + """Calculate execution time in ms from pending call start to the given timestamp.""" + pending = self._pending_calls.get(tool_call_id) + if not pending: + return 0.0 + try: + end_time = datetime.fromisoformat(timestamp_str) + except (ValueError, TypeError): + end_time = datetime.now(timezone.utc) + delta = end_time - pending.started_at + return delta.total_seconds() * 1000 + + @staticmethod + def _to_started_payload(event: dict[str, Any]) -> ToolCallStartedPayload: + return ToolCallStartedPayload( + event_uid=event.get("event_uid", ""), + tool_call_id=event.get("tool_call_id", ""), + tool_name=event.get("tool_name", ""), + tool_type=event.get("tool_type", ""), + tool_subtype=event.get("tool_subtype"), + arguments=event.get("arguments", {}), + timestamp=event.get("timestamp", ""), + ) + + def _to_completed_payload(self, event: dict[str, Any]) -> ToolCallCompletedPayload: + tool_call_id = event.get("tool_call_id", "") + timestamp = event.get("timestamp", "") + return ToolCallCompletedPayload( + event_uid=event.get("event_uid", ""), + tool_call_id=tool_call_id, + tool_name=event.get("tool_name", ""), + tool_type=event.get("tool_type", ""), + tool_subtype=event.get("tool_subtype"), + result=event.get("result"), + execution_time=self._calculate_execution_time(tool_call_id, timestamp), + timestamp=timestamp, + documents_accessed=event.get("documents_accessed"), + ) + + def _to_failed_payload(self, event: dict[str, Any]) -> ToolCallFailedPayload: + tool_call_id = event.get("tool_call_id", "") + timestamp = event.get("timestamp", "") + return ToolCallFailedPayload( + event_uid=event.get("event_uid", ""), + tool_call_id=tool_call_id, + tool_name=event.get("tool_name", ""), + tool_type=event.get("tool_type", ""), + tool_subtype=event.get("tool_subtype"), + error_message=event.get("error_message", ""), + execution_time=self._calculate_execution_time(tool_call_id, timestamp), + timestamp=timestamp, + ) + + def _build_completed_payload( + self, started_event: dict[str, Any], result: Any + ) -> ToolCallCompletedPayload: + """Build a completed payload from a started event (for client tool auto-complete).""" + tool_call_id = started_event.get("tool_call_id", "") + timestamp = started_event.get("timestamp", "") + return ToolCallCompletedPayload( + event_uid=started_event.get("event_uid", ""), + tool_call_id=tool_call_id, + tool_name=started_event.get("tool_name", ""), + tool_type=started_event.get("tool_type", ""), + tool_subtype=started_event.get("tool_subtype"), + result=result, + execution_time=self._calculate_execution_time(tool_call_id, timestamp), + timestamp=timestamp, + ) + + def _build_failed_payload( + self, started_event: dict[str, Any], error_message: str + ) -> ToolCallFailedPayload: + """Build a failed payload from a started event (for client tool auto-fail).""" + tool_call_id = started_event.get("tool_call_id", "") + timestamp = started_event.get("timestamp", "") + return ToolCallFailedPayload( + event_uid=started_event.get("event_uid", ""), + tool_call_id=tool_call_id, + tool_name=started_event.get("tool_name", ""), + tool_type=started_event.get("tool_type", ""), + tool_subtype=started_event.get("tool_subtype"), + error_message=error_message, + execution_time=self._calculate_execution_time(tool_call_id, timestamp), + timestamp=timestamp, + ) diff --git a/src/anam/client.py b/src/anam/client.py index 99936e0..c6b683b 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -15,6 +15,7 @@ from ._api import CoreApiClient from ._streaming import StreamingClient from ._talk_message_stream import TalkMessageStream +from ._tool_call_manager import ToolCallManager from .errors import ConfigurationError, SessionError from .types import ( AgentAudioInputConfig, @@ -27,6 +28,7 @@ PersonaConfig, SessionInfo, SessionOptions, + ToolCallHandler, ) logger = logging.getLogger(__name__) @@ -131,6 +133,9 @@ def __init__( event: [] for event in AnamEvent } + # Tool call manager + self._tool_call_manager = ToolCallManager(emit=self._emit) + # Internal state self._api_client: CoreApiClient | None = None self._session_info: SessionInfo | None = None @@ -316,6 +321,18 @@ async def _handle_data_message(self, data: dict[str, Any]) -> None: AnamEvent.MESSAGE_HISTORY_UPDATED, self._message_history.copy() ) + elif message_type == "toolCallStarted": + event_data = data.get("data", data) + await self._tool_call_manager.process_started_event(event_data) + + elif message_type == "toolCallCompleted": + event_data = data.get("data", data) + await self._tool_call_manager.process_completed_event(event_data) + + elif message_type == "toolCallFailed": + event_data = data.get("data", data) + await self._tool_call_manager.process_failed_event(event_data) + def _process_message_stream_event(self, event: MessageStreamEvent, timestamp: str) -> None: """Process a message stream event and update message history.""" # Find existing message with same ID (for both user and persona messages) @@ -364,6 +381,43 @@ async def _handle_connection_closed(self, code: str, reason: str | None) -> None self._is_streaming = False await self._emit(AnamEvent.CONNECTION_CLOSED, code, reason) + def register_tool_call_handler( + self, tool_name: str, handler: ToolCallHandler + ) -> Callable[[], None]: + """Register a handler for tool call lifecycle events. + + The handler receives callbacks when the named tool is started, + completed, or fails. For **client** tools, returning a string from + ``handler.on_start()`` automatically completes the call with that + result; raising an exception automatically fails it. + + Args: + tool_name: The name of the tool to handle. + handler: An object implementing the ToolCallHandler protocol + (with ``on_start``, ``on_complete``, and/or ``on_fail`` methods). + + Returns: + An unsubscribe function that removes the handler when called. + + Example: + ```python + class RedirectHandler: + async def on_start(self, payload): + print(f"Redirecting with args: {payload.arguments}") + return "redirect_success" + + async def on_complete(self, payload): + print(f"Redirect completed in {payload.execution_time}ms") + + async def on_fail(self, payload): + print(f"Redirect failed: {payload.error_message}") + + unsubscribe = client.register_tool_call_handler("redirect", RedirectHandler()) + # Later: unsubscribe() to remove the handler + ``` + """ + return self._tool_call_manager.register_handler(tool_name, handler) + def create_agent_audio_input_stream( self, config: AgentAudioInputConfig ) -> AgentAudioInputStream: @@ -386,6 +440,7 @@ 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() 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 6ab936d..06b74ca 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -1,8 +1,10 @@ """Type definitions for the Anam SDK.""" +from __future__ import annotations + from dataclasses import dataclass, field from enum import Enum -from typing import Any +from typing import Any, Protocol, runtime_checkable class AnamEvent(str, Enum): @@ -21,6 +23,11 @@ class AnamEvent(str, Enum): # Persona events TALK_STREAM_INTERRUPTED = "talk_stream_interrupted" + # Tool call events + TOOL_CALL_STARTED = "tool_call_started" + TOOL_CALL_COMPLETED = "tool_call_completed" + TOOL_CALL_FAILED = "tool_call_failed" + # Error events ERROR = "error" SERVER_WARNING = "server_warning" @@ -259,3 +266,113 @@ def from_api_response(cls, data: dict[str, Any]) -> "SessionInfo": max_reconnection_attempts=client_config.get("maxWsReconnectionAttempts", 5), ice_servers=client_config.get("iceServers", []), ) + + +# --- Tool Call Types --- + + +@dataclass +class ToolCallStartedPayload: + """Payload emitted when a tool call starts. + + Attributes: + event_uid: Unique event identifier. + 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"). + tool_subtype: Subtype for server tools (e.g., "webhook", "rag"). + arguments: Arguments passed to the tool. + timestamp: ISO timestamp of the event. + """ + + event_uid: str + tool_call_id: str + tool_name: str + tool_type: str + tool_subtype: str | None + arguments: dict[str, Any] + timestamp: str + + +@dataclass +class ToolCallCompletedPayload: + """Payload emitted when a tool call completes successfully. + + Attributes: + event_uid: Unique event identifier. + 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"). + tool_subtype: Subtype for server tools (e.g., "webhook", "rag"). + result: The result returned by the tool. + execution_time: Time in milliseconds between started and completed. + timestamp: ISO timestamp of the event. + documents_accessed: Documents accessed by RAG tools (if applicable). + """ + + event_uid: str + tool_call_id: str + tool_name: str + tool_type: str + tool_subtype: str | None + result: Any + execution_time: float + timestamp: str + documents_accessed: list[str] | None = None + + +@dataclass +class ToolCallFailedPayload: + """Payload emitted when a tool call fails. + + Attributes: + event_uid: Unique event identifier. + 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"). + tool_subtype: Subtype for server tools (e.g., "webhook", "rag"). + error_message: Description of the error. + execution_time: Time in milliseconds between started and failed. + timestamp: ISO timestamp of the event. + """ + + event_uid: str + tool_call_id: str + tool_name: str + tool_type: str + tool_subtype: str | None + error_message: str + execution_time: float + timestamp: str + + +@runtime_checkable +class ToolCallHandler(Protocol): + """Protocol for tool call lifecycle handlers. + + Register handlers via ``AnamClient.register_tool_call_handler()`` to + respond to tool call lifecycle events for a specific tool name. + + For **client** tools, returning a string from ``on_start`` automatically + sends the result back to the engine. Raising an exception in ``on_start`` + automatically reports the failure. + + For **server** tools, ``on_start`` is informational only — the engine + handles execution and sends completed/failed events. + """ + + 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 with that + result. Return ``None`` to handle completion separately. + """ + ... + + async def on_complete(self, payload: ToolCallCompletedPayload) -> None: + """Called when a tool call completes successfully.""" + ... + + async def on_fail(self, payload: ToolCallFailedPayload) -> None: + """Called when a tool call fails.""" + ... diff --git a/tests/test_tool_call_manager.py b/tests/test_tool_call_manager.py new file mode 100644 index 0000000..c1085bb --- /dev/null +++ b/tests/test_tool_call_manager.py @@ -0,0 +1,449 @@ +"""Tests for ToolCallManager.""" + +import asyncio + +import pytest + +from anam._tool_call_manager import ToolCallManager +from anam.types import ( + AnamEvent, + ToolCallCompletedPayload, + ToolCallFailedPayload, + ToolCallStartedPayload, +) + + +def _make_started_event(**overrides): + """Create a toolCallStarted event dict.""" + event = { + "event_uid": "evt-1", + "session_id": "sess-1", + "tool_call_id": "tc-1", + "tool_name": "redirect", + "tool_type": "client", + "tool_subtype": None, + "arguments": {"url": "https://example.com"}, + "timestamp": "2025-01-01T00:00:00+00:00", + "timestamp_user_action": "2025-01-01T00:00:00+00:00", + "user_action_correlation_id": "corr-1", + "used_outside_engine": True, + } + event.update(overrides) + return event + + +def _make_completed_event(**overrides): + """Create a toolCallCompleted event dict.""" + event = _make_started_event() + event.update( + { + "result": "success", + "timestamp": "2025-01-01T00:00:01+00:00", # 1 second later + } + ) + event.update(overrides) + return event + + +def _make_failed_event(**overrides): + """Create a toolCallFailed event dict.""" + event = _make_started_event() + event.update( + { + "error_message": "something went wrong", + "timestamp": "2025-01-01T00:00:02+00:00", # 2 seconds later + } + ) + event.update(overrides) + return event + + +class TestToolCallManager: + """Tests for ToolCallManager.""" + + @pytest.fixture + def emitted_events(self): + return [] + + @pytest.fixture + def manager(self, emitted_events): + async def emit(event, *args, **kwargs): + emitted_events.append((event, args, kwargs)) + + return ToolCallManager(emit=emit) + + @pytest.mark.asyncio + async def test_started_event_emits_public_event(self, manager, emitted_events): + """Test that processing a started event emits TOOL_CALL_STARTED.""" + await manager.process_started_event(_make_started_event()) + + assert len(emitted_events) == 1 + event_type, args, _ = emitted_events[0] + assert event_type == AnamEvent.TOOL_CALL_STARTED + payload = args[0] + assert isinstance(payload, ToolCallStartedPayload) + assert payload.tool_call_id == "tc-1" + assert payload.tool_name == "redirect" + assert payload.tool_type == "client" + assert payload.arguments == {"url": "https://example.com"} + + @pytest.mark.asyncio + async def test_completed_event_emits_public_event(self, manager, emitted_events): + """Test that processing a completed event emits TOOL_CALL_COMPLETED.""" + # First process started to track pending call + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + await manager.process_completed_event(_make_completed_event()) + + assert len(emitted_events) == 1 + event_type, args, _ = emitted_events[0] + assert event_type == AnamEvent.TOOL_CALL_COMPLETED + payload = args[0] + assert isinstance(payload, ToolCallCompletedPayload) + assert payload.tool_call_id == "tc-1" + assert payload.result == "success" + + @pytest.mark.asyncio + async def test_failed_event_emits_public_event(self, manager, emitted_events): + """Test that processing a failed event emits TOOL_CALL_FAILED.""" + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + await manager.process_failed_event(_make_failed_event()) + + assert len(emitted_events) == 1 + event_type, args, _ = emitted_events[0] + assert event_type == AnamEvent.TOOL_CALL_FAILED + payload = args[0] + assert isinstance(payload, ToolCallFailedPayload) + assert payload.tool_call_id == "tc-1" + assert payload.error_message == "something went wrong" + + @pytest.mark.asyncio + async def test_execution_time_calculation(self, manager, emitted_events): + """Test that execution time is calculated from started to completed.""" + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + await manager.process_completed_event(_make_completed_event()) + + payload = emitted_events[0][1][0] + # Started at T+0, completed at T+1s = 1000ms + assert payload.execution_time == pytest.approx(1000.0, abs=1.0) + + @pytest.mark.asyncio + async def test_execution_time_on_failure(self, manager, emitted_events): + """Test that execution time is calculated on failure.""" + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + await manager.process_failed_event(_make_failed_event()) + + payload = emitted_events[0][1][0] + # Started at T+0, failed at T+2s = 2000ms + 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 + ): + """Test that client tools auto-complete when handler returns a string.""" + + class MyHandler: + async def on_start(self, payload): + return "my_result" + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + pass + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + + # Should emit STARTED then COMPLETED + assert len(emitted_events) == 2 + assert emitted_events[0][0] == AnamEvent.TOOL_CALL_STARTED + assert emitted_events[1][0] == AnamEvent.TOOL_CALL_COMPLETED + completed_payload = emitted_events[1][1][0] + assert completed_payload.result == "my_result" + + @pytest.mark.asyncio + async def test_client_tool_auto_fail_on_handler_exception( + self, manager, emitted_events + ): + """Test that client tools auto-fail when handler raises.""" + + class MyHandler: + async def on_start(self, payload): + raise ValueError("handler error") + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + pass + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + + # Should emit STARTED then FAILED + assert len(emitted_events) == 2 + assert emitted_events[0][0] == AnamEvent.TOOL_CALL_STARTED + assert emitted_events[1][0] == AnamEvent.TOOL_CALL_FAILED + failed_payload = emitted_events[1][1][0] + assert "handler error" in failed_payload.error_message + + @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.""" + + class MyHandler: + async def on_start(self, payload): + return "ignored_for_server" + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + pass + + manager.register_handler("search", MyHandler()) + await manager.process_started_event( + _make_started_event(tool_name="search", tool_type="server") + ) + + # Should only emit STARTED (no auto-complete for server tools) + assert len(emitted_events) == 1 + assert emitted_events[0][0] == AnamEvent.TOOL_CALL_STARTED + + @pytest.mark.asyncio + async def test_register_handler_returns_unsubscribe(self, manager, emitted_events): + """Test that register_handler returns a working unsubscribe function.""" + call_count = 0 + + class MyHandler: + async def on_start(self, payload): + nonlocal call_count + call_count += 1 + return "result" + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + pass + + unsubscribe = manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + assert call_count == 1 + + unsubscribe() + emitted_events.clear() + await manager.process_started_event( + _make_started_event(tool_call_id="tc-2") + ) + # Handler should not be called after unsubscribe + assert call_count == 1 + + @pytest.mark.asyncio + async def test_completed_handler_called(self, manager, emitted_events): + """Test that on_complete handler is called for completed events.""" + completed_payloads = [] + + class MyHandler: + async def on_start(self, payload): + pass + + async def on_complete(self, payload): + completed_payloads.append(payload) + + async def on_fail(self, payload): + pass + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + await manager.process_completed_event(_make_completed_event()) + + assert len(completed_payloads) == 1 + assert completed_payloads[0].result == "success" + + @pytest.mark.asyncio + async def test_failed_handler_called(self, manager, emitted_events): + """Test that on_fail handler is called for failed events.""" + failed_payloads = [] + + class MyHandler: + async def on_start(self, payload): + pass + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + failed_payloads.append(payload) + + manager.register_handler("redirect", MyHandler()) + await manager.process_started_event(_make_started_event()) + await manager.process_failed_event(_make_failed_event()) + + assert len(failed_payloads) == 1 + assert failed_payloads[0].error_message == "something went wrong" + + @pytest.mark.asyncio + async def test_clear_pending_calls(self, manager, emitted_events): + """Test that clear_pending_calls removes tracked calls.""" + await manager.process_started_event(_make_started_event()) + assert len(manager._pending_calls) == 1 + + manager.clear_pending_calls() + assert len(manager._pending_calls) == 0 + + @pytest.mark.asyncio + async def test_no_handler_still_emits_events(self, manager, emitted_events): + """Test that events are emitted even without a registered handler.""" + await manager.process_started_event(_make_started_event()) + await manager.process_completed_event(_make_completed_event()) + await manager.process_failed_event(_make_failed_event()) + + event_types = [e[0] for e in emitted_events] + assert AnamEvent.TOOL_CALL_STARTED in event_types + assert AnamEvent.TOOL_CALL_COMPLETED in event_types + assert AnamEvent.TOOL_CALL_FAILED in event_types + + @pytest.mark.asyncio + async def test_tool_subtype_passed_through(self, manager, emitted_events): + """Test that tool_subtype is included in payloads.""" + await manager.process_started_event( + _make_started_event(tool_type="server", tool_subtype="webhook") + ) + + payload = emitted_events[0][1][0] + assert payload.tool_subtype == "webhook" + + @pytest.mark.asyncio + async def test_documents_accessed_on_completed(self, manager, emitted_events): + """Test that documents_accessed is included in completed payloads.""" + await manager.process_started_event(_make_started_event()) + emitted_events.clear() + + await manager.process_completed_event( + _make_completed_event(documents_accessed=["doc1.pdf", "doc2.pdf"]) + ) + + payload = emitted_events[0][1][0] + assert payload.documents_accessed == ["doc1.pdf", "doc2.pdf"] + + +class TestAnamClientToolCallIntegration: + """Test tool call integration in AnamClient.""" + + def test_register_tool_call_handler(self): + """Test that register_tool_call_handler works on AnamClient.""" + from anam import AnamClient + + client = AnamClient(api_key="test-key", persona_id="test-persona") + + class MyHandler: + async def on_start(self, payload): + return "result" + + async def on_complete(self, payload): + pass + + async def on_fail(self, payload): + pass + + unsubscribe = client.register_tool_call_handler("redirect", MyHandler()) + assert callable(unsubscribe) + + # Verify handler is registered + assert "redirect" in client._tool_call_manager._handlers + + # Unsubscribe and verify + unsubscribe() + assert "redirect" not in client._tool_call_manager._handlers + + @pytest.mark.asyncio + async def test_handle_data_message_tool_call_started(self): + """Test that _handle_data_message routes toolCallStarted messages.""" + from anam import AnamClient + + client = AnamClient(api_key="test-key", persona_id="test-persona") + + received = [] + + @client.on(AnamEvent.TOOL_CALL_STARTED) + async def handler(payload): + received.append(payload) + + await client._handle_data_message( + { + "messageType": "toolCallStarted", + "data": _make_started_event(), + } + ) + + assert len(received) == 1 + assert received[0].tool_name == "redirect" + + @pytest.mark.asyncio + async def test_handle_data_message_tool_call_completed(self): + """Test that _handle_data_message routes toolCallCompleted messages.""" + from anam import AnamClient + + client = AnamClient(api_key="test-key", persona_id="test-persona") + + received = [] + + @client.on(AnamEvent.TOOL_CALL_COMPLETED) + async def handler(payload): + received.append(payload) + + # First start, then complete + await client._handle_data_message( + { + "messageType": "toolCallStarted", + "data": _make_started_event(), + } + ) + await client._handle_data_message( + { + "messageType": "toolCallCompleted", + "data": _make_completed_event(), + } + ) + + assert len(received) == 1 + assert received[0].result == "success" + + @pytest.mark.asyncio + async def test_handle_data_message_tool_call_failed(self): + """Test that _handle_data_message routes toolCallFailed messages.""" + from anam import AnamClient + + client = AnamClient(api_key="test-key", persona_id="test-persona") + + received = [] + + @client.on(AnamEvent.TOOL_CALL_FAILED) + async def handler(payload): + received.append(payload) + + await client._handle_data_message( + { + "messageType": "toolCallStarted", + "data": _make_started_event(), + } + ) + await client._handle_data_message( + { + "messageType": "toolCallFailed", + "data": _make_failed_event(), + } + ) + + assert len(received) == 1 + assert received[0].error_message == "something went wrong" From dffbb8f107d0c1536eb22199918e897e2777f08a Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 16 Mar 2026 13:10:00 +0900 Subject: [PATCH 02/10] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/anam/_tool_call_manager.py | 44 ++++++++++++++++++++++------------ src/anam/client.py | 10 +++++--- src/anam/types.py | 14 +++++++---- 3 files changed, 45 insertions(+), 23 deletions(-) diff --git a/src/anam/_tool_call_manager.py b/src/anam/_tool_call_manager.py index f9ca3e0..83f58a7 100644 --- a/src/anam/_tool_call_manager.py +++ b/src/anam/_tool_call_manager.py @@ -80,17 +80,23 @@ async def process_started_event(self, event: dict[str, Any]) -> None: tool_type = event.get("tool_type", "") timestamp_str = event.get("timestamp", "") - # Track the pending call + # Track the pending call (only if we have a non-empty tool_call_id) try: started_at = datetime.fromisoformat(timestamp_str) except (ValueError, TypeError): started_at = datetime.now(timezone.utc) - self._pending_calls[tool_call_id] = _PendingToolCall( - tool_call_id=tool_call_id, - tool_name=tool_name, - started_at=started_at, - ) + if tool_call_id: + self._pending_calls[tool_call_id] = _PendingToolCall( + tool_call_id=tool_call_id, + tool_name=tool_name, + started_at=started_at, + ) + else: + logger.warning( + "Received toolCallStarted event without a valid tool_call_id; " + "skipping pending-call tracking." + ) # Build public payload payload = self._to_started_payload(event) @@ -111,14 +117,16 @@ async def process_started_event(self, event: dict[str, Any]) -> None: event, result=result ) await self._emit(AnamEvent.TOOL_CALL_COMPLETED, completed_payload) - self._pending_calls.pop(tool_call_id, None) + if tool_call_id: + self._pending_calls.pop(tool_call_id, None) except Exception as e: # Auto-fail failed_payload = self._build_failed_payload( event, error_message=str(e) ) await self._emit(AnamEvent.TOOL_CALL_FAILED, failed_payload) - self._pending_calls.pop(tool_call_id, None) + if tool_call_id: + self._pending_calls.pop(tool_call_id, None) else: # For server tools, just call on_start informally try: @@ -197,7 +205,13 @@ def _calculate_execution_time(self, tool_call_id: str, timestamp_str: str) -> fl end_time = datetime.fromisoformat(timestamp_str) except (ValueError, TypeError): end_time = datetime.now(timezone.utc) - delta = end_time - pending.started_at + # Normalize both datetimes to UTC if they are timezone-naive to avoid TypeError + if end_time.tzinfo is None: + end_time = end_time.replace(tzinfo=timezone.utc) + started_at = pending.started_at + if started_at.tzinfo is None: + started_at = started_at.replace(tzinfo=timezone.utc) + delta = end_time - started_at return delta.total_seconds() * 1000 @staticmethod @@ -246,7 +260,7 @@ def _build_completed_payload( ) -> ToolCallCompletedPayload: """Build a completed payload from a started event (for client tool auto-complete).""" tool_call_id = started_event.get("tool_call_id", "") - timestamp = started_event.get("timestamp", "") + completion_time = datetime.now(timezone.utc).isoformat() return ToolCallCompletedPayload( event_uid=started_event.get("event_uid", ""), tool_call_id=tool_call_id, @@ -254,8 +268,8 @@ def _build_completed_payload( tool_type=started_event.get("tool_type", ""), tool_subtype=started_event.get("tool_subtype"), result=result, - execution_time=self._calculate_execution_time(tool_call_id, timestamp), - timestamp=timestamp, + execution_time=self._calculate_execution_time(tool_call_id, completion_time), + timestamp=completion_time, ) def _build_failed_payload( @@ -263,7 +277,7 @@ def _build_failed_payload( ) -> ToolCallFailedPayload: """Build a failed payload from a started event (for client tool auto-fail).""" tool_call_id = started_event.get("tool_call_id", "") - timestamp = started_event.get("timestamp", "") + completion_time = datetime.now(timezone.utc).isoformat() return ToolCallFailedPayload( event_uid=started_event.get("event_uid", ""), tool_call_id=tool_call_id, @@ -271,6 +285,6 @@ def _build_failed_payload( tool_type=started_event.get("tool_type", ""), tool_subtype=started_event.get("tool_subtype"), error_message=error_message, - execution_time=self._calculate_execution_time(tool_call_id, timestamp), - timestamp=timestamp, + execution_time=self._calculate_execution_time(tool_call_id, completion_time), + timestamp=completion_time, ) diff --git a/src/anam/client.py b/src/anam/client.py index c6b683b..c7a2a6a 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -388,13 +388,17 @@ 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()`` automatically completes the call with that - result; raising an exception automatically fails it. + ``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. Args: tool_name: The name of the tool to handle. handler: An object implementing the ToolCallHandler protocol - (with ``on_start``, ``on_complete``, and/or ``on_fail`` methods). + (implementing ``on_start``, ``on_complete``, and ``on_fail`` methods). Returns: An unsubscribe function that removes the handler when called. diff --git a/src/anam/types.py b/src/anam/types.py index 06b74ca..ca575fd 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -353,9 +353,11 @@ class ToolCallHandler(Protocol): Register handlers via ``AnamClient.register_tool_call_handler()`` to respond to tool call lifecycle events for a specific tool name. - For **client** tools, returning a string from ``on_start`` automatically - sends the result back to the engine. Raising an exception in ``on_start`` - automatically reports the failure. + 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. For **server** tools, ``on_start`` is informational only — the engine handles execution and sends completed/failed events. @@ -364,8 +366,10 @@ class ToolCallHandler(Protocol): 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 with that - result. Return ``None`` to handle completion separately. + For client tools, return a string to auto-complete the call by + emitting a local ``TOOL_CALL_COMPLETED`` event with that result. + Return ``None`` to handle completion separately (for example, by + emitting the completion or failure event at a later time). """ ... From c33a056f644be33e5a3996f8884a7e282cff17a3 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 16 Mar 2026 13:24:06 +0900 Subject: [PATCH 03/10] allow part implementation of ToolCallHandler --- src/anam/types.py | 13 ++++++------- uv.lock | 2 +- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/src/anam/types.py b/src/anam/types.py index ca575fd..e14dc8a 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -4,7 +4,7 @@ from dataclasses import dataclass, field from enum import Enum -from typing import Any, Protocol, runtime_checkable +from typing import Any class AnamEvent(str, Enum): @@ -346,9 +346,8 @@ class ToolCallFailedPayload: timestamp: str -@runtime_checkable -class ToolCallHandler(Protocol): - """Protocol for tool call lifecycle handlers. +class ToolCallHandler: + """Base class for tool call lifecycle handlers. Register handlers via ``AnamClient.register_tool_call_handler()`` to respond to tool call lifecycle events for a specific tool name. @@ -371,12 +370,12 @@ async def on_start(self, payload: ToolCallStartedPayload) -> str | None: Return ``None`` to handle completion separately (for example, by emitting the completion or failure event at a later time). """ - ... + return None async def on_complete(self, payload: ToolCallCompletedPayload) -> None: """Called when a tool call completes successfully.""" - ... + pass async def on_fail(self, payload: ToolCallFailedPayload) -> None: """Called when a tool call fails.""" - ... + pass diff --git a/uv.lock b/uv.lock index 89f9fe8..b98206d 100644 --- a/uv.lock +++ b/uv.lock @@ -188,7 +188,7 @@ wheels = [ [[package]] name = "anam" -version = "0.3.0" +version = "0.4.0a1" source = { editable = "." } dependencies = [ { name = "aiohttp" }, From f87e18b7365596178a69fb8af65ff452b4e9d082 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 16 Mar 2026 13:47:18 +0900 Subject: [PATCH 04/10] fix tool call handler docs --- src/anam/_tool_call_manager.py | 10 +++------- src/anam/client.py | 6 +++--- tests/test_tool_call_manager.py | 12 +++--------- 3 files changed, 9 insertions(+), 19 deletions(-) diff --git a/src/anam/_tool_call_manager.py b/src/anam/_tool_call_manager.py index 83f58a7..fa8d587 100644 --- a/src/anam/_tool_call_manager.py +++ b/src/anam/_tool_call_manager.py @@ -55,7 +55,7 @@ def register_handler(self, tool_name: str, handler: ToolCallHandler) -> Callable Args: tool_name: The name of the tool to handle. - handler: An object implementing the ToolCallHandler protocol. + handler: A :class:`ToolCallHandler` subclass instance. Returns: An unsubscribe function that removes the handler when called. @@ -113,17 +113,13 @@ async def process_started_event(self, event: dict[str, Any]) -> None: 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 - ) + 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) except Exception as e: # Auto-fail - failed_payload = self._build_failed_payload( - event, error_message=str(e) - ) + 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) diff --git a/src/anam/client.py b/src/anam/client.py index c7a2a6a..22585b9 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -397,15 +397,15 @@ def register_tool_call_handler( Args: tool_name: The name of the tool to handle. - handler: An object implementing the ToolCallHandler protocol - (implementing ``on_start``, ``on_complete``, and ``on_fail`` methods). + handler: A :class:`ToolCallHandler` subclass instance. Override + only the methods you need (``on_start``, ``on_complete``, ``on_fail``). Returns: An unsubscribe function that removes the handler when called. Example: ```python - class RedirectHandler: + class RedirectHandler(ToolCallHandler): async def on_start(self, payload): print(f"Redirecting with args: {payload.arguments}") return "redirect_success" diff --git a/tests/test_tool_call_manager.py b/tests/test_tool_call_manager.py index c1085bb..3a7b532 100644 --- a/tests/test_tool_call_manager.py +++ b/tests/test_tool_call_manager.py @@ -145,9 +145,7 @@ 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): """Test that client tools auto-complete when handler returns a string.""" class MyHandler: @@ -171,9 +169,7 @@ async def on_fail(self, payload): assert completed_payload.result == "my_result" @pytest.mark.asyncio - async def test_client_tool_auto_fail_on_handler_exception( - self, manager, emitted_events - ): + async def test_client_tool_auto_fail_on_handler_exception(self, manager, emitted_events): """Test that client tools auto-fail when handler raises.""" class MyHandler: @@ -242,9 +238,7 @@ async def on_fail(self, payload): unsubscribe() emitted_events.clear() - await manager.process_started_event( - _make_started_event(tool_call_id="tc-2") - ) + await manager.process_started_event(_make_started_event(tool_call_id="tc-2")) # Handler should not be called after unsubscribe assert call_count == 1 From f552d550032d17ff434feb93bdf3450f5da9cbc8 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 26 Mar 2026 21:09:19 +0900 Subject: [PATCH 05/10] example script for tool call handlers --- examples/tool_call_test.py | 482 +++++++++++++++++++++++++++++++++++++ 1 file changed, 482 insertions(+) create mode 100644 examples/tool_call_test.py diff --git a/examples/tool_call_test.py b/examples/tool_call_test.py new file mode 100644 index 0000000..0badc61 --- /dev/null +++ b/examples/tool_call_test.py @@ -0,0 +1,482 @@ +"""Test app for tool call lifecycle events. + +Tests the ToolCallHandler registration and lifecycle callbacks +(on_start, on_complete, on_fail) with both client and server tools. + +Requires a persona with tools configured. Set up tools at https://lab.anam.ai/tools. + +Requirements: + uv sync --extra display + # or: pip install opencv-python sounddevice + +Usage: + export ANAM_API_KEY="your-api-key" + export ANAM_PERSONA_ID="your-persona-id" # Must have tools configured + uv run --extra display python examples/tool_call_test.py +""" + +import asyncio +import json +import logging +import os +import sys +from collections.abc import Callable +from pathlib import Path + +from dotenv import load_dotenv + +from anam import AnamClient, AnamEvent, ClientOptions +from anam.types import ( + PersonaConfig, + ToolCallCompletedPayload, + ToolCallFailedPayload, + ToolCallHandler, + ToolCallStartedPayload, +) + +sys.path.insert(0, str(Path(__file__).parent.parent)) +from examples.utils import AudioPlayer, VideoDisplay, async_input + +load_dotenv() + +# Configure logging +logging.basicConfig( + level=logging.WARNING, + format="%(levelname)s: %(message)s", +) +logging.getLogger("anam").setLevel(logging.DEBUG) +logging.getLogger("websockets").setLevel(logging.WARNING) +logging.getLogger("aiortc").setLevel(logging.WARNING) +logging.getLogger("aioice").setLevel(logging.WARNING) + + +# --- Tool Call Handlers --- + + +class LoggingHandler(ToolCallHandler): + """Logs all tool call lifecycle events. Works for any tool type.""" + + def __init__(self, name: str) -> None: + self.name = name + + async def on_start(self, payload: ToolCallStartedPayload) -> str | None: + print(f"\n{'='*60}") + print(f"TOOL STARTED: {self.name}") + print(f" tool_call_id: {payload.tool_call_id}") + print(f" tool_type: {payload.tool_type}") + print(f" tool_subtype: {payload.tool_subtype}") + print(f" arguments: {json.dumps(payload.arguments, indent=2)}") + print(f" timestamp: {payload.timestamp}") + print(f"{'='*60}") + return None # Don't auto-complete + + async def on_complete(self, payload: ToolCallCompletedPayload) -> None: + print(f"\n{'='*60}") + print(f"TOOL COMPLETED: {self.name}") + print(f" tool_call_id: {payload.tool_call_id}") + print(f" result: {payload.result}") + print(f" execution_time: {payload.execution_time:.1f}ms") + print(f" docs_accessed: {payload.documents_accessed}") + print(f" timestamp: {payload.timestamp}") + print(f"{'='*60}") + + async def on_fail(self, payload: ToolCallFailedPayload) -> None: + print(f"\n{'='*60}") + print(f"TOOL FAILED: {self.name}") + print(f" tool_call_id: {payload.tool_call_id}") + print(f" error_message: {payload.error_message}") + print(f" execution_time: {payload.execution_time:.1f}ms") + print(f" timestamp: {payload.timestamp}") + print(f"{'='*60}") + + +class AutoCompleteHandler(ToolCallHandler): + """Dummy handler that forces a TOOL_CALL_COMPLETED event for testing. + + For client tools only. Returning a string from on_start causes the SDK's + ToolCallManager to immediately emit a local TOOL_CALL_COMPLETED event + with that string as the result. This lets you verify the completed + lifecycle path without needing a real tool implementation or engine + round-trip. + """ + + def __init__(self, name: str, result: str) -> None: + self.name = name + self.result = result + + async def on_start(self, payload: ToolCallStartedPayload) -> str | None: + print(f"\n AUTO-COMPLETE [{self.name}]: returning '{self.result}'") + print(f" arguments: {json.dumps(payload.arguments, indent=2)}") + return self.result + + async def on_complete(self, payload: ToolCallCompletedPayload) -> None: + print(f" AUTO-COMPLETE [{self.name}]: completed in {payload.execution_time:.1f}ms") + print(f" result: {payload.result}") + + +class AutoFailHandler(ToolCallHandler): + """Dummy handler that forces a TOOL_CALL_FAILED event for testing. + + For client tools only. Raising an exception from on_start causes the + SDK's ToolCallManager to immediately emit a local TOOL_CALL_FAILED event + with the exception message as the error. This lets you verify the failure + lifecycle path without needing a real tool error. + """ + + def __init__(self, name: str, error: str) -> None: + self.name = name + self.error = error + + async def on_start(self, _payload: ToolCallStartedPayload) -> str | None: + print(f"\n AUTO-FAIL [{self.name}]: raising error '{self.error}'") + raise RuntimeError(self.error) + + async def on_fail(self, payload: ToolCallFailedPayload) -> None: + print(f" AUTO-FAIL [{self.name}]: failed in {payload.execution_time:.1f}ms") + print(f" error: {payload.error_message}") + + +class OnStartOnlyHandler(ToolCallHandler): + """Dummy handler that only implements on_start (tests partial implementation). + + Verifies that the SDK handles handlers with missing on_complete/on_fail + methods gracefully. Returns None so no auto-complete is triggered. + """ + + async def on_start(self, payload: ToolCallStartedPayload) -> str | None: + print(f"\n ON_START_ONLY: {payload.tool_name} started") + return None + + +# Track registered handlers for unsubscribe testing +_unsubscribers: dict[str, Callable[[], None]] = {} + + +def register_handlers(client: AnamClient, tool_names: list[str]) -> None: + """Register handlers based on user input. + + By default, registers a LoggingHandler for each tool name provided. + Special prefixes change behavior: + - "auto:=" -> AutoCompleteHandler + - "fail:=" -> AutoFailHandler + - "partial:" -> OnStartOnlyHandler + """ + for spec in tool_names: + if spec.startswith("auto:"): + # auto:tool_name=result_string + rest = spec[5:] + name, _, result = rest.partition("=") + handler = AutoCompleteHandler(name, result or "ok") + unsub = client.register_tool_call_handler(name, handler) + _unsubscribers[name] = unsub + print(f" Registered AutoCompleteHandler for '{name}' -> '{result or 'ok'}'") + elif spec.startswith("fail:"): + rest = spec[5:] + name, _, error = rest.partition("=") + handler = AutoFailHandler(name, error or "handler error") + unsub = client.register_tool_call_handler(name, handler) + _unsubscribers[name] = unsub + print(f" Registered AutoFailHandler for '{name}' -> '{error or 'handler error'}'") + elif spec.startswith("partial:"): + name = spec[8:] + handler = OnStartOnlyHandler() + unsub = client.register_tool_call_handler(name, handler) + _unsubscribers[name] = unsub + print(f" Registered OnStartOnlyHandler for '{name}'") + else: + handler = LoggingHandler(spec) + unsub = client.register_tool_call_handler(spec, handler) + _unsubscribers[spec] = unsub + print(f" Registered LoggingHandler for '{spec}'") + + +async def interactive_loop(session, client: AnamClient, display: VideoDisplay) -> None: + """Interactive command loop with tool-testing commands.""" + print("\n" + "=" * 60) + print("Tool Call Test App") + print("=" * 60) + print("Commands:") + print(" m - Send message (trigger tool calls via LLM)") + print(" t - Talk (bypass LLM, direct TTS)") + print(" i - Interrupt") + print(" reg [tool2] ... - Register LoggingHandler(s)") + print(" reg auto:= - Register AutoCompleteHandler (client tools)") + print(" reg fail:= - Register AutoFailHandler (client tools)") + print(" reg partial: - Register OnStartOnlyHandler") + print(" unreg - Unsubscribe handler") + print(" handlers - List registered handlers") + print(" q - Quit") + print("=" * 60) + print("\nTip: Send a message that triggers a tool call to test the handlers.") + print(" e.g., 'm What is the weather in Dublin?' (if weather tool is configured)\n") + + while True: + try: + user_input = await async_input(">> ") + parts = user_input.strip().split() + if not parts: + continue + + command = parts[0].lower() + + if command == "q": + print("Exiting...") + display.stop() + break + + elif command == "m": + if len(parts) < 2: + print("Usage: m ") + continue + text = " ".join(parts[1:]) + try: + await session.send_message(text) + print(f"Sent: {text}") + except Exception as e: + print(f"Error: {e}") + + elif command == "t": + if len(parts) < 2: + print("Usage: t ") + continue + text = " ".join(parts[1:]) + try: + await session.send_talk_stream(text) + print(f"Talk: {text}") + except Exception as e: + print(f"Error: {e}") + + elif command == "i": + try: + await session.interrupt() + print("Interrupted") + except Exception as e: + print(f"Error: {e}") + + elif command == "reg": + if len(parts) < 2: + print("Usage: reg [tool_name2] ...") + print(" reg auto:=") + print(" reg fail:=") + print(" reg partial:") + continue + register_handlers(client, parts[1:]) + + elif command == "unreg": + if len(parts) < 2: + print("Usage: unreg ") + continue + name = parts[1] + if name in _unsubscribers: + _unsubscribers[name]() + del _unsubscribers[name] + print(f"Unsubscribed handler for '{name}'") + else: + print(f"No handler registered for '{name}'") + + elif command == "handlers": + if _unsubscribers: + print("Registered handlers:") + for name in _unsubscribers: + print(f" - {name}") + else: + print("No handlers registered") + + else: + print(f"Unknown command: {command}") + + except KeyboardInterrupt: + print("\nInterrupted by user") + break + except Exception as e: + print(f"Error: {e}") + + +async def stream_session( + client: AnamClient, + display: VideoDisplay, + audio_player: AudioPlayer, +) -> None: + """Run the streaming session.""" + + @client.on(AnamEvent.CONNECTION_ESTABLISHED) + async def on_connected() -> None: + print("Connected!") + + @client.on(AnamEvent.CONNECTION_CLOSED) + async def on_closed(code: str, reason: str | None) -> None: + print(f"Connection closed: {code} - {reason or 'User initiated'}") + + # Tool call event listeners (in addition to handlers, these fire for ALL tools) + @client.on(AnamEvent.TOOL_CALL_STARTED) + async def on_tool_started(payload: ToolCallStartedPayload) -> None: + print(f"\n[EVENT] TOOL_CALL_STARTED: {payload.tool_name} ({payload.tool_type})") + + @client.on(AnamEvent.TOOL_CALL_COMPLETED) + async def on_tool_completed(payload: ToolCallCompletedPayload) -> None: + print( + f"[EVENT] TOOL_CALL_COMPLETED: {payload.tool_name} " + f"({payload.execution_time:.1f}ms)" + ) + + @client.on(AnamEvent.TOOL_CALL_FAILED) + async def on_tool_failed(payload: ToolCallFailedPayload) -> None: + print( + f"[EVENT] TOOL_CALL_FAILED: {payload.tool_name} - {payload.error_message} " + f"({payload.execution_time:.1f}ms)" + ) + + @client.on(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED) + async def on_message_stream(event) -> None: + if event.content_index == 0: + role = "USER" if event.role.value == "user" else "AVATAR" + print(f"\n[{role}] ", end="", flush=True) + print(event.content, end="", flush=True) + if event.end_of_speech: + print() + + async def consume_video(session) -> None: + try: + async for frame in session.video_frames(): + display.update(frame) + except Exception as e: + logging.getLogger(__name__).error(f"Video error: {e}") + + async def consume_audio(session) -> None: + try: + async for frame in session.audio_frames(): + audio_player.add_frame(frame) + except Exception as e: + logging.getLogger(__name__).error(f"Audio error: {e}") + + async with client.connect() as session: + print(f"Session: {session.session_id}") + + video_task = asyncio.create_task(consume_video(session)) + audio_task = asyncio.create_task(consume_audio(session)) + interactive_task = asyncio.create_task(interactive_loop(session, client, display)) + + while display.is_running(): + if not session.is_active: + break + if interactive_task.done(): + break + await asyncio.sleep(0.1) + + video_task.cancel() + audio_task.cancel() + if not interactive_task.done(): + interactive_task.cancel() + + for task in [video_task, audio_task, interactive_task]: + try: + await task + except asyncio.CancelledError: + pass + except Exception as e: + logging.getLogger(__name__).error(f"Task error: {e}") + + if session.is_active: + try: + await session.close() + except Exception as e: + logging.getLogger(__name__).error(f"Close error: {e}") + + +def main() -> None: + api_key = os.environ.get("ANAM_API_KEY", "").strip().strip('"') + persona_id = os.environ.get("ANAM_PERSONA_ID", "").strip().strip('"') + api_base_url = os.environ.get("ANAM_API_BASE_URL", "https://api.anam.ai").strip().strip('"') + + # Support either persona_id or ephemeral config + avatar_id = os.environ.get("ANAM_AVATAR_ID", "").strip().strip('"') + voice_id = os.environ.get("ANAM_VOICE_ID", "").strip().strip('"') + llm_id = os.environ.get("ANAM_LLM_ID", "").strip().strip('"') + + if not api_key: + raise ValueError("Set ANAM_API_KEY environment variable") + + if persona_id: + print(f"Using persona_id: {persona_id}") + client = AnamClient( + api_key=api_key, + persona_id=persona_id, + options=ClientOptions(api_base_url=api_base_url), + ) + elif avatar_id and voice_id and llm_id: + persona_config = PersonaConfig( + avatar_id=avatar_id, + voice_id=voice_id, + llm_id=llm_id, + system_prompt=( + "You are a helpful assistant. When the user asks you to do something " + "that could involve a tool, use the appropriate tool. Keep responses short." + ), + ) + print(f"Using ephemeral persona: avatar={avatar_id}, voice={voice_id}, llm={llm_id}") + client = AnamClient( + api_key=api_key, + persona_config=persona_config, + options=ClientOptions(api_base_url=api_base_url), + ) + else: + raise ValueError( + "Set either ANAM_PERSONA_ID, or all of ANAM_AVATAR_ID + ANAM_VOICE_ID + ANAM_LLM_ID" + ) + + # Pre-register handlers from env var (comma-separated tool specs) + tool_specs = os.environ.get("ANAM_TOOL_HANDLERS", "").strip() + if tool_specs: + print("Pre-registering tool handlers from ANAM_TOOL_HANDLERS:") + register_handlers(client, tool_specs.split(",")) + + display = VideoDisplay() + audio_player = AudioPlayer() + audio_player.start() + + import threading + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + stream_task = loop.create_task(stream_session(client, display, audio_player)) + + def run_async() -> None: + try: + loop.run_until_complete(stream_task) + except asyncio.CancelledError: + pass + except Exception as e: + print(f"Error: {e}") + + thread = threading.Thread(target=run_async, daemon=True) + thread.start() + + try: + display.run() + except KeyboardInterrupt: + print("\nInterrupted") + finally: + display.stop() + audio_player.stop() + if not stream_task.done(): + stream_task.cancel() + thread.join(timeout=2.0) + if not loop.is_running(): + try: + pending = [t for t in asyncio.all_tasks(loop) if not t.done()] + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + except RuntimeError: + pass + finally: + try: + if not loop.is_closed(): + loop.close() + except RuntimeError: + pass + + +if __name__ == "__main__": + main() From 6372e3a2c2824519125b4bf7e40a87f6274e2f0f Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 16 Apr 2026 12:10:32 +0100 Subject: [PATCH 06/10] add ephemeral tools setup --- examples/tool_call_handler.py | 307 ++++++++++++++++++++++++ {examples => scripts}/tool_call_test.py | 4 +- src/anam/__init__.py | 10 + src/anam/_tool_call_manager.py | 20 ++ src/anam/client.py | 15 +- src/anam/types.py | 122 ++++++++++ 6 files changed, 467 insertions(+), 11 deletions(-) create mode 100644 examples/tool_call_handler.py rename {examples => scripts}/tool_call_test.py (99%) diff --git a/examples/tool_call_handler.py b/examples/tool_call_handler.py new file mode 100644 index 0000000..7a56379 --- /dev/null +++ b/examples/tool_call_handler.py @@ -0,0 +1,307 @@ +"""Tool call handler example — client-side logging for tool events. + +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 + +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. + +Requirements: + uv sync --extra display + # or: pip install opencv-python sounddevice + +Usage: + export ANAM_API_KEY="your-api-key" + export ANAM_AVATAR_ID="your-avatar-id" + export ANAM_VOICE_ID="your-voice-id" + export ANAM_LLM_ID="your-llm-id" + uv run --extra display python examples/tool_call_handler.py +""" + +import asyncio +import json +import logging +import os +import sys +from pathlib import Path + +from dotenv import load_dotenv + +from anam import AnamClient, AnamEvent, ClientOptions +from anam.types import ( + ClientToolConfig, + PersonaConfig, + ToolCallCompletedPayload, + ToolCallFailedPayload, + ToolCallHandler, + ToolCallStartedPayload, + ToolParametersConfig, +) + +sys.path.insert(0, str(Path(__file__).parent.parent)) +from examples.utils import AudioPlayer, VideoDisplay, async_input + +load_dotenv() + +REQUIRED_ENV_VARS = ["ANAM_API_KEY", "ANAM_AVATAR_ID", "ANAM_VOICE_ID", "ANAM_LLM_ID"] +missing = [v for v in REQUIRED_ENV_VARS if not os.getenv(v)] +if missing: + raise EnvironmentError(f"Missing required environment variables: {', '.join(missing)}") + +logging.basicConfig(level=logging.WARNING, format="%(levelname)s: %(message)s") +logging.getLogger("anam").setLevel(logging.WARNING) +logging.getLogger("websockets").setLevel(logging.WARNING) +logging.getLogger("aiortc").setLevel(logging.WARNING) +logging.getLogger("aioice").setLevel(logging.WARNING) + + +class WeatherToolHandler(ToolCallHandler): + """Handles get_weather tool calls on the client side. + + Logs the tool call lifecycle and returns a dummy weather result. + In a real application you would call an external API here. + """ + + 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. + return json.dumps({ + "city": city, + "temperature_celsius": 18, + "condition": "Partly cloudy", + }) + + async def on_complete(self, payload: ToolCallCompletedPayload) -> None: + print(f"[get_weather] Completed in {payload.execution_time:.0f}ms") + + async def on_fail(self, payload: ToolCallFailedPayload) -> None: + print(f"[get_weather] Failed: {payload.error_message}") + + +async def interactive_loop(session, display: VideoDisplay) -> None: + """Interactive command loop.""" + print("\n" + "=" * 60) + print("Tool Call Handler Example") + print("=" * 60) + print("Commands:") + print(" m - Send message (trigger tool calls via LLM)") + print(" t - Talk (bypass LLM, direct TTS)") + print(" i - Interrupt") + print(" q - Quit") + print() + print("Try: 'm What is the weather in Dublin?'") + print("=" * 60 + "\n") + + while True: + try: + user_input = await async_input(">> ") + parts = user_input.strip().split() + if not parts: + continue + + command = parts[0].lower() + + if command == "q": + print("Exiting...") + display.stop() + break + elif command == "m": + if len(parts) < 2: + print("Usage: m ") + continue + text = " ".join(parts[1:]) + await session.send_message(text) + print(f"Sent: {text}") + elif command == "t": + if len(parts) < 2: + print("Usage: t ") + continue + text = " ".join(parts[1:]) + await session.send_talk_stream(text) + print(f"Talk: {text}") + elif command == "i": + await session.interrupt() + print("Interrupted") + else: + print(f"Unknown command: {command}") + + except KeyboardInterrupt: + print("\nInterrupted by user") + break + except Exception as e: + print(f"Error: {e}") + + +async def stream_session( + client: AnamClient, + display: VideoDisplay, + audio_player: AudioPlayer, +) -> None: + """Run the streaming session.""" + + @client.on(AnamEvent.CONNECTION_ESTABLISHED) + async def on_connected() -> None: + print("Connected!") + + @client.on(AnamEvent.CONNECTION_CLOSED) + async def on_closed(code: str, reason: str | None) -> None: + print(f"Connection closed: {code} - {reason or 'User initiated'}") + + @client.on(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED) + async def on_message_stream(event) -> None: + if event.content_index == 0: + role = "USER" if event.role.value == "user" else "AVATAR" + print(f"\n[{role}] ", end="", flush=True) + print(event.content, end="", flush=True) + if event.end_of_speech: + print() + + async def consume_video(session) -> None: + try: + async for frame in session.video_frames(): + display.update(frame) + except Exception as e: + logging.getLogger(__name__).error(f"Video error: {e}") + + async def consume_audio(session) -> None: + try: + async for frame in session.audio_frames(): + audio_player.add_frame(frame) + except Exception as e: + logging.getLogger(__name__).error(f"Audio error: {e}") + + async with client.connect() as session: + print(f"Session: {session.session_id}") + + video_task = asyncio.create_task(consume_video(session)) + audio_task = asyncio.create_task(consume_audio(session)) + interactive_task = asyncio.create_task(interactive_loop(session, display)) + + while display.is_running(): + if not session.is_active: + break + if interactive_task.done(): + break + await asyncio.sleep(0.1) + + video_task.cancel() + audio_task.cancel() + if not interactive_task.done(): + interactive_task.cancel() + + for task in [video_task, audio_task, interactive_task]: + try: + await task + except asyncio.CancelledError: + pass + except Exception as e: + logging.getLogger(__name__).error(f"Task error: {e}") + + if session.is_active: + try: + await session.close() + except Exception as e: + logging.getLogger(__name__).error(f"Close error: {e}") + + +def main() -> None: + api_key = os.environ.get("ANAM_API_KEY", "").strip().strip('"') + avatar_id = os.environ.get("ANAM_AVATAR_ID", "").strip().strip('"') + voice_id = os.environ.get("ANAM_VOICE_ID", "").strip().strip('"') + llm_id = os.environ.get("ANAM_LLM_ID", "").strip().strip('"') + avatar_model = os.environ.get("ANAM_AVATAR_MODEL") + 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. + get_weather_tool = ClientToolConfig( + name="get_weather", + description="Get the current weather for a city.", + parameters=ToolParametersConfig( + properties={ + "city": { + "type": "string", + "description": "The city to get weather for", + }, + }, + required=["city"], + ), + ) + + persona_config = PersonaConfig( + avatar_id=avatar_id, + voice_id=voice_id, + llm_id=llm_id, + avatar_model=avatar_model, + system_prompt=( + "You are a helpful assistant. When the user asks about the weather " + "in a city, use the get_weather tool. Keep responses short." + ), + tools=[get_weather_tool], + ) + print(f"Using persona: avatar={avatar_id}, voice={voice_id}, llm={llm_id}") + + client = AnamClient( + api_key=api_key, + persona_config=persona_config, + options=ClientOptions(api_base_url=api_base_url), + ) + + # Register the handler. When the LLM calls get_weather, + # WeatherToolHandler.on_start() logs and returns a dummy result. + client.register_tool_call_handler("get_weather", WeatherToolHandler()) + print("Registered WeatherToolHandler for 'get_weather'") + + display = VideoDisplay() + audio_player = AudioPlayer() + audio_player.start() + + import threading + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + stream_task = loop.create_task(stream_session(client, display, audio_player)) + + def run_async() -> None: + try: + loop.run_until_complete(stream_task) + except asyncio.CancelledError: + pass + except Exception as e: + print(f"Error: {e}") + + thread = threading.Thread(target=run_async, daemon=True) + thread.start() + + try: + display.run() + except KeyboardInterrupt: + print("\nInterrupted") + finally: + display.stop() + audio_player.stop() + if not stream_task.done(): + stream_task.cancel() + thread.join(timeout=2.0) + if not loop.is_running(): + try: + pending = [t for t in asyncio.all_tasks(loop) if not t.done()] + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + except RuntimeError: + pass + finally: + try: + if not loop.is_closed(): + loop.close() + except RuntimeError: + pass + + +if __name__ == "__main__": + main() diff --git a/examples/tool_call_test.py b/scripts/tool_call_test.py similarity index 99% rename from examples/tool_call_test.py rename to scripts/tool_call_test.py index 0badc61..bfb214f 100644 --- a/examples/tool_call_test.py +++ b/scripts/tool_call_test.py @@ -12,7 +12,7 @@ Usage: export ANAM_API_KEY="your-api-key" export ANAM_PERSONA_ID="your-persona-id" # Must have tools configured - uv run --extra display python examples/tool_call_test.py + uv run --extra display python scripts/tool_call_test.py """ import asyncio @@ -428,7 +428,7 @@ def main() -> None: tool_specs = os.environ.get("ANAM_TOOL_HANDLERS", "").strip() if tool_specs: print("Pre-registering tool handlers from ANAM_TOOL_HANDLERS:") - register_handlers(client, tool_specs.split(",")) + register_handlers(client, [s.strip() for s in tool_specs.split(",") if s.strip()]) display = VideoDisplay() audio_player = AudioPlayer() diff --git a/src/anam/__init__.py b/src/anam/__init__.py index 126c274..68a7045 100644 --- a/src/anam/__init__.py +++ b/src/anam/__init__.py @@ -58,7 +58,9 @@ async def consume_audio(): AgentAudioInputConfig, AnamEvent, ClientOptions, + ClientToolConfig, ConnectionClosedCode, + KnowledgeToolConfig, Message, MessageRole, MessageStreamEvent, @@ -69,6 +71,9 @@ async def consume_audio(): ToolCallFailedPayload, ToolCallHandler, ToolCallStartedPayload, + ToolConfig, + ToolParametersConfig, + WebhookToolConfig, ) __all__ = [ @@ -83,7 +88,9 @@ async def consume_audio(): "AnamEvent", "AudioFrame", "ClientOptions", + "ClientToolConfig", "ConnectionClosedCode", + "KnowledgeToolConfig", "Message", "MessageRole", "MessageStreamEvent", @@ -94,6 +101,9 @@ async def consume_audio(): "ToolCallFailedPayload", "ToolCallHandler", "ToolCallStartedPayload", + "ToolConfig", + "ToolParametersConfig", + "WebhookToolConfig", "VideoFrame", # Errors "AnamError", diff --git a/src/anam/_tool_call_manager.py b/src/anam/_tool_call_manager.py index fa8d587..5b47b90 100644 --- a/src/anam/_tool_call_manager.py +++ b/src/anam/_tool_call_manager.py @@ -117,12 +117,32 @@ async def process_started_event(self, event: dict[str, Any]) -> None: 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 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) + # Dispatch to handler's on_fail callback + if hasattr(handler, "on_fail") and handler.on_fail is not None: + try: + await handler.on_fail(failed_payload) + except Exception as cb_err: + logger.error( + "Error in on_fail handler for tool '%s': %s", + tool_name, + cb_err, + ) else: # For server tools, just call on_start informally try: diff --git a/src/anam/client.py b/src/anam/client.py index d131f94..defbb26 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -331,25 +331,22 @@ async def _handle_data_message(self, data: dict[str, Any]) -> None: await self._emit(AnamEvent.USER_SPEECH_STARTED, correlation_id) elif message_type == "userSpeechEnded": await self._emit(AnamEvent.USER_SPEECH_ENDED, correlation_id) - - @staticmethod - def _extract_correlation_id(data: dict[str, Any]) -> str | None: - """Extract a turn correlation ID from backend event payloads.""" - correlation_id = data.get("user_action_correlation_id") or data.get("correlationId") - return correlation_id if isinstance(correlation_id, str) else None - elif message_type == "toolCallStarted": event_data = data.get("data", data) await self._tool_call_manager.process_started_event(event_data) - elif message_type == "toolCallCompleted": event_data = data.get("data", data) await self._tool_call_manager.process_completed_event(event_data) - elif message_type == "toolCallFailed": event_data = data.get("data", data) await self._tool_call_manager.process_failed_event(event_data) + @staticmethod + def _extract_correlation_id(data: dict[str, Any]) -> str | None: + """Extract a turn correlation ID from backend event payloads.""" + correlation_id = data.get("user_action_correlation_id") or data.get("correlationId") + return correlation_id if isinstance(correlation_id, str) else None + def _process_message_stream_event(self, event: MessageStreamEvent, timestamp: str) -> None: """Process a message stream event and update message history.""" # Find existing message with same ID (for both user and persona messages) diff --git a/src/anam/types.py b/src/anam/types.py index d8d7c33..dee05ef 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -54,6 +54,120 @@ class MessageRole(str, Enum): SYSTEM = "system" +@dataclass +class ToolParametersConfig: + """JSON Schema definition for tool parameters (OpenAI-compatible function calling). + + Args: + properties: Map of parameter names to their schema definitions. + required: List of required parameter names. + """ + + properties: dict[str, dict[str, Any]] + required: list[str] | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = {"type": "object", "properties": self.properties} + if self.required is not None: + result["required"] = self.required + return result + + +@dataclass +class ClientToolConfig: + """Configuration for a client tool that triggers events on the client SDK. + + Args: + 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. + """ + + name: str + description: str + parameters: ToolParametersConfig | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "type": "client", + "name": self.name, + "description": self.description, + } + if self.parameters is not None: + result["parameters"] = self.parameters.to_dict() + return result + + +@dataclass +class WebhookToolConfig: + """Configuration for a server webhook tool that calls an external URL. + + Args: + name: Tool name (must be unique within the persona). + description: Description of what the tool does (shown to the LLM). + url: The webhook URL to call. + method: HTTP method (GET, POST, PUT, PATCH, DELETE). Defaults to POST. + headers: Optional HTTP headers to include. + query_parameters: JSON Schema for query parameters (LLM fills values). + parameters: JSON Schema for body parameters (LLM fills values). + await_response: Whether to wait for the webhook response. Defaults to True. + """ + + name: str + description: str + url: str + method: str = "POST" + headers: dict[str, str] | None = None + query_parameters: ToolParametersConfig | None = None + parameters: ToolParametersConfig | None = None + await_response: bool = True + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "type": "server", + "subtype": "webhook", + "name": self.name, + "description": self.description, + "url": self.url, + "method": self.method, + "awaitResponse": self.await_response, + } + if self.headers is not None: + result["headers"] = self.headers + if self.query_parameters is not None: + result["queryParameters"] = self.query_parameters.to_dict() + if self.parameters is not None: + result["parameters"] = self.parameters.to_dict() + return result + + +@dataclass +class KnowledgeToolConfig: + """Configuration for a server knowledge (RAG) tool. + + Args: + name: Tool name (must be unique within the persona). + description: Description of what the tool does (shown to the LLM). + document_folder_ids: List of knowledge folder UUIDs to search. + """ + + name: str + description: str + document_folder_ids: list[str] + + def to_dict(self) -> dict[str, Any]: + return { + "type": "server", + "subtype": "knowledge", + "name": self.name, + "description": self.description, + "documentFolderIds": self.document_folder_ids, + } + + +ToolConfig = ClientToolConfig | WebhookToolConfig | KnowledgeToolConfig + + @dataclass class PersonaConfig: """Configuration for an Anam persona. @@ -71,6 +185,8 @@ class PersonaConfig: max_session_length_seconds: Maximum session duration (optional). enable_audio_passthrough: If True, bypasses Anam's orchestration layer and allows to ingest TTS audio directly through the socket. + tools: List of inline tool configurations for ephemeral personas. + tool_ids: List of pre-created tool IDs (from https://lab.anam.ai/tools). """ persona_id: str | None = None @@ -83,6 +199,8 @@ class PersonaConfig: llm_id: str | None = None max_session_length_seconds: int | None = None enable_audio_passthrough: bool | None = False + tools: list[ToolConfig] | None = None + tool_ids: list[str] | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" @@ -107,6 +225,10 @@ def to_dict(self) -> dict[str, Any]: result["maxSessionLengthSeconds"] = self.max_session_length_seconds if self.enable_audio_passthrough is not None: result["enableAudioPassthrough"] = self.enable_audio_passthrough + if self.tools is not None: + result["tools"] = [t.to_dict() for t in self.tools] + if self.tool_ids is not None: + result["toolIds"] = self.tool_ids return result From fa6169b8c895bf9ce142edda297f07420d7baad2 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 16 Apr 2026 12:14:24 +0100 Subject: [PATCH 07/10] Update src/anam/types.py Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- src/anam/types.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/anam/types.py b/src/anam/types.py index dee05ef..d5473e3 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -494,8 +494,9 @@ async def on_start(self, payload: ToolCallStartedPayload) -> str | None: For client tools, return a string to auto-complete the call by emitting a local ``TOOL_CALL_COMPLETED`` event with that result. - Return ``None`` to handle completion separately (for example, by - emitting the completion or failure event at a later time). + Return ``None`` if the handler only needs the start notification. + The SDK does not currently expose a public API for completing or + failing a client tool call later. """ return None From eeb3134e073499aced8e412ec20dc27127da3b64 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 16 Apr 2026 12:19:15 +0100 Subject: [PATCH 08/10] fix comment --- src/anam/types.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/anam/types.py b/src/anam/types.py index d5473e3..a3b04bc 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -495,9 +495,7 @@ async def on_start(self, payload: ToolCallStartedPayload) -> str | None: 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 expose a public API for completing or - failing a client tool call later. - """ + The SDK does not currently pass the result of client tool calls back to the engine""" return None async def on_complete(self, payload: ToolCallCompletedPayload) -> None: From adcc0a5bd4bf2d18a6b2bea01ecdf006ac1dc9d1 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 16 Apr 2026 12:30:34 +0100 Subject: [PATCH 09/10] fix tool_call_test --- scripts/tool_call_test.py | 73 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 71 insertions(+), 2 deletions(-) diff --git a/scripts/tool_call_test.py b/scripts/tool_call_test.py index bfb214f..6ab3f02 100644 --- a/scripts/tool_call_test.py +++ b/scripts/tool_call_test.py @@ -3,15 +3,32 @@ Tests the ToolCallHandler registration and lifecycle callbacks (on_start, on_complete, on_fail) with both client and server tools. -Requires a persona with tools configured. Set up tools at https://lab.anam.ai/tools. +Supports either a pre-defined persona (ANAM_PERSONA_ID) or an ephemeral +persona (ANAM_AVATAR_ID + ANAM_VOICE_ID + ANAM_LLM_ID). For ephemeral +personas, tools can be defined inline via ANAM_TOOLS (JSON) or referenced +by ID via ANAM_TOOL_IDS. Requirements: uv sync --extra display # or: pip install opencv-python sounddevice Usage: + # With persona_id (tools configured in Lab): export ANAM_API_KEY="your-api-key" - export ANAM_PERSONA_ID="your-persona-id" # Must have tools configured + export ANAM_PERSONA_ID="your-persona-id" + uv run --extra display python scripts/tool_call_test.py + + # With ephemeral persona + inline tools: + export ANAM_API_KEY="your-api-key" + export ANAM_AVATAR_ID="your-avatar-id" + export ANAM_VOICE_ID="your-voice-id" + export ANAM_LLM_ID="your-llm-id" + export ANAM_TOOLS='[{"name":"get_weather","description":"Get weather for a city","parameters":{"properties":{"city":{"type":"string"}},"required":["city"]}}]' + export ANAM_TOOL_HANDLERS="auto:get_weather=sunny" + uv run --extra display python scripts/tool_call_test.py + + # With ephemeral persona + pre-created tool IDs: + export ANAM_TOOL_IDS="uuid-1,uuid-2" uv run --extra display python scripts/tool_call_test.py """ @@ -27,11 +44,13 @@ from anam import AnamClient, AnamEvent, ClientOptions from anam.types import ( + ClientToolConfig, PersonaConfig, ToolCallCompletedPayload, ToolCallFailedPayload, ToolCallHandler, ToolCallStartedPayload, + ToolParametersConfig, ) sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -396,6 +415,54 @@ def main() -> None: if not api_key: raise ValueError("Set ANAM_API_KEY environment variable") + # Default client tool used when no ANAM_TOOLS or ANAM_TOOL_IDS are set + default_tool = ClientToolConfig( + name="get_weather", + description="Get the current weather for a city.", + parameters=ToolParametersConfig( + properties={ + "city": { + "type": "string", + "description": "The city to get weather for", + }, + }, + required=["city"], + ), + ) + + # Parse inline tool definitions from ANAM_TOOLS env var (JSON array) + # Example: ANAM_TOOLS='[{"name":"get_weather","description":"Get weather","parameters":{"properties":{"city":{"type":"string"}},"required":["city"]}}]' + tools = None + tools_json = os.environ.get("ANAM_TOOLS", "").strip() + if tools_json: + raw_tools = json.loads(tools_json) + tools = [] + for t in raw_tools: + params = None + if t.get("parameters"): + params = ToolParametersConfig( + properties=t["parameters"]["properties"], + required=t["parameters"].get("required"), + ) + tools.append(ClientToolConfig( + name=t["name"], + description=t["description"], + parameters=params, + )) + print(f"Inline tools: {[t.name for t in tools]}") + + # Parse pre-created tool IDs from ANAM_TOOL_IDS env var (comma-separated) + tool_ids = None + tool_ids_raw = os.environ.get("ANAM_TOOL_IDS", "").strip() + if tool_ids_raw: + tool_ids = [s.strip() for s in tool_ids_raw.split(",") if s.strip()] + print(f"Tool IDs: {tool_ids}") + + # Fall back to default get_weather tool if nothing else is configured + if not tools and not tool_ids and not persona_id: + tools = [default_tool] + print(f"Using default tool: {default_tool.name}") + if persona_id: print(f"Using persona_id: {persona_id}") client = AnamClient( @@ -412,6 +479,8 @@ def main() -> None: "You are a helpful assistant. When the user asks you to do something " "that could involve a tool, use the appropriate tool. Keep responses short." ), + tools=tools, + tool_ids=tool_ids, ) print(f"Using ephemeral persona: avatar={avatar_id}, voice={voice_id}, llm={llm_id}") client = AnamClient( From 1aebb31cb506c226a21e3631fe2d66e5d958426d Mon Sep 17 00:00:00 2001 From: sr-anam Date: Thu, 16 Apr 2026 12:36:09 +0100 Subject: [PATCH 10/10] Apply suggestions from code review Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- scripts/tool_call_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/tool_call_test.py b/scripts/tool_call_test.py index 6ab3f02..f1ab702 100644 --- a/scripts/tool_call_test.py +++ b/scripts/tool_call_test.py @@ -434,7 +434,7 @@ def main() -> None: # Example: ANAM_TOOLS='[{"name":"get_weather","description":"Get weather","parameters":{"properties":{"city":{"type":"string"}},"required":["city"]}}]' tools = None tools_json = os.environ.get("ANAM_TOOLS", "").strip() - if tools_json: + if tools_json and not persona_id: raw_tools = json.loads(tools_json) tools = [] for t in raw_tools: