SignalWire AI Agents SDK - Complete API Reference 🔗 ↑ TOC

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

Table of Contents 🔗 ↑ TOC

  1. AgentBase Class - Core agent functionality
  2. SwaigFunctionResult Class - Function response handling
  3. DataMap Class - Serverless API integration
  4. Context System - Structured workflows
  5. State Management - Persistent state
  6. Skills System - Modular capabilities
  7. Utility Classes - Supporting classes

AgentBase Class 🔗 ↑ TOC

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

Constructor 🔗 ↑ TOC

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:

Core Methods 🔗 ↑ TOC

Deployment and Execution 🔗 ↑ TOC

run(event=None, context=None, force_mode=None, host=None, port=None) 🔗 ↑ TOC

Auto-detects deployment environment and runs the agent appropriately.

Parameters:

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) 🔗 ↑ TOC

Explicitly run as HTTP server using FastAPI/Uvicorn.

Parameters:

Usage:

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

Prompt Configuration 🔗 ↑ TOC

Text-Based Prompts 🔗 ↑ TOC

set_prompt_text(text: str) -> AgentBase 🔗 ↑ TOC

Set the agent's prompt as raw text.

Parameters:

Usage:

agent.set_prompt_text("You are a helpful customer service agent.")
set_post_prompt(text: str) -> AgentBase 🔗 ↑ TOC

Set additional text to append after the main prompt.

Parameters:

Usage:

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

Structured Prompts (POM) 🔗 ↑ TOC

prompt_add_section 🔗 ↑ TOC
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:

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 🔗 ↑ TOC
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:

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 🔗 ↑ TOC
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:

Usage:

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

Voice and Language Configuration 🔗 ↑ TOC

add_language 🔗 ↑ TOC
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:

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 🔗 ↑ TOC

Set multiple language configurations at once.

Parameters:

Usage:

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

Speech Recognition Configuration 🔗 ↑ TOC

add_hint(hint: str) -> AgentBase 🔗 ↑ TOC

Add a single speech recognition hint.

Parameters:

Usage:

agent.add_hint("SignalWire")
add_hints(hints: List[str]) -> AgentBase 🔗 ↑ TOC

Add multiple speech recognition hints.

Parameters:

Usage:

agent.add_hints(["SignalWire", "SWML", "API", "webhook", "SIP"])
add_pattern_hint 🔗 ↑ TOC
def add_pattern_hint(
    hint: str, 
    pattern: str, 
    replace: str, 
    ignore_case: bool = False
) -> AgentBase

Add a pattern-based hint for speech recognition.

Parameters:

Usage:

agent.add_pattern_hint(
    "phone number",
    r"(\d{3})-(\d{3})-(\d{4})",
    r"(\1) \2-\3"
)
add_pronunciation 🔗 ↑ TOC
def add_pronunciation(
    replace: str, 
    with_text: str, 
    ignore_case: bool = False
) -> AgentBase

Add pronunciation rules for text-to-speech.

Parameters:

Usage:

agent.add_pronunciation("API", "A P I")
agent.add_pronunciation("SWML", "swim-el")
set_pronunciations 🔗 ↑ TOC
def set_pronunciations(
    pronunciations: List[Dict[str, Any]]
) -> AgentBase

Set multiple pronunciation rules at once.

Parameters:

Usage:

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

AI Parameters Configuration 🔗 ↑ TOC

set_param(key: str, value: Any) -> AgentBase 🔗 ↑ TOC

Set a single AI parameter.

Parameters:

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 🔗 ↑ TOC

Set multiple AI parameters at once.

Parameters:

Common Parameters:

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 🔗 ↑ TOC

set_global_data(data: Dict[str, Any]) -> AgentBase 🔗 ↑ TOC

Set global data available to the AI and functions.

Parameters:

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 🔗 ↑ TOC

Update existing global data (merge with existing).

Parameters:

Usage:

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

Function Definition 🔗 ↑ TOC

define_tool 🔗 ↑ TOC
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:

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) 🔗 ↑ TOC

Decorator for defining tools as class methods.

Parameters:

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 🔗 ↑ TOC
def register_swaig_function(
    function_dict: Dict[str, Any]
) -> AgentBase

Register a pre-built SWAIG function dictionary.

