SignalWire AI Agents SDK - Complete API Reference

This document provides a comprehensive reference for all public APIs in the SignalWire AI Agents SDK.

AgentBase Class

The AgentBase class is the foundation for creating AI agents. It extends SWMLService and provides comprehensive functionality for building conversational AI agents.

Constructor

AgentBase(
    name: str,
    route: str = "/",
    host: str = "0.0.0.0",
    port: int = 3000,
    basic_auth: Optional[Tuple[str, str]] = None,
    use_pom: bool = True,
    enable_state_tracking: bool = False,
    token_expiry_secs: int = 3600,
    auto_answer: bool = True,
    record_call: bool = False,
    record_format: str = "mp4",
    record_stereo: bool = True,
    state_manager: Optional[StateManager] = None,
    default_webhook_url: Optional[str] = None,
    agent_id: Optional[str] = None,
    native_functions: Optional[List[str]] = None,
    schema_path: Optional[str] = None,
    suppress_logs: bool = False,
    enable_post_prompt_override: bool = False,
    check_for_input_override: bool = False
)

Parameters:

name (str): Human-readable name for the agent
route (str): HTTP route path for the agent (default: "/")
host (str): Host address to bind to (default: "0.0.0.0")
port (int): Port number to listen on (default: 3000)
basic_auth (Optional[Tuple[str, str]]): Username/password for HTTP basic auth
use_pom (bool): Whether to use Prompt Object Model (default: True)
enable_state_tracking (bool): Enable persistent state management (default: False)
token_expiry_secs (int): Security token expiration time (default: 3600)
auto_answer (bool): Automatically answer incoming calls (default: True)
record_call (bool): Record calls by default (default: False)
record_format (str): Recording format: "mp4", "wav", "mp3" (default: "mp4")
record_stereo (bool): Record in stereo (default: True)
state_manager (Optional[StateManager]): Custom state manager instance
default_webhook_url (Optional[str]): Default webhook URL for functions
agent_id (Optional[str]): Unique identifier for the agent
native_functions (Optional[List[str]]): List of native function names to enable
schema_path (Optional[str]): Path to custom SWML schema file
suppress_logs (bool): Suppress logging output (default: False)
enable_post_prompt_override (bool): Allow post-prompt URL override (default: False)
check_for_input_override (bool): Allow check-for-input URL override (default: False)

Core Methods

Deployment and Execution

`run(event=None, context=None, force_mode=None, host=None, port=None)`

Auto-detects deployment environment and runs the agent appropriately.

Parameters:

event: Event object for serverless environments
context: Context object for serverless environments
force_mode (str): Force specific mode: "server", "lambda", "cgi", "cloud_function"
host (Optional[str]): Override host address
port (Optional[int]): Override port number

Usage:

# Auto-detect environment
agent.run()

# Force server mode
agent.run(force_mode="server", host="localhost", port=8080)

# Lambda handler
def lambda_handler(event, context):
    return agent.run(event, context)

`serve(host=None, port=None)`

Explicitly run as HTTP server using FastAPI/Uvicorn.

Parameters:

host (Optional[str]): Host address to bind to
port (Optional[int]): Port number to listen on

Usage:

agent.serve()  # Use constructor defaults
agent.serve(host="0.0.0.0", port=3000)

Prompt Configuration

Text-Based Prompts

`set_prompt_text(text: str) -> AgentBase`

Set the agent's prompt as raw text.

Parameters:

text (str): The complete prompt text

Usage:

agent.set_prompt_text("You are a helpful customer service agent.")

`set_post_prompt(text: str) -> AgentBase`

Set additional text to append after the main prompt.

Parameters:

text (str): Text to append after main prompt

Usage:

agent.set_post_prompt("Always be polite and professional.")

Structured Prompts (POM)

`prompt_add_section`

def prompt_add_section(
    title: str, 
    body: str = "", 
    bullets: Optional[List[str]] = None, 
    numbered: bool = False, 
    numbered_bullets: bool = False, 
    subsections: Optional[List[Dict[str, Any]]] = None
) -> AgentBase

Add a structured section to the prompt using Prompt Object Model.

Parameters:

title (str): Section title/heading
body (str): Main section content (default: "")
bullets (Optional[List[str]]): List of bullet points
numbered (bool): Use numbered sections (default: False)
numbered_bullets (bool): Use numbered bullet points (default: False)
subsections (Optional[List[Dict]]): Nested subsections

Usage:

# Simple section
agent.prompt_add_section("Role", "You are a customer service representative.")

# Section with bullets
agent.prompt_add_section(
    "Guidelines", 
    "Follow these principles:",
    bullets=["Be helpful", "Stay professional", "Listen carefully"]
)

# Numbered bullets
agent.prompt_add_section(
    "Process",
    "Follow these steps:",
    bullets=["Greet the customer", "Identify their need", "Provide solution"],
    numbered_bullets=True
)

`prompt_add_to_section`

def prompt_add_to_section(
    title: str, 
    body: Optional[str] = None, 
    bullet: Optional[str] = None, 
    bullets: Optional[List[str]] = None
) -> AgentBase

Add content to an existing prompt section.

Parameters:

title (str): Title of existing section to modify
body (Optional[str]): Additional body text to append
bullet (Optional[str]): Single bullet point to add
bullets (Optional[List[str]]): Multiple bullet points to add

Usage:

# Add body text to existing section
agent.prompt_add_to_section("Guidelines", "Remember to always verify customer identity.")

# Add single bullet
agent.prompt_add_to_section("Process", bullet="Document the interaction")

# Add multiple bullets
agent.prompt_add_to_section("Process", bullets=["Follow up", "Close ticket"])

`prompt_add_subsection`

def prompt_add_subsection(
    parent_title: str, 
    title: str, 
    body: str = "", 
    bullets: Optional[List[str]] = None
) -> AgentBase

Add a subsection to an existing prompt section.

Parameters:

parent_title (str): Title of parent section
title (str): Subsection title
body (str): Subsection content (default: "")
bullets (Optional[List[str]]): Subsection bullet points

Usage:

agent.prompt_add_subsection(
    "Guidelines",
    "Escalation Rules", 
    "Escalate when:",
    bullets=["Customer is angry", "Technical issue beyond scope"]
)

Voice and Language Configuration

`add_language`

def add_language(
    name: str, 
    code: str, 
    voice: str, 
    speech_fillers: Optional[List[str]] = None, 
    function_fillers: Optional[List[str]] = None, 
    engine: Optional[str] = None, 
    model: Optional[str] = None
) -> AgentBase

Configure voice and language settings for the agent.

Parameters:

name (str): Human-readable language name
code (str): Language code (e.g., "en-US", "es-ES")
voice (str): Voice identifier (e.g., "rime.spore", "nova.luna")
speech_fillers (Optional[List[str]]): Filler phrases during speech processing
function_fillers (Optional[List[str]]): Filler phrases during function execution
engine (Optional[str]): TTS engine to use
model (Optional[str]): AI model to use

Usage:

# Basic language setup
agent.add_language("English", "en-US", "rime.spore")

# With custom fillers
agent.add_language(
    "English", 
    "en-US", 
    "nova.luna",
    speech_fillers=["Let me think...", "One moment..."],
    function_fillers=["Processing...", "Looking that up..."]
)

`set_languages(languages: List[Dict[str, Any]]) -> AgentBase`

Set multiple language configurations at once.

Parameters:

languages (List[Dict]): List of language configuration dictionaries

Usage:

agent.set_languages([
    {"name": "English", "code": "en-US", "voice": "rime.spore"},
    {"name": "Spanish", "code": "es-ES", "voice": "nova.luna"}
])

Speech Recognition Configuration

`add_hint(hint: str) -> AgentBase`

Add a single speech recognition hint.

Parameters:

hint (str): Word or phrase to improve recognition accuracy

Usage:

agent.add_hint("SignalWire")

`add_hints(hints: List[str]) -> AgentBase`

Add multiple speech recognition hints.

Parameters:

hints (List[str]): List of words/phrases for better recognition

Usage:

agent.add_hints(["SignalWire", "SWML", "API", "webhook", "SIP"])

`add_pattern_hint`

def add_pattern_hint(
    hint: str, 
    pattern: str, 
    replace: str, 
    ignore_case: bool = False
) -> AgentBase

Add a pattern-based hint for speech recognition.

Parameters:

hint (str): The hint phrase
pattern (str): Regex pattern to match
replace (str): Replacement text
ignore_case (bool): Case-insensitive matching (default: False)

Usage:

agent.add_pattern_hint(
    "phone number",
    r"(\d{3})-(\d{3})-(\d{4})",
    r"(\1) \2-\3"
)

`add_pronunciation`

def add_pronunciation(
    replace: str, 
    with_text: str, 
    ignore_case: bool = False
) -> AgentBase

Add pronunciation rules for text-to-speech.

Parameters:

replace (str): Text to replace
with_text (str): Replacement pronunciation
ignore_case (bool): Case-insensitive replacement (default: False)

Usage:

agent.add_pronunciation("API", "A P I")
agent.add_pronunciation("SWML", "swim-el")

`set_pronunciations`

def set_pronunciations(
    pronunciations: List[Dict[str, Any]]
) -> AgentBase

Set multiple pronunciation rules at once.

Parameters:

pronunciations (List[Dict]): List of pronunciation rule dictionaries

Usage:

agent.set_pronunciations([
    {"replace": "API", "with": "A P I"},
    {"replace": "SWML", "with": "swim-el", "ignore_case": True}
])

AI Parameters Configuration

`set_param(key: str, value: Any) -> AgentBase`

Set a single AI parameter.

Parameters:

key (str): Parameter name
value (Any): Parameter value

Usage:

agent.set_param("ai_model", "gpt-4.1-nano")
agent.set_param("end_of_speech_timeout", 500)

`set_params(params: Dict[str, Any]) -> AgentBase`

Set multiple AI parameters at once.

Parameters:

params (Dict[str, Any]): Dictionary of parameter key-value pairs

Common Parameters:

ai_model: AI model to use ("gpt-4.1-nano", "gpt-4.1-mini", etc.)
end_of_speech_timeout: Milliseconds to wait for speech end (default: 1000)
attention_timeout: Milliseconds before attention timeout (default: 30000)
background_file_volume: Volume for background audio (-60 to 0 dB)
temperature: AI creativity/randomness (0.0 to 2.0)
max_tokens: Maximum response length
top_p: Nucleus sampling parameter (0.0 to 1.0)

Usage:

agent.set_params({
    "ai_model": "gpt-4.1-nano",
    "end_of_speech_timeout": 500,
    "attention_timeout": 15000,
    "background_file_volume": -20,
    "temperature": 0.7
})

Global Data Management

`set_global_data(data: Dict[str, Any]) -> AgentBase`

Set global data available to the AI and functions.

Parameters:

