Skip to main content

Advanced features

State Management​

Enable state tracking to persist information across interactions:

# Enable state tracking in the constructor
super().__init__(
    name="stateful-agent",
    enable_state_tracking=True,  # Automatically registers startup_hook and hangup_hook
    state_manager=FileStateManager(storage_dir="./state")  # Optional custom state manager
)

# Access and update state
@AgentBase.tool(
    name="save_preference",
    description="Save a user preference",
    parameters={
        "key": {
            "type": "string",
            "description": "The preference key"
        },
        "value": {
            "type": "string",
            "description": "The preference value"
        }
    }
)
def save_preference(self, args, raw_data):
    # Get the call ID from the raw data
    call_id = raw_data.get("call_id")
    
    if call_id:
        # Get current state or empty dict if none exists
        state = self.get_state(call_id) or {}
        
        # Update the state
        preferences = state.get("preferences", {})
        preferences[args.get("key")] = args.get("value")
        state["preferences"] = preferences
        
        # Save the updated state
        self.update_state(call_id, state)
        
        return SwaigFunctionResult("Preference saved")
    else:
        return SwaigFunctionResult("Could not save preference: No call ID")

SIP Routing​

SIP routing allows your agents to receive voice calls via SIP addresses. The SDK supports both individual agent-level routing and centralized server-level routing.

Individual Agent SIP Routing​

Enable SIP routing on a single agent:

# Enable SIP routing with automatic username mapping based on agent name
agent.enable_sip_routing(auto_map=True)

# Register additional SIP usernames for this agent
agent.register_sip_username("support_agent")
agent.register_sip_username("help_desk")

When auto_map=True, the agent automatically registers SIP usernames based on:

  • The agent's name (e.g., support@domain)
  • The agent's route path (e.g., /support becomes support@domain)
  • Common variations (e.g., removing vowels for shorter dialing)

Server-Level SIP Routing (Multi-Agent)​

For multi-agent setups, centralized routing is more efficient:

# Create an AgentServer
server = AgentServer(host="0.0.0.0", port=3000)

# Register multiple agents
server.register(registration_agent)  # Route: /register
server.register(support_agent)       # Route: /support

# Set up central SIP routing
server.setup_sip_routing(route="/sip", auto_map=True)

# Register additional SIP username mappings
server.register_sip_username("signup", "/register")    # signup@domain → registration agent
server.register_sip_username("help", "/support")       # help@domain → support agent

With server-level routing:

  • Each agent is reachable via its name (when auto_map=True)
  • Additional SIP usernames can be mapped to specific agent routes
  • All SIP routing is handled at a single endpoint (/sip by default)

How SIP Routing Works​

  1. A SIP call comes in with a username (e.g., support@yourdomain)
  2. The SDK extracts the username part (support)
  3. The system checks if this username is registered:
    • In individual routing: The current agent checks its own username list
    • In server routing: The server checks its central mapping table
  4. If a match is found, the call is routed to the appropriate agent

Custom Routing​

You can dynamically handle requests to different paths using routing callbacks:

# Enable custom routing in the constructor or anytime after initialization
self.register_routing_callback(self.handle_customer_route, path="/customer")
self.register_routing_callback(self.handle_product_route, path="/product")

# Define the routing handlers
def handle_customer_route(self, request, body):
    """
    Process customer-related requests
    
    Args:
        request: FastAPI Request object
        body: Parsed JSON body as dictionary
        
    Returns:
        Optional[str]: A URL to redirect to, or None to process normally
    """
    # Extract any relevant data
    customer_id = body.get("customer_id")
    
    # You can redirect to another agent/service if needed
    if customer_id and customer_id.startswith("vip-"):
        return f"/vip-handler/{customer_id}"
        
    # Or return None to process the request with on_swml_request
    return None
    
# Customize SWML based on the route in on_swml_request
def on_swml_request(self, request_data=None, callback_path=None):
    """
    Customize SWML based on the request and path
    
    Args:
        request_data: The request body data
        callback_path: The path that triggered the routing callback
    """
    if callback_path == "/customer":
        # Serve customer-specific content
        return {
            "sections": {
                "main": [
                    {"answer": {}},
                    {"play": {"url": "say:Welcome to customer service!"}}
                ]
            }
        }
    # Other path handling...
    return None