Parameters:

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 🔗 ↑ TOC

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

Add a modular skill to the agent.

Parameters:

Available Skills:

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 🔗 ↑ TOC

Remove a skill from the agent.

Parameters:

Usage:

agent.remove_skill("web_search")
list_skills() -> List[str] 🔗 ↑ TOC

Get list of currently added skills.

Returns:

Usage:

active_skills = agent.list_skills()
print(f"Active skills: {active_skills}")
has_skill(skill_name: str) -> bool 🔗 ↑ TOC

Check if a skill is currently added.

Parameters:

Returns:

Usage:

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

Native Functions 🔗 ↑ TOC

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

Enable specific native SWML functions.

Parameters:

Available Native Functions:

Usage:

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

Function Includes 🔗 ↑ TOC

add_function_include 🔗 ↑ TOC
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:

Usage:

agent.add_function_include(
    "https://external-service.com/swaig",
    ["external_function1", "external_function2"],
    meta_data={"service": "external", "version": "1.0"}
)
set_function_includes 🔗 ↑ TOC
def set_function_includes(
    includes: List[Dict[str, Any]]
) -> AgentBase

Set multiple function includes at once.

Parameters:

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 🔗 ↑ TOC

set_web_hook_url(url: str) -> AgentBase 🔗 ↑ TOC

Set default webhook URL for SWAIG functions.

Parameters:

Usage:

agent.set_web_hook_url("https://myserver.com/webhook")
set_post_prompt_url(url: str) -> AgentBase 🔗 ↑ TOC

Set URL for post-prompt processing.

Parameters:

Usage:

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

Dynamic Configuration 🔗 ↑ TOC

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

Set callback for per-request dynamic configuration.

Parameters:

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 🔗 ↑ TOC

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

Enable SIP-based routing for voice calls.

Parameters:

Usage:

agent.enable_sip_routing()
register_sip_username(sip_username: str) -> AgentBase 🔗 ↑ TOC

Register a specific SIP username for this agent.

Parameters:

Usage:

agent.register_sip_username("support")
agent.register_sip_username("sales")
register_routing_callback 🔗 ↑ TOC
def register_routing_callback(
    callback_fn: Callable[[Request, Dict[str, Any]], Optional[str]], 
    path: str = "/sip"
) -> None

Register custom routing logic for SIP calls.

Parameters:

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 🔗 ↑ TOC

get_name() -> str 🔗 ↑ TOC

Get the agent's name.

Returns:

get_app() 🔗 ↑ TOC

Get the FastAPI application instance.

Returns:

as_router() -> APIRouter 🔗 ↑ TOC

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

Returns:

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 🔗 ↑ TOC

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

Override to handle conversation summaries.

Parameters:

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 🔗 ↑ TOC
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:

Returns:

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 🔗 ↑ TOC
def on_request(
    request_data: Optional[dict] = None, 
    callback_path: Optional[str] = None
) -> Optional[dict]

Override to handle general requests.

Parameters:

Returns:

on_swml_request 🔗 ↑ TOC
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:

Returns:

Authentication 🔗 ↑ TOC

validate_basic_auth(username: str, password: str) -> bool 🔗 ↑ TOC

Override to implement custom basic authentication logic.

Parameters:

Returns:

Usage:

class MyAgent(AgentBase):
    def validate_basic_auth(self, username, password):
        # Custom auth logic
        return username == "admin" and password == "secret"
get_basic_auth_credentials 🔗 ↑ TOC
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:

Returns:

Context System 🔗 ↑ TOC

define_contexts() -> ContextBuilder 🔗 ↑ TOC

Define structured workflow contexts for the agent.

Returns:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

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

Parameters:

Post-processing Behavior:

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 🔗 ↑ TOC

Response Configuration 🔗 ↑ TOC

set_response(response: str) -> SwaigFunctionResult 🔗 ↑ TOC

Set or update the natural language response text.

Parameters:

Usage:

result = SwaigFunctionResult()
result.set_response("I found your order information")
set_post_process(post_process: bool) -> SwaigFunctionResult 🔗 ↑ TOC

Enable or disable post-processing for this result.

Parameters:

Usage:

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

Action Management 🔗 ↑ TOC

add_action(name: str, data: Any) -> SwaigFunctionResult 🔗 ↑ TOC