data (Dict[str, Any]): Global data dictionary

Usage:

agent.set_global_data({
    "company_name": "Acme Corp",
    "support_hours": "9 AM - 5 PM EST",
    "escalation_number": "+1-555-0123"
})

`update_global_data(data: Dict[str, Any]) -> AgentBase`

Update existing global data (merge with existing).

Parameters:

data (Dict[str, Any]): Data to merge with existing global data

Usage:

agent.update_global_data({
    "current_promotion": "20% off all services",
    "promotion_expires": "2024-12-31"
})

Function Definition

`define_tool`

def define_tool(
    name: str, 
    description: str, 
    parameters: Dict[str, Any], 
    handler: Callable, 
    secure: bool = True, 
    fillers: Optional[Dict[str, List[str]]] = None, 
    webhook_url: Optional[str] = None, 
    **swaig_fields
) -> AgentBase

Define a custom SWAIG function/tool.

Parameters:

name (str): Function name
description (str): Function description for AI
parameters (Dict[str, Any]): JSON schema for function parameters
handler (Callable): Function to execute when called
secure (bool): Require security token (default: True)
fillers (Optional[Dict[str, List[str]]]): Language-specific filler phrases
webhook_url (Optional[str]): Custom webhook URL
**swaig_fields: Additional SWAIG function properties

Usage:

def get_weather(args, raw_data):
    location = args.get("location", "Unknown")
    return SwaigFunctionResult(f"The weather in {location} is sunny and 75°F")

agent.define_tool(
    name="get_weather",
    description="Get current weather for a location",
    parameters={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City name"
            }
        },
        "required": ["location"]
    },
    handler=get_weather,
    fillers={"en-US": ["Checking weather...", "Looking up forecast..."]}
)

`@AgentBase.tool(name=None, **kwargs)` (Class Decorator)

Decorator for defining tools as class methods.

Parameters:

name (Optional[str]): Function name (defaults to method name)
**kwargs: Same parameters as define_tool()

Usage:

class MyAgent(AgentBase):
    @AgentBase.tool(
        description="Get current time",
        parameters={"type": "object", "properties": {}}
    )
    def get_time(self, args, raw_data):
        import datetime
        return SwaigFunctionResult(f"Current time: {datetime.datetime.now()}")

`register_swaig_function`

def register_swaig_function(
    function_dict: Dict[str, Any]
) -> AgentBase

Parameters:

function_dict (Dict[str, Any]): Complete SWAIG function definition

Usage:

# Register a DataMap tool
weather_tool = DataMap('get_weather').webhook('GET', 'https://api.weather.com/...')
agent.register_swaig_function(weather_tool.to_swaig_function())

Skills System

`add_skill`

def add_skill(
    skill_name: str, 
    params: Optional[Dict[str, Any]] = None
) -> AgentBase

Add a modular skill to the agent.

Parameters:

skill_name (str): Name of the skill to add
params (Optional[Dict[str, Any]]): Skill configuration parameters

Available Skills:

datetime: Current date/time information
math: Mathematical calculations
web_search: Google Custom Search integration
datasphere: SignalWire DataSphere search
native_vector_search: Local document search

Usage:

# Simple skill
agent.add_skill("datetime")
agent.add_skill("math")

# Skill with configuration
agent.add_skill("web_search", {
    "api_key": "your-google-api-key",
    "search_engine_id": "your-search-engine-id",
    "num_results": 3
})

# Multiple instances with different names
agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "general-engine",
    "tool_name": "search_general"
})

agent.add_skill("web_search", {
    "api_key": "your-api-key", 
    "search_engine_id": "news-engine",
    "tool_name": "search_news"
})

`remove_skill(skill_name: str) -> AgentBase`

Remove a skill from the agent.

Parameters:

skill_name (str): Name of skill to remove

Usage:

agent.remove_skill("web_search")

`list_skills() -> List[str]`

Get list of currently added skills.

Returns:

List[str]: Names of active skills

Usage:

active_skills = agent.list_skills()
print(f"Active skills: {active_skills}")

`has_skill(skill_name: str) -> bool`

Check if a skill is currently added.

Parameters:

skill_name (str): Name of skill to check

Returns:

bool: True if skill is active

Usage:

if agent.has_skill("web_search"):
    print("Web search is available")

Native Functions

`set_native_functions`

def set_native_functions(
    function_names: List[str]
) -> AgentBase

Enable specific native SWML functions.

Parameters:

function_names (List[str]): List of native function names to enable

Available Native Functions:

transfer: Transfer calls
hangup: End calls
play: Play audio files
record: Record audio
send_sms: Send SMS messages

Usage:

agent.set_native_functions(["transfer", "hangup", "send_sms"])

`set_internal_fillers`

def set_internal_fillers(
    internal_fillers: Dict[str, Dict[str, List[str]]]
) -> AgentBase

Set custom filler phrases for internal/native SWAIG functions.

Parameters:

internal_fillers (Dict[str, Dict[str, List[str]]]): Function name → language code → filler phrases

Available Internal Functions:

next_step: Moving between workflow steps (contexts system)
change_context: Switching contexts in workflows
check_time: Getting current time
wait_for_user: Waiting for user input
wait_seconds: Pausing for specified duration
get_visual_input: Processing visual data

Usage:

agent.set_internal_fillers({
    "next_step": {
        "en-US": ["Moving to the next step...", "Let's continue..."],
        "es": ["Pasando al siguiente paso...", "Continuemos..."]
    },
    "check_time": {
        "en-US": ["Let me check the time...", "Getting current time..."]
    }
})

`add_internal_filler`

def add_internal_filler(
    function_name: str, 
    language_code: str, 
    fillers: List[str]
) -> AgentBase

Add internal fillers for a specific function and language.

Parameters:

function_name (str): Name of the internal function
language_code (str): Language code (e.g., "en-US", "es", "fr")
fillers (List[str]): List of filler phrases

Usage:

agent.add_internal_filler("next_step", "en-US", [
    "Great! Let's move to the next step...",
    "Perfect! Moving forward..."
])

Function Includes

`add_function_include`

def add_function_include(
    url: str, 
    functions: List[str], 
    meta_data: Optional[Dict[str, Any]] = None
) -> AgentBase

Include external SWAIG functions from another service.

Parameters:

url (str): URL of external SWAIG service
functions (List[str]): List of function names to include
meta_data (Optional[Dict[str, Any]]): Additional metadata

Usage:

agent.add_function_include(
    "https://external-service.com/swaig",
    ["external_function1", "external_function2"],
    meta_data={"service": "external", "version": "1.0"}
)

`set_function_includes`

def set_function_includes(
    includes: List[Dict[str, Any]]
) -> AgentBase

Set multiple function includes at once.

Parameters:

includes (List[Dict[str, Any]]): List of function include configurations

Usage:

agent.set_function_includes([
    {
        "url": "https://service1.com/swaig",
        "functions": ["func1", "func2"]
    },
    {
        "url": "https://service2.com/swaig", 
        "functions": ["func3"],
        "meta_data": {"priority": "high"}
    }
])

Webhook Configuration

`set_web_hook_url(url: str) -> AgentBase`

Set default webhook URL for SWAIG functions.

Parameters:

url (str): Default webhook URL

Usage:

agent.set_web_hook_url("https://myserver.com/webhook")

`set_post_prompt_url(url: str) -> AgentBase`

Set URL for post-prompt processing.

Parameters:

url (str): Post-prompt webhook URL

Usage:

agent.set_post_prompt_url("https://myserver.com/post-prompt")

Dynamic Configuration

`set_dynamic_config_callback`

def set_dynamic_config_callback(
    callback: Callable[[dict, dict, dict, EphemeralAgentConfig], None]
) -> AgentBase

Set callback for per-request dynamic configuration.

Parameters:

callback (Callable): Function that receives (query_params, headers, body, config)

Usage:

def configure_agent(query_params, headers, body, config):
    # Configure based on request
    if query_params.get("language") == "spanish":
        config.add_language("Spanish", "es-ES", "nova.luna")
    
    # Set customer-specific data
    customer_id = headers.get("X-Customer-ID")
    if customer_id:
        config.set_global_data({"customer_id": customer_id})

agent.set_dynamic_config_callback(configure_agent)

SIP Integration

`enable_sip_routing`

def enable_sip_routing(
    auto_map: bool = True, 
    path: str = "/sip"
) -> AgentBase

Enable SIP-based routing for voice calls.

Parameters:

auto_map (bool): Automatically map SIP usernames (default: True)
path (str): SIP routing endpoint path (default: "/sip")

Usage:

agent.enable_sip_routing()

`register_sip_username(sip_username: str) -> AgentBase`

Parameters:

sip_username (str): SIP username to register

Usage:

agent.register_sip_username("support")
agent.register_sip_username("sales")

`register_routing_callback`

def register_routing_callback(
    callback_fn: Callable[[Request, Dict[str, Any]], Optional[str]], 
    path: str = "/sip"
) -> None

Parameters:

callback_fn (Callable): Function that returns agent route based on request
path (str): Routing endpoint path (default: "/sip")

Usage:

def route_call(request, body):
    sip_username = body.get("sip_username")
    if sip_username == "support":
        return "/support-agent"
    elif sip_username == "sales":
        return "/sales-agent"
    return None

agent.register_routing_callback(route_call)

Utility Methods

`get_name() -> str`

Get the agent's name.

Returns:

str: Agent name

`get_app()`

Get the FastAPI application instance.

Returns:

FastAPI: The underlying FastAPI app

`as_router() -> APIRouter`

Get the agent as a FastAPI router for embedding in larger applications.

Returns:

APIRouter: FastAPI router instance

Usage:

# Embed agent in larger FastAPI app
main_app = FastAPI()
agent_router = agent.as_router()
main_app.include_router(agent_router, prefix="/agent")

Event Handlers

`on_summary`

def on_summary(
    summary: Optional[Dict[str, Any]], 
    raw_data: Optional[Dict[str, Any]] = None
) -> None

Override to handle conversation summaries.

Parameters:

summary (Optional[Dict[str, Any]]): Summary data
raw_data (Optional[Dict[str, Any]]): Raw request data

Usage:

class MyAgent(AgentBase):
    def on_summary(self, summary, raw_data):
        print(f"Conversation summary: {summary}")
        # Save to database, send notification, etc.

`on_function_call`

def on_function_call(
    name: str, 
    args: Dict[str, Any], 
    raw_data: Optional[Dict[str, Any]] = None
) -> Any

Override to handle function calls with custom logic.

Parameters:

name (str): Function name being called
args (Dict[str, Any]): Function arguments
raw_data (Optional[Dict[str, Any]]): Raw request data

Returns:

Any: Function result (typically SwaigFunctionResult)

Usage:

class MyAgent(AgentBase):
    def on_function_call(self, name, args, raw_data):
        if name == "get_weather":
            location = args.get("location")
            # Custom weather logic
            return SwaigFunctionResult(f"Weather in {location}: Sunny")
        return super().on_function_call(name, args, raw_data)

`on_request`

def on_request(
    request_data: Optional[dict] = None, 
    callback_path: Optional[str] = None
) -> Optional[dict]

Override to handle general requests.

Parameters:

request_data (Optional[dict]): Request data
callback_path (Optional[str]): Callback path

Returns:

Optional[dict]: Response modifications

`on_swml_request`

def on_swml_request(
    request_data: Optional[dict] = None, 
    callback_path: Optional[str] = None, 
    request: Optional[Request] = None
) -> Optional[dict]

Override to handle SWML generation requests.

Parameters:

request_data (Optional[dict]): Request data
callback_path (Optional[str]): Callback path
request (Optional[Request]): FastAPI request object

Returns:

Optional[dict]: SWML modifications

Authentication

`validate_basic_auth(username: str, password: str) -> bool`

Override to implement custom basic authentication logic.

Parameters:

username (str): Username from basic auth
password (str): Password from basic auth

Returns:

bool: True if credentials are valid

Usage:

class MyAgent(AgentBase):
    def validate_basic_auth(self, username, password):
        # Custom auth logic
        return username == "admin" and password == "secret"

`get_basic_auth_credentials`

def get_basic_auth_credentials(
    include_source: bool = False
) -> Union[Tuple[str, str], Tuple[str, str, str]]

Get basic auth credentials from environment or constructor.

Parameters:

include_source (bool): Include source information (default: False)

Returns:

Tuple: (username, password) or (username, password, source)

Context System

`define_contexts() -> ContextBuilder`

Define structured workflow contexts for the agent.

Returns:

ContextBuilder: Builder for creating contexts and steps

Usage:

contexts = agent.define_contexts()
contexts.add_context("greeting") \
    .add_step("welcome", "Welcome! How can I help?") \
    .on_completion_go_to("main_menu")

contexts.add_context("main_menu") \
    .add_step("menu", "Choose: 1) Support 2) Sales 3) Billing") \
    .allow_functions(["transfer_to_support", "transfer_to_sales"])

This concludes Part 1 of the API reference covering the AgentBase class. The document will continue with SwaigFunctionResult, DataMap, and other components in subsequent parts.

SwaigFunctionResult Class

The SwaigFunctionResult class is used to create structured responses from SWAIG functions. It handles both natural language responses and structured actions that the agent should execute.

Constructor

SwaigFunctionResult(
    response: Optional[str] = None, 
    post_process: bool = False
)

Parameters:

response (Optional[str]): Natural language response text for the AI to speak
post_process (bool): Whether to let AI take another turn before executing actions (default: False)

Post-processing Behavior:

post_process=False (default): Execute actions immediately after AI response
post_process=True: Let AI respond to user one more time, then execute actions

Usage:

# Simple response
result = SwaigFunctionResult("The weather is sunny and 75°F")

# Response with post-processing enabled
result = SwaigFunctionResult("I'll transfer you now", post_process=True)

# Empty response (actions only)
result = SwaigFunctionResult()

Core Methods

Response Configuration

`set_response(response: str) -> SwaigFunctionResult`

Set or update the natural language response text.

Parameters:

response (str): The text the AI should speak

Usage:

result = SwaigFunctionResult()
result.set_response("I found your order information")

`set_post_process(post_process: bool) -> SwaigFunctionResult`

Enable or disable post-processing for this result.

Parameters:

post_process (bool): True to let AI respond once more before executing actions

Usage:

result = SwaigFunctionResult("I'll help you with that")
result.set_post_process(True)  # Let AI handle follow-up questions first

Action Management

`add_action(name: str, data: Any) -> SwaigFunctionResult`

Add a structured action to execute.

Parameters:

name (str): Action name/type (e.g., "play", "transfer", "set_global_data")
data (Any): Action data - can be string, boolean, object, or array

Usage:

# Simple action with boolean
result.add_action("hangup", True)

# Action with string data
result.add_action("play", "welcome.mp3")

# Action with object data
result.add_action("set_global_data", {"customer_id": "12345", "status": "verified"})

# Action with array data
result.add_action("send_sms", ["+15551234567", "Your order is ready!"])

`add_actions(actions: List[Dict[str, Any]]) -> SwaigFunctionResult`

Add multiple actions at once.

Parameters:

actions (List[Dict[str, Any]]): List of action dictionaries

Usage:

result.add_actions([
    {"play": "hold_music.mp3"},
    {"set_global_data": {"status": "on_hold"}},
    {"wait": 5000}
])

Call Control Actions

Call Transfer and Connection

`connect(destination: str, final: bool = True, from_addr: Optional[str] = None) -> SwaigFunctionResult`

Transfer or connect the call to another destination.

Parameters:

destination (str): Phone number, SIP address, or other destination
final (bool): Permanent transfer (True) vs temporary transfer (False) (default: True)
from_addr (Optional[str]): Override caller ID

Transfer Types:

final=True: Permanent transfer - call exits agent completely
final=False: Temporary transfer - call returns to agent if far end hangs up

Usage:

# Permanent transfer to phone number
result.connect("+15551234567", final=True)

# Temporary transfer to SIP address with custom caller ID
result.connect("support@company.com", final=False, from_addr="+15559876543")

# Transfer with response
result = SwaigFunctionResult("Transferring you to our sales team")
result.connect("sales@company.com")

`swml_transfer(dest: str, ai_response: str) -> SwaigFunctionResult`

Create a SWML-based transfer with AI response setup.

Parameters:

dest (str): Transfer destination
ai_response (str): AI response when transfer completes

Usage:

result.swml_transfer(
    "+15551234567", 
    "You've been transferred back to me. How else can I help?"
)

`sip_refer(to_uri: str) -> SwaigFunctionResult`

Perform a SIP REFER transfer.

Parameters:

to_uri (str): SIP URI to transfer to

Usage:

result.sip_refer("sip:support@company.com")

Call Management

`hangup() -> SwaigFunctionResult`

End the call immediately.

Usage:

result = SwaigFunctionResult("Thank you for calling. Goodbye!")
result.hangup()

`hold(timeout: int = 300) -> SwaigFunctionResult`

Put the call on hold.

Parameters:

timeout (int): Hold timeout in seconds (default: 300)

Usage:

result = SwaigFunctionResult("Please hold while I look that up")
result.hold(timeout=60)

`stop() -> SwaigFunctionResult`

Stop current audio playback or recording.

Usage:

result.stop()

Audio Control

`say(text: str) -> SwaigFunctionResult`

Add text for the AI to speak.

Parameters:

text (str): Text to speak

Usage:

result.say("Please wait while I process your request")

`play_background_file(filename: str, wait: bool = False) -> SwaigFunctionResult`

Play an audio file in the background.

Parameters:

filename (str): Audio file path or URL
wait (bool): Wait for file to finish before continuing (default: False)

Usage:

# Play hold music in background
result.play_background_file("hold_music.mp3")

# Play announcement and wait for completion
result.play_background_file("important_announcement.wav", wait=True)

`stop_background_file() -> SwaigFunctionResult`

Stop background audio playback.

Usage:

result.stop_background_file()

Data Management Actions

`set_global_data(data: Dict[str, Any]) -> SwaigFunctionResult`

Set global data for the conversation.

Parameters:

data (Dict[str, Any]): Global data to set

Usage:

result.set_global_data({
    "customer_id": "12345",
    "order_status": "shipped",
    "tracking_number": "1Z999AA1234567890"
})

`update_global_data(data: Dict[str, Any]) -> SwaigFunctionResult`

Update existing global data (merge with existing).

Parameters:

data (Dict[str, Any]): Data to merge

Usage:

result.update_global_data({
    "last_interaction": "2024-01-15T10:30:00Z",
    "agent_notes": "Customer satisfied with resolution"
})

`remove_global_data(keys: Union[str, List[str]]) -> SwaigFunctionResult`

Remove specific keys from global data.

Parameters:

keys (Union[str, List[str]]): Key name or list of key names to remove

Usage:

# Remove single key
result.remove_global_data("temporary_data")

# Remove multiple keys
result.remove_global_data(["temp1", "temp2", "cache_data"])

`set_metadata(data: Dict[str, Any]) -> SwaigFunctionResult`

Set metadata for the conversation.

Parameters:

data (Dict[str, Any]): Metadata to set

Usage:

result.set_metadata({
    "call_type": "support",
    "priority": "high",
    "department": "technical"
})

`remove_metadata(keys: Union[str, List[str]]) -> SwaigFunctionResult`

Remove specific metadata keys.

Parameters:

keys (Union[str, List[str]]): Key name or list of key names to remove

Usage:

result.remove_metadata(["temporary_flag", "debug_info"])

AI Behavior Control

`set_end_of_speech_timeout(milliseconds: int) -> SwaigFunctionResult`

Adjust how long to wait for speech to end.

Parameters:

milliseconds (int): Timeout in milliseconds

Usage:

# Shorter timeout for quick responses
result.set_end_of_speech_timeout(300)

# Longer timeout for thoughtful responses
result.set_end_of_speech_timeout(2000)

`set_speech_event_timeout(milliseconds: int) -> SwaigFunctionResult`

Set timeout for speech events.

Parameters:

milliseconds (int): Timeout in milliseconds

Usage:

result.set_speech_event_timeout(5000)

`wait_for_user(enabled: Optional[bool] = None, timeout: Optional[int] = None, answer_first: bool = False) -> SwaigFunctionResult`

Control whether to wait for user input.

Parameters:

enabled (Optional[bool]): Enable/disable waiting for user
timeout (Optional[int]): Timeout in milliseconds
answer_first (bool): Answer call before waiting (default: False)

Usage:

# Wait for user input with 10 second timeout
result.wait_for_user(enabled=True, timeout=10000)

# Don't wait for user (immediate response)
result.wait_for_user(enabled=False)

`toggle_functions(function_toggles: List[Dict[str, Any]]) -> SwaigFunctionResult`

Enable or disable specific functions.

Parameters:

function_toggles (List[Dict[str, Any]]): List of function toggle configurations

Usage:

result.toggle_functions([
    {"name": "transfer_to_sales", "enabled": True},
    {"name": "end_call", "enabled": False},
    {"name": "escalate", "enabled": True, "timeout": 30000}
])

`enable_functions_on_timeout(enabled: bool = True) -> SwaigFunctionResult`

Control whether functions are enabled when timeout occurs.

Parameters:

enabled (bool): Enable functions on timeout (default: True)

Usage:

result.enable_functions_on_timeout(False)  # Disable functions on timeout

`enable_extensive_data(enabled: bool = True) -> SwaigFunctionResult`

Enable extensive data collection.

Parameters:

enabled (bool): Enable extensive data (default: True)

Usage:

