This document provides a comprehensive reference for all public APIs in the SignalWire AI Agents SDK.
- AgentBase Class - Core agent functionality
- SwaigFunctionResult Class - SWAIG (SignalWire AI Gateway) function response handling
- DataMap Class - Serverless API tools that execute on SignalWire's servers
- Context System - Structured workflows
- State Management - Persistent state
- Skills System - Modular capabilities
- Utility Classes - Supporting classes
The AgentBase class is the foundation for creating AI agents. It extends SWMLService (the base class for generating SWML -- SignalWire Markup Language -- documents) and provides comprehensive functionality for building conversational AI agents.
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,
token_expiry_secs: int = 3600,
auto_answer: bool = True,
record_call: bool = False,
record_format: str = "mp4",
record_stereo: bool = True,
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,
config_file: Optional[str] = None
)Parameters:
name(str): Human-readable name for the agentroute(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 authuse_pom(bool): Whether to use Prompt Object Model (default: True)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)default_webhook_url(Optional[str]): Default webhook URL for functionsagent_id(Optional[str]): Unique identifier for the agentnative_functions(Optional[List[str]]): List of native function names to enableschema_path(Optional[str]): Path to custom SWML schema filesuppress_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)config_file(Optional[str]): Path to JSON configuration file with environment variable substitution support. See Configuration Guide for details.
Auto-detects deployment environment and runs the agent appropriately.
Parameters:
event: Event object for serverless environmentscontext: Context object for serverless environmentsforce_mode(str): Force specific mode: "server", "lambda", "cgi", "cloud_function"host(Optional[str]): Override host addressport(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)Explicitly run as HTTP server using FastAPI/Uvicorn.
Parameters:
host(Optional[str]): Host address to bind toport(Optional[int]): Port number to listen on
Usage:
agent.serve() # Use constructor defaults
agent.serve(host="0.0.0.0", port=3000)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 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.")def set_prompt_llm_params(**params) -> AgentBaseSet Language Model parameters for the main prompt. Accepts any parameters which will be passed through to the SignalWire server. The server validates and applies parameters based on the target model's capabilities.
Common Parameters:
temperature: Controls randomness. Lower = more focusedtop_p: Nucleus sampling thresholdbarge_confidence: ASR confidence to interruptpresence_penalty: Topic diversity controlfrequency_penalty: Repetition control
Note: No defaults are sent unless explicitly set. Invalid parameters for the selected model will be handled/ignored by the server.
Usage:
# Configure for consistent, professional responses
agent.set_prompt_llm_params(
temperature=0.3,
top_p=0.9,
barge_confidence=0.7,
presence_penalty=0.1,
frequency_penalty=0.2
)def set_post_prompt_llm_params(**params) -> AgentBaseSet Language Model parameters for the post-prompt. Accepts any parameters which will be passed through to the SignalWire server. The server validates and applies parameters based on the target model's capabilities.
Common Parameters:
temperature: Controls randomness. Lower = more focusedtop_p: Nucleus sampling thresholdpresence_penalty: Topic diversity controlfrequency_penalty: Repetition control
Note: barge_confidence is not applicable to post-prompt. No defaults are sent unless explicitly set.
Usage:
# Configure for focused summaries
agent.set_post_prompt_llm_params(
temperature=0.2,
top_p=0.9
)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
) -> AgentBaseAdd a structured section to the prompt using Prompt Object Model.
Parameters:
title(str): Section title/headingbody(str): Main section content (default: "")bullets(Optional[List[str]]): List of bullet pointsnumbered(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
)def prompt_add_to_section(
title: str,
body: Optional[str] = None,
bullet: Optional[str] = None,
bullets: Optional[List[str]] = None
) -> AgentBaseAdd content to an existing prompt section.
Parameters:
title(str): Title of existing section to modifybody(Optional[str]): Additional body text to appendbullet(Optional[str]): Single bullet point to addbullets(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"])def prompt_add_subsection(
parent_title: str,
title: str,
body: str = "",
bullets: Optional[List[str]] = None
) -> AgentBaseAdd a subsection to an existing prompt section.
Parameters:
parent_title(str): Title of parent sectiontitle(str): Subsection titlebody(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"]
)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
) -> AgentBaseConfigure voice and language settings for the agent.
Parameters:
name(str): Human-readable language namecode(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 processingfunction_fillers(Optional[List[str]]): Filler phrases during function executionengine(Optional[str]): TTS engine to usemodel(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 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"}
])Add a single speech recognition hint.
Parameters:
hint(str): Word or phrase to improve recognition accuracy
Usage:
agent.add_hint("SignalWire")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"])def add_pattern_hint(
hint: str,
pattern: str,
replace: str,
ignore_case: bool = False
) -> AgentBaseAdd a pattern-based hint for speech recognition.
Parameters:
hint(str): The hint phrasepattern(str): Regex pattern to matchreplace(str): Replacement textignore_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"
)def add_pronunciation(
replace: str,
with_text: str,
ignore_case: bool = False
) -> AgentBaseAdd pronunciation rules for text-to-speech.
Parameters:
replace(str): Text to replacewith_text(str): Replacement pronunciationignore_case(bool): Case-insensitive replacement (default: False)
Usage:
agent.add_pronunciation("API", "A P I")
agent.add_pronunciation("SWML", "swim-el")def set_pronunciations(
pronunciations: List[Dict[str, Any]]
) -> AgentBaseSet 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}
])Set a single AI parameter.
Parameters:
key(str): Parameter namevalue(Any): Parameter value
Usage:
agent.set_param("ai_model", "gpt-4.1-nano")
agent.set_param("end_of_speech_timeout", 500)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 lengthtop_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
})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 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"
})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,
is_typed_handler: bool = False,
**swaig_fields
) -> AgentBaseDefine a custom SWAIG function/tool.
Parameters:
name(str): Function namedescription(str): Function description for AIparameters(Dict[str, Any]): JSON schema for function parameters. If omitted when using the decorator and the handler has type-hinted parameters, the schema is inferred automatically from the type hints.handler(Callable): Function to execute when calledsecure(bool): Require security token (default: True)fillers(Optional[Dict[str, List[str]]]): Language-specific filler phraseswebhook_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..."]}
)Decorator for defining tools as class methods.
Parameters:
name(Optional[str]): Function name (defaults to method name)**kwargs: Same parameters asdefine_tool()
When parameters is omitted and the handler has type-hinted parameters (beyond self), the schema is inferred automatically from the type hints. The description is extracted from the docstring's first line, and per-parameter descriptions come from the Args: block.
Usage (explicit schema):
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()}")Usage (type-hinted, schema inferred):
class MyAgent(AgentBase):
@AgentBase.tool(name="get_weather")
def get_weather(self, city: str, units: str = "celsius"):
"""Get the weather forecast.
Args:
city: Name of the city
units: Temperature units
"""
return SwaigFunctionResult(f"Weather in {city}")def register_swaig_function(
function_dict: Dict[str, Any]
) -> AgentBaseRegister a pre-built SWAIG function dictionary.
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())SignalWire AI agents support special SWAIG functions that are automatically called at specific points in the conversation lifecycle:
Called when a new conversation/call begins.
Implementation:
@AgentBase.tool(
name="startup_hook",
description="Called when a new conversation starts to initialize state",
parameters={}
)
def startup_hook(self, args, raw_data):
call_id = raw_data.get("call_id")
# Initialize session resources, load user data, etc.
return SwaigFunctionResult("Session initialized")Called when a conversation/call ends.
Implementation:
@AgentBase.tool(
name="hangup_hook",
description="Called when conversation ends to clean up resources",
parameters={}
)
def hangup_hook(self, args, raw_data):
call_id = raw_data.get("call_id")
# Clean up resources, save session data, etc.
return SwaigFunctionResult("Session ended")Common Use Cases:
- Loading user preferences at session start
- Initializing session-specific resources
- Logging conversation metrics
- Cleaning up temporary data
- Saving conversation summaries
def add_skill(
skill_name: str,
params: Optional[Dict[str, Any]] = None
) -> AgentBaseAdd a modular skill to the agent.
Parameters:
skill_name(str): Name of the skill to addparams(Optional[Dict[str, Any]]): Skill configuration parameters
Available Skills:
datetime: Current date/time informationmath: Mathematical calculationsweb_search: Google Custom Search integrationdatasphere: SignalWire DataSphere searchnative_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 a skill from the agent.
Parameters:
skill_name(str): Name of skill to remove
Usage:
agent.remove_skill("web_search")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}")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")def set_native_functions(
function_names: List[str]
) -> AgentBaseEnable specific native SWML functions.
Parameters:
function_names(List[str]): List of native function names to enable
Available Native Functions:
transfer: Transfer callshangup: End callsplay: Play audio filesrecord: Record audiosend_sms: Send SMS messages
Usage:
agent.set_native_functions(["transfer", "hangup", "send_sms"])def set_internal_fillers(
internal_fillers: Dict[str, Dict[str, List[str]]]
) -> AgentBaseSet 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 workflowscheck_time: Getting current timewait_for_user: Waiting for user inputwait_seconds: Pausing for specified durationget_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..."]
}
})def add_internal_filler(
function_name: str,
language_code: str,
fillers: List[str]
) -> AgentBaseAdd internal fillers for a specific function and language.
Parameters:
function_name(str): Name of the internal functionlanguage_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..."
])def add_function_include(
url: str,
functions: List[str],
meta_data: Optional[Dict[str, Any]] = None
) -> AgentBaseInclude external SWAIG functions from another service.
Parameters:
url(str): URL of external SWAIG servicefunctions(List[str]): List of function names to includemeta_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"}
)def set_function_includes(
includes: List[Dict[str, Any]]
) -> AgentBaseSet 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"}
}
])Set default webhook URL for SWAIG functions.
Parameters:
url(str): Default webhook URL
Usage:
agent.set_web_hook_url("https://myserver.com/webhook")Set URL for post-prompt processing.
Parameters:
url(str): Post-prompt webhook URL
Usage:
agent.set_post_prompt_url("https://myserver.com/post-prompt")Add query parameters to be included in all SWAIG webhook URLs.
This is useful for preserving dynamic configuration state across SWAIG callbacks. For example, if your dynamic config adds skills based on query parameters, you can pass those same parameters through to the SWAIG webhook so the same configuration is applied.
Parameters:
params(dict): Dictionary of query parameter key-value pairs
Usage:
# In dynamic config callback, preserve configuration parameters
def configure_agent(query_params, headers, body, agent):
customer_id = query_params.get("customer_id")
if customer_id:
# Pass through to SWAIG callbacks
agent.add_swaig_query_params({"customer_id": customer_id})
agent.add_skill("customer_lookup", {"customer_id": customer_id})
agent.set_dynamic_config_callback(configure_agent)Clear all SWAIG query parameters.
Usage:
agent.clear_swaig_query_params()def enable_debug_events(level: int = 1) -> AgentBaseEnable the debug event webhook for this agent. When enabled, the AI module will POST real-time debug events to a /debug_events endpoint on this agent during calls. Events are automatically logged via the agent's structured logger and can optionally be handled with a custom callback via on_debug_event().
Parameters:
level(int): Debug event verbosity level.1= high-level events (barge, errors, session start/end, step changes).2+= adds high-volume events (every LLM request/response, conversation_add). Default:1
Usage:
agent.enable_debug_events() # level 1 (default)
agent.enable_debug_events(level=2) # include high-volume eventsHow it works:
- Registers a
/debug_eventsPOST endpoint on the agent's HTTP server - Auto-sets
debug_webhook_urlanddebug_webhook_levelin the SWMLparamsduring rendering - The URL is built automatically using the same auth/proxy logic as other webhook URLs
- No manual URL configuration needed
Event types at level 1:
| Event label | Description |
|---|---|
session_start |
AI session started (model, TTS engine, voice, language) |
session_end |
AI session ended (reason, duration, token counts) |
barge |
User interrupted AI speech (barge type, elapsed ms) |
step_change |
Conversation step changed |
context_change |
Conversation context changed |
llm_error |
LLM error (fatal, retry, max_retries) |
voice_error |
TTS voice configuration or runtime error |
hold |
Call placed on hold or taken off hold |
filler |
Filler phrase spoken (thinking or function filler) |
consolidation |
Token consolidation triggered |
process_action |
Webhook action being processed |
gather_start |
Gather flow started |
gather_complete |
Gather flow completed |
Additional events at level 2+:
| Event label | Description |
|---|---|
llm_request |
LLM API request initiated (input tokens) |
llm_response |
LLM API response received (duration, output tokens) |
conversation_add |
Entry added to conversation history |
These methods allow you to customize the SWML call flow by inserting verbs at different stages of the call lifecycle.
Add a verb to run before the call is answered (while still ringing).
Safe pre-answer verbs: transfer, execute, return, label, goto, request, switch, cond, if, eval, set, unset, hangup, send_sms, sleep, stop_record_call, stop_denoise, stop_tap
Parameters:
verb_name(str): The SWML verb nameconfig(dict): Verb configuration dictionary
Usage:
# Send SMS before answering
agent.add_pre_answer_verb("send_sms", {
"to": "+15551234567",
"from": "+15559876543",
"body": "Incoming call from AI agent"
})
# Set variables before answer
agent.add_pre_answer_verb("set", {"call_start": "${system.timestamp}"})Configure the answer verb that connects the call.
Parameters:
config(dict, optional): Answer verb configuration (e.g.,{"max_duration": 3600})
Usage:
# Set maximum call duration to 1 hour
agent.add_answer_verb({"max_duration": 3600})Add a verb to run after the call is answered but before the AI starts.
Parameters:
verb_name(str): The SWML verb name (e.g., "play", "sleep")config(dict): Verb configuration dictionary
Usage:
# Play welcome message before AI starts
agent.add_post_answer_verb("play", {
"url": "say:Welcome to our AI assistant. This call may be recorded."
})
# Add a brief pause
agent.add_post_answer_verb("sleep", {"duration": 1})Add a verb to run after the AI conversation ends.
Parameters:
verb_name(str): The SWML verb name (e.g., "hangup", "transfer", "request")config(dict): Verb configuration dictionary
Usage:
# Clean hangup after AI ends
agent.add_post_ai_verb("hangup", {})
# Transfer to human after AI conversation
agent.add_post_ai_verb("transfer", {"to": "+15551234567"})
# Log call completion
agent.add_post_ai_verb("request", {
"url": "https://myserver.com/call-complete",
"method": "POST"
})Remove all pre-answer verbs.
Remove all post-answer verbs.
Remove all post-AI verbs.
Method Chaining Example:
agent.add_pre_answer_verb("set", {"source": "ai_agent"}) \
.add_answer_verb({"max_duration": 1800}) \
.add_post_answer_verb("play", {"url": "say:Hello!"}) \
.add_post_ai_verb("hangup", {})def set_dynamic_config_callback(
callback: Callable[[dict, dict, dict, AgentBase], None]
) -> AgentBaseSet 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)def enable_sip_routing(
auto_map: bool = True,
path: str = "/sip"
) -> AgentBaseEnable 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 a specific SIP username for this agent.
Parameters:
sip_username(str): SIP username to register
Usage:
agent.register_sip_username("support")
agent.register_sip_username("sales")def register_routing_callback(
callback_fn: Callable[[Request, Dict[str, Any]], Optional[str]],
path: str = "/sip"
) -> NoneRegister custom routing logic for SIP calls.
Parameters:
callback_fn(Callable): Function that returns agent route based on requestpath(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)Get the agent's name.
Returns:
- str: Agent name
Get the FastAPI application instance.
Returns:
- FastAPI: The underlying FastAPI app
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")def on_summary(
summary: Optional[Dict[str, Any]],
raw_data: Optional[Dict[str, Any]] = None
) -> NoneOverride to handle conversation summaries. This callback is triggered when the AI generates a summary based on your post_prompt configuration.
Parameters:
summary(Optional[Dict[str, Any]]): Parsed summary data (frompost_prompt_data.parsed[0])raw_data(Optional[Dict[str, Any]]): Complete raw POST data includingpost_prompt_datawith bothrawandparsedfields
Usage:
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="summary-agent", route="/agent")
# Configure post-prompt to request JSON summary
self.set_post_prompt("""
Return a JSON summary of the conversation:
{
"topic": "MAIN_TOPIC",
"satisfied": true/false,
"follow_up_needed": true/false,
"key_points": ["point1", "point2"]
}
""")
def on_summary(self, summary, raw_data):
"""Handle conversation summaries after call ends"""
if summary:
# Access parsed JSON fields directly
topic = summary.get("topic", "Unknown")
satisfied = summary.get("satisfied", False)
print(f"Call about: {topic}, Customer satisfied: {satisfied}")
# Save to database, send to CRM, trigger follow-up, etc.
if summary.get("follow_up_needed"):
self.schedule_follow_up(summary)
# Access raw summary text if needed
if raw_data and 'post_prompt_data' in raw_data:
raw_text = raw_data['post_prompt_data'].get('raw', '')
print(f"Raw summary: {raw_text}")def on_debug_event(handler: Callable) -> CallableRegister a handler for debug webhook events. Use as a decorator. Requires enable_debug_events() to be called first.
The handler receives:
event_type(str): The event label (e.g."barge","llm_error","session_start")data(dict): The full event payload includingcall_id,label, and event-specific fields
The handler may be sync or async.
Usage (decorator style):
agent = AgentBase("my_agent")
agent.enable_debug_events()
@agent.on_debug_event
def handle_debug(event_type, data):
call_id = data.get("call_id")
if event_type == "llm_error":
print(f"LLM error on call {call_id}: {data.get('event')}")
elif event_type == "barge":
print(f"Barge after {data.get('barge_elapsed_ms')}ms")
elif event_type == "session_end":
print(f"Call ended: {data.get('reason')}, duration: {data.get('duration_ms')}ms")Usage (subclass style):
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="debug-agent", route="/agent")
self.enable_debug_events(level=2)
self.on_debug_event(self.handle_debug)
def handle_debug(self, event_type, data):
if event_type == "llm_error":
self.alert_ops_team(data)Note: Even without registering a handler, all debug events are automatically logged via the agent's structured logger when
enable_debug_events()is called.
def on_function_call(
name: str,
args: Dict[str, Any],
raw_data: Optional[Dict[str, Any]] = None
) -> AnyOverride to handle function calls with custom logic.
Parameters:
name(str): Function name being calledargs(Dict[str, Any]): Function argumentsraw_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)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 datacallback_path(Optional[str]): Callback path
Returns:
- Optional[dict]: Response modifications
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 datacallback_path(Optional[str]): Callback pathrequest(Optional[Request]): FastAPI request object
Returns:
- Optional[dict]: SWML modifications
Override to implement custom basic authentication logic.
Parameters:
username(str): Username from basic authpassword(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"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)
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.
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.
SwaigFunctionResult(
response: Optional[str] = None,
post_process: bool = False
)Parameters:
response(Optional[str]): Natural language response text for the AI to speakpost_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 responsepost_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()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")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 firstAdd 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 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}
])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 destinationfinal(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 completelyfinal=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")Create a SWML-based transfer with AI response setup.
Parameters:
dest(str): Transfer destinationai_response(str): AI response when transfer completes
Usage:
result.swml_transfer(
"+15551234567",
"You've been transferred back to me. How else can I help?"
)Perform a SIP REFER transfer.
Parameters:
to_uri(str): SIP URI to transfer to
Usage:
result.sip_refer("sip:support@company.com")End the call immediately.
Usage:
result = SwaigFunctionResult("Thank you for calling. Goodbye!")
result.hangup()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 current audio playback or recording.
Usage:
result.stop()Add text for the AI to speak.
Parameters:
text(str): Text to speak
Usage:
result.say("Please wait while I process your request")Play an audio file in the background.
Parameters:
filename(str): Audio file path or URLwait(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 audio playback.
Usage:
result.stop_background_file()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 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 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 for the conversation.
Parameters:
data(Dict[str, Any]): Metadata to set
Usage:
result.set_metadata({
"call_type": "support",
"priority": "high",
"department": "technical"
})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"])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 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 usertimeout(Optional[int]): Timeout in millisecondsanswer_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)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}
])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 timeoutEnable extensive data collection.
Parameters:
enabled(bool): Enable extensive data (default: True)
Usage:
result.enable_extensive_data(True)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
})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 promptuser_prompt(Optional[str]): New user promptconsolidate(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 for testing or automation.
Parameters:
text(str): Text to simulate as user input
Usage:
result.simulate_user_input("I need help with my order")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 numberfrom_number(str): Sender phone numberbody(Optional[str]): SMS message textmedia(Optional[List[str]]): List of media URLstags(Optional[List[str]]): Message tagsregion(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"]
)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 recordingstereo(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 recordingbeep(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 secondsstatus_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 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")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 namemuted(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/contentmax_participants(int): Maximum participants (default: 250)record(str): Recording setting (default: "do-not-record")region(Optional[str]): SignalWire regiontrim(str): Trim setting for recordings (default: "trim-silence")coach(Optional[str]): Coach participant identifierstatus_callback_event(Optional[str]): Status callback eventsstatus_callback(Optional[str]): Status callback URLstatus_callback_method(str): Status callback HTTP method (default: "POST")recording_status_callback(Optional[str]): Recording status callback URLrecording_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
)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 URLinput_method(str): Input method: "dtmf", "speech" (default: "dtmf")status_url(Optional[str]): Payment status webhook URLpayment_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 chargecurrency(str): Currency code (default: "usd")language(str): Language for prompts (default: "en-US")voice(str): Voice for prompts (default: "woman")description(Optional[str]): Payment descriptionvalid_card_types(str): Accepted card types (default: "visa mastercard amex")parameters(Optional[List[Dict[str, str]]]): Additional parametersprompts(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"
)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 tocontrol_id(Optional[str]): Unique identifier for this tapdirection(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 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")Execute custom SWML content.
Parameters:
swml_content: SWML document or content to executetransfer(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)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"}]}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 identifieractions(List[Dict[str, str]]): List of action configurationscard_type(Optional[str]): Card type for promptserror_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 a payment action configuration.
Parameters:
action_type(str): Action typephrase(str): Action phrase
Usage:
action = SwaigFunctionResult.create_payment_action("say", "Enter your card number")Create a payment parameter configuration.
Parameters:
name(str): Parameter namevalue(str): Parameter value
Usage:
param = SwaigFunctionResult.create_payment_parameter("merchant_id", "12345")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.
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.
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')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')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(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 nameparam_type(str): JSON schema type: "string", "number", "boolean", "array", "object"description(str): Parameter description for the AIrequired(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')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 sendform_param(Optional[str]): Send JSON body as single form parameter with this nameinput_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
)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}'
})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'
})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'))
)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}')
)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')
)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
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 patternpattern(Union[str, Pattern]): Regex pattern or compiled Pattern objectoutput(SwaigFunctionResult): Response when pattern matchesnomatch_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
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 errorSet 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'])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}'
}
}
])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_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'))
)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')
)
)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)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 nameurl(str): API endpoint URLresponse_template(str): Response template stringparameters(Optional[Dict[str, Dict]]): Parameter definitionsmethod(str): HTTP method (default: "GET")headers(Optional[Dict[str, str]]): HTTP headersbody(Optional[Dict[str, Any]]): Request bodyerror_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 namepatterns(Dict[str, Tuple[str, SwaigFunctionResult]]): Pattern mappingsparameters(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())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.
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.
The ContextBuilder is accessed via agent.define_contexts() and provides the main interface for creating structured workflows.
# 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"])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")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 acontext_switchSWAIG action
Prompt Hierarchy: Base Agent Prompt → Context Prompt → Step Prompt
# 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)The Skills System provides modular, reusable capabilities that can be easily added to any agent.
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"
})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})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 instancesdelay(Optional[float]): Delay between requests in secondsno_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
})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 searchtool_name(Optional[str]): Custom tool name for multiple instancescount(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"]
})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 0.0-1.0 (default: 0.0). Higher values are stricter, lower values are more permissive. Typical range: 0.2-0.4 for all-MiniLM-L6-v2, 0.3-0.5 for all-mpnet-base-v2
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.25
})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"]
}]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, andREADME.md - Implement your skill class in
skill.py - The skill will be automatically available
Represents a SWAIG function definition with metadata and validation.
SWAIGFunction(
function: str,
description: str,
parameters: Dict[str, Any],
**kwargs
)Parameters:
function(str): Function namedescription(str): Function descriptionparameters(Dict[str, Any]): JSON schema for parameters**kwargs: Additional SWAIG properties
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())Base class providing SWML document generation and HTTP service capabilities. AgentBase extends this class.
Generate the complete SWML document for the service.
Handle incoming HTTP requests and generate appropriate responses.
The dynamic configuration callback receives the agent instance directly, allowing you to configure it based on request data.
Usage:
def dynamic_config(query_params, body_params, headers, agent):
# Configure based on request
if query_params.get("lang") == "es":
agent.add_language("Spanish", "es-ES", "nova.luna")
# Customer-specific configuration
customer_id = headers.get("X-Customer-ID")
if customer_id:
agent.set_global_data({"customer_id": customer_id})
agent.prompt_add_section("Customer Context", f"You are helping customer {customer_id}")
# Add skills dynamically
if query_params.get("enable_search") == "true":
agent.add_skill("web_search", {"provider": "google"})
agent.set_dynamic_config_callback(dynamic_config)The SDK supports various environment variables for configuration:
SWML_BASIC_AUTH_USER: Basic auth usernameSWML_BASIC_AUTH_PASSWORD: Basic auth password
SWML_SSL_ENABLED: Enable SSL (true/false)SWML_SSL_CERT_PATH: Path to SSL certificateSWML_SSL_KEY_PATH: Path to SSL private keySWML_DOMAIN: Domain name for SSL
SWML_PROXY_URL_BASE: Base URL for proxy server
GOOGLE_SEARCH_API_KEY: Google Custom Search API keyGOOGLE_SEARCH_ENGINE_ID: Google Custom Search Engine IDDATASPHERE_SPACE_NAME: DataSphere space nameDATASPHERE_PROJECT_ID: DataSphere project IDDATASPHERE_TOKEN: DataSphere access token
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
})Here's a comprehensive example using multiple SDK components:
from signalwire_agents import AgentBase, SwaigFunctionResult, DataMap
class ComprehensiveAgent(AgentBase):
def __init__(self):
super().__init__(
name="Comprehensive Agent",
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 deployment across multiple environments.