Add a structured action to execute.

Parameters:

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 🔗 ↑ TOC

Add multiple actions at once.

Parameters:

Usage:

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

Call Control Actions 🔗 ↑ TOC

Call Transfer and Connection 🔗 ↑ TOC

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

Transfer or connect the call to another destination.

Parameters:

Transfer Types:

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 🔗 ↑ TOC

Create a SWML-based transfer with AI response setup.

Parameters:

Usage:

result.swml_transfer(
    "+15551234567", 
    "You've been transferred back to me. How else can I help?"
)
sip_refer(to_uri: str) -> SwaigFunctionResult 🔗 ↑ TOC

Perform a SIP REFER transfer.

Parameters:

Usage:

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

Call Management 🔗 ↑ TOC

hangup() -> SwaigFunctionResult 🔗 ↑ TOC

End the call immediately.

Usage:

result = SwaigFunctionResult("Thank you for calling. Goodbye!")
result.hangup()
hold(timeout: int = 300) -> SwaigFunctionResult 🔗 ↑ TOC

Put the call on hold.

Parameters:

Usage:

result = SwaigFunctionResult("Please hold while I look that up")
result.hold(timeout=60)
stop() -> SwaigFunctionResult 🔗 ↑ TOC

Stop current audio playback or recording.

Usage:

result.stop()

Audio Control 🔗 ↑ TOC

say(text: str) -> SwaigFunctionResult 🔗 ↑ TOC

Add text for the AI to speak.

Parameters:

Usage:

result.say("Please wait while I process your request")
play_background_file(filename: str, wait: bool = False) -> SwaigFunctionResult 🔗 ↑ TOC

Play an audio file in the background.

Parameters:

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 🔗 ↑ TOC

Stop background audio playback.

Usage:

result.stop_background_file()

Data Management Actions 🔗 ↑ TOC

set_global_data(data: Dict[str, Any]) -> SwaigFunctionResult 🔗 ↑ TOC

Set global data for the conversation.

Parameters:

Usage:

result.set_global_data({
    "customer_id": "12345",
    "order_status": "shipped",
    "tracking_number": "1Z999AA1234567890"
})
update_global_data(data: Dict[str, Any]) -> SwaigFunctionResult 🔗 ↑ TOC

Update existing global data (merge with existing).

Parameters:

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 🔗 ↑ TOC

Remove specific keys from global data.

Parameters:

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 🔗 ↑ TOC

Set metadata for the conversation.

Parameters:

Usage:

result.set_metadata({
    "call_type": "support",
    "priority": "high",
    "department": "technical"
})
remove_metadata(keys: Union[str, List[str]]) -> SwaigFunctionResult 🔗 ↑ TOC

Remove specific metadata keys.

Parameters:

Usage:

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

AI Behavior Control 🔗 ↑ TOC

set_end_of_speech_timeout(milliseconds: int) -> SwaigFunctionResult 🔗 ↑ TOC

Adjust how long to wait for speech to end.

Parameters:

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 🔗 ↑ TOC

Set timeout for speech events.

Parameters:

Usage:

result.set_speech_event_timeout(5000)
wait_for_user(enabled: Optional[bool] = None, timeout: Optional[int] = None, answer_first: bool = False) -> SwaigFunctionResult 🔗 ↑ TOC

Control whether to wait for user input.

Parameters:

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 🔗 ↑ TOC

Enable or disable specific functions.

Parameters:

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 🔗 ↑ TOC

Control whether functions are enabled when timeout occurs.

Parameters:

Usage:

result.enable_functions_on_timeout(False)  # Disable functions on timeout
enable_extensive_data(enabled: bool = True) -> SwaigFunctionResult 🔗 ↑ TOC

Enable extensive data collection.

Parameters:

Usage:

result.enable_extensive_data(True)
update_settings(settings: Dict[str, Any]) -> SwaigFunctionResult 🔗 ↑ TOC

Update various AI settings.

Parameters:

Usage:

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

Context and Conversation Control 🔗 ↑ TOC

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

Switch conversation context or reset the conversation.

Parameters:

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 🔗 ↑ TOC

Simulate user input for testing or automation.

Parameters:

Usage:

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

Communication Actions 🔗 ↑ TOC

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 🔗 ↑ TOC