result.enable_extensive_data(True)

`update_settings(settings: Dict[str, Any]) -> SwaigFunctionResult`

Update various AI settings.

Parameters:

settings (Dict[str, Any]): Settings to update

Usage:

result.update_settings({
    "temperature": 0.8,
    "max_tokens": 150,
    "end_of_speech_timeout": 800
})

Context and Conversation Control

`switch_context(system_prompt: Optional[str] = None, user_prompt: Optional[str] = None, consolidate: bool = False, full_reset: bool = False) -> SwaigFunctionResult`

Switch conversation context or reset the conversation.

Parameters:

system_prompt (Optional[str]): New system prompt
user_prompt (Optional[str]): New user prompt
consolidate (bool): Consolidate conversation history (default: False)
full_reset (bool): Completely reset conversation (default: False)

Usage:

# Switch to technical support context
result.switch_context(
    system_prompt="You are now a technical support specialist",
    user_prompt="The customer needs technical help"
)

# Reset conversation completely
result.switch_context(full_reset=True)

# Consolidate conversation history
result.switch_context(consolidate=True)

`simulate_user_input(text: str) -> SwaigFunctionResult`

Simulate user input for testing or automation.

Parameters:

text (str): Text to simulate as user input

Usage:

result.simulate_user_input("I need help with my order")

Communication Actions

`send_sms(to_number: str, from_number: str, body: Optional[str] = None, media: Optional[List[str]] = None, tags: Optional[List[str]] = None, region: Optional[str] = None) -> SwaigFunctionResult`

Send an SMS message.

Parameters:

to_number (str): Recipient phone number
from_number (str): Sender phone number
body (Optional[str]): SMS message text
media (Optional[List[str]]): List of media URLs
tags (Optional[List[str]]): Message tags
region (Optional[str]): SignalWire region

Usage:

# Simple text message
result.send_sms(
    to_number="+15551234567",
    from_number="+15559876543", 
    body="Your order #12345 has shipped!"
)

# Message with media and tags
result.send_sms(
    to_number="+15551234567",
    from_number="+15559876543",
    body="Here's your receipt",
    media=["https://example.com/receipt.pdf"],
    tags=["receipt", "order_12345"]
)

Recording and Media

`record_call(control_id: Optional[str] = None, stereo: bool = False, format: str = "wav", direction: str = "both", terminators: Optional[str] = None, beep: bool = False, input_sensitivity: float = 44.0, initial_timeout: float = 0.0, end_silence_timeout: float = 0.0, max_length: Optional[float] = None, status_url: Optional[str] = None) -> SwaigFunctionResult`

Start call recording.

Parameters:

control_id (Optional[str]): Unique identifier for this recording
stereo (bool): Record in stereo (default: False)
format (str): Recording format: "wav", "mp3", "mp4" (default: "wav")
direction (str): Recording direction: "both", "inbound", "outbound" (default: "both")
terminators (Optional[str]): DTMF keys to stop recording
beep (bool): Play beep before recording (default: False)
input_sensitivity (float): Input sensitivity level (default: 44.0)
initial_timeout (float): Initial timeout in seconds (default: 0.0)
end_silence_timeout (float): End silence timeout in seconds (default: 0.0)
max_length (Optional[float]): Maximum recording length in seconds
status_url (Optional[str]): Webhook URL for recording status

Usage:

# Basic recording
result.record_call(format="mp3", direction="both")

# Recording with control ID and settings
result.record_call(
    control_id="customer_call_001",
    stereo=True,
    format="wav",
    beep=True,
    max_length=300.0,
    terminators="#*"
)

`stop_record_call(control_id: Optional[str] = None) -> SwaigFunctionResult`

Stop call recording.

Parameters:

control_id (Optional[str]): Control ID of recording to stop

Usage:

result.stop_record_call()
result.stop_record_call(control_id="customer_call_001")

Conference and Room Management

`join_room(name: str) -> SwaigFunctionResult`

Join a SignalWire room.

Parameters:

name (str): Room name to join

Usage:

result.join_room("support_room_1")

join_conference(name: str, muted: bool = False, beep: str = "true", start_on_enter: bool = True, end_on_exit: bool = False, wait_url: Optional[str] = None, max_participants: int = 250, record: str = "do-not-record", region: Optional[str] = None, trim: str = "trim-silence", coach: Optional[str] = None, status_callback_event: Optional[str] = None, status_callback: Optional[str] = None, status_callback_method: str = "POST", recording_status_callback: Optional[str] = None, recording_status_callback_method: str = "POST", recording_status_callback_event: str = "completed", result: Optional[Any] = None) -> SwaigFunctionResult

Join a conference call.

Parameters:

name (str): Conference name
muted (bool): Join muted (default: False)
beep (str): Beep setting: "true", "false", "onEnter", "onExit" (default: "true")
start_on_enter (bool): Start conference when this participant enters (default: True)
end_on_exit (bool): End conference when this participant exits (default: False)
wait_url (Optional[str]): URL for hold music/content
max_participants (int): Maximum participants (default: 250)
record (str): Recording setting (default: "do-not-record")
region (Optional[str]): SignalWire region
trim (str): Trim setting for recordings (default: "trim-silence")
coach (Optional[str]): Coach participant identifier
status_callback_event (Optional[str]): Status callback events
status_callback (Optional[str]): Status callback URL
status_callback_method (str): Status callback HTTP method (default: "POST")
recording_status_callback (Optional[str]): Recording status callback URL
recording_status_callback_method (str): Recording status callback method (default: "POST")
recording_status_callback_event (str): Recording status callback events (default: "completed")

Usage:

# Basic conference join
result.join_conference("sales_meeting")

# Conference with recording and settings
result.join_conference(
    name="support_conference",
    muted=False,
    beep="onEnter",
    record="record-from-start",
    max_participants=10
)

Payment Processing

pay(payment_connector_url: str, input_method: str = "dtmf", status_url: Optional[str] = None, payment_method: str = "credit-card", timeout: int = 5, max_attempts: int = 1, security_code: bool = True, postal_code: Union[bool, str] = True, min_postal_code_length: int = 0, token_type: str = "reusable", charge_amount: Optional[str] = None, currency: str = "usd", language: str = "en-US", voice: str = "woman", description: Optional[str] = None, valid_card_types: str = "visa mastercard amex", parameters: Optional[List[Dict[str, str]]] = None, prompts: Optional[List[Dict[str, Any]]] = None) -> SwaigFunctionResult

Process a payment through the call.

Parameters:

payment_connector_url (str): Payment processor webhook URL
input_method (str): Input method: "dtmf", "speech" (default: "dtmf")
status_url (Optional[str]): Payment status webhook URL
payment_method (str): Payment method: "credit-card" (default: "credit-card")
timeout (int): Input timeout in seconds (default: 5)
max_attempts (int): Maximum retry attempts (default: 1)
security_code (bool): Require security code (default: True)
postal_code (Union[bool, str]): Require postal code (default: True)
min_postal_code_length (int): Minimum postal code length (default: 0)
token_type (str): Token type: "reusable", "one-time" (default: "reusable")
charge_amount (Optional[str]): Amount to charge
currency (str): Currency code (default: "usd")
language (str): Language for prompts (default: "en-US")
voice (str): Voice for prompts (default: "woman")
description (Optional[str]): Payment description
valid_card_types (str): Accepted card types (default: "visa mastercard amex")
parameters (Optional[List[Dict[str, str]]]): Additional parameters
prompts (Optional[List[Dict[str, Any]]]): Custom prompts

Usage:

# Basic payment processing
result.pay(
    payment_connector_url="https://payment-processor.com/webhook",
    charge_amount="29.99",
    description="Monthly subscription"
)

# Payment with custom settings
result.pay(
    payment_connector_url="https://payment-processor.com/webhook",
    input_method="speech",
    timeout=10,
    max_attempts=3,
    security_code=True,
    postal_code=True,
    charge_amount="149.99",
    currency="usd",
    description="Premium service upgrade"
)

Call Monitoring

`tap(uri: str, control_id: Optional[str] = None, direction: str = "both", codec: str = "PCMU", rtp_ptime: int = 20, status_url: Optional[str] = None) -> SwaigFunctionResult`

Start call tapping/monitoring.

Parameters:

uri (str): URI to send tapped audio to
control_id (Optional[str]): Unique identifier for this tap
direction (str): Tap direction: "both", "inbound", "outbound" (default: "both")
codec (str): Audio codec: "PCMU", "PCMA", "G722" (default: "PCMU")
rtp_ptime (int): RTP packet time in milliseconds (default: 20)
status_url (Optional[str]): Status webhook URL

Usage:

# Basic call tapping
result.tap("sip:monitor@company.com")

# Tap with specific settings
result.tap(
    uri="sip:quality@company.com",
    control_id="quality_monitor_001",
    direction="both",
    codec="G722"
)

`stop_tap(control_id: Optional[str] = None) -> SwaigFunctionResult`

Stop call tapping.

Parameters:

control_id (Optional[str]): Control ID of tap to stop

Usage:

result.stop_tap()
result.stop_tap(control_id="quality_monitor_001")

Advanced SWML Execution

`execute_swml(swml_content, transfer: bool = False) -> SwaigFunctionResult`

Execute custom SWML content.

Parameters:

swml_content: SWML document or content to execute
transfer (bool): Whether this is a transfer operation (default: False)

Usage:

# Execute custom SWML
custom_swml = {
    "version": "1.0.0",
    "sections": {
        "main": [
            {"play": {"url": "https://example.com/custom.mp3"}},
            {"say": {"text": "Custom SWML execution"}}
        ]
    }
}
result.execute_swml(custom_swml)

Utility Methods

`to_dict() -> Dict[str, Any]`

Convert the result to a dictionary for serialization.

Returns:

Dict[str, Any]: Dictionary representation of the result

Usage:

result = SwaigFunctionResult("Hello world")
result.add_action("play", "music.mp3")
result_dict = result.to_dict()
print(result_dict)
# Output: {"response": "Hello world", "action": [{"play": "music.mp3"}]}

Static Helper Methods

`create_payment_prompt(for_situation: str, actions: List[Dict[str, str]], card_type: Optional[str] = None, error_type: Optional[str] = None) -> Dict[str, Any]`

Create a payment prompt configuration.

Parameters:

for_situation (str): Situation identifier
actions (List[Dict[str, str]]): List of action configurations
card_type (Optional[str]): Card type for prompts
error_type (Optional[str]): Error type for error prompts

Usage:

prompt = SwaigFunctionResult.create_payment_prompt(
    for_situation="card_number",
    actions=[
        SwaigFunctionResult.create_payment_action("say", "Please enter your card number")
    ]
)

`create_payment_action(action_type: str, phrase: str) -> Dict[str, str]`

Create a payment action configuration.

Parameters:

action_type (str): Action type
phrase (str): Action phrase

