This document provides a comprehensive reference for all public APIs in the SignalWire AI Agents SDK.
The AgentBase
class is the foundation for creating AI agents. It extends SWMLService
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,
enable_state_tracking: bool = False,
token_expiry_secs: int = 3600,
auto_answer: bool = True,
record_call: bool = False,
record_format: str = "mp4",
record_stereo: bool = True,
state_manager: Optional[StateManager] = None,
default_webhook_url: Optional[str] = None,
agent_id: Optional[str] = None,
native_functions: Optional[List[str]] = None,
schema_path: Optional[str] = None,
suppress_logs: bool = False,
enable_post_prompt_override: bool = False,
check_for_input_override: bool = False
)
Parameters:
name
(str): Human-readable name for the 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)enable_state_tracking
(bool): Enable persistent state management (default: False)token_expiry_secs
(int): Security token expiration time (default: 3600)auto_answer
(bool): Automatically answer incoming calls (default: True)record_call
(bool): Record calls by default (default: False)record_format
(str): Recording format: "mp4", "wav", "mp3" (default: "mp4")record_stereo
(bool): Record in stereo (default: True)state_manager
(Optional[StateManager]): Custom state manager instancedefault_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)run(event=None, context=None, force_mode=None, host=None, port=None)
🔗 ↑ TOCAuto-detects deployment environment and runs the agent appropriately.
Parameters:
event
: Event object for serverless environmentscontext
: Context object for serverless environments force_mode
(str): Force specific mode: "server", "lambda", "cgi", "cloud_function"host
(Optional[str]): Override host addressport
(Optional[int]): Override port numberUsage:
# 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)
🔗 ↑ TOCExplicitly run as HTTP server using FastAPI/Uvicorn.
Parameters:
host
(Optional[str]): Host address to bind toport
(Optional[int]): Port number to listen onUsage:
agent.serve() # Use constructor defaults
agent.serve(host="0.0.0.0", port=3000)
set_prompt_text(text: str) -> AgentBase
🔗 ↑ TOCSet the agent's prompt as raw text.
Parameters:
text
(str): The complete prompt textUsage:
agent.set_prompt_text("You are a helpful customer service agent.")
set_post_prompt(text: str) -> AgentBase
🔗 ↑ TOCSet additional text to append after the main prompt.
Parameters:
text
(str): Text to append after main promptUsage:
agent.set_post_prompt("Always be polite and professional.")
prompt_add_section
🔗 ↑ TOCdef prompt_add_section(
title: str,
body: str = "",
bullets: Optional[List[str]] = None,
numbered: bool = False,
numbered_bullets: bool = False,
subsections: Optional[List[Dict[str, Any]]] = None
) -> AgentBase
Add a structured section to the prompt using Prompt Object Model.
Parameters:
title
(str): Section title/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 subsectionsUsage:
# 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
🔗 ↑ TOCdef prompt_add_to_section(
title: str,
body: Optional[str] = None,
bullet: Optional[str] = None,
bullets: Optional[List[str]] = None
) -> AgentBase
Add content to an existing prompt section.
Parameters:
title
(str): Title of existing section to modifybody
(Optional[str]): Additional body text to appendbullet
(Optional[str]): Single bullet point to addbullets
(Optional[List[str]]): Multiple bullet points to addUsage:
# 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
🔗 ↑ TOCdef prompt_add_subsection(
parent_title: str,
title: str,
body: str = "",
bullets: Optional[List[str]] = None
) -> AgentBase
Add a subsection to an existing prompt section.
Parameters:
parent_title
(str): Title of parent sectiontitle
(str): Subsection titlebody
(str): Subsection content (default: "")bullets
(Optional[List[str]]): Subsection bullet pointsUsage:
agent.prompt_add_subsection(
"Guidelines",
"Escalation Rules",
"Escalate when:",
bullets=["Customer is angry", "Technical issue beyond scope"]
)
add_language
🔗 ↑ TOCdef add_language(
name: str,
code: str,
voice: str,
speech_fillers: Optional[List[str]] = None,
function_fillers: Optional[List[str]] = None,
engine: Optional[str] = None,
model: Optional[str] = None
) -> AgentBase
Configure voice and language settings for the agent.
Parameters:
name
(str): Human-readable language 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 useUsage:
# 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
🔗 ↑ TOCSet multiple language configurations at once.
Parameters:
languages
(List[Dict]): List of language configuration dictionariesUsage:
agent.set_languages([
{"name": "English", "code": "en-US", "voice": "rime.spore"},
{"name": "Spanish", "code": "es-ES", "voice": "nova.luna"}
])
add_hint(hint: str) -> AgentBase
🔗 ↑ TOCAdd a single speech recognition hint.
Parameters:
hint
(str): Word or phrase to improve recognition accuracyUsage:
agent.add_hint("SignalWire")
add_hints(hints: List[str]) -> AgentBase
🔗 ↑ TOCAdd multiple speech recognition hints.
Parameters:
hints
(List[str]): List of words/phrases for better recognitionUsage:
agent.add_hints(["SignalWire", "SWML", "API", "webhook", "SIP"])
add_pattern_hint
🔗 ↑ TOCdef add_pattern_hint(
hint: str,
pattern: str,
replace: str,
ignore_case: bool = False
) -> AgentBase
Add a pattern-based hint for speech recognition.
Parameters:
hint
(str): The hint 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"
)
add_pronunciation
🔗 ↑ TOCdef add_pronunciation(
replace: str,
with_text: str,
ignore_case: bool = False
) -> AgentBase
Add 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")
set_pronunciations
🔗 ↑ TOCdef set_pronunciations(
pronunciations: List[Dict[str, Any]]
) -> AgentBase
Set multiple pronunciation rules at once.
Parameters:
pronunciations
(List[Dict]): List of pronunciation rule dictionariesUsage:
agent.set_pronunciations([
{"replace": "API", "with": "A P I"},
{"replace": "SWML", "with": "swim-el", "ignore_case": True}
])
set_param(key: str, value: Any) -> AgentBase
🔗 ↑ TOCSet a single AI parameter.
Parameters:
key
(str): Parameter namevalue
(Any): Parameter valueUsage:
agent.set_param("ai_model", "gpt-4.1-nano")
agent.set_param("end_of_speech_timeout", 500)
set_params(params: Dict[str, Any]) -> AgentBase
🔗 ↑ TOCSet multiple AI parameters at once.
Parameters:
params
(Dict[str, Any]): Dictionary of parameter key-value pairsCommon 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(data: Dict[str, Any]) -> AgentBase
🔗 ↑ TOCSet global data available to the AI and functions.
Parameters:
data
(Dict[str, Any]): Global data dictionaryUsage:
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
🔗 ↑ TOCUpdate existing global data (merge with existing).
Parameters:
data
(Dict[str, Any]): Data to merge with existing global dataUsage:
agent.update_global_data({
"current_promotion": "20% off all services",
"promotion_expires": "2024-12-31"
})
define_tool
🔗 ↑ TOCdef define_tool(
name: str,
description: str,
parameters: Dict[str, Any],
handler: Callable,
secure: bool = True,
fillers: Optional[Dict[str, List[str]]] = None,
webhook_url: Optional[str] = None,
**swaig_fields
) -> AgentBase
Define a custom SWAIG function/tool.
Parameters:
name
(str): Function namedescription
(str): Function description for AIparameters
(Dict[str, Any]): JSON schema for function parametershandler
(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 propertiesUsage:
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) 🔗 ↑ TOCDecorator for defining tools as class methods.
Parameters:
name
(Optional[str]): Function name (defaults to method name)**kwargs
: Same parameters as define_tool()
Usage:
class MyAgent(AgentBase):
@AgentBase.tool(
description="Get current time",
parameters={"type": "object", "properties": {}}
)
def get_time(self, args, raw_data):
import datetime
return SwaigFunctionResult(f"Current time: {datetime.datetime.now()}")
register_swaig_function
🔗 ↑ TOCdef register_swaig_function(
function_dict: Dict[str, Any]
) -> AgentBase
Register a pre-built SWAIG function dictionary.
Parameters:
function_dict
(Dict[str, Any]): Complete SWAIG function definitionUsage:
# Register a DataMap tool
weather_tool = DataMap('get_weather').webhook('GET', 'https://api.weather.com/...')
agent.register_swaig_function(weather_tool.to_swaig_function())
add_skill
🔗 ↑ TOCdef add_skill(
skill_name: str,
params: Optional[Dict[str, Any]] = None
) -> AgentBase
Add a modular skill to the agent.
Parameters:
skill_name
(str): Name of the skill to addparams
(Optional[Dict[str, Any]]): Skill configuration parametersAvailable Skills:
datetime
: Current date/time informationmath
: Mathematical calculationsweb_search
: Google Custom Search integrationdatasphere
: SignalWire DataSphere searchnative_vector_search
: Local document searchUsage:
# 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
🔗 ↑ TOCRemove a skill from the agent.
Parameters:
skill_name
(str): Name of skill to removeUsage:
agent.remove_skill("web_search")
list_skills() -> List[str]
🔗 ↑ TOCGet list of currently added skills.
Returns:
Usage:
active_skills = agent.list_skills()
print(f"Active skills: {active_skills}")
has_skill(skill_name: str) -> bool
🔗 ↑ TOCCheck if a skill is currently added.
Parameters:
skill_name
(str): Name of skill to checkReturns:
Usage:
if agent.has_skill("web_search"):
print("Web search is available")
set_native_functions
🔗 ↑ TOCdef set_native_functions(
function_names: List[str]
) -> AgentBase
Enable specific native SWML functions.
Parameters:
function_names
(List[str]): List of native function names to enableAvailable Native Functions:
transfer
: Transfer callshangup
: End callsplay
: Play audio filesrecord
: Record audiosend_sms
: Send SMS messagesUsage:
agent.set_native_functions(["transfer", "hangup", "send_sms"])
add_function_include
🔗 ↑ TOCdef add_function_include(
url: str,
functions: List[str],
meta_data: Optional[Dict[str, Any]] = None
) -> AgentBase
Include external SWAIG functions from another service.
Parameters:
url
(str): URL of external SWAIG servicefunctions
(List[str]): List of function names to includemeta_data
(Optional[Dict[str, Any]]): Additional metadataUsage:
agent.add_function_include(
"https://external-service.com/swaig",
["external_function1", "external_function2"],
meta_data={"service": "external", "version": "1.0"}
)
set_function_includes
🔗 ↑ TOCdef set_function_includes(
includes: List[Dict[str, Any]]
) -> AgentBase
Set multiple function includes at once.
Parameters:
includes
(List[Dict[str, Any]]): List of function include configurationsUsage:
agent.set_function_includes([
{
"url": "https://service1.com/swaig",
"functions": ["func1", "func2"]
},
{
"url": "https://service2.com/swaig",
"functions": ["func3"],
"meta_data": {"priority": "high"}
}
])
set_web_hook_url(url: str) -> AgentBase
🔗 ↑ TOCSet default webhook URL for SWAIG functions.
Parameters:
url
(str): Default webhook URLUsage:
agent.set_web_hook_url("https://myserver.com/webhook")
set_post_prompt_url(url: str) -> AgentBase
🔗 ↑ TOCSet URL for post-prompt processing.
Parameters:
url
(str): Post-prompt webhook URLUsage:
agent.set_post_prompt_url("https://myserver.com/post-prompt")
set_dynamic_config_callback
🔗 ↑ TOCdef set_dynamic_config_callback(
callback: Callable[[dict, dict, dict, EphemeralAgentConfig], None]
) -> AgentBase
Set callback for per-request dynamic configuration.
Parameters:
callback
(Callable): Function that receives (query_params, headers, body, config)Usage:
def configure_agent(query_params, headers, body, config):
# Configure based on request
if query_params.get("language") == "spanish":
config.add_language("Spanish", "es-ES", "nova.luna")
# Set customer-specific data
customer_id = headers.get("X-Customer-ID")
if customer_id:
config.set_global_data({"customer_id": customer_id})
agent.set_dynamic_config_callback(configure_agent)
enable_sip_routing
🔗 ↑ TOCdef enable_sip_routing(
auto_map: bool = True,
path: str = "/sip"
) -> AgentBase
Enable SIP-based routing for voice calls.
Parameters:
auto_map
(bool): Automatically map SIP usernames (default: True)path
(str): SIP routing endpoint path (default: "/sip")Usage:
agent.enable_sip_routing()
register_sip_username(sip_username: str) -> AgentBase
🔗 ↑ TOCRegister a specific SIP username for this agent.
Parameters:
sip_username
(str): SIP username to registerUsage:
agent.register_sip_username("support")
agent.register_sip_username("sales")
register_routing_callback
🔗 ↑ TOCdef register_routing_callback(
callback_fn: Callable[[Request, Dict[str, Any]], Optional[str]],
path: str = "/sip"
) -> None
Register 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_name() -> str
🔗 ↑ TOCGet the agent's name.
Returns:
get_app()
🔗 ↑ TOCGet the FastAPI application instance.
Returns:
as_router() -> APIRouter
🔗 ↑ TOCGet 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")
on_summary
🔗 ↑ TOCdef on_summary(
summary: Optional[Dict[str, Any]],
raw_data: Optional[Dict[str, Any]] = None
) -> None
Override to handle conversation summaries.
Parameters:
summary
(Optional[Dict[str, Any]]): Summary dataraw_data
(Optional[Dict[str, Any]]): Raw request dataUsage:
class MyAgent(AgentBase):
def on_summary(self, summary, raw_data):
print(f"Conversation summary: {summary}")
# Save to database, send notification, etc.
on_function_call
🔗 ↑ TOCdef on_function_call(
name: str,
args: Dict[str, Any],
raw_data: Optional[Dict[str, Any]] = None
) -> Any
Override to handle function calls with custom logic.
Parameters:
name
(str): Function name being calledargs
(Dict[str, Any]): Function argumentsraw_data
(Optional[Dict[str, Any]]): Raw request dataReturns:
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
🔗 ↑ TOCdef 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 pathReturns:
on_swml_request
🔗 ↑ TOCdef 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 path request
(Optional[Request]): FastAPI request objectReturns:
validate_basic_auth(username: str, password: str) -> bool
🔗 ↑ TOCOverride to implement custom basic authentication logic.
Parameters:
username
(str): Username from basic authpassword
(str): Password from basic authReturns:
Usage:
class MyAgent(AgentBase):
def validate_basic_auth(self, username, password):
# Custom auth logic
return username == "admin" and password == "secret"
get_basic_auth_credentials
🔗 ↑ TOCdef 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:
define_contexts() -> ContextBuilder
🔗 ↑ TOCDefine 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.
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 actionsUsage:
# 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_response(response: str) -> SwaigFunctionResult
🔗 ↑ TOCSet or update the natural language response text.
Parameters:
response
(str): The text the AI should speakUsage:
result = SwaigFunctionResult()
result.set_response("I found your order information")
set_post_process(post_process: bool) -> SwaigFunctionResult
🔗 ↑ TOCEnable or disable post-processing for this result.
Parameters:
post_process
(bool): True to let AI respond once more before executing actionsUsage:
result = SwaigFunctionResult("I'll help you with that")
result.set_post_process(True) # Let AI handle follow-up questions first
add_action(name: str, data: Any) -> SwaigFunctionResult
🔗 ↑ TOCAdd 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 arrayUsage:
# 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
🔗 ↑ TOCAdd multiple actions at once.
Parameters:
actions
(List[Dict[str, Any]]): List of action dictionariesUsage:
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
🔗 ↑ TOCTransfer 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 IDTransfer Types:
final=True
: Permanent transfer - call exits agent completelyfinal=False
: Temporary transfer - call returns to agent if far end hangs upUsage:
# 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
🔗 ↑ TOCCreate a SWML-based transfer with AI response setup.
Parameters:
dest
(str): Transfer destinationai_response
(str): AI response when transfer completesUsage:
result.swml_transfer(
"+15551234567",
"You've been transferred back to me. How else can I help?"
)
sip_refer(to_uri: str) -> SwaigFunctionResult
🔗 ↑ TOCPerform a SIP REFER transfer.
Parameters:
to_uri
(str): SIP URI to transfer toUsage:
result.sip_refer("sip:support@company.com")
hangup() -> SwaigFunctionResult
🔗 ↑ TOCEnd the call immediately.
Usage:
result = SwaigFunctionResult("Thank you for calling. Goodbye!")
result.hangup()
hold(timeout: int = 300) -> SwaigFunctionResult
🔗 ↑ TOCPut the call on hold.
Parameters:
timeout
(int): Hold timeout in seconds (default: 300)Usage:
result = SwaigFunctionResult("Please hold while I look that up")
result.hold(timeout=60)
stop() -> SwaigFunctionResult
🔗 ↑ TOCStop current audio playback or recording.
Usage:
result.stop()
say(text: str) -> SwaigFunctionResult
🔗 ↑ TOCAdd text for the AI to speak.
Parameters:
text
(str): Text to speakUsage:
result.say("Please wait while I process your request")
play_background_file(filename: str, wait: bool = False) -> SwaigFunctionResult
🔗 ↑ TOCPlay 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_file() -> SwaigFunctionResult
🔗 ↑ TOCStop background audio playback.
Usage:
result.stop_background_file()
set_global_data(data: Dict[str, Any]) -> SwaigFunctionResult
🔗 ↑ TOCSet global data for the conversation.
Parameters:
data
(Dict[str, Any]): Global data to setUsage:
result.set_global_data({
"customer_id": "12345",
"order_status": "shipped",
"tracking_number": "1Z999AA1234567890"
})
update_global_data(data: Dict[str, Any]) -> SwaigFunctionResult
🔗 ↑ TOCUpdate existing global data (merge with existing).
Parameters:
data
(Dict[str, Any]): Data to mergeUsage:
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
🔗 ↑ TOCRemove specific keys from global data.
Parameters:
keys
(Union[str, List[str]]): Key name or list of key names to removeUsage:
# 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
🔗 ↑ TOCSet metadata for the conversation.
Parameters:
data
(Dict[str, Any]): Metadata to setUsage:
result.set_metadata({
"call_type": "support",
"priority": "high",
"department": "technical"
})
remove_metadata(keys: Union[str, List[str]]) -> SwaigFunctionResult
🔗 ↑ TOCRemove specific metadata keys.
Parameters:
keys
(Union[str, List[str]]): Key name or list of key names to removeUsage:
result.remove_metadata(["temporary_flag", "debug_info"])
set_end_of_speech_timeout(milliseconds: int) -> SwaigFunctionResult
🔗 ↑ TOCAdjust how long to wait for speech to end.
Parameters:
milliseconds
(int): Timeout in millisecondsUsage:
# 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
🔗 ↑ TOCSet timeout for speech events.
Parameters:
milliseconds
(int): Timeout in millisecondsUsage:
result.set_speech_event_timeout(5000)
wait_for_user(enabled: Optional[bool] = None, timeout: Optional[int] = None, answer_first: bool = False) -> SwaigFunctionResult
🔗 ↑ TOCControl 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)
toggle_functions(function_toggles: List[Dict[str, Any]]) -> SwaigFunctionResult
🔗 ↑ TOCEnable or disable specific functions.
Parameters:
function_toggles
(List[Dict[str, Any]]): List of function toggle configurationsUsage:
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
🔗 ↑ TOCControl whether functions are enabled when timeout occurs.
Parameters:
enabled
(bool): Enable functions on timeout (default: True)Usage:
result.enable_functions_on_timeout(False) # Disable functions on timeout
enable_extensive_data(enabled: bool = True) -> SwaigFunctionResult
🔗 ↑ TOCEnable extensive data collection.
Parameters:
enabled
(bool): Enable extensive data (default: True)Usage:
result.enable_extensive_data(True)
update_settings(settings: Dict[str, Any]) -> SwaigFunctionResult
🔗 ↑ TOCUpdate various AI settings.
Parameters:
settings
(Dict[str, Any]): Settings to updateUsage:
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
🔗 ↑ TOCSwitch 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(text: str) -> SwaigFunctionResult
🔗 ↑ TOCSimulate user input for testing or automation.
Parameters:
text
(str): Text to simulate as user inputUsage:
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
🔗 ↑ TOCSend 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 regionUsage:
# 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
🔗 ↑ TOCStart 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 statusUsage:
# 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
🔗 ↑ TOCStop call recording.
Parameters:
control_id
(Optional[str]): Control ID of recording to stopUsage:
result.stop_record_call()
result.stop_record_call(control_id="customer_call_001")
join_room(name: str) -> SwaigFunctionResult
🔗 ↑ TOCJoin a SignalWire room.
Parameters:
name
(str): Room name to joinUsage:
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
🔗 ↑ TOCJoin 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
🔗 ↑ TOCProcess 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 promptsUsage:
# 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
🔗 ↑ TOCStart 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 URLUsage:
# 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
🔗 ↑ TOCStop call tapping.
Parameters:
control_id
(Optional[str]): Control ID of tap to stopUsage:
result.stop_tap()
result.stop_tap(control_id="quality_monitor_001")
execute_swml(swml_content, transfer: bool = False) -> SwaigFunctionResult
🔗 ↑ TOCExecute 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)
to_dict() -> Dict[str, Any]
🔗 ↑ TOCConvert 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"}]}
create_payment_prompt(for_situation: str, actions: List[Dict[str, str]], card_type: Optional[str] = None, error_type: Optional[str] = None) -> Dict[str, Any]
🔗 ↑ TOCCreate 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 promptsUsage:
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]
🔗 ↑ TOCCreate a payment action configuration.
Parameters:
action_type
(str): Action typephrase
(str): Action phraseUsage:
action = SwaigFunctionResult.create_payment_action("say", "Enter your card number")
create_payment_parameter(name: str, value: str) -> Dict[str, str]
🔗 ↑ TOCCreate a payment parameter configuration.
Parameters:
name
(str): Parameter namevalue
(str): Parameter valueUsage:
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 createUsage:
# Create a new DataMap tool
weather_map = DataMap('get_weather')
search_map = DataMap('search_docs')
purpose(description: str) -> DataMap
🔗 ↑ TOCSet the function description/purpose.
Parameters:
description
(str): Human-readable description of what this function doesUsage:
data_map = DataMap('get_weather').purpose('Get current weather information for any city')
description(description: str) -> DataMap
🔗 ↑ TOCAlias for purpose()
- set the function description.
Parameters:
description
(str): Function descriptionUsage:
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
🔗 ↑ TOCAdd 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 validationUsage:
# 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
🔗 ↑ TOCConfigure 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 presentVariable 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 metadataUsage:
# 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
🔗 ↑ TOCSet the JSON body for POST/PUT requests.
Parameters:
data
(Dict[str, Any]): JSON body data (supports ${variable}
substitution)Usage:
# Static body with parameter substitution
data_map.body({
'query': '${args.search_term}',
'limit': 5,
'filters': {
'category': '${args.category}',
'active': True
}
})
# Body with call-related data (NOT sensitive info)
data_map.body({
'customer_id': '${global_data.customer_id}',
'request_id': '${meta_data.call_id}',
'search': '${args.query}'
})
params(data: Dict[str, Any]) -> DataMap
🔗 ↑ TOCSet 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'))
)
output(result: SwaigFunctionResult) -> DataMap
🔗 ↑ TOCSet the response template for successful API calls.
Parameters:
result
(SwaigFunctionResult): Response template with variable substitutionVariable Substitution in Outputs:
${response.field}
: API response fields${response.nested.field}
: Nested response fields${response.array[0].field}
: Array element fields${args.parameter}
: Original function arguments${global_data.key}
: Call-wide data store (user info, call state)Usage:
# Simple response template
data_map.output(SwaigFunctionResult('Weather in ${args.location}: ${response.current.condition.text}, ${response.current.temp_f}°F'))
# Response with actions
data_map.output(
SwaigFunctionResult('Found ${response.total_results} results')
.update_global_data({'last_search': '${args.query}'})
.add_action('play', 'search_complete.mp3')
)
# Complex response with nested data
data_map.output(
SwaigFunctionResult('Order ${response.order.id} status: ${response.order.status}. Estimated delivery: ${response.order.delivery.estimated_date}')
)
fallback_output(result: SwaigFunctionResult) -> DataMap
🔗 ↑ TOCSet the response when all webhooks fail.
Parameters:
result
(SwaigFunctionResult): Fallback responseUsage:
data_map.fallback_output(
SwaigFunctionResult('Sorry, the service is temporarily unavailable. Please try again later.')
.add_action('play', 'service_unavailable.mp3')
)
foreach(foreach_config: Union[str, Dict[str, Any]]) -> DataMap
🔗 ↑ TOCProcess array responses by iterating over elements.
Parameters:
foreach_config
(Union[str, Dict]): Array path or configuration objectSimple 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 processedexpression(test_value: str, pattern: Union[str, Pattern], output: SwaigFunctionResult, nomatch_output: Optional[SwaigFunctionResult] = None) -> DataMap
🔗 ↑ TOCAdd 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 matchUsage:
# 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 groupserror_keys(keys: List[str]) -> DataMap
🔗 ↑ TOCSpecify response fields that indicate errors.
Parameters:
keys
(List[str]): List of field names that indicate API errorsUsage:
# 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
🔗 ↑ TOCSet global error keys for all webhooks in this DataMap.
Parameters:
keys
(List[str]): Global error field namesUsage:
data_map.global_error_keys(['error', 'message', 'code'])
webhook_expressions(expressions: List[Dict[str, Any]]) -> DataMap
🔗 ↑ TOCAdd expression-based webhook selection.
Parameters:
expressions
(List[Dict[str, Any]]): List of expression configurationsUsage:
# 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')
)
)
to_swaig_function() -> Dict[str, Any]
🔗 ↑ TOCConvert 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)
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
🔗 ↑ TOCCreate 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 namesUsage:
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
🔗 ↑ TOCCreate a pattern-based tool without API calls.
Parameters:
name
(str): Function namepatterns
(Dict[str, Tuple[str, SwaigFunctionResult]]): Pattern mappingsparameters
(Optional[Dict[str, Dict]]): Parameter definitionsUsage:
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 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.
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"])
add_context(name: str) -> Context
🔗 ↑ TOCCreate a new context in the workflow.
Parameters:
name
(str): Unique context nameReturns:
Usage:
# Create multiple contexts
greeting_context = contexts.add_context("greeting")
main_menu_context = contexts.add_context("main_menu")
support_context = contexts.add_context("support")
Represents a single context containing multiple sequential steps.
add_step(name: str) -> Step
🔗 ↑ TOCAdd a new step to this context.
Parameters:
name
(str): Unique step name within this contextReturns:
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
🔗 ↑ TOCSet which contexts can be navigated to from this context.
Parameters:
contexts
(List[str]): List of valid context namesUsage:
# Allow navigation to specific contexts
context.set_valid_contexts(["main_menu", "escalation", "end_call"])
# Allow navigation to any context
context.set_valid_contexts(["*"])
Represents a single step within a context with its own prompt and behavior.
set_text(text: str) -> Step
🔗 ↑ TOCSet the step's prompt as raw text.
Parameters:
text
(str): Complete prompt text for this stepUsage:
step.set_text("Please provide your account number so I can look up your information.")
add_section(title: str, body: str) -> Step
🔗 ↑ TOCAdd a structured prompt section (POM-style).
Parameters:
title
(str): Section titlebody
(str): Section contentUsage:
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
🔗 ↑ TOCAdd a section with bullet points.
Parameters:
title
(str): Section titlebullets
(List[str]): List of bullet pointsUsage:
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.
set_step_criteria(criteria: str) -> Step
🔗 ↑ TOCDefine when this step is considered complete.
Parameters:
criteria
(str): Description of completion criteriaUsage:
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
🔗 ↑ TOCControl which functions are available in this step.
Parameters:
functions
(Union[str, List[str]]): "none" to disable all functions, or list of function namesUsage:
# 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
🔗 ↑ TOCDefine which steps can be navigated to from this step.
Parameters:
steps
(List[str]): List of valid step names (include "next" for sequential flow)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(["*"])
# 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"])
create_simple_context(name: str = "default") -> Context
🔗 ↑ TOCCreate a simple single-context workflow.
Parameters:
name
(str): Context name (default: "default")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")
The State Management system provides persistent storage for conversation data across calls and sessions.
The base interface that all state managers must implement.
store(call_id: str, data: Dict[str, Any]) -> bool
🔗 ↑ TOCStore state data for a call.
Parameters:
call_id
(str): Unique identifier for the calldata
(Dict[str, Any]): State data to storeReturns:
retrieve(call_id: str) -> Optional[Dict[str, Any]]
🔗 ↑ TOCRetrieve state data for a call.
Parameters:
call_id
(str): Unique identifier for the callReturns:
update(call_id: str, data: Dict[str, Any]) -> bool
🔗 ↑ TOCUpdate existing state data (merges with existing).
Parameters:
call_id
(str): Unique identifier for the calldata
(Dict[str, Any]): Data to merge with existing stateReturns:
delete(call_id: str) -> bool
🔗 ↑ TOCDelete state data for a call.
Parameters:
call_id
(str): Unique identifier for the callReturns:
cleanup_expired() -> int
🔗 ↑ TOCClean up expired state data.
Returns:
exists(call_id: str) -> bool
🔗 ↑ TOCCheck if state exists for a call.
Parameters:
call_id
(str): Unique identifier for the callReturns:
File-based state manager implementation that stores state data in JSON files.
FileStateManager(
storage_dir: str = "./agent_state",
expiry_hours: int = 24,
auto_cleanup: bool = True
)
Parameters:
storage_dir
(str): Directory to store state files (default: "./agent_state")expiry_hours
(int): Hours after which state expires (default: 24)auto_cleanup
(bool): Automatically clean up expired files (default: True)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")
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
)
The Skills System provides modular, reusable capabilities that can be easily added to any agent.
datetime
Skill 🔗 ↑ TOCProvides current date and time information.
Parameters:
timezone
(Optional[str]): Timezone for date/time (default: system timezone)format
(Optional[str]): Custom date/time format stringUsage:
# 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 🔗 ↑ TOCSafe mathematical expression evaluation.
Parameters:
precision
(Optional[int]): Decimal precision for results (default: 2)max_expression_length
(Optional[int]): Maximum expression length (default: 100)Usage:
# Basic math skill
agent.add_skill("math")
# With custom precision
agent.add_skill("math", {"precision": 4})
web_search
Skill 🔗 ↑ TOCGoogle 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 foundUsage:
# 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 🔗 ↑ TOCSignalWire 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 tagsUsage:
# 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 🔗 ↑ TOCLocal document search with vector similarity and keyword search.
Parameters:
index_path
(str): Path to search index file (required)tool_name
(Optional[str]): Custom tool name (default: "search_documents")max_results
(Optional[int]): Maximum results to return (default: 5)similarity_threshold
(Optional[float]): Minimum similarity score (default: 0.1)Usage:
# Basic local search
agent.add_skill("native_vector_search", {
"index_path": "./knowledge.swsearch"
})
# With custom settings
agent.add_skill("native_vector_search", {
"index_path": "./docs.swsearch",
"tool_name": "search_docs",
"max_results": 10,
"similarity_threshold": 0.3
})
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:
signalwire_agents/skills/your_skill/
__init__.py
, skill.py
, and README.md
skill.py
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 propertiesfrom 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.
get_swml_document() -> Dict[str, Any]
🔗 ↑ TOCGenerate the complete SWML document for the service.
handle_request(request_data: Dict[str, Any]) -> Dict[str, Any]
🔗 ↑ TOCHandle incoming HTTP requests and generate appropriate responses.
Used in dynamic configuration callbacks to modify agent settings per-request.
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)
The SDK supports various environment variables for configuration:
SWML_BASIC_AUTH_USER
: Basic auth usernameSWML_BASIC_AUTH_PASSWORD
: Basic auth passwordSWML_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 SSLSWML_PROXY_URL_BASE
: Base URL for proxy serverGOOGLE_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 tokenimport 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
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.