Send an SMS message.

Parameters:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

Start call recording.

Parameters:

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 🔗 ↑ TOC

Stop call recording.

Parameters:

Usage:

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

Conference and Room Management 🔗 ↑ TOC

join_room(name: str) -> SwaigFunctionResult 🔗 ↑ TOC

Join a SignalWire room.

Parameters:

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 🔗 ↑ TOC

Join a conference call.

Parameters:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

Process a payment through the call.

Parameters:

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 🔗 ↑ TOC

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

Start call tapping/monitoring.

Parameters:

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 🔗 ↑ TOC

Stop call tapping.

Parameters:

Usage:

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

Advanced SWML Execution 🔗 ↑ TOC

execute_swml(swml_content, transfer: bool = False) -> SwaigFunctionResult 🔗 ↑ TOC

Execute custom SWML content.

Parameters:

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 🔗 ↑ TOC

to_dict() -> Dict[str, Any] 🔗 ↑ TOC

Convert the result to a dictionary for serialization.

Returns:

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 🔗 ↑ TOC

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

Create a payment prompt configuration.

Parameters:

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] 🔗 ↑ TOC

Create a payment action configuration.

Parameters:

Usage:

action = SwaigFunctionResult.create_payment_action("say", "Enter your card number")
create_payment_parameter(name: str, value: str) -> Dict[str, str] 🔗 ↑ TOC

Create a payment parameter configuration.

Parameters:

Usage:

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

Method Chaining 🔗 ↑ TOC

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 🔗 ↑ TOC

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 🔗 ↑ TOC

DataMap(function_name: str)

Parameters:

Usage:

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

Core Configuration Methods 🔗 ↑ TOC

Function Metadata 🔗 ↑ TOC

purpose(description: str) -> DataMap 🔗 ↑ TOC

Set the function description/purpose.

Parameters:

Usage:

data_map = DataMap('get_weather').purpose('Get current weather information for any city')
description(description: str) -> DataMap 🔗 ↑ TOC

Alias for purpose() - set the function description.

Parameters:

Usage:

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

Parameter Definition 🔗 ↑ TOC

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

Add a function parameter with JSON schema validation.

Parameters:

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 🔗 ↑ TOC

HTTP Webhook Configuration 🔗 ↑ TOC

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 🔗 ↑ TOC

Configure an HTTP API call.

Parameters:

Variable Substitution in URLs:

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 🔗 ↑ TOC

Set the JSON body for POST/PUT requests.

Parameters:

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 🔗 ↑ TOC

Set URL query parameters.

Parameters:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

Basic Output 🔗 ↑ TOC

output(result: SwaigFunctionResult) -> DataMap 🔗 ↑ TOC

Set the response template for successful API calls.

Parameters:

Variable Substitution in Outputs:

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 🔗 ↑ TOC

Set the response when all webhooks fail.

Parameters:

Usage:

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

Array Processing 🔗 ↑ TOC

foreach(foreach_config: Union[str, Dict[str, Any]]) -> DataMap 🔗 ↑ TOC

Process array responses by iterating over elements.

Parameters:

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:

Pattern-Based Processing 🔗 ↑ TOC

Expression Matching 🔗 ↑ TOC

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

Add pattern-based responses without API calls.

Parameters:

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:

Error Handling 🔗 ↑ TOC

error_keys(keys: List[str]) -> DataMap 🔗 ↑ TOC

Specify response fields that indicate errors.

Parameters:

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 🔗 ↑ TOC

Set global error keys for all webhooks in this DataMap.

Parameters:

Usage:

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

Advanced Configuration 🔗 ↑ TOC

webhook_expressions(expressions: List[Dict[str, Any]]) -> DataMap 🔗 ↑ TOC

Add expression-based webhook selection.

Parameters:

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 🔗 ↑ TOC

Simple Weather API 🔗 ↑ TOC

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 🔗 ↑ TOC

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) 🔗 ↑ TOC

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 🔗 ↑ TOC

to_swaig_function() -> Dict[str, Any] 🔗 ↑ TOC

Convert the DataMap to a SWAIG function dictionary for registration.

Returns:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

Create a simple API integration tool.

Parameters:

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 🔗 ↑ TOC

Create a pattern-based tool without API calls.