Usage:

action = SwaigFunctionResult.create_payment_action("say", "Enter your card number")

`create_payment_parameter(name: str, value: str) -> Dict[str, str]`

Create a payment parameter configuration.

Parameters:

name (str): Parameter name
value (str): Parameter value

Usage:

param = SwaigFunctionResult.create_payment_parameter("merchant_id", "12345")

Method Chaining

All methods return self, enabling fluent method chaining:

result = (SwaigFunctionResult("I'll help you with that")
    .set_post_process(True)
    .update_global_data({"status": "helping"})
    .set_end_of_speech_timeout(800)
    .add_action("play", "thinking.mp3"))

# Complex workflow
result = (SwaigFunctionResult("Processing your payment")
    .set_post_process(True)
    .update_global_data({"payment_status": "processing"})
    .pay(
        payment_connector_url="https://payments.com/webhook",
        charge_amount="99.99",
        description="Service payment"
    )
    .send_sms(
        to_number="+15551234567",
        from_number="+15559876543",
        body="Payment confirmation will be sent shortly"
    ))

This concludes Part 2 of the API reference covering the SwaigFunctionResult class. The document will continue with DataMap and other components in subsequent parts.

DataMap Class

The DataMap class provides a declarative approach to creating SWAIG tools that integrate with REST APIs without requiring webhook infrastructure. DataMap tools execute on SignalWire's server infrastructure, eliminating the need to expose webhook endpoints.

Constructor

DataMap(function_name: str)

Parameters:

function_name (str): Name of the SWAIG function this DataMap will create

Usage:

# Create a new DataMap tool
weather_map = DataMap('get_weather')
search_map = DataMap('search_docs')

Core Configuration Methods

Function Metadata

`purpose(description: str) -> DataMap`

Set the function description/purpose.

Parameters:

description (str): Human-readable description of what this function does

Usage:

data_map = DataMap('get_weather').purpose('Get current weather information for any city')

`description(description: str) -> DataMap`

Alias for purpose() - set the function description.

Parameters:

description (str): Function description

Usage:

data_map = DataMap('search_api').description('Search our knowledge base for information')

Parameter Definition

`parameter(name: str, param_type: str, description: str, required: bool = False, enum: Optional[List[str]] = None) -> DataMap`

Add a function parameter with JSON schema validation.

Parameters:

name (str): Parameter name
param_type (str): JSON schema type: "string", "number", "boolean", "array", "object"
description (str): Parameter description for the AI
required (bool): Whether parameter is required (default: False)
enum (Optional[List[str]]): List of allowed values for validation

Usage:

# Required string parameter
data_map.parameter('location', 'string', 'City name or ZIP code', required=True)

# Optional number parameter
data_map.parameter('days', 'number', 'Number of forecast days', required=False)

# Enum parameter with allowed values
data_map.parameter('units', 'string', 'Temperature units', 
                  enum=['celsius', 'fahrenheit'], required=False)

# Boolean parameter
data_map.parameter('include_alerts', 'boolean', 'Include weather alerts', required=False)

# Array parameter
data_map.parameter('categories', 'array', 'Search categories to include')

API Integration Methods

HTTP Webhook Configuration

`webhook(method: str, url: str, headers: Optional[Dict[str, str]] = None, form_param: Optional[str] = None, input_args_as_params: bool = False, require_args: Optional[List[str]] = None) -> DataMap`

Configure an HTTP API call.

Parameters:

method (str): HTTP method: "GET", "POST", "PUT", "DELETE", "PATCH"
url (str): API endpoint URL (supports ${variable} substitution)
headers (Optional[Dict[str, str]]): HTTP headers to send
form_param (Optional[str]): Send JSON body as single form parameter with this name
input_args_as_params (bool): Merge function arguments into URL parameters (default: False)
require_args (Optional[List[str]]): Only execute if these arguments are present

Variable Substitution in URLs:

${args.parameter_name}: Function argument values
${global_data.key}: Call-wide data store (user info, call state - NOT credentials)
${meta_data.call_id}: Call and function metadata

Usage:

# Simple GET request with parameter substitution
data_map.webhook('GET', 'https://api.weather.com/v1/current?key=API_KEY&q=${args.location}')

# POST request with authentication headers
data_map.webhook(
    'POST', 
    'https://api.company.com/search',
    headers={
        'Authorization': 'Bearer YOUR_TOKEN',
        'Content-Type': 'application/json'
    }
)

# Webhook that requires specific arguments
data_map.webhook(
    'GET',
    'https://api.service.com/data?id=${args.customer_id}',
    require_args=['customer_id']
)

# Use global data for call-related info (NOT credentials)
data_map.webhook(
    'GET',
    'https://api.service.com/customer/${global_data.customer_id}/orders',
    headers={'Authorization': 'Bearer YOUR_API_TOKEN'}  # Use static credentials
)

`body(data: Dict[str, Any]) -> DataMap`

Set the JSON body for POST/PUT requests.

Parameters:

data (Dict[str, Any]): JSON body data (supports ${variable} substitution)

Usage:

# Static body with parameter substitution
data_map.body({
    'query': '${args.search_term}',
    'limit': 5,
    'filters': {
        'category': '${args.category}',
        'active': True
    }
})

# Body with call-related data (NOT sensitive info)
data_map.body({
    'customer_id': '${global_data.customer_id}',
    'request_id': '${meta_data.call_id}',
    'search': '${args.query}'
})

`params(data: Dict[str, Any]) -> DataMap`

Set URL query parameters.

Parameters:

data (Dict[str, Any]): Query parameters (supports ${variable} substitution)

Usage:

# URL parameters with substitution
data_map.params({
    'api_key': 'YOUR_API_KEY',
    'q': '${args.location}',
    'units': '${args.units}',
    'lang': 'en'
})

Multiple Webhooks and Fallbacks

DataMap supports multiple webhook configurations for fallback scenarios:

# Primary API with fallback
data_map = (DataMap('search_with_fallback')
    .purpose('Search with multiple API fallbacks')
    .parameter('query', 'string', 'Search query', required=True)
    
    # Primary API
    .webhook('GET', 'https://api.primary.com/search?q=${args.query}')
    .output(SwaigFunctionResult('Primary result: ${response.title}'))
    
    # Fallback API
    .webhook('GET', 'https://api.fallback.com/search?q=${args.query}')
    .output(SwaigFunctionResult('Fallback result: ${response.title}'))
    
    # Final fallback if all APIs fail
    .fallback_output(SwaigFunctionResult('Sorry, all search services are currently unavailable'))
)

Response Processing

Basic Output

`output(result: SwaigFunctionResult) -> DataMap`

Set the response template for successful API calls.

Parameters:

result (SwaigFunctionResult): Response template with variable substitution

Variable Substitution in Outputs:

${response.field}: API response fields
${response.nested.field}: Nested response fields
${response.array[0].field}: Array element fields
${args.parameter}: Original function arguments
${global_data.key}: Call-wide data store (user info, call state)

Usage:

# Simple response template
data_map.output(SwaigFunctionResult('Weather in ${args.location}: ${response.current.condition.text}, ${response.current.temp_f}°F'))

# Response with actions
data_map.output(
    SwaigFunctionResult('Found ${response.total_results} results')
    .update_global_data({'last_search': '${args.query}'})
    .add_action('play', 'search_complete.mp3')
)

# Complex response with nested data
data_map.output(
    SwaigFunctionResult('Order ${response.order.id} status: ${response.order.status}. Estimated delivery: ${response.order.delivery.estimated_date}')
)

`fallback_output(result: SwaigFunctionResult) -> DataMap`

Set the response when all webhooks fail.

Parameters:

result (SwaigFunctionResult): Fallback response

Usage:

data_map.fallback_output(
    SwaigFunctionResult('Sorry, the service is temporarily unavailable. Please try again later.')
    .add_action('play', 'service_unavailable.mp3')
)

Array Processing

`foreach(foreach_config: Union[str, Dict[str, Any]]) -> DataMap`

Process array responses by iterating over elements.

Parameters:

foreach_config (Union[str, Dict]): Array path or configuration object

Simple Array Processing:

# Process array of search results
data_map = (DataMap('search_docs')
    .webhook('GET', 'https://api.docs.com/search?q=${args.query}')
    .foreach('${response.results}')  # Iterate over results array
    .output(SwaigFunctionResult('Found: ${foreach.title} - ${foreach.summary}'))
)

Advanced Array Processing:

# Complex foreach configuration
data_map.foreach({
    'array': '${response.items}',
    'limit': 3,  # Process only first 3 items
    'filter': {
        'field': 'status',
        'value': 'active'
    }
})

Foreach Variable Access:

${foreach.field}: Current array element field
${foreach.nested.field}: Nested fields in current element
${foreach_index}: Current iteration index (0-based)
${foreach_count}: Total number of items being processed

Pattern-Based Processing

Expression Matching

`expression(test_value: str, pattern: Union[str, Pattern], output: SwaigFunctionResult, nomatch_output: Optional[SwaigFunctionResult] = None) -> DataMap`

Add pattern-based responses without API calls.

Parameters:

test_value (str): Template string to test against pattern
pattern (Union[str, Pattern]): Regex pattern or compiled Pattern object
output (SwaigFunctionResult): Response when pattern matches
nomatch_output (Optional[SwaigFunctionResult]): Response when pattern doesn't match

Usage:

# Command-based responses
control_map = (DataMap('file_control')
    .purpose('Control file playback')
    .parameter('command', 'string', 'Playback command', required=True)
    .parameter('filename', 'string', 'File to control')
    
    # Start commands
    .expression(
        '${args.command}', 
        r'start|play|begin',
        SwaigFunctionResult('Starting playback')
        .add_action('start_playback', {'file': '${args.filename}'})
    )
    
    # Stop commands
    .expression(
        '${args.command}',
        r'stop|pause|halt',
        SwaigFunctionResult('Stopping playback')
        .add_action('stop_playback', True)
    )
    
    # Volume commands
    .expression(
        '${args.command}',
        r'volume (\d+)',
        SwaigFunctionResult('Setting volume to ${match.1}')
        .add_action('set_volume', '${match.1}')
    )
)

Pattern Matching Variables:

${match.0}: Full match
${match.1}, ${match.2}, etc.: Capture groups
${match.group_name}: Named capture groups

Error Handling

`error_keys(keys: List[str]) -> DataMap`

Specify response fields that indicate errors.

Parameters:

keys (List[str]): List of field names that indicate API errors

Usage:

# Treat these response fields as errors
data_map.error_keys(['error', 'error_message', 'status_code'])

# If API returns {"error": "Not found"}, DataMap will treat this as an error

`global_error_keys(keys: List[str]) -> DataMap`

Set global error keys for all webhooks in this DataMap.

Parameters:

keys (List[str]): Global error field names

Usage:

data_map.global_error_keys(['error', 'message', 'code'])