Customizing SWML Requests​

You can modify the SWML document based on request data by overriding the on_swml_request method:

def on_swml_request(self, request_data=None, callback_path=None):
    """
    Customize the SWML document based on request data
    
    Args:
        request_data: The request data (body for POST or query params for GET)
        callback_path: The path that triggered the routing callback
        
    Returns:
        Optional dict with modifications to apply to the document
    """
    if request_data and "caller_type" in request_data:
        # Example: Return modifications to change the AI behavior based on caller type
        if request_data["caller_type"] == "vip":
            return {
                "sections": {
                    "main": [
                        # Keep the first verb (answer)
                        # Modify the AI verb parameters
                        {
                            "ai": {
                                "params": {
                                    "wait_for_user": False,
                                    "end_of_speech_timeout": 500  # More responsive
                                }
                            }
                        }
                    ]
                }
            }
            
    # You can also use the callback_path to serve different content based on the route
    if callback_path == "/customer":
        return {
            "sections": {
                "main": [
                    {"answer": {}},
                    {"play": {"url": "say:Welcome to our customer service line."}}
                ]
            }
        }
    
    # Return None to use the default document
    return None

Conversation Summary Handling​

Process conversation summaries:

def on_summary(self, summary, raw_data=None):
    """
    Handle the conversation summary
    
    Args:
        summary: The summary object or None if no summary was found
        raw_data: The complete raw POST data from the request
    """
    if summary:
        # Log the summary
        self.log.info("conversation_summary", summary=summary)
        
        # Save the summary to a database, send notifications, etc.
        # ...

Custom Webhook URLs​

You can override the default webhook URLs for SWAIG functions and post-prompt delivery:

# In your agent initialization or setup code:

# Override the webhook URL for all SWAIG functions
agent.set_web_hook_url("https://external-service.example.com/handle-swaig")

# Override the post-prompt delivery URL
agent.set_post_prompt_url("https://analytics.example.com/conversation-summaries")

# These methods allow you to:
# 1. Send function calls to external services instead of handling them locally
# 2. Send conversation summaries to analytics services or other systems
# 3. Use special URLs with pre-configured authentication

External Input Checking​

The SDK provides a check-for-input endpoint that allows agents to check for new input from external systems:

# Example client code that checks for new input
import requests
import json

def check_for_new_input(agent_url, conversation_id, auth):
    """
    Check if there's any new input for a conversation
    
    Args:
        agent_url: Base URL for the agent
        conversation_id: ID of the conversation to check
        auth: (username, password) tuple for basic auth
    
    Returns:
        New messages if any, None otherwise
    """
    url = f"{agent_url}/check_for_input"
    response = requests.post(
        url,
        json={"conversation_id": conversation_id},
        auth=auth
    )
    
    if response.status_code == 200:
        data = response.json()
        if data.get("new_input", False):
            return data.get("messages", [])
    
    return None

By default, the check_for_input endpoint returns an empty response. To implement custom behavior, override the _handle_check_for_input_request method in your agent:

async def _handle_check_for_input_request(self, request):
    # First do basic authentication check
    if not self._check_basic_auth(request):
        return Response(
            content=json.dumps({"error": "Unauthorized"}),
            status_code=401,
            headers={"WWW-Authenticate": "Basic"},
            media_type="application/json"
        )
    
    # Get conversation_id from request
    conversation_id = None
    if request.method == "POST":
        body = await request.json()
        conversation_id = body.get("conversation_id")
    else:
        conversation_id = request.query_params.get("conversation_id")
    
    if not conversation_id:
        return Response(
            content=json.dumps({"error": "Missing conversation_id"}),
            status_code=400,
            media_type="application/json"
        )
    
    # Custom logic to check for new input
    # For example, checking a database or external API
    messages = self._get_new_messages(conversation_id)
    
    return {
        "status": "success",
        "conversation_id": conversation_id,
        "new_input": len(messages) > 0,
        "messages": messages
    }

This endpoint is useful for implementing asynchronous conversations where users might send messages through different channels that need to be incorporated into the agent conversation.