Parameters:

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 🔗 ↑ TOC

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 🔗 ↑ TOC

The Context System provides an alternative to traditional prompt-based agents by allowing structured workflows with sequential steps. Each step contains its own prompt, completion criteria, and function restrictions.

ContextBuilder Class 🔗 ↑ TOC

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

Getting Started 🔗 ↑ TOC

# 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 🔗 ↑ TOC

Create a new context in the workflow.

Parameters:

Returns:

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 🔗 ↑ TOC

Represents a single context containing multiple sequential steps.

Step Management 🔗 ↑ TOC

add_step(name: str) -> Step 🔗 ↑ TOC

Add a new step to this context.

Parameters:

Returns:

Usage:

context = contexts.add_context("customer_service")

# Add multiple steps
welcome_step = context.add_step("welcome")
identify_step = context.add_step("identify_need")
resolve_step = context.add_step("resolve_issue")
set_valid_contexts(contexts: List[str]) -> Context 🔗 ↑ TOC

Set which contexts can be navigated to from this context.

Parameters:

Usage:

# Allow navigation to specific contexts
context.set_valid_contexts(["main_menu", "escalation", "end_call"])

# Allow navigation to any context
context.set_valid_contexts(["*"])

Step Class 🔗 ↑ TOC

Represents a single step within a context with its own prompt and behavior.

Prompt Configuration 🔗 ↑ TOC

set_text(text: str) -> Step 🔗 ↑ TOC

Set the step's prompt as raw text.

Parameters:

Usage:

step.set_text("Please provide your account number so I can look up your information.")
add_section(title: str, body: str) -> Step 🔗 ↑ TOC

Add a structured prompt section (POM-style).

Parameters:

Usage:

step.add_section("Your Role", "You are a technical support specialist")
step.add_section("Current Task", "Help the customer troubleshoot their connection issue")
add_bullets(title: str, bullets: List[str]) -> Step 🔗 ↑ TOC

Add a section with bullet points.

Parameters:

Usage:

step.add_bullets("Available Actions", [
    "Transfer to billing department",
    "Schedule a callback",
    "Escalate to supervisor",
    "End the call"
])

Note: You cannot mix set_text() with add_section()/add_bullets(). Choose one approach per step.

Step Behavior 🔗 ↑ TOC

set_step_criteria(criteria: str) -> Step 🔗 ↑ TOC

Define when this step is considered complete.

Parameters:

Usage:

step.set_step_criteria("Customer has provided their account number and it has been verified")
step.set_step_criteria("Issue has been resolved or escalated appropriately")
set_functions(functions: Union[str, List[str]]) -> Step 🔗 ↑ TOC

Control which functions are available in this step.

Parameters:

Usage:

# Disable all functions
step.set_functions("none")

# Allow specific functions only
step.set_functions(["lookup_account", "verify_identity"])

# Allow all functions (default behavior)
step.set_functions(["*"])
set_valid_steps(steps: List[str]) -> Step 🔗 ↑ TOC

Define which steps can be navigated to from this step.

Parameters:

Usage:

# Sequential flow to next step
step.set_valid_steps(["next"])

# Allow jumping to specific steps
step.set_valid_steps(["verify_account", "escalate", "end_call"])

# Allow any step navigation
step.set_valid_steps(["*"])

Complete Context Example 🔗 ↑ TOC

# Define a customer service workflow
contexts = agent.define_contexts()

# Greeting context
greeting = contexts.add_context("greeting")
greeting.add_step("welcome") \
    .set_text("Hello! Welcome to Acme Support. How can I help you today?") \
    .set_step_criteria("Customer has explained their issue or request") \
    .set_valid_steps(["next"])

greeting.add_step("categorize") \
    .add_section("Your Task", "Categorize the customer's request") \
    .add_bullets("Categories", [
        "Technical support - transfer to tech team",
        "Billing inquiry - transfer to billing", 
        "General question - handle directly"
    ]) \
    .set_functions(["transfer_to_tech", "transfer_to_billing"]) \
    .set_step_criteria("Request has been categorized and appropriate action taken")

greeting.set_valid_contexts(["technical_support", "billing", "general_help"])