Advanced Configuration

`webhook_expressions(expressions: List[Dict[str, Any]]) -> DataMap`

Add expression-based webhook selection.

Parameters:

expressions (List[Dict[str, Any]]): List of expression configurations

Usage:

# Different APIs based on input
data_map.webhook_expressions([
    {
        'test': '${args.type}',
        'pattern': 'weather',
        'webhook': {
            'method': 'GET',
            'url': 'https://weather-api.com/current?q=${args.location}'
        }
    },
    {
        'test': '${args.type}',
        'pattern': 'news',
        'webhook': {
            'method': 'GET', 
            'url': 'https://news-api.com/search?q=${args.query}'
        }
    }
])

Complete DataMap Examples

Simple Weather API

weather_tool = (DataMap('get_weather')
    .purpose('Get current weather information')
    .parameter('location', 'string', 'City name or ZIP code', required=True)
    .parameter('units', 'string', 'Temperature units', enum=['celsius', 'fahrenheit'])
    .webhook('GET', 'https://api.weather.com/v1/current?key=API_KEY&q=${args.location}&units=${args.units}')
    .output(SwaigFunctionResult('Weather in ${args.location}: ${response.current.condition.text}, ${response.current.temp_f}°F'))
    .error_keys(['error'])
)

# Register with agent
agent.register_swaig_function(weather_tool.to_swaig_function())

Search with Array Processing

search_tool = (DataMap('search_knowledge')
    .purpose('Search company knowledge base')
    .parameter('query', 'string', 'Search query', required=True)
    .parameter('category', 'string', 'Search category', enum=['docs', 'faq', 'policies'])
    .webhook(
        'POST', 
        'https://api.company.com/search',
        headers={'Authorization': 'Bearer TOKEN'}
    )
    .body({
        'query': '${args.query}',
        'category': '${args.category}',
        'limit': 5
    })
    .foreach('${response.results}')
    .output(SwaigFunctionResult('Found: ${foreach.title} - ${foreach.summary}'))
    .fallback_output(SwaigFunctionResult('Search service is temporarily unavailable'))
)

Command Processing (No API)

control_tool = (DataMap('system_control')
    .purpose('Control system functions')
    .parameter('action', 'string', 'Action to perform', required=True)
    .parameter('target', 'string', 'Target for the action')
    
    # Restart commands
    .expression(
        '${args.action}',
        r'restart|reboot',
        SwaigFunctionResult('Restarting ${args.target}')
        .add_action('restart_service', {'service': '${args.target}'})
    )
    
    # Status commands
    .expression(
        '${args.action}',
        r'status|check',
        SwaigFunctionResult('Checking status of ${args.target}')
        .add_action('check_status', {'service': '${args.target}'})
    )
    
    # Default for unrecognized commands
    .expression(
        '${args.action}',
        r'.*',
        SwaigFunctionResult('Unknown command: ${args.action}'),
        nomatch_output=SwaigFunctionResult('Please specify a valid action')
    )
)

Conversion and Registration

`to_swaig_function() -> Dict[str, Any]`

Convert the DataMap to a SWAIG function dictionary for registration.

Returns:

Dict[str, Any]: Complete SWAIG function definition

Usage:

# Build DataMap
weather_map = DataMap('get_weather').purpose('Get weather').parameter('location', 'string', 'City', required=True)

# Convert to SWAIG function and register
swaig_function = weather_map.to_swaig_function()
agent.register_swaig_function(swaig_function)

Convenience Functions

The SDK provides helper functions for common DataMap patterns:

`create_simple_api_tool(name: str, url: str, response_template: str, parameters: Optional[Dict[str, Dict]] = None, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Optional[Dict[str, Any]] = None, error_keys: Optional[List[str]] = None) -> DataMap`

Create a simple API integration tool.

Parameters:

name (str): Function name
url (str): API endpoint URL
response_template (str): Response template string
parameters (Optional[Dict[str, Dict]]): Parameter definitions
method (str): HTTP method (default: "GET")
headers (Optional[Dict[str, str]]): HTTP headers
body (Optional[Dict[str, Any]]): Request body
error_keys (Optional[List[str]]): Error field names

Usage:

from signalwire_agents.core.data_map import create_simple_api_tool

weather = create_simple_api_tool(
    name='get_weather',
    url='https://api.weather.com/v1/current?key=API_KEY&q=${location}',
    response_template='Weather in ${location}: ${response.current.condition.text}',
    parameters={
        'location': {
            'type': 'string', 
            'description': 'City name', 
            'required': True
        }
    }
)

agent.register_swaig_function(weather.to_swaig_function())

`create_expression_tool(name: str, patterns: Dict[str, Tuple[str, SwaigFunctionResult]], parameters: Optional[Dict[str, Dict]] = None) -> DataMap`

Create a pattern-based tool without API calls.

Parameters:

name (str): Function name
patterns (Dict[str, Tuple[str, SwaigFunctionResult]]): Pattern mappings
parameters (Optional[Dict[str, Dict]]): Parameter definitions

Usage:

from signalwire_agents.core.data_map import create_expression_tool

file_control = create_expression_tool(
    name='file_control',
    patterns={
        r'start.*': ('${args.command}', SwaigFunctionResult().add_action('start_playback', True)),
        r'stop.*': ('${args.command}', SwaigFunctionResult().add_action('stop_playback', True))
    },
    parameters={
        'command': {
            'type': 'string',
            'description': 'Playback command',
            'required': True
        }
    }
)

agent.register_swaig_function(file_control.to_swaig_function())

Method Chaining

All DataMap methods return self, enabling fluent method chaining:

complete_tool = (DataMap('comprehensive_search')
    .purpose('Comprehensive search with fallbacks')
    .parameter('query', 'string', 'Search query', required=True)
    .parameter('category', 'string', 'Search category', enum=['all', 'docs', 'faq'])
    .webhook('GET', 'https://primary-api.com/search?q=${args.query}&cat=${args.category}')
    .output(SwaigFunctionResult('Primary: ${response.title}'))
    .webhook('GET', 'https://backup-api.com/search?q=${args.query}')
    .output(SwaigFunctionResult('Backup: ${response.title}'))
    .fallback_output(SwaigFunctionResult('All search services unavailable'))
    .error_keys(['error', 'message'])
)

This concludes Part 3 of the API reference covering the DataMap class. The document will continue with Context System and other components in subsequent parts.

Context System

The Context System enhances traditional prompt-based agents by adding structured workflows with sequential steps on top of a base prompt. Each step contains its own guidance, completion criteria, and function restrictions while building upon the agent's foundational prompt.

ContextBuilder Class

The ContextBuilder is accessed via agent.define_contexts() and provides the main interface for creating structured workflows.

Getting Started

# Access the context builder
contexts = agent.define_contexts()

# Create contexts and steps
contexts.add_context("greeting") \
    .add_step("welcome") \
    .set_text("Welcome! How can I help you today?") \
    .set_step_criteria("User has stated their need") \
    .set_valid_steps(["next"])

`add_context(name: str) -> Context`

Create a new context in the workflow.

Parameters:

name (str): Unique context name

Returns:

Context: Context object for method chaining

Usage:

# Create multiple contexts
greeting_context = contexts.add_context("greeting")
main_menu_context = contexts.add_context("main_menu")
support_context = contexts.add_context("support")

Context Class

The Context class represents a conversation context containing multiple steps with enhanced features:

class Context:
    def add_step(self, name: str) -> Step
        """Create a new step in this context"""
    
    def set_valid_contexts(self, contexts: List[str]) -> Context
        """Set which contexts can be accessed from this context"""
        
    # Context entry parameters (for context switching behavior)
    def set_post_prompt(self, post_prompt: str) -> Context
        """Override agent's post prompt when this context is active"""
    
    def set_system_prompt(self, system_prompt: str) -> Context
        """Trigger context switch with new system instructions (makes this a Context Switch Context)"""
        
    def set_consolidate(self, consolidate: bool) -> Context
        """Whether to consolidate conversation history when entering this context"""
        
    def set_full_reset(self, full_reset: bool) -> Context
        """Whether to do complete system prompt replacement vs injection"""
        
    def set_user_prompt(self, user_prompt: str) -> Context
        """User message to inject when entering this context for AI context"""
    
    # Context prompts (guidance for all steps in context)
    def set_prompt(self, prompt: str) -> Context
        """Set simple string prompt that applies to all steps in this context"""
        
    def add_section(self, title: str, body: str) -> Context
        """Add POM-style section to context prompt"""
        
    def add_bullets(self, title: str, bullets: List[str]) -> Context
        """Add POM-style bullet section to context prompt"""

Context Types:

Workflow Container Context (no system_prompt): Organizes steps without conversation state changes
Context Switch Context (has system_prompt): Triggers conversation state changes when entered, processing entry parameters like a context_switch SWAIG action

Prompt Hierarchy: Base Agent Prompt → Context Prompt → Step Prompt

Usage Examples

# Workflow container context (just organizes steps)
main_context = contexts.add_context("main")
main_context.set_prompt("Follow standard customer service protocols")

# Context switch context (changes AI behavior)  
billing_context = contexts.add_context("billing")
billing_context.set_system_prompt("You are now a billing specialist") \
    .set_consolidate(True) \
    .set_user_prompt("Customer needs billing assistance") \
    .add_section("Department", "Billing Department") \
    .add_bullets("Services", ["Account inquiries", "Payments", "Refunds"])

# Full reset context (complete conversation reset)
manager_context = contexts.add_context("manager") 
manager_context.set_system_prompt("You are a senior manager") \
    .set_full_reset(True) \
    .set_consolidate(True)

State Management

The State Management system provides persistent storage for conversation data across calls and sessions.

StateManager (Abstract Base Class)

The base interface that all state managers must implement.

Core Methods

`store(call_id: str, data: Dict[str, Any]) -> bool`

Store state data for a call.

Parameters:

call_id (str): Unique identifier for the call
data (Dict[str, Any]): State data to store

Returns:

bool: True if successful, False otherwise

`retrieve(call_id: str) -> Optional[Dict[str, Any]]`

Retrieve state data for a call.

Parameters:

call_id (str): Unique identifier for the call

Returns:

Optional[Dict[str, Any]]: State data or None if not found

`update(call_id: str, data: Dict[str, Any]) -> bool`

Update existing state data (merges with existing).

Parameters:

call_id (str): Unique identifier for the call
data (Dict[str, Any]): Data to merge with existing state

Returns:

bool: True if successful, False otherwise

`delete(call_id: str) -> bool`

Delete state data for a call.

Parameters:

call_id (str): Unique identifier for the call

Returns:

bool: True if successful, False otherwise

`cleanup_expired() -> int`

Clean up expired state data.

Returns:

int: Number of expired items cleaned up

`exists(call_id: str) -> bool`

Check if state exists for a call.

Parameters:

call_id (str): Unique identifier for the call

Returns:

bool: True if state exists, False otherwise

FileStateManager

File-based state manager implementation that stores state data in JSON files.

Constructor

FileStateManager(
    storage_dir: str = "./agent_state",
    expiry_hours: int = 24,
    auto_cleanup: bool = True
)

Parameters:

storage_dir (str): Directory to store state files (default: "./agent_state")
expiry_hours (int): Hours after which state expires (default: 24)
auto_cleanup (bool): Automatically clean up expired files (default: True)

Usage

from signalwire_agents.core.state import FileStateManager

# Create file-based state manager
state_manager = FileStateManager(
    storage_dir="/var/agent_state",
    expiry_hours=48,  # 2 days
    auto_cleanup=True
)

# Use with agent
agent = AgentBase(
    name="Stateful Agent",
    enable_state_tracking=True,
    state_manager=state_manager
)

# Manual state operations
state_manager.store("call_123", {
    "customer_id": "12345",
    "issue_type": "billing",
    "status": "in_progress"
})

# Retrieve state
state = state_manager.retrieve("call_123")
if state:
    print(f"Customer: {state['customer_id']}")

# Update state
state_manager.update("call_123", {
    "status": "resolved",
    "resolution": "Account credited"
})

# Clean up expired state
cleaned = state_manager.cleanup_expired()
print(f"Cleaned up {cleaned} expired state files")

Custom State Manager

You can implement custom state managers for databases, Redis, etc.:

from signalwire_agents.core.state import StateManager
import redis

class RedisStateManager(StateManager):
    def __init__(self, redis_url: str, expiry_seconds: int = 86400):
        self.redis = redis.from_url(redis_url)
        self.expiry = expiry_seconds
    
    def store(self, call_id: str, data: Dict[str, Any]) -> bool:
        try:
            import json
            self.redis.setex(
                f"agent_state:{call_id}",
                self.expiry,
                json.dumps(data)
            )
            return True
        except Exception:
            return False
    
    def retrieve(self, call_id: str) -> Optional[Dict[str, Any]]:
        try:
            import json
            data = self.redis.get(f"agent_state:{call_id}")
            return json.loads(data) if data else None
        except Exception:
            return None
    
    def update(self, call_id: str, data: Dict[str, Any]) -> bool:
        existing = self.retrieve(call_id)
        if existing:
            existing.update(data)
            return self.store(call_id, existing)
        return self.store(call_id, data)
    
    def delete(self, call_id: str) -> bool:
        try:
            self.redis.delete(f"agent_state:{call_id}")
            return True
        except Exception:
            return False
    
    def cleanup_expired(self) -> int:
        # Redis handles expiry automatically
        return 0

# Use custom state manager
redis_state = RedisStateManager("redis://localhost:6379")
agent = AgentBase(
    name="Redis Agent",
    enable_state_tracking=True,
    state_manager=redis_state
)

Skills System

The Skills System provides modular, reusable capabilities that can be easily added to any agent.

Available Built-in Skills

`datetime` Skill

Provides current date and time information.

Parameters:

timezone (Optional[str]): Timezone for date/time (default: system timezone)
format (Optional[str]): Custom date/time format string

Usage:

# Basic datetime skill
agent.add_skill("datetime")

# With timezone
agent.add_skill("datetime", {"timezone": "America/New_York"})

# With custom format
agent.add_skill("datetime", {
    "timezone": "UTC",
    "format": "%Y-%m-%d %H:%M:%S %Z"
})

`math` Skill

Safe mathematical expression evaluation.

Parameters:

precision (Optional[int]): Decimal precision for results (default: 2)
max_expression_length (Optional[int]): Maximum expression length (default: 100)

Usage:

# Basic math skill
agent.add_skill("math")

# With custom precision
agent.add_skill("math", {"precision": 4})

`web_search` Skill

Google Custom Search API integration with web scraping.

Parameters:

api_key (str): Google Custom Search API key (required)
search_engine_id (str): Google Custom Search Engine ID (required)
num_results (Optional[int]): Number of results to return (default: 3)
tool_name (Optional[str]): Custom tool name for multiple instances
delay (Optional[float]): Delay between requests in seconds
no_results_message (Optional[str]): Custom message when no results found

Usage:

# Basic web search
agent.add_skill("web_search", {
    "api_key": "your-google-api-key",
    "search_engine_id": "your-search-engine-id"
})

# Multiple search instances
agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "general-engine-id",
    "tool_name": "search_general",
    "num_results": 5
})

agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "news-engine-id",
    "tool_name": "search_news",
    "num_results": 3,
    "delay": 0.5
})

`datasphere` Skill

SignalWire DataSphere knowledge search integration.

Parameters:

space_name (str): DataSphere space name (required)
project_id (str): DataSphere project ID (required)
token (str): DataSphere access token (required)
document_id (Optional[str]): Specific document to search
tool_name (Optional[str]): Custom tool name for multiple instances
count (Optional[int]): Number of results to return (default: 3)
tags (Optional[List[str]]): Filter by document tags

Usage:

# Basic DataSphere search
agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project",
    "token": "my-token"
})

# Multiple DataSphere instances
agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project",
    "token": "my-token",
    "document_id": "drinks-menu",
    "tool_name": "search_drinks",
    "count": 5
})

agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project", 
    "token": "my-token",
    "tool_name": "search_policies",
    "tags": ["HR", "Policies"]
})

`native_vector_search` Skill

Local document search with vector similarity and keyword search.

Parameters:

index_path (str): Path to search index file (required)
tool_name (Optional[str]): Custom tool name (default: "search_documents")
max_results (Optional[int]): Maximum results to return (default: 5)
similarity_threshold (Optional[float]): Minimum similarity score (default: 0.1)

Usage:

# Basic local search
agent.add_skill("native_vector_search", {
    "index_path": "./knowledge.swsearch"
})

# With custom settings
agent.add_skill("native_vector_search", {
    "index_path": "./docs.swsearch",
    "tool_name": "search_docs",
    "max_results": 10,
    "similarity_threshold": 0.3
})

Creating Custom Skills

Skill Structure

Create a new skill by extending SkillBase:

from signalwire_agents.core.skill_base import SkillBase
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult

class CustomSkill(SkillBase):
    SKILL_NAME = "custom_skill"
    SKILL_DESCRIPTION = "Description of what this skill does"
    SKILL_VERSION = "1.0.0"
    REQUIRED_PACKAGES = ["requests"]  # Python packages needed
    REQUIRED_ENV_VARS = ["API_KEY"]   # Environment variables needed
    
    def setup(self) -> bool:
        """Validate and store configuration"""
        if not self.params.get("api_key"):
            self.logger.error("api_key parameter is required")
            return False
        
        self.api_key = self.params["api_key"]
        return True
    
    def register_tools(self) -> None:
        """Register skill functions"""
        # DataMap-based tool
        tool = (DataMap("custom_function")
            .description("Custom API integration")
            .parameter("query", "string", "Search query", required=True)
            .webhook("GET", f"https://api.example.com/search?key={self.api_key}&q=${{args.query}}")
            .output(SwaigFunctionResult("Found: ${{response.title}}"))
        )
        
        self.agent.register_swaig_function(tool.to_swaig_function())
    
    def get_hints(self) -> List[str]:
        """Speech recognition hints"""
        return ["custom search", "find information"]
    
    def get_global_data(self) -> Dict[str, Any]:
        """Global data for DataMap"""
        return {"skill_version": self.SKILL_VERSION}
    
    def get_prompt_sections(self) -> List[Dict[str, Any]]:
        """Prompt sections to add"""
        return [{
            "title": "Custom Search Capability",
            "body": "You can search our custom database for information.",
            "bullets": ["Use the custom_function to search", "Results are real-time"]
        }]

Skill Registration

Skills are automatically discovered from the signalwire_agents/skills/ directory. To register a custom skill:

Create directory: signalwire_agents/skills/your_skill/
Add __init__.py, skill.py, and README.md
Implement your skill class in skill.py
The skill will be automatically available

Utility Classes

SWAIGFunction Class

Represents a SWAIG function definition with metadata and validation.

Constructor

SWAIGFunction(
    function: str,
    description: str,
    parameters: Dict[str, Any],
    **kwargs
)

Parameters:

function (str): Function name
description (str): Function description
parameters (Dict[str, Any]): JSON schema for parameters
**kwargs: Additional SWAIG properties

Usage

from signalwire_agents.core.swaig_function import SWAIGFunction

# Create SWAIG function
swaig_func = SWAIGFunction(
    function="get_weather",
    description="Get current weather",
    parameters={
        "type": "object",
        "properties": {
            "location": {"type": "string", "description": "City name"}
        },
        "required": ["location"]
    },
    secure=True,
    fillers={"en-US": ["Checking weather..."]}
)

# Register with agent
agent.register_swaig_function(swaig_func.to_dict())

SWMLService Class

Base class providing SWML document generation and HTTP service capabilities. AgentBase extends this class.

Key Methods

`get_swml_document() -> Dict[str, Any]`

Generate the complete SWML document for the service.

`handle_request(request_data: Dict[str, Any]) -> Dict[str, Any]`

Handle incoming HTTP requests and generate appropriate responses.

EphemeralAgentConfig Class

Used in dynamic configuration callbacks to modify agent settings per-request.

Available Methods

All the same configuration methods as AgentBase:

add_language(), add_hint(), set_params()
prompt_add_section(), set_global_data()
add_function_include(), set_native_functions()

Usage:

def dynamic_config(query_params, headers, body, config):
    # Configure based on request
    if query_params.get("lang") == "es":
        config.add_language("Spanish", "es-ES", "nova.luna")
    
    # Customer-specific configuration
    customer_id = headers.get("X-Customer-ID")
    if customer_id:
        config.set_global_data({"customer_id": customer_id})
        config.prompt_add_section("Customer Context", f"You are helping customer {customer_id}")

agent.set_dynamic_config_callback(dynamic_config)

Environment Variables

The SDK supports various environment variables for configuration:

Authentication

SWML_BASIC_AUTH_USER: Basic auth username
SWML_BASIC_AUTH_PASSWORD: Basic auth password

SSL/HTTPS

SWML_SSL_ENABLED: Enable SSL (true/false)
SWML_SSL_CERT_PATH: Path to SSL certificate
SWML_SSL_KEY_PATH: Path to SSL private key
SWML_DOMAIN: Domain name for SSL

Proxy Support

SWML_PROXY_URL_BASE: Base URL for proxy server

Skills Configuration

GOOGLE_SEARCH_API_KEY: Google Custom Search API key
GOOGLE_SEARCH_ENGINE_ID: Google Custom Search Engine ID
DATASPHERE_SPACE_NAME: DataSphere space name
DATASPHERE_PROJECT_ID: DataSphere project ID
DATASPHERE_TOKEN: DataSphere access token