# Technical support context
tech_support = contexts.add_context("technical_support")
tech_support.add_step("diagnose") \
    .set_text("I'll help you troubleshoot this technical issue. Let me gather some information.") \
    .set_functions(["run_diagnostics", "check_system_status"]) \
    .set_step_criteria("Diagnostic information has been collected") \
    .set_valid_steps(["next"])

tech_support.add_step("resolve") \
    .set_text("Based on the diagnostics, here's what we need to do to fix this issue.") \
    .set_functions(["apply_fix", "schedule_technician", "escalate_to_engineer"]) \
    .set_step_criteria("Issue has been resolved or escalated appropriately")

tech_support.set_valid_contexts(["greeting", "escalation"])

Utility Function 🔗 ↑ TOC

create_simple_context(name: str = "default") -> Context 🔗 ↑ TOC

Create a simple single-context workflow.

Parameters:

Returns:

Usage:

from signalwire_agents.core.contexts import create_simple_context

# Create a simple linear workflow
simple_context = create_simple_context("customer_intake")
simple_context.add_step("greeting").set_text("Welcome to our service!")
simple_context.add_step("collect_info").set_text("Please provide your details")
simple_context.add_step("process").set_text("Let me process your request")

State Management 🔗 ↑ TOC

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

StateManager (Abstract Base Class) 🔗 ↑ TOC

The base interface that all state managers must implement.

Core Methods 🔗 ↑ TOC

store(call_id: str, data: Dict[str, Any]) -> bool 🔗 ↑ TOC

Store state data for a call.

Parameters:

Returns:

retrieve(call_id: str) -> Optional[Dict[str, Any]] 🔗 ↑ TOC

Retrieve state data for a call.

Parameters:

Returns:

update(call_id: str, data: Dict[str, Any]) -> bool 🔗 ↑ TOC

Update existing state data (merges with existing).

Parameters:

Returns:

delete(call_id: str) -> bool 🔗 ↑ TOC

Delete state data for a call.

Parameters:

Returns:

cleanup_expired() -> int 🔗 ↑ TOC

Clean up expired state data.

Returns:

exists(call_id: str) -> bool 🔗 ↑ TOC

Check if state exists for a call.

Parameters:

Returns:

FileStateManager 🔗 ↑ TOC

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

Constructor 🔗 ↑ TOC

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

Parameters:

Usage 🔗 ↑ TOC

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 🔗 ↑ TOC

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 🔗 ↑ TOC

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

Available Built-in Skills 🔗 ↑ TOC

datetime Skill 🔗 ↑ TOC

Provides current date and time information.

Parameters:

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 🔗 ↑ TOC

Safe mathematical expression evaluation.

Parameters:

Usage:

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

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

web_search Skill 🔗 ↑ TOC

Google Custom Search API integration with web scraping.

Parameters:

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 🔗 ↑ TOC

SignalWire DataSphere knowledge search integration.

Parameters:

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 🔗 ↑ TOC

Local document search with vector similarity and keyword search.

Parameters:

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 🔗 ↑ TOC

Skill Structure 🔗 ↑ TOC

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 🔗 ↑ TOC

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

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

Utility Classes 🔗 ↑ TOC

SWAIGFunction Class 🔗 ↑ TOC

Represents a SWAIG function definition with metadata and validation.

Constructor 🔗 ↑ TOC

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

Parameters:

Usage 🔗 ↑ TOC

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 🔗 ↑ TOC

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

Key Methods 🔗 ↑ TOC

get_swml_document() -> Dict[str, Any] 🔗 ↑ TOC

Generate the complete SWML document for the service.

handle_request(request_data: Dict[str, Any]) -> Dict[str, Any] 🔗 ↑ TOC

Handle incoming HTTP requests and generate appropriate responses.

EphemeralAgentConfig Class 🔗 ↑ TOC

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

Available Methods 🔗 ↑ TOC

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 🔗 ↑ TOC

The SDK supports various environment variables for configuration:

Authentication 🔗 ↑ TOC

SSL/HTTPS 🔗 ↑ TOC

Proxy Support 🔗 ↑ TOC

Skills Configuration 🔗 ↑ TOC

Usage 🔗 ↑ TOC

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 🔗 ↑ TOC

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("Task", "Categorize the customer's request") \
            .add_bullets("Options", [
                "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(["next"])

        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.