Usage

import os

# Set environment variables
os.environ["SWML_BASIC_AUTH_USER"] = "admin"
os.environ["SWML_BASIC_AUTH_PASSWORD"] = "secret"
os.environ["GOOGLE_SEARCH_API_KEY"] = "your-api-key"

# Agent will automatically use these
agent = AgentBase("My Agent")
agent.add_skill("web_search", {
    "search_engine_id": "your-engine-id"
    # api_key will be read from environment
})

Complete Example

Here's a comprehensive example using multiple SDK components:

from signalwire_agents import AgentBase, SwaigFunctionResult, DataMap
from signalwire_agents.core.state import FileStateManager

class ComprehensiveAgent(AgentBase):
    def __init__(self):
        # Initialize with state management
        state_manager = FileStateManager(
            storage_dir="./agent_state",
            expiry_hours=48
        )
        
        super().__init__(
            name="Comprehensive Agent",
            enable_state_tracking=True,
            state_manager=state_manager,
            auto_answer=True,
            record_call=True
        )
        
        # Configure voice and language
        self.add_language("English", "en-US", "rime.spore",
                         speech_fillers=["Let me check...", "One moment..."])
        
        # Add speech recognition hints
        self.add_hints(["SignalWire", "customer service", "technical support"])
        
        # Configure AI parameters
        self.set_params({
            "ai_model": "gpt-4.1-nano",
            "end_of_speech_timeout": 800,
            "temperature": 0.7
        })
        
        # Add skills
        self.add_skill("datetime")
        self.add_skill("math")
        self.add_skill("web_search", {
            "api_key": "your-google-api-key",
            "search_engine_id": "your-engine-id",
            "num_results": 3
        })
        
        # Set up structured workflow
        self._setup_contexts()
        
        # Add custom tools
        self._register_custom_tools()
        
        # Set global data
        self.set_global_data({
            "company_name": "Acme Corp",
            "support_hours": "9 AM - 5 PM EST",
            "version": "2.0"
        })
    
    def _setup_contexts(self):
        """Set up structured workflow contexts"""
        contexts = self.define_contexts()
        
        # Greeting context
        greeting = contexts.add_context("greeting")
        greeting.add_step("welcome") \
            .set_text("Hello! Welcome to Acme Corp support. How can I help you today?") \
            .set_step_criteria("Customer has explained their issue") \
            .set_valid_steps(["next"])
        
        greeting.add_step("categorize") \
            .add_section("Current Task", "Categorize the customer's request") \
            .add_bullets("Categories", [
                "Technical issue - use diagnostic tools",
                "Billing question - transfer to billing",
                "General inquiry - handle directly"
            ]) \
            .set_functions(["transfer_to_billing", "run_diagnostics"]) \
            .set_step_criteria("Request categorized and action taken")
        
        # Technical support context
        tech = contexts.add_context("technical_support")
        tech.add_step("diagnose") \
            .set_text("Let me run some diagnostics to identify the issue.") \
            .set_functions(["run_diagnostics", "check_system_status"]) \
            .set_step_criteria("Diagnostics completed") \
            .set_valid_steps(["resolve"])
        
        tech.add_step("resolve") \
            .set_text("Based on the diagnostics, here's how we'll fix this.") \
            .set_functions(["apply_fix", "schedule_technician"]) \
            .set_step_criteria("Issue resolved or escalated")
    
    def _register_custom_tools(self):
        """Register custom DataMap tools"""
        
        # Customer lookup tool
        lookup_tool = (DataMap("lookup_customer")
            .description("Look up customer information")
            .parameter("customer_id", "string", "Customer ID", required=True)
            .webhook("GET", "https://api.company.com/customers/${args.customer_id}",
                    headers={"Authorization": "Bearer YOUR_TOKEN"})
            .output(SwaigFunctionResult("Customer: ${response.name}, Status: ${response.status}"))
            .error_keys(["error"])
        )
        
        self.register_swaig_function(lookup_tool.to_swaig_function())
        
        # System control tool
        control_tool = (DataMap("system_control")
            .description("Control system functions")
            .parameter("action", "string", "Action to perform", required=True)
            .parameter("target", "string", "Target system")
            .expression("${args.action}", r"restart|reboot",
                       SwaigFunctionResult("Restarting ${args.target}")
                       .add_action("restart_system", {"target": "${args.target}"}))
            .expression("${args.action}", r"status|check",
                       SwaigFunctionResult("Checking ${args.target} status")
                       .add_action("check_status", {"target": "${args.target}"}))
        )
        
        self.register_swaig_function(control_tool.to_swaig_function())
    
    @AgentBase.tool(
        description="Transfer call to billing department",
        parameters={"type": "object", "properties": {}}
    )
    def transfer_to_billing(self, args, raw_data):
        """Transfer to billing with state tracking"""
        return (SwaigFunctionResult("Transferring you to our billing department")
                .update_global_data({"last_action": "transfer_to_billing"})
                .connect("billing@company.com", final=False))
    
    def on_summary(self, summary, raw_data):
        """Handle conversation summaries"""
        print(f"Conversation completed: {summary}")
        # Could save to database, send notifications, etc.

# Run the agent
if __name__ == "__main__":
    agent = ComprehensiveAgent()
    agent.run()

This concludes the complete API reference for the SignalWire AI Agents SDK. The SDK provides a comprehensive framework for building sophisticated AI agents with modular capabilities, structured workflows, persistent state, and seamless deployment across multiple environments.

AgentBase Class​

Constructor​

Core Methods​

Deployment and Execution​

run(event=None, context=None, force_mode=None, host=None, port=None)​

serve(host=None, port=None)​

Prompt Configuration​

Text-Based Prompts​

set_prompt_text(text: str) -> AgentBase​

set_post_prompt(text: str) -> AgentBase​

Structured Prompts (POM)​

prompt_add_section​

prompt_add_to_section​

prompt_add_subsection​

Voice and Language Configuration​

add_language​

set_languages(languages: List[Dict[str, Any]]) -> AgentBase​

Speech Recognition Configuration​

add_hint(hint: str) -> AgentBase​

add_hints(hints: List[str]) -> AgentBase​

add_pattern_hint​

add_pronunciation​

set_pronunciations​

AI Parameters Configuration​

set_param(key: str, value: Any) -> AgentBase​

set_params(params: Dict[str, Any]) -> AgentBase​

Global Data Management​

set_global_data(data: Dict[str, Any]) -> AgentBase​

update_global_data(data: Dict[str, Any]) -> AgentBase​

Function Definition​

define_tool​

@AgentBase.tool(name=None, **kwargs) (Class Decorator)​

register_swaig_function​

Skills System​

add_skill​

remove_skill(skill_name: str) -> AgentBase​

list_skills() -> List[str]​

has_skill(skill_name: str) -> bool​

Native Functions​

set_native_functions​

set_internal_fillers​

add_internal_filler​

Function Includes​

add_function_include​

set_function_includes​

Webhook Configuration​

set_web_hook_url(url: str) -> AgentBase​

set_post_prompt_url(url: str) -> AgentBase​

Dynamic Configuration​

set_dynamic_config_callback​

SIP Integration​

enable_sip_routing​

register_sip_username(sip_username: str) -> AgentBase​

register_routing_callback​

Utility Methods​

get_name() -> str​

get_app()​

as_router() -> APIRouter​

Event Handlers​

on_summary​

on_function_call​

on_request​

on_swml_request​

Authentication​

validate_basic_auth(username: str, password: str) -> bool​

get_basic_auth_credentials​

Context System​

define_contexts() -> ContextBuilder​

SwaigFunctionResult Class​

Constructor​

Core Methods​

Response Configuration​

set_response(response: str) -> SwaigFunctionResult​

set_post_process(post_process: bool) -> SwaigFunctionResult​

Action Management​

add_action(name: str, data: Any) -> SwaigFunctionResult​

add_actions(actions: List[Dict[str, Any]]) -> SwaigFunctionResult​

Call Control Actions​

Call Transfer and Connection​

connect(destination: str, final: bool = True, from_addr: Optional[str] = None) -> SwaigFunctionResult​

AgentBase Class

Constructor

Core Methods

Deployment and Execution

`run(event=None, context=None, force_mode=None, host=None, port=None)`

`serve(host=None, port=None)`

Prompt Configuration

Text-Based Prompts

`set_prompt_text(text: str) -> AgentBase`

`set_post_prompt(text: str) -> AgentBase`

Structured Prompts (POM)

`prompt_add_section`

`prompt_add_to_section`

`prompt_add_subsection`

Voice and Language Configuration

`add_language`

`set_languages(languages: List[Dict[str, Any]]) -> AgentBase`

Speech Recognition Configuration

`add_hint(hint: str) -> AgentBase`

`add_hints(hints: List[str]) -> AgentBase`

`add_pattern_hint`

`add_pronunciation`

`set_pronunciations`

AI Parameters Configuration

`set_param(key: str, value: Any) -> AgentBase`

`set_params(params: Dict[str, Any]) -> AgentBase`

Global Data Management

`set_global_data(data: Dict[str, Any]) -> AgentBase`

`update_global_data(data: Dict[str, Any]) -> AgentBase`

Function Definition

`define_tool`

`@AgentBase.tool(name=None, **kwargs)` (Class Decorator)

`register_swaig_function`

Skills System

`add_skill`

`remove_skill(skill_name: str) -> AgentBase`

`list_skills() -> List[str]`

`has_skill(skill_name: str) -> bool`

Native Functions

`set_native_functions`

`set_internal_fillers`

`add_internal_filler`

Function Includes

`add_function_include`

`set_function_includes`

Webhook Configuration

`set_web_hook_url(url: str) -> AgentBase`

`set_post_prompt_url(url: str) -> AgentBase`

Dynamic Configuration

`set_dynamic_config_callback`

SIP Integration

`enable_sip_routing`

`register_sip_username(sip_username: str) -> AgentBase`

`register_routing_callback`

Utility Methods

`get_name() -> str`

`get_app()`

`as_router() -> APIRouter`

Event Handlers

`on_summary`

`on_function_call`

`on_request`

`on_swml_request`

Authentication

`validate_basic_auth(username: str, password: str) -> bool`

`get_basic_auth_credentials`

Context System

`define_contexts() -> ContextBuilder`

SwaigFunctionResult Class

Constructor

Core Methods

Response Configuration

`set_response(response: str) -> SwaigFunctionResult`

`set_post_process(post_process: bool) -> SwaigFunctionResult`

Action Management

`add_action(name: str, data: Any) -> SwaigFunctionResult`

`add_actions(actions: List[Dict[str, Any]]) -> SwaigFunctionResult`

Call Control Actions

Call Transfer and Connection

`connect(destination: str, final: bool = True, from_addr: Optional[str] = None) -> SwaigFunctionResult`