signalwire.core¶
signalwire.core
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Core components for SignalWire AI Agents
__all__ = ['AgentBase', 'FunctionResult', 'SWAIGFunction', 'SWMLBuilder', 'SWMLService', 'SWMLVerbHandler', 'VerbHandlerRegistry']
module-attribute
¶
AgentBase
¶
Bases: AuthMixin, WebMixin, SWMLService, PromptMixin, SkillMixin, AIConfigMixin, ServerlessMixin, StateMixin, MCPServerMixin
Base class for all SignalWire AI Agents.
This class extends SWMLService and provides enhanced functionality for building agents including: - Prompt building and customization - SWML rendering - SWAIG function definition and execution - Web service for serving SWML and handling webhooks - Security and session management
Subclassing options: 1. Simple override of get_prompt() for raw text 2. Using prompt_* methods for structured prompts 3. Declarative PROMPT_SECTIONS class attribute
PROMPT_SECTIONS = None
class-attribute
instance-attribute
¶
log = logger.bind(agent=name)
instance-attribute
¶
agent_id = agent_id or str(uuid.uuid4())
instance-attribute
¶
pom
instance-attribute
¶
signing_key = signing_key or os.environ.get('SIGNALWIRE_SIGNING_KEY')
instance-attribute
¶
tool = self._tool_decorator
instance-attribute
¶
native_functions = native_functions or []
instance-attribute
¶
skill_manager = SkillManager(self)
instance-attribute
¶
__init__(name, route='/', host='0.0.0.0', port=None, basic_auth=None, use_pom=True, token_expiry_secs=3600, auto_answer=True, record_call=False, record_format='mp4', record_stereo=True, default_webhook_url=None, agent_id=None, native_functions=None, schema_path=None, suppress_logs=False, enable_post_prompt_override=False, check_for_input_override=False, config_file=None, schema_validation=True, signing_key=None, trust_proxy_for_signature=False)
¶
Initialize a new agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Agent name/identifier |
required |
route
|
str
|
HTTP route path for this agent |
'/'
|
host
|
str
|
Host to bind the web server to |
'0.0.0.0'
|
port
|
int | None
|
Port to bind the web server to |
None
|
basic_auth
|
tuple[str, str] | None
|
Optional (username, password) tuple for basic auth |
None
|
use_pom
|
bool
|
Whether to use POM for prompt building |
True
|
token_expiry_secs
|
int
|
Seconds until tokens expire |
3600
|
auto_answer
|
bool
|
Whether to automatically answer calls |
True
|
record_call
|
bool
|
Whether to record calls |
False
|
record_format
|
str
|
Recording format |
'mp4'
|
record_stereo
|
bool
|
Whether to record in stereo |
True
|
default_webhook_url
|
str | None
|
Optional default webhook URL for all SWAIG functions |
None
|
agent_id
|
str | None
|
Optional unique ID for this agent, generated if not provided |
None
|
native_functions
|
list[str] | None
|
Optional list of native functions to include in the SWAIG object |
None
|
schema_path
|
str | None
|
Optional path to the schema file |
None
|
suppress_logs
|
bool
|
Whether to suppress structured logs |
False
|
enable_post_prompt_override
|
bool
|
Whether to enable post-prompt override |
False
|
check_for_input_override
|
bool
|
Whether to enable check-for-input override |
False
|
config_file
|
str | None
|
Optional path to configuration file |
None
|
schema_validation
|
bool
|
Enable SWML schema validation. Default True. Can also be disabled via SWML_SKIP_SCHEMA_VALIDATION=1 env var. |
True
|
signing_key
|
str | None
|
Optional SignalWire Signing Key (from Dashboard → API Credentials). When set, webhook signature validation is enforced on POST /, /swaig, /post_prompt — unsigned or invalidly-signed requests get a 403. Falls back to the SIGNALWIRE_SIGNING_KEY env var if not passed. See porting-sdk/webhooks.md for the contract. |
None
|
trust_proxy_for_signature
|
bool
|
If True, honor X-Forwarded-Proto / X-Forwarded-Host when reconstructing the URL during signature validation. Default False — proxy headers are spoofable, so opt in only when you control the proxy chain. |
False
|
get_name()
¶
Get agent name
Returns:
| Type | Description |
|---|---|
str
|
Agent name |
get_full_url(include_auth=False)
¶
Get the full URL for this agent's endpoint
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_auth
|
bool
|
Whether to include authentication credentials in the URL |
False
|
Returns:
| Type | Description |
|---|---|
str
|
Full URL including host, port, and route (with auth if requested) |
on_summary(summary, raw_data=None)
¶
Called when a post-prompt summary is received
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
summary
|
dict[str, Any] | None
|
The summary object or None if no summary was found |
required |
raw_data
|
dict[str, Any] | None
|
The complete raw POST data from the request |
None
|
on_debug_event(handler)
¶
Register a handler for debug webhook events.
Use as a decorator to receive real-time debug events from the AI module during calls. 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 including call_id, label, and event-specific fields
The handler may be sync or async.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
Callable
|
Callback function with signature (event_type: str, data: dict) |
required |
Returns:
| Type | Description |
|---|---|
Callable
|
The handler function (unchanged), for use as a decorator |
Example
@agent.on_debug_event def handle(event_type, data): if event_type == "barge": print(f"Barge detected: {data.get('barge_elapsed_ms')}ms")
add_pre_answer_verb(verb_name, config)
¶
Add a verb to run before the call is answered.
Pre-answer verbs execute while the call is still ringing. Only certain verbs are safe to use before answering:
transfer, execute, return, label, goto, request, switch,
cond, if, eval, set, unset, hangup, send_sms, sleep, stop_record_call, stop_denoise, stop_tap
Verbs with auto_answer option (play, connect): Must include "auto_answer": False in config to prevent automatic answering.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "play", "sleep", "request") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If verb is not safe for pre-answer use |
Example
Play ringback tone before answering¶
agent.add_pre_answer_verb("play", { "urls": ["ring:us"], "auto_answer": False })
add_answer_verb(config=None)
¶
Configure the answer verb.
The answer verb connects the call. Use this method to customize answer behavior, such as setting max_duration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any] | None
|
Optional answer verb configuration (e.g., {"max_duration": 3600}) |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_post_answer_verb(verb_name, config)
¶
Add a verb to run after the call is answered but before the AI starts.
Post-answer verbs run after the call is connected. Common uses include welcome messages, legal disclaimers, and hold music.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "play", "sleep") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_post_ai_verb(verb_name, config)
¶
Add a verb to run after the AI conversation ends.
Post-AI verbs run when the AI completes its conversation. Common uses include clean disconnects, transfers, and logging.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "hangup", "transfer", "request") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
Log call completion and hang up¶
agent.add_post_ai_verb("request", { "url": "https://api.example.com/call-complete", "method": "POST" }) agent.add_post_ai_verb("hangup", {})
clear_pre_answer_verbs()
¶
clear_post_answer_verbs()
¶
clear_post_ai_verbs()
¶
enable_sip_routing(auto_map=True, path='/sip')
¶
Enable SIP-based routing for this agent
This allows the agent to automatically route SIP requests based on SIP usernames. When enabled, an endpoint at the specified path is automatically created that will handle SIP requests and deliver them to this agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
auto_map
|
bool
|
Whether to automatically map common SIP usernames to this agent (based on the agent name and route path) |
True
|
path
|
str
|
The path to register the SIP routing endpoint (default: "/sip") |
'/sip'
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
register_sip_username(sip_username)
¶
Register a SIP username that should be routed to this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sip_username
|
str
|
SIP username to register |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
auto_map_sip_usernames()
¶
Automatically register common SIP usernames based on this agent's name and route
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_web_hook_url(url)
¶
Override the default web_hook_url with a supplied URL string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The URL to use for SWAIG function webhooks |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_post_prompt_url(url)
¶
Override the default post_prompt_url with a supplied URL string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The URL to use for post-prompt summary delivery |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_swaig_query_params(params)
¶
Add query parameters that will be included in all SWAIG webhook URLs
This is particularly 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:
| Name | Type | Description | Default |
|---|---|---|---|
params
|
dict[str, str]
|
Dictionary of query parameters to add to SWAIG URLs |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
def dynamic_config(query_params, body_params, headers, agent): if query_params.get('tier') == 'premium': agent.add_skill('advanced_search') # Preserve the tier param so SWAIG callbacks work agent.add_swaig_query_params({'tier': 'premium'})
FunctionResult
¶
Wrapper around SWAIG function responses that handles proper formatting of response text and actions.
The result object has three main components: 1. response: Text the AI should say back to the user 2. action: List of structured actions to execute 3. post_process: Whether to let AI take another turn before executing actions
Post-processing behavior: - post_process=False (default): Execute actions immediately after AI response - post_process=True: Let AI respond to user one more time, then execute actions
This is useful for confirmation workflows like: "I'll transfer you to sales. Do you have any other questions first?" (AI can handle follow-up, then execute the transfer)
Example
return FunctionResult("Found your order")
With actions¶
return ( FunctionResult("I'll transfer you to support") .add_action("transfer", {"dest": "support"}) )
With simple action value¶
return ( FunctionResult("I'll confirm that") .add_action("confirm", True) )
With multiple actions¶
return ( FunctionResult("Processing your request") .add_actions([ {"set_global_data": {"key": "value"}}, {"play": {"url": "music.mp3"}} ]) )
With post-processing enabled¶
return ( FunctionResult("Let me transfer you to billing", post_process=True) .connect("+15551234567", final=True) )
Using the connect helper¶
return ( FunctionResult("I'll transfer you to our sales team now") .connect("sales@company.com", final=False, from_addr="+15559876543") )
response = response if response is not None else ''
instance-attribute
¶
action = []
instance-attribute
¶
post_process = post_process
instance-attribute
¶
__init__(response=None, post_process=False)
¶
Initialize a new SWAIG function result
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str | None
|
Optional natural language response to include |
None
|
post_process
|
bool
|
Whether to let AI take another turn before executing actions. Defaults to False (execute actions immediately after response). |
False
|
set_response(response)
¶
Set the natural language response text
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str
|
The text the AI should say |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
set_post_process(post_process)
¶
Set whether to enable post-processing for this result.
Post-processing allows the AI to take one more turn with the user before executing any actions. This is useful for confirmation workflows.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
post_process
|
bool
|
True to let AI respond once more before executing actions, False to execute actions immediately after the response. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
add_action(name, data)
¶
Add a structured action to the response
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name/type of the action (e.g., "play", "transfer") |
required |
data
|
Any
|
The data for the action - can be a string, boolean, object, or array |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
add_actions(actions)
¶
Add multiple structured actions to the response
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
actions
|
list[dict[str, Any]]
|
List of action objects to add to the response |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
connect(destination, final=True, from_addr=None)
¶
Add a connect action to transfer/connect the call to another destination.
This is a convenience method that abstracts the SWML connect verb, so users don't need to manually construct SWML documents.
Transfer behavior: - final=True: Permanent transfer - call exits the agent completely, SWML replaces the agent and call continues there - final=False: Temporary transfer - if far end hangs up, call returns to the agent to continue the conversation
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
destination
|
str
|
Where to connect the call (phone number, SIP address, etc.) |
required |
final
|
bool
|
Whether this is a permanent transfer (True) or temporary (False). Defaults to True for permanent transfers. |
True
|
from_addr
|
str | None
|
Optional caller ID override (phone number or SIP address). If None, uses the current call's from address. |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
swml_transfer(dest, ai_response, final=True)
¶
Add a SWML transfer action with AI response setup for when transfer completes.
This is a virtual helper that generates SWML to transfer the call to another destination and sets up an AI response for when the transfer completes and control returns to the agent.
For transfers, you typically want to enable post-processing so the AI speaks the response first before executing the transfer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dest
|
str
|
Destination URL for the transfer (SWML endpoint, SIP address, etc.) |
required |
ai_response
|
str
|
Message the AI should say when transfer completes and control returns |
required |
final
|
bool
|
Whether this is a permanent transfer (True) or temporary (False). Defaults to True for permanent transfers (same as connect method). |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
Example
Permanent transfer (default)¶
result = ( FunctionResult("I'm transferring you to support", post_process=True) .swml_transfer( "https://support.example.com/swml", "Goodbye!" # Won't be used since final=True by default ) )
Temporary transfer with return¶
result.swml_transfer( dest, "The support call is complete. How else can I help?", final=False )
update_global_data(data)
¶
Update global agent data variables.
This is a convenience method that abstracts the set_global_data action. Global data persists across the entire agent session and is available in prompt variables and can be accessed by all functions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of key-value pairs to set/update in global data |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
swml_user_event(event_data)
¶
Send a user event through SWML to update the client UI.
This is a convenience method for sending user events to connected clients, commonly used for real-time UI updates in interactive applications.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_data
|
dict[str, Any]
|
Dictionary containing the event type and any associated data Example: {"type": "cards_dealt", "player_hand": [...], "score": 21} |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
Example
result = ( FunctionResult("You have blackjack!") .swml_user_event({ "type": "cards_dealt", "player_hand": player_cards, "dealer_hand": dealer_cards, "player_score": 21 }) )
swml_change_step(step_name)
¶
Force the conversation into a specific step in the current context.
This is a workflow-level transition driven by your webhook, not by the
model. Webhook-triggered step changes bypass any valid_steps clamp
on the current step (the model's next_step tool is the constrained
path; this is the unconstrained one). Use it when something the user
said unambiguously requires moving the flow forward.
Communicating intent to the destination step¶
When the model arrives at the new step, the next thing it sees in
history is the tool result that contained this action — i.e., your
FunctionResult's response text. Put your "why" there. The model
reads it before the new step's injected instructions:
FunctionResult(
"Premium plan confirmed. Now collecting payment details."
).swml_change_step("collect_payment")
For structured data the destination step needs (account number,
plan tier, etc.), pair this with update_global_data() and reference
the values from the destination step's text via ${var} expansion:
(FunctionResult("Premium plan confirmed.")
.update_global_data({"plan": "premium", "billing_cycle": "annual"})
.swml_change_step("collect_payment"))
And in the destination step's text:
step.set_text(
"Collect payment for the ${plan} plan, billed ${billing_cycle}. "
"Confirm the amount with the user before proceeding."
)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
step_name
|
str
|
Name of the step to transition to. The step must exist in the current context. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining. |
Example
result = ( FunctionResult("Starting a new hand") .update_global_data({"chips": 1000}) .swml_change_step("betting") )
swml_change_context(context_name)
¶
Force the conversation into a different context.
Webhook-triggered context changes bypass any valid_contexts clamp.
Step state resets (cur_step = 0) on entry. The new context's first
step's instructions are injected on the next turn.
Communicating intent to the destination context¶
Same pattern as swml_change_step: the model reads your FunctionResult's
response text as the tool result before it sees the destination
context's first step. Put your reason there:
FunctionResult(
"I'm transferring you to billing because your question is "
"about an invoice charge."
).swml_change_context("billing")
For structured carry-over, use update_global_data() and reference the values via ${var} expansion in the destination context's step text.
Note that conversation history is preserved across context switches
unless the destination context is marked isolated=True. The model
retains everything the user said in the previous context — you do
not need to re-state it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context_name
|
str
|
Name of the context to switch to. Must exist in the agent's context map. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining. |
Example
result = ( FunctionResult("Transferring you to technical support.") .update_global_data({"original_issue": user_problem}) .swml_change_context("technical_support") )
execute_swml(swml_content, transfer=False)
¶
Execute SWML content with optional transfer behavior.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
swml_content
|
Can be: - String: Raw SWML JSON text - Dict: SWML data structure - SWML object: SignalWire SWML SDK object with .to_dict() method |
required | |
transfer
|
bool
|
Boolean - whether call should exit agent after execution |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
hangup()
¶
hold(timeout=300)
¶
Put the call on hold with optional timeout.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timeout
|
int
|
Timeout in seconds (max 900, default 300) |
300
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
wait_for_user(enabled=None, timeout=None, answer_first=False)
¶
Control how agent waits for user input.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool | None
|
Boolean to enable/disable waiting |
None
|
timeout
|
int | None
|
Number of seconds to wait |
None
|
answer_first
|
bool
|
Special "answer_first" mode |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop()
¶
say(text)
¶
Make the agent speak specific text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text for agent to speak |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
play_background_file(filename, wait=False)
¶
Play audio or video file in background.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Audio/video filename/path |
required |
wait
|
bool
|
Whether to suppress attention-getting behavior during playback |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop_background_file()
¶
Stop currently playing background file.
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
add_dynamic_hints(hints)
¶
Add dynamic speech recognition hints during a call.
Hints improve speech recognition accuracy for domain-specific terms. Each hint can be a simple string or a pronunciation pattern object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hints
|
list[str | dict[str, Any]]
|
List of hints, where each entry is either: - A string: Simple hint word/phrase (e.g., "Cabby") - A dict with pronunciation pattern: - "pattern": Regex pattern to match in recognized speech - "replace": Replacement text when pattern matches - "ignore_case": Optional bool for case-insensitive matching |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("I'll listen for that name") .add_dynamic_hints([ "Cabby", {"pattern": "cab bee", "replace": "Cabby", "ignore_case": True} ]) )
clear_dynamic_hints()
¶
Clear all dynamic speech recognition hints.
Removes all hints previously added via add_dynamic_hints().
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_end_of_speech_timeout(milliseconds)
¶
Adjust end of speech timeout - milliseconds of silence after speaking has been detected to finalize speech recognition.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
milliseconds
|
int
|
Timeout in milliseconds |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_speech_event_timeout(milliseconds)
¶
Adjust speech event timeout - milliseconds since last speech detection event to finalize recognition. Works better in noisy environments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
milliseconds
|
int
|
Timeout in milliseconds |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
remove_global_data(keys)
¶
Remove global agent data variables.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
Single key string or list of keys to remove |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_metadata(data)
¶
Set metadata scoped to current function's meta_data_token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of key-value pairs for metadata |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
remove_metadata(keys)
¶
Remove metadata from current function's meta_data_token scope.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
Single key string or list of keys to remove |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
toggle_functions(function_toggles)
¶
Enable/disable specific SWAIG functions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_toggles
|
list[dict[str, Any]]
|
List of dicts with 'function' and 'active' keys |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
enable_functions_on_timeout(enabled=True)
¶
Enable function calls on speaker timeout.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool
|
Whether to enable functions on timeout |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
enable_extensive_data(enabled=True)
¶
Send full data to LLM for this turn only, then use smaller replacement in subsequent turns.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool
|
Whether to send extensive data this turn only |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
replace_in_history(text=True)
¶
After first send, replace tool_call+result pair in conversation history.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str | bool
|
String = replace tool_call with an assistant message containing this text. True = remove the tool_call+result pair from history entirely. |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
update_settings(settings)
¶
Update agent runtime settings.
Supported settings: - frequency-penalty: Float (-2.0 to 2.0) - presence-penalty: Float (-2.0 to 2.0) - max-tokens: Integer (0 to 4096) - top-p: Float (0.0 to 1.0) - confidence: Float (0.0 to 1.0) - barge-confidence: Float (0.0 to 1.0) - temperature: Float (0.0 to 2.0, clamped to 1.5)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
settings
|
dict[str, Any]
|
Dictionary of settings to update |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
switch_context(system_prompt=None, user_prompt=None, consolidate=False, full_reset=False)
¶
Change agent context/prompt during conversation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
system_prompt
|
str | None
|
New system prompt |
None
|
user_prompt
|
str | None
|
User message to add |
None
|
consolidate
|
bool
|
Whether to summarize existing conversation |
False
|
full_reset
|
bool
|
Whether to do complete context reset |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
simulate_user_input(text)
¶
Queue simulated user input.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text to simulate as user input |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
send_sms(to_number, from_number, body=None, media=None, tags=None, region=None)
¶
Send a text message to a PSTN phone number using SWML.
This is a virtual helper that generates SWML to send SMS messages. Either body or media (or both) must be provided.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_number
|
str
|
Phone number in E.164 format to send to |
required |
from_number
|
str
|
Phone number in E.164 format to send from |
required |
body
|
str | None
|
Body text of the message (optional if media provided) |
None
|
media
|
list[str] | None
|
Array of URLs to send in the message (optional if body provided) |
None
|
tags
|
list[str] | None
|
Array of tags to associate with the message for UI searching |
None
|
region
|
str | None
|
Region to originate the message from |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither body nor media is provided |
pay(payment_connector_url, input_method='dtmf', status_url=None, payment_method='credit-card', timeout=5, max_attempts=1, security_code=True, postal_code=True, min_postal_code_length=0, token_type='reusable', charge_amount=None, currency='usd', language='en-US', voice='woman', description=None, valid_card_types='visa mastercard amex', parameters=None, prompts=None, ai_response='The payment status is ${pay_result}, do not mention anything else about collecting payment if successful.')
¶
Process payment using SWML pay action.
This is a virtual helper that generates SWML for payment processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_connector_url
|
str
|
URL to make payment requests to (required) |
required |
input_method
|
str
|
Method to collect payment details. The SWML pay verb
only accepts "dtmf" (schema |
'dtmf'
|
status_url
|
str | None
|
URL for status change notifications |
None
|
payment_method
|
str
|
Payment method ("credit-card" currently supported) |
'credit-card'
|
timeout
|
int
|
Seconds to wait for next digit (default: 5) |
5
|
max_attempts
|
int
|
Number of retry attempts (default: 1) |
1
|
security_code
|
bool
|
Whether to prompt for security code (default: True) |
True
|
postal_code
|
bool | str
|
Whether to prompt for postal code, or actual postcode |
True
|
min_postal_code_length
|
int
|
Minimum postal code digits (default: 0) |
0
|
token_type
|
str
|
Payment type ("one-time" or "reusable", default: "reusable") |
'reusable'
|
charge_amount
|
str | None
|
Amount to charge as decimal string |
None
|
currency
|
str
|
Currency code (default: "usd") |
'usd'
|
language
|
str
|
Language for prompts (default: "en-US") |
'en-US'
|
voice
|
str
|
TTS voice to use (default: "woman") |
'woman'
|
description
|
str | None
|
Custom payment description |
None
|
valid_card_types
|
str
|
Space-separated card types (default: "visa mastercard amex") |
'visa mastercard amex'
|
parameters
|
list[dict[str, str]] | None
|
Array of name/value pairs for payment connector |
None
|
prompts
|
list[dict[str, Any]] | None
|
Array of custom prompt configurations |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
record_call(control_id=None, stereo=False, format='wav', direction='both', terminators=None, beep=False, input_sensitivity=44.0, initial_timeout=None, end_silence_timeout=None, max_length=None, status_url=None)
¶
Start background call recording using SWML.
This is a virtual helper that generates SWML to start recording the call in the background. Unlike foreground recording, the script continues executing while recording happens in the background.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
Identifier for this recording (for use with stop_record_call) |
None
|
stereo
|
bool
|
Record in stereo (default: False) |
False
|
format
|
Literal['wav', 'mp3', 'mp4']
|
Recording format - "wav", "mp3", or "mp4" (default: "wav") |
'wav'
|
direction
|
Literal['speak', 'listen', 'both']
|
Audio direction - "speak", "listen", or "both" (default: "both") |
'both'
|
terminators
|
str | None
|
Digits that stop recording when pressed |
None
|
beep
|
bool
|
Play beep before recording (default: False) |
False
|
input_sensitivity
|
float
|
Input sensitivity for recording (default: 44.0) |
44.0
|
initial_timeout
|
float | None
|
Time in seconds to wait for speech start (for voicemail-style recording) |
None
|
end_silence_timeout
|
float | None
|
Time in seconds to wait in silence before ending (for voicemail-style recording) |
None
|
max_length
|
float | None
|
Maximum recording length in seconds |
None
|
status_url
|
str | None
|
URL to send recording status events to |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop_record_call(control_id=None)
¶
Stop an active background call recording using SWML.
This is a virtual helper that generates SWML to stop a recording that was started with record_call().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
Identifier for the recording to stop. If not provided, the most recent recording will be stopped. |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
join_room(name)
¶
Join a RELAY room using SWML.
This is a virtual helper that generates SWML to join a RELAY room, which enables multi-party communication and collaboration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the room to join (required) |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
sip_refer(to_uri)
¶
Send SIP REFER to a SIP call using SWML.
This is a virtual helper that generates SWML to send a SIP REFER message, which is used for call transfer in SIP environments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_uri
|
str
|
The SIP URI to send the REFER to (required) |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
join_conference(name, muted=False, beep='true', start_on_enter=True, end_on_exit=False, wait_url=None, max_participants=250, record='do-not-record', region=None, trim='trim-silence', coach=None, status_callback_event=None, status_callback=None, status_callback_method='POST', recording_status_callback=None, recording_status_callback_method='POST', recording_status_callback_event='completed', result=None)
¶
Join an ad-hoc audio conference with RELAY and CXML calls using SWML.
This is a virtual helper that generates SWML to join audio conferences with extensive configuration options for call management and recording.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of conference (required) |
required |
muted
|
bool
|
Whether to join muted (default: False) |
False
|
beep
|
str
|
Beep configuration - "true", "false", "onEnter", "onExit" (default: "true") |
'true'
|
start_on_enter
|
bool
|
Whether conference starts when this participant enters (default: True) |
True
|
end_on_exit
|
bool
|
Whether conference ends when this participant exits (default: False) |
False
|
wait_url
|
str | None
|
SWML URL for hold music (default: None for default hold music) |
None
|
max_participants
|
int
|
Maximum participants <= 250 (default: 250) |
250
|
record
|
str
|
Recording mode - "do-not-record", "record-from-start" (default: "do-not-record") |
'do-not-record'
|
region
|
str | None
|
Conference region (default: None) |
None
|
trim
|
str
|
Trim silence - "trim-silence", "do-not-trim" (default: "trim-silence") |
'trim-silence'
|
coach
|
str | None
|
SWML Call ID or CXML CallSid for coaching (default: None) |
None
|
status_callback_event
|
str | None
|
Events to report - "start end join leave mute hold modify speaker announcement" (default: None) |
None
|
status_callback
|
str | None
|
URL for status callbacks (default: None) |
None
|
status_callback_method
|
str
|
HTTP method - "GET", "POST" (default: "POST") |
'POST'
|
recording_status_callback
|
str | None
|
URL for recording status callbacks (default: None) |
None
|
recording_status_callback_method
|
str
|
HTTP method - "GET", "POST" (default: "POST") |
'POST'
|
recording_status_callback_event
|
str
|
Recording events - "in-progress completed absent" (default: "completed") |
'completed'
|
result
|
Any | None
|
Switch on return_value when object {} or cond when array [] (default: None) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If beep value is invalid or max_participants exceeds 250 |
tap(uri, control_id=None, direction='both', codec='PCMU', rtp_ptime=20, status_url=None)
¶
Start background call tap using SWML.
This is a virtual helper that generates SWML to start background call tapping. Media is streamed over Websocket or RTP to customer controlled URI.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
str
|
Destination of tap media stream (required) Formats: rtp://IP:port, ws://example.com, or wss://example.com |
required |
control_id
|
str | None
|
Identifier for this tap to use with stop_tap (optional) Default is generated and stored in tap_control_id variable |
None
|
direction
|
Literal['speak', 'hear', 'both']
|
Direction of audio to tap (default: "both") "speak" = what party says "hear" = what party hears "both" = what party hears and says |
'both'
|
codec
|
Literal['PCMU', 'PCMA']
|
Codec for tap media stream - "PCMU" or "PCMA" (default: "PCMU") |
'PCMU'
|
rtp_ptime
|
int
|
Packetization time in milliseconds for RTP (default: 20) |
20
|
status_url
|
str | None
|
URL for status change requests (optional) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If direction or codec values are invalid |
stop_tap(control_id=None)
¶
Stop an active tap stream using SWML.
This is a virtual helper that generates SWML to stop a tap stream that was started with tap().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
ID of the tap to stop (optional) If not set, the last tap started will be stopped |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
execute_rpc(method, params=None, call_id=None, node_id=None)
¶
Execute an RPC method on a call using SWML.
This is a generic helper for executing RPC commands. For common operations, consider using the specific helpers: rpc_dial(), rpc_ai_message(), rpc_ai_unhold().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
The RPC method to execute (e.g., "dial", "ai_message", "ai_unhold") |
required |
params
|
dict[str, Any] | None
|
Parameters for the RPC method (optional) |
None
|
call_id
|
str | None
|
Target call ID for the RPC (optional) |
None
|
node_id
|
str | None
|
Target node ID for the RPC (optional) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Executing RPC") .execute_rpc( method="ai_message", call_id="some-call-id", params={"role": "system", "message_text": "Hello"} ) )
rpc_dial(to_number, from_number, dest_swml, device_type='phone')
¶
Dial out to a number with a destination SWML URL using execute_rpc.
This is commonly used in call screening scenarios where you place a caller on hold and dial out to a human, with the dest_swml specifying what agent handles the outbound leg.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_number
|
str
|
Phone number to dial (E.164 format) |
required |
from_number
|
str
|
Caller ID to use (E.164 format) |
required |
dest_swml
|
str
|
URL to the SWML that handles the outbound call |
required |
device_type
|
str
|
Device type, typically "phone" (default: "phone") |
'phone'
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Please hold while I connect you.") .hold(timeout=120) .rpc_dial( to_number="+15551234567", from_number="+15559876543", dest_swml="https://example.com/call-agent?caller=John" ) )
rpc_ai_message(call_id, message_text, role='system')
¶
Inject a message into an AI agent on another call using execute_rpc.
This is useful for cross-call communication, such as notifying a held caller's AI agent about a status change or instructing it to relay a message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str
|
The call ID of the target call |
required |
message_text
|
str
|
The message text to inject into the AI conversation |
required |
role
|
str
|
The role for the message, typically "system" (default: "system") |
'system'
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("I'll let them know.") .rpc_ai_message( call_id=original_call_id, message_text="The person you were trying to reach is unavailable. Please take a message." ) )
rpc_ai_unhold(call_id)
¶
Unhold another call using execute_rpc.
This releases a call from hold state, typically used after injecting a message to the held caller's AI agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str
|
The call ID of the call to unhold |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Understood, I'll let them know.") .rpc_ai_message(call_id, "No one is available. Please take a message.") .rpc_ai_unhold(call_id) )
create_payment_prompt(for_situation, actions, card_type=None, error_type=None)
staticmethod
¶
Create a payment prompt structure for use with pay() method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
for_situation
|
str
|
Situation to use prompt for (e.g., "payment-card-number") |
required |
actions
|
list[dict[str, str]]
|
List of actions with 'type' and 'phrase' keys |
required |
card_type
|
str | None
|
Space-separated card types for this prompt |
None
|
error_type
|
str | None
|
Space-separated error types for this prompt |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary representing the prompt structure |
create_payment_action(action_type, phrase)
staticmethod
¶
Create a payment action for use in payment prompts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
action_type
|
str
|
"Say" for text-to-speech or "Play" for audio file |
required |
phrase
|
str
|
Sentence to say or URL to play |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dictionary representing the action |
create_payment_parameter(name, value)
staticmethod
¶
Create a payment parameter for use with pay() method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Parameter name |
required |
value
|
str
|
Parameter value |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dictionary representing the parameter |
to_dict()
¶
Convert to the JSON structure expected by SWAIG
The result must have at least one of: - 'response': Text to be spoken by the AI - 'action': Array of action objects
Optional: - 'post_process': Boolean controlling when actions execute
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary in SWAIG function response format |
SWAIGFunction
¶
Represents a SWAIG function — i.e., a tool the AI model can call.
A SWAIG function is exactly the same concept as a "tool" in native OpenAI / Anthropic tool calling. Each SWAIGFunction is rendered into the OpenAI tool schema format and sent to the model on every turn:
{"type": "function", "function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}}
The model parses description (and the description inside each
parameter) to decide WHEN to call the tool and HOW to fill in the
arguments. Both fields are prompt-engineered text — the quality of
your descriptions directly drives tool-selection accuracy.
Use AgentBase.define_tool() to create one (it builds the SWAIGFunction for you and registers it). Use the @agent.tool() decorator for the same thing in class-based agents.
name = name
instance-attribute
¶
handler = handler
instance-attribute
¶
description = description
instance-attribute
¶
parameters = parameters or {}
instance-attribute
¶
secure = secure
instance-attribute
¶
fillers = fillers
instance-attribute
¶
wait_file = wait_file
instance-attribute
¶
wait_file_loops = wait_file_loops
instance-attribute
¶
webhook_url = webhook_url
instance-attribute
¶
required = required or []
instance-attribute
¶
is_typed_handler = is_typed_handler
instance-attribute
¶
extra_swaig_fields = extra_swaig_fields
instance-attribute
¶
is_external = webhook_url is not None
instance-attribute
¶
__init__(name, handler, description, parameters=None, secure=False, fillers=None, wait_file=None, wait_file_loops=None, webhook_url=None, required=None, is_typed_handler=False, **extra_swaig_fields)
¶
Initialize a new SWAIG function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name. Becomes the |
required |
handler
|
Callable
|
Python callable invoked when the model calls this tool. |
required |
description
|
str
|
LLM-facing description. The model reads this on every turn to decide whether to call the tool. Be specific about WHEN to use it and what makes it the right choice over sibling tools — vague descriptions are the most common cause of "model has the right tool but doesn't call it" failures. |
required |
parameters
|
dict[str, dict] | None
|
JSON Schema for the arguments. Per-property
|
None
|
secure
|
bool
|
Whether this function requires SWAIG token validation. |
False
|
fillers
|
dict[str, list[str]] | None
|
Optional dictionary of filler phrases by language code (deprecated, use wait_file). |
None
|
wait_file
|
str | None
|
Optional URL to audio file to play while function executes. |
None
|
wait_file_loops
|
int | None
|
Optional number of times to loop the wait_file. |
None
|
webhook_url
|
str | None
|
Optional external webhook URL to use instead of local handling. |
None
|
required
|
list[str] | None
|
Optional list of required parameter names. |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters (auto-wrapped). |
False
|
**extra_swaig_fields
|
Additional SWAIG-only fields (meta_data_token, web_hook_auth_*, etc.) to include in the generated definition. |
{}
|
__call__(*args, **kwargs)
¶
Call the underlying handler function
execute(args, raw_data=None)
¶
Execute the function with the given arguments
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args
|
dict[str, Any]
|
Parsed arguments for the function |
required |
raw_data
|
dict[str, Any] | None
|
Optional raw request data |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Function result as a dictionary (from FunctionResult.to_dict()) |
validate_args(args)
¶
Validate the arguments against the parameter schema.
Uses jsonschema_rs if available for fast validation, falls back to jsonschema if available, otherwise skips validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args
|
dict[str, Any]
|
Arguments to validate |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
Tuple of (is_valid: bool, errors: list[str]) |
tuple
|
If no validation library is available, returns (True, []) |
to_swaig(base_url, token=None, call_id=None, include_auth=True)
¶
Convert this function to a SWAIG-compatible JSON object for SWML
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base_url
|
str
|
Base URL for the webhook |
required |
token
|
str | None
|
Optional auth token to include |
None
|
call_id
|
str | None
|
Optional call ID for session tracking |
None
|
include_auth
|
bool
|
Whether to include auth credentials in URL |
True
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary representation for the SWAIG array in SWML |
SWMLService
¶
Bases: ToolMixin
Base class for creating and serving SWML documents.
This class provides core functionality for: - Loading and validating SWML schema - Creating SWML documents - Setting up web endpoints for serving SWML - Managing authentication - Registering SWML functions
It serves as the foundation for more specialized services like AgentBase.
name = name
instance-attribute
¶
route = route.rstrip('/')
instance-attribute
¶
host = host
instance-attribute
¶
port = port if port is not None else int(os.environ.get('PORT', 3000))
instance-attribute
¶
log = logger.bind(service=name)
instance-attribute
¶
security = SecurityConfig(config_file=config_file, service_name=name)
instance-attribute
¶
ssl_enabled = self.security.ssl_enabled
instance-attribute
¶
domain = self.security.domain
instance-attribute
¶
ssl_cert_path = self.security.ssl_cert_path
instance-attribute
¶
ssl_key_path = self.security.ssl_key_path
instance-attribute
¶
schema_utils = SchemaUtils(schema_path, schema_validation=(self._schema_validation))
instance-attribute
¶
verb_registry = VerbHandlerRegistry()
instance-attribute
¶
full_validation_enabled
property
¶
Check if full JSON Schema validation is enabled.
Returns:
| Type | Description |
|---|---|
bool
|
True if schema validator is initialized |
__init__(name, route='/', host='0.0.0.0', port=None, basic_auth=None, schema_path=None, config_file=None, schema_validation=True)
¶
Initialize a new SWML service
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Service name/identifier |
required |
route
|
str
|
HTTP route path for this service |
'/'
|
host
|
str
|
Host to bind the web server to |
'0.0.0.0'
|
port
|
int | None
|
Port to bind the web server to |
None
|
basic_auth
|
tuple[str, str] | None
|
Optional (username, password) tuple for basic auth |
None
|
schema_path
|
str | None
|
Optional path to the schema file |
None
|
config_file
|
str | None
|
Optional path to configuration file |
None
|
schema_validation
|
bool
|
Enable schema validation. Default True. Can also be disabled via SWML_SKIP_SCHEMA_VALIDATION=1 env var. |
True
|
__getattr__(name)
¶
Dynamically generate and return SWML verb methods when accessed
This method is called when an attribute lookup fails through the normal mechanisms. It checks if the attribute name corresponds to a SWML verb defined in the schema, and if so, dynamically creates a method for that verb.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the attribute being accessed |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The dynamically created verb method if name is a valid SWML verb, |
Any
|
otherwise raises AttributeError |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If name is not a valid SWML verb |
reset_document()
¶
Reset the current document to an empty state
add_verb(verb_name, config)
¶
Add a verb to the main section of the current document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb to add |
required |
config
|
dict[str, Any] | int
|
Configuration for the verb or direct value for certain verbs (e.g., sleep) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the verb was added successfully, False otherwise |
add_section(section_name)
¶
Add a new section to the document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the section was added, False if it already exists |
add_verb_to_section(section_name, verb_name, config)
¶
Add a verb to a specific section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add to |
required |
verb_name
|
str
|
The name of the verb to add |
required |
config
|
dict[str, Any] | int
|
Configuration for the verb or direct value for certain verbs (e.g., sleep) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the verb was added successfully, False otherwise |
get_document()
¶
Get the current SWML document
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The current SWML document as a dictionary |
render_document()
¶
Render the current SWML document as a JSON string
Returns:
| Type | Description |
|---|---|
str
|
The current SWML document as a JSON string |
register_verb_handler(handler)
¶
Register a custom verb handler
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
SWMLVerbHandler
|
The verb handler to register |
required |
as_router()
¶
Create a FastAPI router for this service
Returns:
| Name | Type | Description |
|---|---|---|
APIRouter |
APIRouter
|
FastAPI router |
register_routing_callback(callback_fn, path='/sip')
¶
Register a callback function that will be called to determine routing based on POST data.
When a routing callback is registered, an endpoint at the specified path is automatically created that will handle requests. This endpoint will use the callback to determine if the request should be processed by this service or redirected.
The callback should take a request object and request body dictionary and return: - A route string if it should be routed to a different endpoint - None if normal processing should continue
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback_fn
|
Callable[[Request, dict[str, Any]], str | None]
|
The callback function to register |
required |
path
|
str
|
The path where this callback should be registered (default: "/sip") |
'/sip'
|
extract_sip_username(request_body)
staticmethod
¶
Extract SIP username from request body
This extracts the username portion of a SIP URI from the 'to' field in the call data of a request body.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_body
|
dict[str, Any]
|
The parsed JSON body of the request |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The extracted SIP username, or None if not found |
on_request(request_data=None, callback_path=None)
¶
Called when SWML is requested, with request data when available
Subclasses can override this to inspect or modify SWML based on the request
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
Optional dict to modify/augment the SWML document |
serve(host=None, port=None, ssl_cert=None, ssl_key=None, ssl_enabled=None, domain=None)
¶
Start a web server for this service
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str | None
|
Host to bind to (defaults to self.host) |
None
|
port
|
int | None
|
Port to bind to (defaults to self.port) |
None
|
ssl_cert
|
str | None
|
Path to SSL certificate file |
None
|
ssl_key
|
str | None
|
Path to SSL key file |
None
|
ssl_enabled
|
bool | None
|
Whether to enable SSL |
None
|
domain
|
str | None
|
Domain name for SSL certificate |
None
|
stop()
¶
Stop the web server
get_basic_auth_credentials(include_source=False)
¶
Get the basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_source
|
bool
|
Whether to include the source of the credentials |
False
|
Returns:
| Type | Description |
|---|---|
tuple[str, str] | tuple[str, str, str]
|
(username, password) tuple or (username, password, source) tuple if include_source is True |
manual_set_proxy_url(proxy_url)
¶
Manually set the proxy URL base for webhook callbacks
This can be called at runtime to set or update the proxy URL
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
proxy_url
|
str
|
The base URL to use for webhooks (e.g., https://example.ngrok.io) |
required |
SWMLVerbHandler
¶
Bases: ABC
Base interface for SWML verb handlers
This abstract class defines the interface that all SWML verb handlers must implement. Verb handlers provide specialized logic for complex SWML verbs that cannot be handled generically.
get_verb_name()
abstractmethod
¶
Get the name of the verb this handler handles
Returns:
| Type | Description |
|---|---|
str
|
The verb name as a string |
validate_config(config)
abstractmethod
¶
Validate the configuration for this verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
The configuration dictionary for this verb |
required |
Returns:
| Type | Description |
|---|---|
tuple[bool, list[str]]
|
(is_valid, error_messages) tuple |
build_config(**kwargs)
abstractmethod
¶
Build a configuration for this verb from the provided arguments
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Keyword arguments specific to this verb |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Configuration dictionary |
VerbHandlerRegistry
¶
Registry for SWML verb handlers
This class maintains a registry of handlers for special SWML verbs and provides methods for accessing and using them.
__init__()
¶
Initialize the registry with default handlers
register_handler(handler)
¶
Register a new verb handler
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
SWMLVerbHandler
|
The handler to register |
required |
get_handler(verb_name)
¶
Get the handler for a specific verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb |
required |
Returns:
| Type | Description |
|---|---|
SWMLVerbHandler | None
|
The handler if found, None otherwise |
has_handler(verb_name)
¶
Check if a handler exists for a specific verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if a handler exists, False otherwise |
SWMLBuilder
¶
Fluent builder for SWML documents
This class provides a fluent interface for building SWML documents by chaining method calls. It delegates to an underlying SWMLService instance for the actual document creation.
service = service
instance-attribute
¶
__init__(service)
¶
Initialize with a SWMLService instance
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service
|
SWMLService
|
The SWMLService to delegate to |
required |
answer(max_duration=None, codecs=None)
¶
Add an 'answer' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_duration
|
int | None
|
Maximum duration in seconds |
None
|
codecs
|
str | None
|
Comma-separated list of codecs |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
hangup(reason=None)
¶
Add a 'hangup' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reason
|
str | None
|
Optional reason for hangup |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
ai(prompt_text=None, prompt_pom=None, post_prompt=None, post_prompt_url=None, swaig=None, **kwargs)
¶
Add an 'ai' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt_text
|
str | None
|
Text prompt for the AI (mutually exclusive with prompt_pom) |
None
|
prompt_pom
|
list[dict[str, Any]] | None
|
POM structure for the AI prompt (mutually exclusive with prompt_text) |
None
|
post_prompt
|
str | None
|
Optional post-prompt text |
None
|
post_prompt_url
|
str | None
|
Optional URL for post-prompt processing |
None
|
swaig
|
dict[str, Any] | None
|
Optional SWAIG configuration |
None
|
**kwargs
|
Additional AI parameters |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
play(url=None, urls=None, volume=None, say_voice=None, say_language=None, say_gender=None, auto_answer=None)
¶
Add a 'play' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str | None
|
Single URL to play (mutually exclusive with urls) |
None
|
urls
|
list[str] | None
|
List of URLs to play (mutually exclusive with url) |
None
|
volume
|
float | None
|
Volume level (-40 to 40) |
None
|
say_voice
|
str | None
|
Voice for text-to-speech |
None
|
say_language
|
str | None
|
Language for text-to-speech |
None
|
say_gender
|
str | None
|
Gender for text-to-speech |
None
|
auto_answer
|
bool | None
|
Whether to auto-answer the call |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
say(text, voice=None, language=None, gender=None, volume=None)
¶
Add a 'play' verb with say: prefix for text-to-speech
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text to speak |
required |
voice
|
str | None
|
Voice for text-to-speech |
None
|
language
|
str | None
|
Language for text-to-speech |
None
|
gender
|
str | None
|
Gender for text-to-speech |
None
|
volume
|
float | None
|
Volume level (-40 to 40) |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
add_section(section_name)
¶
Add a new section to the document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add |
required |
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
build()
¶
Build and return the SWML document
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The complete SWML document as a dictionary |
render()
¶
Build and render the SWML document as a JSON string
Returns:
| Type | Description |
|---|---|
str
|
The complete SWML document as a JSON string |
reset()
¶
Reset the document to an empty state
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
__getattr__(name)
¶
Dynamically generate and return SWML verb methods when accessed
This method is called when an attribute lookup fails through the normal mechanisms. It checks if the attribute name corresponds to a SWML verb defined in the schema, and if so, dynamically creates a method for that verb.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the attribute being accessed |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The dynamically created verb method if name is a valid SWML verb, |
Any
|
otherwise raises AttributeError |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If name is not a valid SWML verb |
agent
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
__all__ = []
module-attribute
¶
config
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
__all__ = []
module-attribute
¶
deployment
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
handlers
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
prompt
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Prompt management modules.
__all__ = ['PromptManager']
module-attribute
¶
PromptManager
¶
Manages prompt building and configuration.
agent = agent
instance-attribute
¶
__init__(agent)
¶
Initialize PromptManager with reference to parent agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
Parent AgentBase instance |
required |
define_contexts(contexts)
¶
Define contexts for the agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
dict[str, Any] | Any
|
Context configuration (dict or ContextBuilder) |
required |
set_prompt_text(text)
¶
Set the agent's prompt as raw text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Prompt text |
required |
set_post_prompt(text)
¶
Set the post-prompt text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Post-prompt text |
required |
set_prompt_pom(pom)
¶
Set the prompt as a POM dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pom
|
list[dict[str, Any]]
|
POM dictionary structure |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If use_pom is False |
prompt_add_section(title, body='', bullets=None, numbered=False, numbered_bullets=False, subsections=None)
¶
Add a section to the prompt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Optional section body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
numbered
|
bool
|
Whether this section should be numbered |
False
|
numbered_bullets
|
bool
|
Whether bullets should be numbered |
False
|
subsections
|
list[dict[str, Any]] | None
|
Optional list of subsection objects |
None
|
prompt_add_to_section(title, body=None, bullet=None, bullets=None)
¶
Add content to an existing section (creating it if needed).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str | None
|
Optional text to append to section body |
None
|
bullet
|
str | None
|
Optional single bullet point to add |
None
|
bullets
|
list[str] | None
|
Optional list of bullet points to add |
None
|
prompt_add_subsection(parent_title, title, body='', bullets=None)
¶
Add a subsection to an existing section (creating parent if needed).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent_title
|
str
|
Parent section title |
required |
title
|
str
|
Subsection title |
required |
body
|
str
|
Optional subsection body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
prompt_has_section(title)
¶
Check if a section exists in the prompt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if section exists, False otherwise |
get_prompt()
¶
Get the prompt configuration.
Returns:
| Type | Description |
|---|---|
str | list[dict[str, Any]] | None
|
Prompt text or sections or None |
get_raw_prompt()
¶
Get the raw prompt text if set.
Returns:
| Type | Description |
|---|---|
str | None
|
Raw prompt text or None |
get_post_prompt()
¶
Get the post-prompt text.
Returns:
| Type | Description |
|---|---|
str | None
|
Post-prompt text or None |
get_contexts()
¶
Get the contexts configuration.
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
Contexts dict or None |
manager
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Prompt management functionality for AgentBase.
logger = get_logger(__name__)
module-attribute
¶
PromptManager
¶
Manages prompt building and configuration.
agent = agent
instance-attribute
¶__init__(agent)
¶Initialize PromptManager with reference to parent agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
Parent AgentBase instance |
required |
define_contexts(contexts)
¶Define contexts for the agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
dict[str, Any] | Any
|
Context configuration (dict or ContextBuilder) |
required |
set_prompt_text(text)
¶Set the agent's prompt as raw text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Prompt text |
required |
set_post_prompt(text)
¶Set the post-prompt text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Post-prompt text |
required |
set_prompt_pom(pom)
¶Set the prompt as a POM dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pom
|
list[dict[str, Any]]
|
POM dictionary structure |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If use_pom is False |
prompt_add_section(title, body='', bullets=None, numbered=False, numbered_bullets=False, subsections=None)
¶Add a section to the prompt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Optional section body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
numbered
|
bool
|
Whether this section should be numbered |
False
|
numbered_bullets
|
bool
|
Whether bullets should be numbered |
False
|
subsections
|
list[dict[str, Any]] | None
|
Optional list of subsection objects |
None
|
prompt_add_to_section(title, body=None, bullet=None, bullets=None)
¶Add content to an existing section (creating it if needed).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str | None
|
Optional text to append to section body |
None
|
bullet
|
str | None
|
Optional single bullet point to add |
None
|
bullets
|
list[str] | None
|
Optional list of bullet points to add |
None
|
prompt_add_subsection(parent_title, title, body='', bullets=None)
¶Add a subsection to an existing section (creating parent if needed).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent_title
|
str
|
Parent section title |
required |
title
|
str
|
Subsection title |
required |
body
|
str
|
Optional subsection body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
prompt_has_section(title)
¶Check if a section exists in the prompt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if section exists, False otherwise |
get_prompt()
¶Get the prompt configuration.
Returns:
| Type | Description |
|---|---|
str | list[dict[str, Any]] | None
|
Prompt text or sections or None |
get_raw_prompt()
¶Get the raw prompt text if set.
Returns:
| Type | Description |
|---|---|
str | None
|
Raw prompt text or None |
get_post_prompt()
¶Get the post-prompt text.
Returns:
| Type | Description |
|---|---|
str | None
|
Post-prompt text or None |
get_contexts()
¶Get the contexts configuration.
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
Contexts dict or None |
routing
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
security
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
swml
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
tools
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Tool management modules.
__all__ = ['ToolDecorator', 'ToolRegistry', 'create_typed_handler_wrapper', 'infer_schema']
module-attribute
¶
ToolRegistry
¶
Manages SWAIG function registration.
agent = agent
instance-attribute
¶
__init__(agent)
¶
Initialize ToolRegistry with reference to parent agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
Parent AgentBase instance |
required |
define_tool(name, description, parameters, handler, secure=True, fillers=None, wait_file=None, wait_file_loops=None, webhook_url=None, required=None, is_typed_handler=False, **swaig_fields)
¶
Define a SWAIG function that the AI can call.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name (must be unique) |
required |
description
|
str
|
Function description for the AI |
required |
parameters
|
dict[str, Any]
|
JSON Schema of parameters |
required |
handler
|
Callable
|
Function to call when invoked |
required |
secure
|
bool
|
Whether to require token validation |
True
|
fillers
|
dict[str, list[str]] | None
|
Optional dict mapping language codes to arrays of filler phrases (deprecated) |
None
|
wait_file
|
str | None
|
Optional URL to audio file to play while function executes |
None
|
wait_file_loops
|
int | None
|
Optional number of times to loop the wait_file |
None
|
webhook_url
|
str | None
|
Optional external webhook URL to use instead of local handling |
None
|
required
|
list[str] | None
|
Optional list of required parameter names |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters |
False
|
**swaig_fields
|
Additional SWAIG fields to include in function definition |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If tool name already exists |
register_swaig_function(function_dict)
¶
Register a raw SWAIG function dictionary (e.g., from DataMap.to_swaig_function()).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_dict
|
dict[str, Any]
|
Complete SWAIG function definition dictionary |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If function name missing or already exists |
register_class_decorated_tools()
¶
Register tools defined with @AgentBase.tool class decorator.
This method scans the class for methods decorated with @AgentBase.tool and registers them automatically. If parameters are not explicitly provided and the function has type-hinted parameters, the schema is inferred from the type hints.
get_function(name)
¶
Get a registered function by name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
SWAIGFunction | dict[str, Any] | None
|
SWAIGFunction instance or raw function dict, or None if not found |
get_all_functions()
¶
Get all registered functions.
Returns:
| Type | Description |
|---|---|
dict[str, SWAIGFunction | dict[str, Any]]
|
Dictionary of function name to function object/dict |
has_function(name)
¶
Check if a function is registered.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if function exists, False otherwise |
remove_function(name)
¶
Remove a registered function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if removed, False if not found |
ToolDecorator
¶
Handles tool decoration logic.
create_instance_decorator(registry)
staticmethod
¶
Create instance tool decorator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry
|
ToolRegistry instance to register with |
required |
Returns:
| Type | Description |
|---|---|
|
Decorator function |
create_class_decorator()
classmethod
¶
Create class tool decorator.
Returns:
| Type | Description |
|---|---|
|
Decorator function |
infer_schema(func)
¶
Inspect a function's signature and type hints to infer a JSON Schema for SWAIG tool parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
The function to inspect. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, dict]
|
A tuple of (parameters, required, description, is_typed, has_raw_data): |
list[str]
|
|
str | None
|
|
bool
|
|
bool
|
|
tuple[dict[str, dict], list[str], str | None, bool, bool]
|
|
create_typed_handler_wrapper(func, has_raw_data)
¶
Wrap a typed handler function so it can be called with the standard SWAIG calling convention (args_dict, raw_data).
The wrapper unpacks the args dict into keyword arguments for the original function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
The original typed handler function. |
required | |
has_raw_data
|
bool
|
If True, pass raw_data as a keyword argument. |
required |
Returns:
| Type | Description |
|---|---|
|
A wrapper function with signature (args, raw_data). |
decorator
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Tool decorator functionality.
logger = get_logger(__name__)
module-attribute
¶
ToolDecorator
¶
Handles tool decoration logic.
create_instance_decorator(registry)
staticmethod
¶Create instance tool decorator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry
|
ToolRegistry instance to register with |
required |
Returns:
| Type | Description |
|---|---|
|
Decorator function |
create_class_decorator()
classmethod
¶Create class tool decorator.
Returns:
| Type | Description |
|---|---|
|
Decorator function |
registry
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Tool registration and management.
logger = get_logger(__name__)
module-attribute
¶
ToolRegistry
¶
Manages SWAIG function registration.
agent = agent
instance-attribute
¶__init__(agent)
¶Initialize ToolRegistry with reference to parent agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
Parent AgentBase instance |
required |
define_tool(name, description, parameters, handler, secure=True, fillers=None, wait_file=None, wait_file_loops=None, webhook_url=None, required=None, is_typed_handler=False, **swaig_fields)
¶Define a SWAIG function that the AI can call.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name (must be unique) |
required |
description
|
str
|
Function description for the AI |
required |
parameters
|
dict[str, Any]
|
JSON Schema of parameters |
required |
handler
|
Callable
|
Function to call when invoked |
required |
secure
|
bool
|
Whether to require token validation |
True
|
fillers
|
dict[str, list[str]] | None
|
Optional dict mapping language codes to arrays of filler phrases (deprecated) |
None
|
wait_file
|
str | None
|
Optional URL to audio file to play while function executes |
None
|
wait_file_loops
|
int | None
|
Optional number of times to loop the wait_file |
None
|
webhook_url
|
str | None
|
Optional external webhook URL to use instead of local handling |
None
|
required
|
list[str] | None
|
Optional list of required parameter names |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters |
False
|
**swaig_fields
|
Additional SWAIG fields to include in function definition |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If tool name already exists |
register_swaig_function(function_dict)
¶Register a raw SWAIG function dictionary (e.g., from DataMap.to_swaig_function()).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_dict
|
dict[str, Any]
|
Complete SWAIG function definition dictionary |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If function name missing or already exists |
register_class_decorated_tools()
¶Register tools defined with @AgentBase.tool class decorator.
This method scans the class for methods decorated with @AgentBase.tool and registers them automatically. If parameters are not explicitly provided and the function has type-hinted parameters, the schema is inferred from the type hints.
get_function(name)
¶Get a registered function by name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
SWAIGFunction | dict[str, Any] | None
|
SWAIGFunction instance or raw function dict, or None if not found |
get_all_functions()
¶Get all registered functions.
Returns:
| Type | Description |
|---|---|
dict[str, SWAIGFunction | dict[str, Any]]
|
Dictionary of function name to function object/dict |
has_function(name)
¶Check if a function is registered.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if function exists, False otherwise |
remove_function(name)
¶Remove a registered function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if removed, False if not found |
type_inference
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Type-hint-based schema inference for SWAIG tool functions.
infer_schema(func)
¶
Inspect a function's signature and type hints to infer a JSON Schema for SWAIG tool parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
The function to inspect. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, dict]
|
A tuple of (parameters, required, description, is_typed, has_raw_data): |
list[str]
|
|
str | None
|
|
bool
|
|
bool
|
|
tuple[dict[str, dict], list[str], str | None, bool, bool]
|
|
create_typed_handler_wrapper(func, has_raw_data)
¶
Wrap a typed handler function so it can be called with the standard SWAIG calling convention (args_dict, raw_data).
The wrapper unpacks the args dict into keyword arguments for the original function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
The original typed handler function. |
required | |
has_raw_data
|
bool
|
If True, pass raw_data as a keyword argument. |
required |
Returns:
| Type | Description |
|---|---|
|
A wrapper function with signature (args, raw_data). |
agent_base
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Base class for all SignalWire AI Agents
logger = get_logger('agent_base')
module-attribute
¶
AgentBase
¶
Bases: AuthMixin, WebMixin, SWMLService, PromptMixin, SkillMixin, AIConfigMixin, ServerlessMixin, StateMixin, MCPServerMixin
Base class for all SignalWire AI Agents.
This class extends SWMLService and provides enhanced functionality for building agents including: - Prompt building and customization - SWML rendering - SWAIG function definition and execution - Web service for serving SWML and handling webhooks - Security and session management
Subclassing options: 1. Simple override of get_prompt() for raw text 2. Using prompt_* methods for structured prompts 3. Declarative PROMPT_SECTIONS class attribute
PROMPT_SECTIONS = None
class-attribute
instance-attribute
¶
log = logger.bind(agent=name)
instance-attribute
¶
agent_id = agent_id or str(uuid.uuid4())
instance-attribute
¶
pom
instance-attribute
¶
signing_key = signing_key or os.environ.get('SIGNALWIRE_SIGNING_KEY')
instance-attribute
¶
tool = self._tool_decorator
instance-attribute
¶
native_functions = native_functions or []
instance-attribute
¶
skill_manager = SkillManager(self)
instance-attribute
¶
__init__(name, route='/', host='0.0.0.0', port=None, basic_auth=None, use_pom=True, token_expiry_secs=3600, auto_answer=True, record_call=False, record_format='mp4', record_stereo=True, default_webhook_url=None, agent_id=None, native_functions=None, schema_path=None, suppress_logs=False, enable_post_prompt_override=False, check_for_input_override=False, config_file=None, schema_validation=True, signing_key=None, trust_proxy_for_signature=False)
¶
Initialize a new agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Agent name/identifier |
required |
route
|
str
|
HTTP route path for this agent |
'/'
|
host
|
str
|
Host to bind the web server to |
'0.0.0.0'
|
port
|
int | None
|
Port to bind the web server to |
None
|
basic_auth
|
tuple[str, str] | None
|
Optional (username, password) tuple for basic auth |
None
|
use_pom
|
bool
|
Whether to use POM for prompt building |
True
|
token_expiry_secs
|
int
|
Seconds until tokens expire |
3600
|
auto_answer
|
bool
|
Whether to automatically answer calls |
True
|
record_call
|
bool
|
Whether to record calls |
False
|
record_format
|
str
|
Recording format |
'mp4'
|
record_stereo
|
bool
|
Whether to record in stereo |
True
|
default_webhook_url
|
str | None
|
Optional default webhook URL for all SWAIG functions |
None
|
agent_id
|
str | None
|
Optional unique ID for this agent, generated if not provided |
None
|
native_functions
|
list[str] | None
|
Optional list of native functions to include in the SWAIG object |
None
|
schema_path
|
str | None
|
Optional path to the schema file |
None
|
suppress_logs
|
bool
|
Whether to suppress structured logs |
False
|
enable_post_prompt_override
|
bool
|
Whether to enable post-prompt override |
False
|
check_for_input_override
|
bool
|
Whether to enable check-for-input override |
False
|
config_file
|
str | None
|
Optional path to configuration file |
None
|
schema_validation
|
bool
|
Enable SWML schema validation. Default True. Can also be disabled via SWML_SKIP_SCHEMA_VALIDATION=1 env var. |
True
|
signing_key
|
str | None
|
Optional SignalWire Signing Key (from Dashboard → API Credentials). When set, webhook signature validation is enforced on POST /, /swaig, /post_prompt — unsigned or invalidly-signed requests get a 403. Falls back to the SIGNALWIRE_SIGNING_KEY env var if not passed. See porting-sdk/webhooks.md for the contract. |
None
|
trust_proxy_for_signature
|
bool
|
If True, honor X-Forwarded-Proto / X-Forwarded-Host when reconstructing the URL during signature validation. Default False — proxy headers are spoofable, so opt in only when you control the proxy chain. |
False
|
get_name()
¶
Get agent name
Returns:
| Type | Description |
|---|---|
str
|
Agent name |
get_full_url(include_auth=False)
¶
Get the full URL for this agent's endpoint
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_auth
|
bool
|
Whether to include authentication credentials in the URL |
False
|
Returns:
| Type | Description |
|---|---|
str
|
Full URL including host, port, and route (with auth if requested) |
on_summary(summary, raw_data=None)
¶
Called when a post-prompt summary is received
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
summary
|
dict[str, Any] | None
|
The summary object or None if no summary was found |
required |
raw_data
|
dict[str, Any] | None
|
The complete raw POST data from the request |
None
|
on_debug_event(handler)
¶
Register a handler for debug webhook events.
Use as a decorator to receive real-time debug events from the AI module during calls. 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 including call_id, label, and event-specific fields
The handler may be sync or async.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
Callable
|
Callback function with signature (event_type: str, data: dict) |
required |
Returns:
| Type | Description |
|---|---|
Callable
|
The handler function (unchanged), for use as a decorator |
Example
@agent.on_debug_event def handle(event_type, data): if event_type == "barge": print(f"Barge detected: {data.get('barge_elapsed_ms')}ms")
add_pre_answer_verb(verb_name, config)
¶
Add a verb to run before the call is answered.
Pre-answer verbs execute while the call is still ringing. Only certain verbs are safe to use before answering:
transfer, execute, return, label, goto, request, switch,
cond, if, eval, set, unset, hangup, send_sms, sleep, stop_record_call, stop_denoise, stop_tap
Verbs with auto_answer option (play, connect): Must include "auto_answer": False in config to prevent automatic answering.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "play", "sleep", "request") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If verb is not safe for pre-answer use |
Example
Play ringback tone before answering¶
agent.add_pre_answer_verb("play", { "urls": ["ring:us"], "auto_answer": False })
add_answer_verb(config=None)
¶
Configure the answer verb.
The answer verb connects the call. Use this method to customize answer behavior, such as setting max_duration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any] | None
|
Optional answer verb configuration (e.g., {"max_duration": 3600}) |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_post_answer_verb(verb_name, config)
¶
Add a verb to run after the call is answered but before the AI starts.
Post-answer verbs run after the call is connected. Common uses include welcome messages, legal disclaimers, and hold music.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "play", "sleep") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_post_ai_verb(verb_name, config)
¶
Add a verb to run after the AI conversation ends.
Post-AI verbs run when the AI completes its conversation. Common uses include clean disconnects, transfers, and logging.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The SWML verb name (e.g., "hangup", "transfer", "request") |
required |
config
|
dict[str, Any]
|
Verb configuration dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
Log call completion and hang up¶
agent.add_post_ai_verb("request", { "url": "https://api.example.com/call-complete", "method": "POST" }) agent.add_post_ai_verb("hangup", {})
clear_pre_answer_verbs()
¶
clear_post_answer_verbs()
¶
clear_post_ai_verbs()
¶
enable_sip_routing(auto_map=True, path='/sip')
¶
Enable SIP-based routing for this agent
This allows the agent to automatically route SIP requests based on SIP usernames. When enabled, an endpoint at the specified path is automatically created that will handle SIP requests and deliver them to this agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
auto_map
|
bool
|
Whether to automatically map common SIP usernames to this agent (based on the agent name and route path) |
True
|
path
|
str
|
The path to register the SIP routing endpoint (default: "/sip") |
'/sip'
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
register_sip_username(sip_username)
¶
Register a SIP username that should be routed to this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sip_username
|
str
|
SIP username to register |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
auto_map_sip_usernames()
¶
Automatically register common SIP usernames based on this agent's name and route
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_web_hook_url(url)
¶
Override the default web_hook_url with a supplied URL string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The URL to use for SWAIG function webhooks |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_post_prompt_url(url)
¶
Override the default post_prompt_url with a supplied URL string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The URL to use for post-prompt summary delivery |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_swaig_query_params(params)
¶
Add query parameters that will be included in all SWAIG webhook URLs
This is particularly 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:
| Name | Type | Description | Default |
|---|---|---|---|
params
|
dict[str, str]
|
Dictionary of query parameters to add to SWAIG URLs |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
def dynamic_config(query_params, body_params, headers, agent): if query_params.get('tier') == 'premium': agent.add_skill('advanced_search') # Preserve the tier param so SWAIG callbacks work agent.add_swaig_query_params({'tier': 'premium'})
auth_handler
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
logger = get_logger('auth_handler')
module-attribute
¶
AuthHandler
¶
Unified authentication handler supporting multiple auth methods.
This class provides a clean pattern for handling Basic Auth, Bearer tokens, and API keys across all SignalWire services.
security_config = security_config
instance-attribute
¶
basic_auth = HTTPBasic(auto_error=False) if HTTPBasic is not None else None
instance-attribute
¶
bearer_auth = HTTPBearer(auto_error=False) if HTTPBearer is not None else None
instance-attribute
¶
__init__(security_config)
¶
Initialize auth handler with security configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
security_config
|
SecurityConfig
|
SecurityConfig instance with auth settings |
required |
verify_basic_auth(credentials)
¶
Verify basic auth credentials
verify_bearer_token(credentials)
¶
Verify bearer token
verify_api_key(api_key)
¶
Verify API key
get_fastapi_dependency(optional=False)
¶
Get FastAPI dependency for authentication.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
optional
|
bool
|
If True, authentication is optional |
False
|
Returns:
| Type | Description |
|---|---|
|
FastAPI dependency function |
flask_decorator(f)
¶
Flask decorator for authentication.
This provides compatibility with Flask-based services like MCP Gateway.
get_auth_info()
¶
Get information about configured auth methods
config_loader
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
logger = get_logger('config_loader')
module-attribute
¶
ConfigLoader
¶
Configuration loader with environment variable substitution.
Supports ${VAR|default} syntax for referencing environment variables within JSON configuration files. This provides a clean pattern for configuration across all SignalWire services.
config_paths = config_paths or self._get_default_paths()
instance-attribute
¶
__init__(config_paths=None)
¶
Initialize config loader.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config_paths
|
list[str] | None
|
Optional list of config file paths to check. If not provided, uses default search paths. |
None
|
has_config()
¶
Check if a configuration was loaded.
get_config_file()
¶
Get the path of the loaded config file.
get_config()
¶
Get the raw configuration (before substitution).
substitute_vars(value, max_depth=10)
¶
Recursively substitute environment variables in configuration values.
Supports ${VAR|default} syntax where: - VAR is the environment variable name - default is the fallback value if VAR is not set
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
The value to process (can be string, dict, list, etc.) |
required |
max_depth
|
int
|
Maximum recursion depth to prevent infinite loops |
10
|
Returns:
| Type | Description |
|---|---|
Any
|
The value with all environment variables substituted |
get(key_path, default=None)
¶
Get a configuration value by dot-notation path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_path
|
str
|
Dot-separated path (e.g., "security.ssl_enabled") |
required |
default
|
Any
|
Default value if path not found |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
The configuration value with variables substituted |
get_section(section)
¶
Get an entire configuration section.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section
|
str
|
The section name (e.g., "security", "server") |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The configuration section with all variables substituted |
merge_with_env(env_prefix='SWML_')
¶
Merge configuration with environment variables.
Config file takes precedence over environment variables, but config can reference env vars via substitution.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env_prefix
|
str
|
Prefix for environment variables to consider |
'SWML_'
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Merged configuration dictionary |
find_config_file(service_name=None, additional_paths=None)
staticmethod
¶
Static method to find a config file for a service.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_name
|
str | None
|
Optional service name for service-specific config |
None
|
additional_paths
|
list[str] | None
|
Additional paths to check |
None
|
Returns:
| Type | Description |
|---|---|
str | None
|
Path to the first config file found, or None |
contexts
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Contexts and Steps System for SignalWire Agents
This module provides an alternative to traditional POM-based prompts by allowing agents to be defined as structured contexts with sequential steps. Each step contains its own prompt, completion criteria, and function restrictions.
MAX_CONTEXTS = 50
module-attribute
¶
MAX_STEPS_PER_CONTEXT = 100
module-attribute
¶
RESERVED_NATIVE_TOOL_NAMES = frozenset({'next_step', 'change_context', 'gather_submit'})
module-attribute
¶
GatherQuestion
¶
Represents a single question in a gather_info configuration
key = key
instance-attribute
¶
question = question
instance-attribute
¶
type = type
instance-attribute
¶
confirm = confirm
instance-attribute
¶
prompt = prompt
instance-attribute
¶
functions = functions
instance-attribute
¶
__init__(key, question, type='string', confirm=False, prompt=None, functions=None)
¶
to_dict()
¶
Convert question to dictionary for SWML generation
GatherInfo
¶
Configuration for gathering information in a step via the C-side gather_info system.
This produces zero tool_call/tool_result entries in LLM-visible history, instead using dynamic step instruction re-injection to present one question at a time.
__init__(output_key=None, completion_action=None, prompt=None)
¶
add_question(key, question, **kwargs)
¶
Add a question to gather.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key name for storing the answer in global_data |
required |
question
|
str
|
The question text to ask |
required |
**kwargs
|
Optional fields - type, confirm, prompt, functions |
{}
|
Returns:
| Type | Description |
|---|---|
GatherInfo
|
Self for method chaining |
to_dict()
¶
Convert to dictionary for SWML generation
Step
¶
Represents a single step within a context
name = name
instance-attribute
¶
__init__(name)
¶
set_text(text)
¶
Set the step's prompt text directly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The prompt text for this step |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
add_section(title, body)
¶
Add a POM section to the step
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Section body text |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
add_bullets(title, bullets)
¶
Add a POM section with bullet points
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
bullets
|
list[str]
|
List of bullet points |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_step_criteria(criteria)
¶
Set the criteria for determining when this step is complete
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
criteria
|
str
|
Description of step completion criteria |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_functions(functions)
¶
Set which non-internal functions are callable while this step is active.
IMPORTANT — keep the per-step active set small: LLM tool selection accuracy degrades noticeably once the per-call tool list grows past ~7-8 entries. Symptoms: the model has the right tool available, the user's request clearly matches it, but the model doesn't call it (or calls a near-neighbor instead). Same prompt may pass on rerun. This is the LLM's native tool-selection behavior — not something the SDK can fix.
Mitigations:
- Whitelist only the tools the current step actually needs.
A 4-tool step is more reliable than a 12-tool step.
- Split a busy step into two narrower steps so the relevant
tools are partitioned by phase of the conversation.
- Remember that `next_step` and `change_context` are
auto-injected when valid_steps / valid_contexts are set —
they count against the budget too. Don't list them in
`functions`; the runtime adds them separately.
IMPORTANT — inheritance behavior:
If you do NOT call this method, the step inherits whichever function
set was active on the previous step (or the previous context's last
step). The server-side runtime only resets the active set when a
step explicitly declares its functions field. This is by design,
but it is the most common source of bugs in multi-step agents:
forgetting set_functions() on a later step lets the previous step's
tools leak through.
Best practice: call set_functions() explicitly on every step that
should have a different toolset than the previous one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
functions
|
str | list[str]
|
One of: - List[str] — whitelist of function names allowed in this step. Functions not in the list become inactive. - [] — explicit disable-all (no user functions callable). - "none" — synonym for [], same effect. |
required |
Internal functions (startup_hook, hangup_hook, check_for_input,
summarize_conversation, gather_submit, get_ideal_strategy) are
ALWAYS protected and cannot be deactivated by this whitelist.
The native navigation tools next_step and change_context are
injected automatically when valid_steps / valid_contexts is set; they
are not affected by this list and do not need to appear in it.
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining. |
Examples:
step.set_functions(["lookup_account", "check_balance"]) # whitelist step.set_functions([]) # disable all step.set_functions("none") # disable all (alt)
set_valid_steps(steps)
¶
Set which steps can be navigated to from this step
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
steps
|
list[str]
|
List of valid step names (include "next" for sequential flow) |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_valid_contexts(contexts)
¶
Set which contexts can be navigated to from this step
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
list[str]
|
List of valid context names |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_end(end)
¶
Mark this step as terminal for the step flow.
IMPORTANT: end=True does NOT end the conversation or hang up the
call. It exits step mode entirely after this step executes — clearing
the steps list, current step index, valid_steps, and valid_contexts.
The agent keeps running, but operates only under the base system
prompt and the context-level prompt; no more step instructions are
injected and no more next_step tool is offered.
To actually end the call, call a hangup tool or define a hangup_hook.
Combine with set_reset_*() if you also want to reset/consolidate
the conversation when this step exits.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
end
|
bool
|
True to exit step mode after this step. |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining. |
set_skip_user_turn(skip)
¶
Set whether to skip waiting for user input after this step
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skip
|
bool
|
Whether to skip the user turn after this step |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_skip_to_next_step(skip)
¶
Set whether to automatically advance to the next step
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skip
|
bool
|
Whether to skip to the next step automatically |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_gather_info(output_key=None, completion_action=None, prompt=None)
¶
Enable info gathering for this step. Questions are presented one at a time via dynamic step instruction re-injection, producing zero tool_call/tool_result entries in LLM-visible history.
After calling this, use add_gather_question() to define questions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
output_key
|
str | None
|
Key in global_data to store answers under (default: top-level) |
None
|
completion_action
|
str | None
|
Where to go when all questions are answered. Can be: - "next_step" to auto-advance to the next sequential step - A specific step name (e.g. "process_results") to jump to that step - None (default) to return to normal step mode after gathering The target must be valid: "next_step" requires a following step, and named steps must exist in the same context. |
None
|
prompt
|
str | None
|
Preamble text injected once when entering the gather step, giving the model personality/context for why it is asking these questions |
None
|
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
add_gather_question(key, question, type='string', confirm=False, prompt=None, functions=None)
¶
Add a question to this step's gather_info configuration. set_gather_info() must be called before this method.
IMPORTANT — gather mode locks function access: While the model is asking gather questions, the runtime forcibly deactivates ALL of the step's other functions. The only callable tools during a gather question are:
- `gather_submit` (the native answer-submission tool)
- Whatever names you list in this question's `functions` arg
`next_step` and `change_context` are also filtered out — the model
cannot navigate away until the gather completes. This is by design:
it forces a tight ask → submit → next-question loop.
If a question needs to call out to a tool (e.g. validate an email,
geocode a ZIP), list that tool name in this question's `functions`.
Functions listed here are active ONLY for this question.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key name for storing the answer in global_data. |
required |
question
|
str
|
The question text the model is instructed to ask. |
required |
type
|
str
|
JSON schema type for the answer ("string", "integer", "number", "boolean"). Default: "string". |
'string'
|
confirm
|
bool
|
If True, the model must read the answer back and obtain
explicit user confirmation before submitting (the gather_submit
schema gains a required |
False
|
prompt
|
str | None
|
Extra instruction text appended after the question. |
None
|
functions
|
list[str] | None
|
Names of functions to unlock for this question only.
These are activated on top of |
None
|
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining. |
clear_sections()
¶
Remove all POM sections and direct text from this step, allowing it to be repopulated with new content.
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_reset_system_prompt(system_prompt)
¶
Set system prompt for context switching when this step navigates to a context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
system_prompt
|
str
|
New system prompt for context switching |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_reset_user_prompt(user_prompt)
¶
Set user prompt for context switching when this step navigates to a context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_prompt
|
str
|
User message to inject for context switching |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_reset_consolidate(consolidate)
¶
Set whether to consolidate conversation when this step switches contexts
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
consolidate
|
bool
|
Whether to consolidate previous conversation |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
set_reset_full_reset(full_reset)
¶
Set whether to do full reset when this step switches contexts
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
full_reset
|
bool
|
Whether to completely rewrite system prompt vs inject |
required |
Returns:
| Type | Description |
|---|---|
Step
|
Self for method chaining |
to_dict()
¶
Convert step to dictionary for SWML generation
Context
¶
A single context containing an ordered list of steps.
Conversation history across context switches¶
By default (isolated=False), switching from one context to another
via change_context PRESERVES the entire conversation history. The
user's prior turns and the model's prior responses remain visible
on the next LLM call. The only thing that changes is which step
instructions get injected.
A common confusion: "the AI re-asked for information the user already
gave." If you see this, history loss is almost never the cause —
history is preserved unless you set isolated=True. The real cause
is usually one of:
- The destination step's
textliterally tells the model to ask ("Ask the user for their account number"). The model follows instructions; rephrase to "Confirm the user's account number" or have the step instructions check global_data first. - The relevant info was never extracted into global_data, so ${var} expansion has nothing to inject and the step prompt looks generic. Add a webhook that captures the field.
- You explicitly called
set_isolated(True)on the destination context. Isolated contexts wipe the conversation array on entry. Pair withset_consolidate(True)if you want a summary instead.
See Context.set_isolated() for the wipe semantics, and the SDK's FunctionResult.swml_change_step / swml_change_context docstrings for how to communicate transition intent through tool response text and global_data.
name = name
instance-attribute
¶
__init__(name)
¶
add_step(name, *, task=None, bullets=None, criteria=None, functions=None, valid_steps=None)
¶
Add a new step to this context.
When called with only name the returned Step can be configured
with the usual method-chaining API. When the optional keyword
arguments are supplied the step is fully configured in one call:
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Step name (must be unique within the context). |
required |
task
|
str | None
|
Text for the "Task" section (equivalent to
|
None
|
bullets
|
list[str] | None
|
List of bullet strings for the "Process" section
(equivalent to |
None
|
criteria
|
str | None
|
Step-completion criteria (equivalent to
|
None
|
functions
|
str | list[str] | None
|
Tool names the step may call, or |
None
|
valid_steps
|
list[str] | None
|
Names of steps the agent may transition to
(equivalent to |
None
|
Returns:
| Type | Description |
|---|---|
Step
|
The configured Step object for optional further chaining. |
get_step(name)
¶
Get an existing step by name for inspection or modification.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Step name |
required |
Returns:
| Type | Description |
|---|---|
Optional[Step]
|
Step object if found, None otherwise |
remove_step(name)
¶
Remove a step from this context entirely.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Step name to remove |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
move_step(name, position)
¶
Move an existing step to a specific position in the step order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Step name to move |
required |
position
|
int
|
Target index in the step order (0 = first) |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_valid_contexts(contexts)
¶
Set which contexts can be navigated to from this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
list[str]
|
List of valid context names |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_valid_steps(steps)
¶
Set which steps can be navigated to from any step in this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
steps
|
list[str]
|
List of valid step names (include "next" for sequential flow) |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_initial_step(step_name)
¶
Set which step the context starts on when entered.
By default, a context starts on its first step (index 0). If the
context has a preamble step that should only run on first entry
(e.g. a greeting), later entries via change_context can skip
it by setting initial_step to the name of the step to start
from instead.
initial_step is honoured both at conversation creation (when
the context is first activated) and when switching to this context
via change_context during the conversation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
step_name
|
str
|
Name of the step to start on. Must exist in this context's step list; validated by ContextBuilder.validate(). |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining. |
Example
ctx = contexts.add_context("support") ctx.add_step("greeting").set_text("Welcome!") ctx.add_step("triage").set_text("What do you need help with?") ctx.set_initial_step("triage") # skip greeting on re-entry
set_post_prompt(post_prompt)
¶
Set post prompt override for this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
post_prompt
|
str
|
Post prompt text to use when this context is active |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_system_prompt(system_prompt)
¶
Set system prompt for context switching (triggers context reset)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
system_prompt
|
str
|
New system prompt for when this context is entered |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_consolidate(consolidate)
¶
Set whether to consolidate conversation history when entering this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
consolidate
|
bool
|
Whether to consolidate previous conversation |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_full_reset(full_reset)
¶
Set whether to do full reset when entering this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
full_reset
|
bool
|
Whether to completely rewrite system prompt vs inject |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_user_prompt(user_prompt)
¶
Set user prompt to inject when entering this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_prompt
|
str
|
User message to inject for context |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_isolated(isolated)
¶
Mark this context as isolated — entering it wipes conversation history.
When isolated=True and the context is entered via change_context,
the runtime calls ai_conversation_restart() and the entire conversation
array is wiped. The model starts fresh with only the new context's
system_prompt + step instructions, with no memory of prior turns.
EXCEPTION — reset overrides the wipe:
If the context also has a reset configuration (set via the
Step.set_reset_*() methods on a step that switches into this
context, or via set_consolidate() / set_full_reset() on the
context itself), the wipe is skipped in favor of the reset
behavior. Use reset with consolidate=True to summarize prior
history into a single message instead of dropping it entirely.
Use cases
- Switching to a sensitive billing flow that should not see prior small-talk
- Handing off to a different agent persona
- Resetting after a long off-topic detour
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
isolated
|
bool
|
True to wipe conversation history on context entry (subject to the reset exception above). |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining. |
add_system_section(title, body)
¶
Add a POM section to the system prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Section body text |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
add_system_bullets(title, bullets)
¶
Add a POM section with bullet points to the system prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
bullets
|
list[str]
|
List of bullet points |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_prompt(prompt)
¶
Set the context's prompt text directly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt
|
str
|
The prompt text for this context |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
add_section(title, body)
¶
Add a POM section to the context prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Section body text |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
add_bullets(title, bullets)
¶
Add a POM section with bullet points to the context prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
bullets
|
list[str]
|
List of bullet points |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_enter_fillers(enter_fillers)
¶
Set fillers that the AI says when entering this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enter_fillers
|
dict[str, list[str]]
|
Dictionary mapping language codes (or "default") to lists of filler phrases Example: {"en-US": ["Welcome...", "Hello..."], "default": ["Entering..."]} |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
set_exit_fillers(exit_fillers)
¶
Set fillers that the AI says when exiting this context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exit_fillers
|
dict[str, list[str]]
|
Dictionary mapping language codes (or "default") to lists of filler phrases Example: {"en-US": ["Goodbye...", "Thank you..."], "default": ["Exiting..."]} |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
add_enter_filler(language_code, fillers)
¶
Add enter fillers for a specific language
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
language_code
|
str
|
Language code (e.g., "en-US", "es") or "default" for catch-all |
required |
fillers
|
list[str]
|
List of filler phrases for entering this context |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
add_exit_filler(language_code, fillers)
¶
Add exit fillers for a specific language
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
language_code
|
str
|
Language code (e.g., "en-US", "es") or "default" for catch-all |
required |
fillers
|
list[str]
|
List of filler phrases for exiting this context |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Self for method chaining |
to_dict()
¶
Convert context to dictionary for SWML generation
ContextBuilder
¶
Builder for multi-step, multi-context AI agent workflows.
A ContextBuilder owns one or more Contexts; each Context owns an ordered list of Steps. Only one context and one step is active at a time. Per chat turn, the runtime injects the current step's instructions as a system message, then asks the LLM for a response.
Native tools auto-injected by the runtime¶
When a step (or its enclosing context) declares valid_steps or
valid_contexts, the runtime auto-injects two native tools so the model
can navigate the flow:
next_step(step: enum)— present when valid_steps is setchange_context(context: enum)— present when valid_contexts is set
Their enum schemas are rewritten on every turn to match whatever
valid_steps / valid_contexts apply to the current step. You do NOT need
to define these tools yourself; they appear automatically.
A third native tool — gather_submit — is injected during gather_info
questioning (see Step.set_gather_info / add_gather_question).
These three names — next_step, change_context, gather_submit
— are reserved. ContextBuilder.validate() will reject any agent that
defines a SWAIG tool with one of these names.
Function whitelisting (Step.set_functions)¶
Each step may declare a functions whitelist. The whitelist is applied
in-memory at the start of each LLM turn. CRITICALLY: if a step does NOT
declare a functions field, it INHERITS the previous step's active set.
See Step.set_functions() for details and examples.
Validation¶
Call validate() (or to_dict(), which calls it) to check that:
- At least one context is defined
- A single context must be named "default"
- Every context has at least one step
- valid_steps references resolve to real step names (or "next")
- valid_contexts references resolve to real context names
- gather_info questions are non-empty and have unique keys
- gather_info completion_action targets a reachable step
- No user-defined SWAIG tool collides with a reserved native name
__init__(agent)
¶
reset()
¶
Remove all contexts, returning the builder to its initial state.
Use this in a dynamic config callback when you need to rebuild contexts from scratch for a specific request — e.g. skipping a greeting context on transfers.
Returns:
| Type | Description |
|---|---|
ContextBuilder
|
Self for method chaining. |
Example::
def on_dynamic_config(query, body, headers, agent):
if query.get("transfer"):
agent.define_contexts().reset()
ctx = agent.define_contexts().add_context("default")
ctx.add_step("route").set_text("Route the caller.")
add_context(name)
¶
Add a new context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Context name |
required |
Returns:
| Type | Description |
|---|---|
Context
|
Context object for method chaining |
get_context(name)
¶
Get an existing context by name for inspection or modification.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Context name |
required |
Returns:
| Type | Description |
|---|---|
Context | None
|
Context object if found, None otherwise |
validate()
¶
Validate the contexts configuration
to_dict()
¶
Convert all contexts to dictionary for SWML generation
data_map
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
DataMap class for building SWAIG data_map configurations
DataMap
¶
Builder class for creating SWAIG data_map configurations.
This provides a fluent interface for building data_map tools that execute on the SignalWire server without requiring webhook endpoints. Works similar to FunctionResult but for building data_map structures.
Example usage
Simple API call - output goes inside webhook¶
data_map = (DataMap('get_weather') .purpose('Get current weather information') .parameter('location', 'string', 'City name', required=True) .webhook('GET', 'https://api.weather.com/v1/current?key=API_KEY&q=${location}') .output(FunctionResult('Weather in ${location}: ${response.current.condition.text}, ${response.current.temp_f}°F')) )
Multiple webhooks with fallback¶
data_map = (DataMap('search_multi') .purpose('Search with fallback APIs') .parameter('query', 'string', 'Search query', required=True) .webhook('GET', 'https://api.primary.com/search?q=${query}') .output(FunctionResult('Primary result: ${response.title}')) .webhook('GET', 'https://api.fallback.com/search?q=${query}') .output(FunctionResult('Fallback result: ${response.title}')) .fallback_output(FunctionResult('Sorry, all search APIs are unavailable')) )
Expression-based responses (no API calls)¶
data_map = (DataMap('file_control') .purpose('Control file playback') .parameter('command', 'string', 'Playback command') .parameter('filename', 'string', 'File to control', required=False) .expression('${args.command}', r'start.', FunctionResult().add_action('start_playbook', {'file': '${args.filename}'})) .expression('${args.command}', r'stop.', FunctionResult().add_action('stop_playback', True)) )
API with array processing¶
data_map = (DataMap('search_docs') .purpose('Search documentation') .parameter('query', 'string', 'Search query', required=True) .webhook('POST', 'https://api.docs.com/search', headers={'Authorization': 'Bearer TOKEN'}) .body({'query': '${query}', 'limit': 3}) .output(FunctionResult('Found: ${response.results[0].title} - ${response.results[0].summary}')) .foreach('${response.results}') )
function_name = function_name
instance-attribute
¶
__init__(function_name)
¶
Initialize a new DataMap builder
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the SWAIG function this data_map will create |
required |
purpose(description)
¶
Set the function description that the LLM will read.
A DataMap creates a SWAIG function that gets sent to the model in
OpenAI tool-schema format. This description field is what the
model reads on every turn to decide WHEN to call the tool. It is
prompt-engineered text, not developer documentation:
- Bad: "Search function"
- Good: "Search the company's knowledge base for help articles matching a user query. Use this when the user asks a product or how-to question that the base prompt does not cover."
Vague descriptions are the most common cause of "the model has the right tool but doesn't call it" failures.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
description
|
str
|
LLM-facing description of what this function does and when to use it. See above. |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining. |
description(description)
¶
Set the function description (alias for purpose).
See purpose() for guidance on writing description text the LLM can act on.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
description
|
str
|
LLM-facing description of what this function does and when to use it. |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining. |
parameter(name, param_type, description, required=False, enum=None)
¶
Add a function parameter.
Just like the function-level description, this parameter
description is sent to the LLM as part of the tool schema and
is read by the model when deciding HOW to fill in the argument.
Write it as an instruction to the model:
- Bad: "the id"
- Good: "The customer's 8-digit account number, no dashes or spaces. Ask the user if they don't provide it."
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Parameter name. Becomes a key in the tool schema's
|
required |
param_type
|
str
|
JSON schema type (string, number, boolean, array, object). |
required |
description
|
str
|
LLM-facing parameter description. See above — this should tell the model what value to put here, in what format, and where to source it. |
required |
required
|
bool
|
Whether parameter is required. |
False
|
enum
|
list[str] | None
|
Optional list of allowed values. The model will only emit values from this list. |
None
|
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining. |
expression(test_value, pattern, output, nomatch_output=None)
¶
Add an expression pattern for pattern-based responses
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
test_value
|
str
|
Template string to test (e.g., "${args.command}") |
required |
pattern
|
str | Pattern
|
Regex pattern string or compiled Pattern object to match against |
required |
output
|
FunctionResult
|
FunctionResult to return when pattern matches |
required |
nomatch_output
|
FunctionResult | None
|
Optional FunctionResult to return when pattern doesn't match |
None
|
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
webhook(method, url, headers=None, form_param=None, input_args_as_params=False, require_args=None)
¶
Add a webhook API call
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
HTTP method (GET, POST, PUT, DELETE, etc.) |
required |
url
|
str
|
API endpoint URL (can include ${variable} substitutions) |
required |
headers
|
dict[str, str] | None
|
Optional HTTP headers |
None
|
form_param
|
str | None
|
Send JSON body as single form parameter with this name |
None
|
input_args_as_params
|
bool
|
Merge function arguments into params |
False
|
require_args
|
list[str] | None
|
Only execute if these arguments are present |
None
|
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
webhook_expressions(expressions)
¶
Add expressions that run after the most recent webhook completes
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
expressions
|
list[dict[str, Any]]
|
List of expression definitions to check post-webhook |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
body(data)
¶
Set request body for the last added webhook (POST/PUT requests)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Request body data (can include ${variable} substitutions) |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
params(data)
¶
Set request params for the last added webhook (alias for body)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Request params data (can include ${variable} substitutions) |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
foreach(foreach_config)
¶
Process an array from the webhook response using foreach mechanism
Args:
foreach_config: Either:
- Dict: Foreach configuration with keys:
- input_key: Key in API response containing the array
- output_key: Name for the built string variable
- max: Maximum number of items to process (optional)
- append: Template string to append for each item
Returns:
Self for method chaining
Example:
.foreach({
"input_key": "results",
"output_key": "formatted_results",
"max": 3,
"append": "Result: ${this.title} - ${this.summary}
" })
output(result)
¶
Set the output result for the most recent webhook
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result
|
FunctionResult
|
FunctionResult defining the response for this webhook |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
fallback_output(result)
¶
Set a fallback output result at the top level (used when all webhooks fail)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result
|
FunctionResult
|
FunctionResult defining the fallback response |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
error_keys(keys)
¶
Set error keys for the most recent webhook (if webhooks exist) or top-level
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
list[str]
|
List of JSON keys whose presence indicates an error |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
global_error_keys(keys)
¶
Set top-level error keys (applies to all webhooks)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
list[str]
|
List of JSON keys whose presence indicates an error |
required |
Returns:
| Type | Description |
|---|---|
DataMap
|
Self for method chaining |
to_swaig_function()
¶
Convert this DataMap to a SWAIG function definition
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with function definition and data_map instead of url |
create_simple_api_tool(name, url, response_template, parameters=None, method='GET', headers=None, body=None, error_keys=None)
¶
Create a simple API tool with minimal configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
url
|
str
|
API endpoint URL |
required |
response_template
|
str
|
Template for formatting the response |
required |
parameters
|
dict[str, dict] | None
|
Optional parameter definitions |
None
|
method
|
str
|
HTTP method (default: GET) |
'GET'
|
headers
|
dict[str, str] | None
|
Optional HTTP headers |
None
|
body
|
dict[str, Any] | None
|
Optional request body (for POST/PUT) |
None
|
error_keys
|
list[str] | None
|
Optional list of error indicator keys |
None
|
Returns:
| Type | Description |
|---|---|
DataMap
|
Configured DataMap object |
create_expression_tool(name, patterns, parameters=None)
¶
Create an expression-based tool for pattern matching responses
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
patterns
|
dict[str, tuple[str, FunctionResult]]
|
Dictionary mapping test_values to (pattern, FunctionResult) tuples |
required |
parameters
|
dict[str, dict] | None
|
Optional parameter definitions |
None
|
Returns:
| Type | Description |
|---|---|
DataMap
|
Configured DataMap object |
function_result
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
FunctionResult class for handling the response format of SWAIG function calls
SwaigFunctionResult = FunctionResult
module-attribute
¶
FunctionResult
¶
Wrapper around SWAIG function responses that handles proper formatting of response text and actions.
The result object has three main components: 1. response: Text the AI should say back to the user 2. action: List of structured actions to execute 3. post_process: Whether to let AI take another turn before executing actions
Post-processing behavior: - post_process=False (default): Execute actions immediately after AI response - post_process=True: Let AI respond to user one more time, then execute actions
This is useful for confirmation workflows like: "I'll transfer you to sales. Do you have any other questions first?" (AI can handle follow-up, then execute the transfer)
Example
return FunctionResult("Found your order")
With actions¶
return ( FunctionResult("I'll transfer you to support") .add_action("transfer", {"dest": "support"}) )
With simple action value¶
return ( FunctionResult("I'll confirm that") .add_action("confirm", True) )
With multiple actions¶
return ( FunctionResult("Processing your request") .add_actions([ {"set_global_data": {"key": "value"}}, {"play": {"url": "music.mp3"}} ]) )
With post-processing enabled¶
return ( FunctionResult("Let me transfer you to billing", post_process=True) .connect("+15551234567", final=True) )
Using the connect helper¶
return ( FunctionResult("I'll transfer you to our sales team now") .connect("sales@company.com", final=False, from_addr="+15559876543") )
response = response if response is not None else ''
instance-attribute
¶
action = []
instance-attribute
¶
post_process = post_process
instance-attribute
¶
__init__(response=None, post_process=False)
¶
Initialize a new SWAIG function result
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str | None
|
Optional natural language response to include |
None
|
post_process
|
bool
|
Whether to let AI take another turn before executing actions. Defaults to False (execute actions immediately after response). |
False
|
set_response(response)
¶
Set the natural language response text
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str
|
The text the AI should say |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
set_post_process(post_process)
¶
Set whether to enable post-processing for this result.
Post-processing allows the AI to take one more turn with the user before executing any actions. This is useful for confirmation workflows.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
post_process
|
bool
|
True to let AI respond once more before executing actions, False to execute actions immediately after the response. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
add_action(name, data)
¶
Add a structured action to the response
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name/type of the action (e.g., "play", "transfer") |
required |
data
|
Any
|
The data for the action - can be a string, boolean, object, or array |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
add_actions(actions)
¶
Add multiple structured actions to the response
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
actions
|
list[dict[str, Any]]
|
List of action objects to add to the response |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
connect(destination, final=True, from_addr=None)
¶
Add a connect action to transfer/connect the call to another destination.
This is a convenience method that abstracts the SWML connect verb, so users don't need to manually construct SWML documents.
Transfer behavior: - final=True: Permanent transfer - call exits the agent completely, SWML replaces the agent and call continues there - final=False: Temporary transfer - if far end hangs up, call returns to the agent to continue the conversation
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
destination
|
str
|
Where to connect the call (phone number, SIP address, etc.) |
required |
final
|
bool
|
Whether this is a permanent transfer (True) or temporary (False). Defaults to True for permanent transfers. |
True
|
from_addr
|
str | None
|
Optional caller ID override (phone number or SIP address). If None, uses the current call's from address. |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
swml_transfer(dest, ai_response, final=True)
¶
Add a SWML transfer action with AI response setup for when transfer completes.
This is a virtual helper that generates SWML to transfer the call to another destination and sets up an AI response for when the transfer completes and control returns to the agent.
For transfers, you typically want to enable post-processing so the AI speaks the response first before executing the transfer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dest
|
str
|
Destination URL for the transfer (SWML endpoint, SIP address, etc.) |
required |
ai_response
|
str
|
Message the AI should say when transfer completes and control returns |
required |
final
|
bool
|
Whether this is a permanent transfer (True) or temporary (False). Defaults to True for permanent transfers (same as connect method). |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
Example
Permanent transfer (default)¶
result = ( FunctionResult("I'm transferring you to support", post_process=True) .swml_transfer( "https://support.example.com/swml", "Goodbye!" # Won't be used since final=True by default ) )
Temporary transfer with return¶
result.swml_transfer( dest, "The support call is complete. How else can I help?", final=False )
update_global_data(data)
¶
Update global agent data variables.
This is a convenience method that abstracts the set_global_data action. Global data persists across the entire agent session and is available in prompt variables and can be accessed by all functions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of key-value pairs to set/update in global data |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
swml_user_event(event_data)
¶
Send a user event through SWML to update the client UI.
This is a convenience method for sending user events to connected clients, commonly used for real-time UI updates in interactive applications.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_data
|
dict[str, Any]
|
Dictionary containing the event type and any associated data Example: {"type": "cards_dealt", "player_hand": [...], "score": 21} |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining |
Example
result = ( FunctionResult("You have blackjack!") .swml_user_event({ "type": "cards_dealt", "player_hand": player_cards, "dealer_hand": dealer_cards, "player_score": 21 }) )
swml_change_step(step_name)
¶
Force the conversation into a specific step in the current context.
This is a workflow-level transition driven by your webhook, not by the
model. Webhook-triggered step changes bypass any valid_steps clamp
on the current step (the model's next_step tool is the constrained
path; this is the unconstrained one). Use it when something the user
said unambiguously requires moving the flow forward.
Communicating intent to the destination step¶
When the model arrives at the new step, the next thing it sees in
history is the tool result that contained this action — i.e., your
FunctionResult's response text. Put your "why" there. The model
reads it before the new step's injected instructions:
FunctionResult(
"Premium plan confirmed. Now collecting payment details."
).swml_change_step("collect_payment")
For structured data the destination step needs (account number,
plan tier, etc.), pair this with update_global_data() and reference
the values from the destination step's text via ${var} expansion:
(FunctionResult("Premium plan confirmed.")
.update_global_data({"plan": "premium", "billing_cycle": "annual"})
.swml_change_step("collect_payment"))
And in the destination step's text:
step.set_text(
"Collect payment for the ${plan} plan, billed ${billing_cycle}. "
"Confirm the amount with the user before proceeding."
)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
step_name
|
str
|
Name of the step to transition to. The step must exist in the current context. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining. |
Example
result = ( FunctionResult("Starting a new hand") .update_global_data({"chips": 1000}) .swml_change_step("betting") )
swml_change_context(context_name)
¶
Force the conversation into a different context.
Webhook-triggered context changes bypass any valid_contexts clamp.
Step state resets (cur_step = 0) on entry. The new context's first
step's instructions are injected on the next turn.
Communicating intent to the destination context¶
Same pattern as swml_change_step: the model reads your FunctionResult's
response text as the tool result before it sees the destination
context's first step. Put your reason there:
FunctionResult(
"I'm transferring you to billing because your question is "
"about an invoice charge."
).swml_change_context("billing")
For structured carry-over, use update_global_data() and reference the values via ${var} expansion in the destination context's step text.
Note that conversation history is preserved across context switches
unless the destination context is marked isolated=True. The model
retains everything the user said in the previous context — you do
not need to re-state it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context_name
|
str
|
Name of the context to switch to. Must exist in the agent's context map. |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
Self for method chaining. |
Example
result = ( FunctionResult("Transferring you to technical support.") .update_global_data({"original_issue": user_problem}) .swml_change_context("technical_support") )
execute_swml(swml_content, transfer=False)
¶
Execute SWML content with optional transfer behavior.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
swml_content
|
Can be: - String: Raw SWML JSON text - Dict: SWML data structure - SWML object: SignalWire SWML SDK object with .to_dict() method |
required | |
transfer
|
bool
|
Boolean - whether call should exit agent after execution |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
hangup()
¶
hold(timeout=300)
¶
Put the call on hold with optional timeout.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timeout
|
int
|
Timeout in seconds (max 900, default 300) |
300
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
wait_for_user(enabled=None, timeout=None, answer_first=False)
¶
Control how agent waits for user input.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool | None
|
Boolean to enable/disable waiting |
None
|
timeout
|
int | None
|
Number of seconds to wait |
None
|
answer_first
|
bool
|
Special "answer_first" mode |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop()
¶
say(text)
¶
Make the agent speak specific text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text for agent to speak |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
play_background_file(filename, wait=False)
¶
Play audio or video file in background.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Audio/video filename/path |
required |
wait
|
bool
|
Whether to suppress attention-getting behavior during playback |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop_background_file()
¶
Stop currently playing background file.
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
add_dynamic_hints(hints)
¶
Add dynamic speech recognition hints during a call.
Hints improve speech recognition accuracy for domain-specific terms. Each hint can be a simple string or a pronunciation pattern object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hints
|
list[str | dict[str, Any]]
|
List of hints, where each entry is either: - A string: Simple hint word/phrase (e.g., "Cabby") - A dict with pronunciation pattern: - "pattern": Regex pattern to match in recognized speech - "replace": Replacement text when pattern matches - "ignore_case": Optional bool for case-insensitive matching |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("I'll listen for that name") .add_dynamic_hints([ "Cabby", {"pattern": "cab bee", "replace": "Cabby", "ignore_case": True} ]) )
clear_dynamic_hints()
¶
Clear all dynamic speech recognition hints.
Removes all hints previously added via add_dynamic_hints().
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_end_of_speech_timeout(milliseconds)
¶
Adjust end of speech timeout - milliseconds of silence after speaking has been detected to finalize speech recognition.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
milliseconds
|
int
|
Timeout in milliseconds |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_speech_event_timeout(milliseconds)
¶
Adjust speech event timeout - milliseconds since last speech detection event to finalize recognition. Works better in noisy environments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
milliseconds
|
int
|
Timeout in milliseconds |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
remove_global_data(keys)
¶
Remove global agent data variables.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
Single key string or list of keys to remove |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
set_metadata(data)
¶
Set metadata scoped to current function's meta_data_token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of key-value pairs for metadata |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
remove_metadata(keys)
¶
Remove metadata from current function's meta_data_token scope.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
Single key string or list of keys to remove |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
toggle_functions(function_toggles)
¶
Enable/disable specific SWAIG functions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_toggles
|
list[dict[str, Any]]
|
List of dicts with 'function' and 'active' keys |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
enable_functions_on_timeout(enabled=True)
¶
Enable function calls on speaker timeout.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool
|
Whether to enable functions on timeout |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
enable_extensive_data(enabled=True)
¶
Send full data to LLM for this turn only, then use smaller replacement in subsequent turns.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled
|
bool
|
Whether to send extensive data this turn only |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
replace_in_history(text=True)
¶
After first send, replace tool_call+result pair in conversation history.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str | bool
|
String = replace tool_call with an assistant message containing this text. True = remove the tool_call+result pair from history entirely. |
True
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
update_settings(settings)
¶
Update agent runtime settings.
Supported settings: - frequency-penalty: Float (-2.0 to 2.0) - presence-penalty: Float (-2.0 to 2.0) - max-tokens: Integer (0 to 4096) - top-p: Float (0.0 to 1.0) - confidence: Float (0.0 to 1.0) - barge-confidence: Float (0.0 to 1.0) - temperature: Float (0.0 to 2.0, clamped to 1.5)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
settings
|
dict[str, Any]
|
Dictionary of settings to update |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
switch_context(system_prompt=None, user_prompt=None, consolidate=False, full_reset=False)
¶
Change agent context/prompt during conversation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
system_prompt
|
str | None
|
New system prompt |
None
|
user_prompt
|
str | None
|
User message to add |
None
|
consolidate
|
bool
|
Whether to summarize existing conversation |
False
|
full_reset
|
bool
|
Whether to do complete context reset |
False
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
simulate_user_input(text)
¶
Queue simulated user input.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text to simulate as user input |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
send_sms(to_number, from_number, body=None, media=None, tags=None, region=None)
¶
Send a text message to a PSTN phone number using SWML.
This is a virtual helper that generates SWML to send SMS messages. Either body or media (or both) must be provided.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_number
|
str
|
Phone number in E.164 format to send to |
required |
from_number
|
str
|
Phone number in E.164 format to send from |
required |
body
|
str | None
|
Body text of the message (optional if media provided) |
None
|
media
|
list[str] | None
|
Array of URLs to send in the message (optional if body provided) |
None
|
tags
|
list[str] | None
|
Array of tags to associate with the message for UI searching |
None
|
region
|
str | None
|
Region to originate the message from |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither body nor media is provided |
pay(payment_connector_url, input_method='dtmf', status_url=None, payment_method='credit-card', timeout=5, max_attempts=1, security_code=True, postal_code=True, min_postal_code_length=0, token_type='reusable', charge_amount=None, currency='usd', language='en-US', voice='woman', description=None, valid_card_types='visa mastercard amex', parameters=None, prompts=None, ai_response='The payment status is ${pay_result}, do not mention anything else about collecting payment if successful.')
¶
Process payment using SWML pay action.
This is a virtual helper that generates SWML for payment processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_connector_url
|
str
|
URL to make payment requests to (required) |
required |
input_method
|
str
|
Method to collect payment details. The SWML pay verb
only accepts "dtmf" (schema |
'dtmf'
|
status_url
|
str | None
|
URL for status change notifications |
None
|
payment_method
|
str
|
Payment method ("credit-card" currently supported) |
'credit-card'
|
timeout
|
int
|
Seconds to wait for next digit (default: 5) |
5
|
max_attempts
|
int
|
Number of retry attempts (default: 1) |
1
|
security_code
|
bool
|
Whether to prompt for security code (default: True) |
True
|
postal_code
|
bool | str
|
Whether to prompt for postal code, or actual postcode |
True
|
min_postal_code_length
|
int
|
Minimum postal code digits (default: 0) |
0
|
token_type
|
str
|
Payment type ("one-time" or "reusable", default: "reusable") |
'reusable'
|
charge_amount
|
str | None
|
Amount to charge as decimal string |
None
|
currency
|
str
|
Currency code (default: "usd") |
'usd'
|
language
|
str
|
Language for prompts (default: "en-US") |
'en-US'
|
voice
|
str
|
TTS voice to use (default: "woman") |
'woman'
|
description
|
str | None
|
Custom payment description |
None
|
valid_card_types
|
str
|
Space-separated card types (default: "visa mastercard amex") |
'visa mastercard amex'
|
parameters
|
list[dict[str, str]] | None
|
Array of name/value pairs for payment connector |
None
|
prompts
|
list[dict[str, Any]] | None
|
Array of custom prompt configurations |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
record_call(control_id=None, stereo=False, format='wav', direction='both', terminators=None, beep=False, input_sensitivity=44.0, initial_timeout=None, end_silence_timeout=None, max_length=None, status_url=None)
¶
Start background call recording using SWML.
This is a virtual helper that generates SWML to start recording the call in the background. Unlike foreground recording, the script continues executing while recording happens in the background.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
Identifier for this recording (for use with stop_record_call) |
None
|
stereo
|
bool
|
Record in stereo (default: False) |
False
|
format
|
Literal['wav', 'mp3', 'mp4']
|
Recording format - "wav", "mp3", or "mp4" (default: "wav") |
'wav'
|
direction
|
Literal['speak', 'listen', 'both']
|
Audio direction - "speak", "listen", or "both" (default: "both") |
'both'
|
terminators
|
str | None
|
Digits that stop recording when pressed |
None
|
beep
|
bool
|
Play beep before recording (default: False) |
False
|
input_sensitivity
|
float
|
Input sensitivity for recording (default: 44.0) |
44.0
|
initial_timeout
|
float | None
|
Time in seconds to wait for speech start (for voicemail-style recording) |
None
|
end_silence_timeout
|
float | None
|
Time in seconds to wait in silence before ending (for voicemail-style recording) |
None
|
max_length
|
float | None
|
Maximum recording length in seconds |
None
|
status_url
|
str | None
|
URL to send recording status events to |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
stop_record_call(control_id=None)
¶
Stop an active background call recording using SWML.
This is a virtual helper that generates SWML to stop a recording that was started with record_call().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
Identifier for the recording to stop. If not provided, the most recent recording will be stopped. |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
join_room(name)
¶
Join a RELAY room using SWML.
This is a virtual helper that generates SWML to join a RELAY room, which enables multi-party communication and collaboration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the room to join (required) |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
sip_refer(to_uri)
¶
Send SIP REFER to a SIP call using SWML.
This is a virtual helper that generates SWML to send a SIP REFER message, which is used for call transfer in SIP environments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_uri
|
str
|
The SIP URI to send the REFER to (required) |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
join_conference(name, muted=False, beep='true', start_on_enter=True, end_on_exit=False, wait_url=None, max_participants=250, record='do-not-record', region=None, trim='trim-silence', coach=None, status_callback_event=None, status_callback=None, status_callback_method='POST', recording_status_callback=None, recording_status_callback_method='POST', recording_status_callback_event='completed', result=None)
¶
Join an ad-hoc audio conference with RELAY and CXML calls using SWML.
This is a virtual helper that generates SWML to join audio conferences with extensive configuration options for call management and recording.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of conference (required) |
required |
muted
|
bool
|
Whether to join muted (default: False) |
False
|
beep
|
str
|
Beep configuration - "true", "false", "onEnter", "onExit" (default: "true") |
'true'
|
start_on_enter
|
bool
|
Whether conference starts when this participant enters (default: True) |
True
|
end_on_exit
|
bool
|
Whether conference ends when this participant exits (default: False) |
False
|
wait_url
|
str | None
|
SWML URL for hold music (default: None for default hold music) |
None
|
max_participants
|
int
|
Maximum participants <= 250 (default: 250) |
250
|
record
|
str
|
Recording mode - "do-not-record", "record-from-start" (default: "do-not-record") |
'do-not-record'
|
region
|
str | None
|
Conference region (default: None) |
None
|
trim
|
str
|
Trim silence - "trim-silence", "do-not-trim" (default: "trim-silence") |
'trim-silence'
|
coach
|
str | None
|
SWML Call ID or CXML CallSid for coaching (default: None) |
None
|
status_callback_event
|
str | None
|
Events to report - "start end join leave mute hold modify speaker announcement" (default: None) |
None
|
status_callback
|
str | None
|
URL for status callbacks (default: None) |
None
|
status_callback_method
|
str
|
HTTP method - "GET", "POST" (default: "POST") |
'POST'
|
recording_status_callback
|
str | None
|
URL for recording status callbacks (default: None) |
None
|
recording_status_callback_method
|
str
|
HTTP method - "GET", "POST" (default: "POST") |
'POST'
|
recording_status_callback_event
|
str
|
Recording events - "in-progress completed absent" (default: "completed") |
'completed'
|
result
|
Any | None
|
Switch on return_value when object {} or cond when array [] (default: None) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If beep value is invalid or max_participants exceeds 250 |
tap(uri, control_id=None, direction='both', codec='PCMU', rtp_ptime=20, status_url=None)
¶
Start background call tap using SWML.
This is a virtual helper that generates SWML to start background call tapping. Media is streamed over Websocket or RTP to customer controlled URI.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
str
|
Destination of tap media stream (required) Formats: rtp://IP:port, ws://example.com, or wss://example.com |
required |
control_id
|
str | None
|
Identifier for this tap to use with stop_tap (optional) Default is generated and stored in tap_control_id variable |
None
|
direction
|
Literal['speak', 'hear', 'both']
|
Direction of audio to tap (default: "both") "speak" = what party says "hear" = what party hears "both" = what party hears and says |
'both'
|
codec
|
Literal['PCMU', 'PCMA']
|
Codec for tap media stream - "PCMU" or "PCMA" (default: "PCMU") |
'PCMU'
|
rtp_ptime
|
int
|
Packetization time in milliseconds for RTP (default: 20) |
20
|
status_url
|
str | None
|
URL for status change requests (optional) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If direction or codec values are invalid |
stop_tap(control_id=None)
¶
Stop an active tap stream using SWML.
This is a virtual helper that generates SWML to stop a tap stream that was started with tap().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
control_id
|
str | None
|
ID of the tap to stop (optional) If not set, the last tap started will be stopped |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
execute_rpc(method, params=None, call_id=None, node_id=None)
¶
Execute an RPC method on a call using SWML.
This is a generic helper for executing RPC commands. For common operations, consider using the specific helpers: rpc_dial(), rpc_ai_message(), rpc_ai_unhold().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
The RPC method to execute (e.g., "dial", "ai_message", "ai_unhold") |
required |
params
|
dict[str, Any] | None
|
Parameters for the RPC method (optional) |
None
|
call_id
|
str | None
|
Target call ID for the RPC (optional) |
None
|
node_id
|
str | None
|
Target node ID for the RPC (optional) |
None
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Executing RPC") .execute_rpc( method="ai_message", call_id="some-call-id", params={"role": "system", "message_text": "Hello"} ) )
rpc_dial(to_number, from_number, dest_swml, device_type='phone')
¶
Dial out to a number with a destination SWML URL using execute_rpc.
This is commonly used in call screening scenarios where you place a caller on hold and dial out to a human, with the dest_swml specifying what agent handles the outbound leg.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_number
|
str
|
Phone number to dial (E.164 format) |
required |
from_number
|
str
|
Caller ID to use (E.164 format) |
required |
dest_swml
|
str
|
URL to the SWML that handles the outbound call |
required |
device_type
|
str
|
Device type, typically "phone" (default: "phone") |
'phone'
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Please hold while I connect you.") .hold(timeout=120) .rpc_dial( to_number="+15551234567", from_number="+15559876543", dest_swml="https://example.com/call-agent?caller=John" ) )
rpc_ai_message(call_id, message_text, role='system')
¶
Inject a message into an AI agent on another call using execute_rpc.
This is useful for cross-call communication, such as notifying a held caller's AI agent about a status change or instructing it to relay a message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str
|
The call ID of the target call |
required |
message_text
|
str
|
The message text to inject into the AI conversation |
required |
role
|
str
|
The role for the message, typically "system" (default: "system") |
'system'
|
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("I'll let them know.") .rpc_ai_message( call_id=original_call_id, message_text="The person you were trying to reach is unavailable. Please take a message." ) )
rpc_ai_unhold(call_id)
¶
Unhold another call using execute_rpc.
This releases a call from hold state, typically used after injecting a message to the held caller's AI agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str
|
The call ID of the call to unhold |
required |
Returns:
| Type | Description |
|---|---|
FunctionResult
|
self for method chaining |
Example
result = ( FunctionResult("Understood, I'll let them know.") .rpc_ai_message(call_id, "No one is available. Please take a message.") .rpc_ai_unhold(call_id) )
create_payment_prompt(for_situation, actions, card_type=None, error_type=None)
staticmethod
¶
Create a payment prompt structure for use with pay() method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
for_situation
|
str
|
Situation to use prompt for (e.g., "payment-card-number") |
required |
actions
|
list[dict[str, str]]
|
List of actions with 'type' and 'phrase' keys |
required |
card_type
|
str | None
|
Space-separated card types for this prompt |
None
|
error_type
|
str | None
|
Space-separated error types for this prompt |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary representing the prompt structure |
create_payment_action(action_type, phrase)
staticmethod
¶
Create a payment action for use in payment prompts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
action_type
|
str
|
"Say" for text-to-speech or "Play" for audio file |
required |
phrase
|
str
|
Sentence to say or URL to play |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dictionary representing the action |
create_payment_parameter(name, value)
staticmethod
¶
Create a payment parameter for use with pay() method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Parameter name |
required |
value
|
str
|
Parameter value |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dictionary representing the parameter |
to_dict()
¶
Convert to the JSON structure expected by SWAIG
The result must have at least one of: - 'response': Text to be spoken by the AI - 'action': Array of action objects
Optional: - 'post_process': Boolean controlling when actions execute
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary in SWAIG function response format |
logging_config
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Central logging configuration for SignalWire Agents SDK
This module provides a single point of control for all logging across the SDK. All components should use get_logger() instead of direct logging module usage or print() statements.
Uses structlog with stdlib integration so that: - All SDK loggers go through structlog processors (bind, JSON, colors) - pytest caplog still works (structlog writes through stdlib) - Third-party loggers are unaffected (handler on signalwire only)
strip_control_chars(logger, method_name, event_dict)
¶
Strip control characters from log event values to prevent log injection.
get_execution_mode()
¶
Determine the execution mode based on environment variables
Returns:
| Name | Type | Description |
|---|---|---|
str |
'server', 'cgi', 'lambda', 'google_cloud_function', 'azure_function', or 'unknown' |
reset_logging_configuration()
¶
Reset the logging configuration flag to allow reconfiguration
This is useful when environment variables change after initial configuration.
configure_logging()
¶
Configure logging system once, globally, based on environment variables
Environment Variables
SIGNALWIRE_LOG_MODE: off, stderr, default, auto SIGNALWIRE_LOG_LEVEL: debug, info, warning, error, critical SIGNALWIRE_LOG_FORMAT: console, json (default: console)
get_logger(name)
¶
Get a logger instance for the specified name with structured logging support
This is the single entry point for all logging in the SDK. All modules should use this instead of direct logging module usage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
Logger name, typically name |
required |
Returns:
| Type | Description |
|---|---|
|
A structlog BoundLogger that supports .bind(), .info(), .debug(), etc. |
mixins
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
__all__ = ['AIConfigMixin', 'AuthMixin', 'PromptMixin', 'ServerlessMixin', 'SkillMixin', 'StateMixin', 'ToolMixin', 'WebMixin']
module-attribute
¶
PromptMixin
¶
Bases: _HostTyped
Mixin class containing all prompt-related methods for AgentBase
contexts
property
¶
Get the ContextBuilder for this agent
Returns:
| Type | Description |
|---|---|
ContextBuilder
|
ContextBuilder instance for defining contexts |
define_contexts(contexts=None)
¶
Define contexts and steps for this agent (alternative to POM/prompt)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
Optional context configuration (dict or ContextBuilder) |
None
|
Returns:
| Type | Description |
|---|---|
Union[AgentBase, ContextBuilder]
|
ContextBuilder for method chaining if no contexts provided |
Note
Contexts can coexist with traditional prompts. The restriction is only that you can't mix POM sections with raw text in the main prompt.
reset_contexts()
¶
Remove all contexts, returning the agent to a no-contexts state.
This is a convenience wrapper around define_contexts().reset().
Use it in a dynamic config callback when you need to rebuild
contexts from scratch for a specific request.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example::
def on_dynamic_config(query, body, headers, agent):
if query.get("transfer"):
agent.reset_contexts()
ctx = agent.define_contexts().add_context("default")
ctx.add_step("route").set_text("Route the caller.")
set_prompt_text(text)
¶
Set the prompt as raw text instead of using POM
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The raw prompt text |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_post_prompt(text)
¶
Set the post-prompt text for summary generation
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The post-prompt text |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_prompt_pom(pom)
¶
Set the prompt as a POM dictionary
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pom
|
list[dict[str, Any]]
|
POM dictionary structure |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_section(title, body='', bullets=None, numbered=False, numbered_bullets=False, subsections=None)
¶
Add a section to the prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Optional section body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
numbered
|
bool
|
Whether this section should be numbered |
False
|
numbered_bullets
|
bool
|
Whether bullets should be numbered |
False
|
subsections
|
list[dict[str, Any]] | None
|
Optional list of subsection objects |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_to_section(title, body=None, bullet=None, bullets=None)
¶
Add content to an existing section (creating it if needed)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str | None
|
Optional text to append to section body |
None
|
bullet
|
str | None
|
Optional single bullet point to add |
None
|
bullets
|
list[str] | None
|
Optional list of bullet points to add |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_subsection(parent_title, title, body='', bullets=None)
¶
Add a subsection to an existing section (creating parent if needed)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent_title
|
str
|
Parent section title |
required |
title
|
str
|
Subsection title |
required |
body
|
str
|
Optional subsection body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_has_section(title)
¶
Check if a section exists in the prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if section exists, False otherwise |
get_prompt()
¶
Get the prompt for the agent
Returns:
| Type | Description |
|---|---|
str | list[dict[str, Any]]
|
Either a string prompt or a POM object as list of dicts |
get_post_prompt()
¶
Get the post-prompt for the agent
Returns:
| Type | Description |
|---|---|
str | None
|
Post-prompt text or None if not set |
ToolMixin
¶
Bases: _HostTyped
Mixin class containing all tool/function-related methods for AgentBase
define_tool(name, description, parameters, handler, secure=True, fillers=None, webhook_url=None, required=None, is_typed_handler=False, **swaig_fields)
¶
Define a SWAIG function the AI can call.
How this becomes a tool the model sees¶
SWAIG functions are NOT a separate concept from "LLM tools" — they ARE the tools, exposed to the model in exactly the same shape as native OpenAI / Anthropic tool calling. Under the hood, the SDK renders each SWAIG function into an OpenAI-format tool schema:
{"type": "function", "function": {
"name": "<your name>",
"description": "<your description>",
"parameters": {<your JSON schema>}
}}
That schema is sent to the model on every turn. The model reads
the description field and the per-parameter description
strings inside parameters to decide WHEN to call this tool and
HOW to fill in the arguments.
Treat description text as prompt engineering, not as developer documentation:
- Bad: "Lookup function"
- Bad: "Internal helper for account fetching"
-
Good: "Look up a customer's account details by their account number. Use this BEFORE quoting any account-specific information (balance, plan, status, billing date)."
-
Bad parameter: "the id"
- Good parameter: "The customer's account number, exactly 8 digits, no dashes or spaces. Ask the user if they don't provide it."
Vague descriptions are the #1 cause of "the model has the right tool but doesn't call it" failures. Be specific about WHEN to use the tool, what makes it the right choice over alternatives, and any preconditions on the arguments.
Tool count matters too: LLM tool selection accuracy degrades past ~7-8 simultaneously-active tools per call. If you have many tools, partition them across steps using Step.set_functions() so only the relevant subset is active at any moment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name. Must be unique across all SWAIG functions
on this agent. Becomes the |
required |
description
|
str
|
The function description as the LLM will read it. See above — this is prompt-engineered text, not internal docs. Explain what the tool does, when to use it, and what NOT to use it for if there are sibling tools the model might confuse it with. |
required |
parameters
|
dict[str, Any]
|
JSON Schema for the function's arguments. The
per-property |
required |
handler
|
Callable
|
Python callable invoked when the model calls this tool. Receives parsed args and the raw request data. Should return a FunctionResult or dict; anything else logs a warning and is stringified. |
required |
secure
|
bool
|
Whether to require SWAIG token validation on the webhook callback. |
True
|
fillers
|
dict[str, list[str]] | None
|
Optional dict mapping language codes to arrays of filler phrases the AI speaks while the function executes. Format: {"en-US": ["one moment...", "checking..."]} |
None
|
webhook_url
|
str | None
|
Optional external webhook URL. If set, the SDK does not handle the call locally — the model's tool call is forwarded to this URL. |
None
|
required
|
list[str] | None
|
Optional list of required parameter names. May also
be set inside |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters (auto-wrapped from inferred schema). |
False
|
**swaig_fields
|
Additional SWAIG-only fields (e.g. meta_data_token, web_hook_auth_user) included in the generated function definition. |
{}
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.define_tool( name="lookup_account", description=( "Look up a customer's account details by their account " "number. Use this BEFORE quoting any account-specific " "information. Do not call it for general questions about " "the product." ), parameters={ "type": "object", "properties": { "account_number": { "type": "string", "description": ( "The customer's 8-digit account number, " "no dashes. Ask the user if not provided." ), } }, "required": ["account_number"], }, handler=self.handle_lookup_account, )
register_swaig_function(function_dict)
¶
Register a raw SWAIG function dictionary (e.g., from DataMap.to_swaig_function())
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_dict
|
dict[str, Any]
|
Complete SWAIG function definition dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
tool(name=None, **kwargs)
classmethod
¶
Class method decorator for defining SWAIG tools
Used as:
@AgentBase.tool(name="example_function", parameters={...}) def example_function(self, param1): # ...
define_tools()
¶
Define the tools this agent can use
Returns:
| Type | Description |
|---|---|
list[SWAIGFunction | dict[str, Any]]
|
List of SWAIGFunction objects or raw dictionaries (for data_map tools) |
This method can be overridden by subclasses.
on_function_call(name, args, raw_data=None)
¶
Called when a SWAIG function is invoked
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
args
|
dict[str, Any]
|
Function arguments |
required |
raw_data
|
dict[str, Any] | None
|
Raw request data |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Function result |
WebMixin
¶
Bases: _HostTyped
Mixin class containing all web server and routing-related methods for AgentBase
get_app()
¶
Get the FastAPI application instance for deployment adapters like Lambda/Mangum
This method ensures the FastAPI app is properly initialized and configured, then returns it for use with deployment adapters like Mangum for AWS Lambda.
Returns:
| Name | Type | Description |
|---|---|---|
FastAPI |
The configured FastAPI application instance |
as_router()
¶
Get a FastAPI router for this agent
Returns:
| Type | Description |
|---|---|
APIRouter
|
FastAPI router |
serve(host=None, port=None)
¶
Start a web server for this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str | None
|
Optional host to override the default |
None
|
port
|
int | None
|
Optional port to override the default |
None
|
run(event=None, context=None, force_mode=None, host=None, port=None)
¶
Smart run method that automatically detects environment and handles accordingly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
Serverless event object (Lambda, Cloud Functions) |
None
|
|
context
|
Serverless context object (Lambda, Cloud Functions) |
None
|
|
force_mode
|
Override automatic mode detection for testing |
None
|
|
host
|
str | None
|
Host override for server mode |
None
|
port
|
int | None
|
Port override for server mode |
None
|
Returns:
| Type | Description |
|---|---|
|
Response for serverless modes, None for server mode |
on_request(request_data=None, callback_path=None)
¶
Called when SWML is requested, with request data when available
This method overrides SWMLService's on_request to properly handle SWML generation for AI Agents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
None to use the default SWML rendering (which will call _render_swml) |
on_swml_request(request_data=None, callback_path=None, request=None)
¶
Customization point for subclasses to modify SWML based on request data
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
request
|
Request | None
|
Optional FastAPI Request object for accessing query params, headers, etc. |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
Optional dict with modifications to apply to the SWML document |
register_routing_callback(callback_fn, path='/sip')
¶
Register a callback function that will be called to determine routing based on POST data.
When a routing callback is registered, an endpoint at the specified path is automatically created that will handle requests. This endpoint will use the callback to determine if the request should be processed by this service or redirected.
The callback should take a request object and request body dictionary and return: - A route string if it should be routed to a different endpoint - None if normal processing should continue
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback_fn
|
Callable[[Request, dict[str, Any]], str | None]
|
The callback function to register |
required |
path
|
str
|
The path where this callback should be registered (default: "/sip") |
'/sip'
|
set_dynamic_config_callback(callback)
¶
Set a callback function for dynamic agent configuration
This callback receives the actual agent instance, allowing you to dynamically configure ANY aspect of the agent including adding skills, modifying prompts, changing parameters, etc. based on request data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
Callable[[dict, dict, dict, AgentBase], None]
|
Function that takes (query_params, body_params, headers, agent) and configures the agent using any available methods like: - agent.add_skill(...) - agent.add_language(...) - agent.prompt_add_section(...) - agent.set_params(...) - agent.set_global_data(...) - agent.define_tool(...) |
required |
Example
def my_config(query_params, body_params, headers, agent): if query_params.get('tier') == 'premium': agent.add_skill("advanced_search") agent.add_language("English", "en-US", "premium_voice") agent.set_params({"end_of_speech_timeout": 500}) agent.set_global_data({"tier": query_params.get('tier', 'standard')})
my_agent.set_dynamic_config_callback(my_config)
manual_set_proxy_url(proxy_url)
¶
Manually set the proxy URL base for webhook callbacks
This can be called at runtime to set or update the proxy URL
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
proxy_url
|
str
|
The base URL to use for webhooks (e.g., https://example.ngrok.io) |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
setup_graceful_shutdown()
¶
Setup signal handlers for graceful shutdown (useful for Kubernetes)
AuthMixin
¶
Bases: _HostTyped
Mixin class containing all authentication-related methods for AgentBase
validate_basic_auth(username, password)
¶
Validate basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
username
|
str
|
Username from request |
required |
password
|
str
|
Password from request |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if valid, False otherwise |
This method can be overridden by subclasses.
get_basic_auth_credentials(include_source=False)
¶
Get the basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_source
|
bool
|
Whether to include the source of the credentials |
False
|
Returns:
| Type | Description |
|---|---|
tuple[str, str] | tuple[str, str, str]
|
If include_source is False: (username, password) tuple |
tuple[str, str] | tuple[str, str, str]
|
If include_source is True: (username, password, source) tuple, where source is one of: "provided", "environment", or "generated" |
SkillMixin
¶
Bases: _HostTyped
Mixin class containing all skill management methods for AgentBase
add_skill(skill_name, params=None)
¶
Add a skill to this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skill_name
|
str
|
Name of the skill to add |
required |
params
|
dict[str, Any] | None
|
Optional parameters to pass to the skill for configuration |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If skill not found or failed to load with detailed error message |
remove_skill(skill_name)
¶
Remove a skill from this agent
list_skills()
¶
List currently loaded skills
has_skill(skill_name)
¶
Check if skill is loaded
AIConfigMixin
¶
Bases: _HostTyped
Mixin class containing all AI configuration methods for AgentBase
SUPPORTED_INTERNAL_FILLER_NAMES = frozenset({'hangup', 'check_time', 'wait_for_user', 'wait_seconds', 'adjust_response_latency', 'next_step', 'change_context', 'get_visual_input', 'get_ideal_strategy'})
class-attribute
instance-attribute
¶
add_hint(hint)
¶
Add a simple string hint to help the AI agent understand certain words better
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hint
|
str
|
The hint string to add |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_hints(hints)
¶
Add multiple string hints
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hints
|
list[str]
|
List of hint strings |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_pattern_hint(hint, pattern, replace, ignore_case=False)
¶
Add a complex hint with pattern matching
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hint
|
str
|
The hint to match |
required |
pattern
|
str
|
Regular expression pattern |
required |
replace
|
str
|
Text to replace the hint with |
required |
ignore_case
|
bool
|
Whether to ignore case when matching |
False
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_language(name, code, voice, speech_fillers=None, function_fillers=None, engine=None, model=None, params=None)
¶
Add a language configuration to support multilingual conversations
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of the language (e.g., "English", "French") |
required |
code
|
str
|
Language code (e.g., "en-US", "fr-FR") |
required |
voice
|
str
|
TTS voice to use. Can be a simple name (e.g., "en-US-Neural2-F") or a combined format "engine.voice:model" (e.g., "elevenlabs.josh:eleven_turbo_v2_5") |
required |
speech_fillers
|
list[str] | None
|
Optional list of filler phrases for natural speech |
None
|
function_fillers
|
list[str] | None
|
Optional list of filler phrases during function calls |
None
|
engine
|
str | None
|
Optional explicit engine name (e.g., "elevenlabs", "rime") |
None
|
model
|
str | None
|
Optional explicit model name (e.g., "eleven_turbo_v2_5", "arcana") |
None
|
params
|
dict[str, Any] | None
|
Optional per-language params dict (engine-specific tuning,
voice settings, etc.). Emitted as the language object's
|
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Examples:
Simple voice name¶
agent.add_language("English", "en-US", "en-US-Neural2-F")
Explicit parameters¶
agent.add_language("English", "en-US", "josh", engine="elevenlabs", model="eleven_turbo_v2_5")
Combined format¶
agent.add_language("English", "en-US", "elevenlabs.josh:eleven_turbo_v2_5")
Per-language params (engine-specific knobs)¶
agent.add_language("English", "en-US", "josh", engine="elevenlabs", params={"stability": 0.5, "similarity_boost": 0.75})
set_language_params(code, params)
¶
Set (or replace) the per-language params dict on an already-added
language. Useful when language entries are built up via add_language()
first and engine-specific tuning is added later (e.g., from a config
loader). Returns self for chaining.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Language code as previously passed to |
required |
params
|
dict[str, Any]
|
Engine-specific params dict to attach. Empty dict removes the key. |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. No-op if the code isn't found. |
get_language_params(code)
¶
Read the per-language params dict for a previously-added language.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Language code as previously passed to |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
The params dict if set, |
set_languages(languages)
¶
Set all language configurations at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
languages
|
list[dict[str, Any]]
|
List of language configuration dictionaries |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_pronunciation(replace, with_text, ignore_case=False)
¶
Add a pronunciation rule to help the AI speak certain words correctly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
replace
|
str
|
The expression to replace |
required |
with_text
|
str
|
The phonetic spelling to use instead |
required |
ignore_case
|
bool
|
Whether to ignore case when matching |
False
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_pronunciations(pronunciations)
¶
Set all pronunciation rules at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pronunciations
|
list[dict[str, Any]]
|
List of pronunciation rule dictionaries |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_param(key, value)
¶
Set a single AI parameter
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Parameter name |
required |
value
|
Any
|
Parameter value |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_params(params)
¶
Set multiple AI parameters at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
params
|
dict[str, Any]
|
Dictionary of parameter name/value pairs |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_global_data(data)
¶
Merge data into the global data available to the AI throughout the conversation.
This merges (not replaces) so that skills and other callers can each contribute keys without clobbering each other.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of global data to merge |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
update_global_data(data)
¶
Update the global data with new values
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of global data to update |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_native_functions(function_names)
¶
Set the list of native functions to enable
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_names
|
list[str]
|
List of native function names |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_internal_fillers(internal_fillers)
¶
Set internal fillers for native SWAIG functions.
Internal fillers are short phrases the AI agent speaks (via TTS) while an internal/native function is running, so the caller doesn't hear dead air during transitions or background work.
Supported function names (matches the SWAIGInternalFiller schema):
hangup — when the agent is hanging up
check_time — when checking the time
wait_for_user — when waiting for user input
wait_seconds — during deliberate pauses
adjust_response_latency — when adjusting response timing
next_step — transitioning between steps in prompt.contexts
change_context — switching between contexts in prompt.contexts
get_visual_input — processing visual input (enable_vision=True)
get_ideal_strategy — thinking (enable_thinking=True)
Notably NOT supported: change_step, gather_submit, or arbitrary user-defined SWAIG function names. The runtime only honors fillers for the names listed above; everything else is silently ignored at the SWML level. This method warns at registration time if you pass an unknown name so you catch the typo early.
Without internal fillers, callers hear dead air during context switches and other internal transitions — set this on every agent that uses contexts/steps with audio.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
internal_fillers
|
dict[str, dict[str, list[str]]]
|
Dictionary mapping function names to language-specific filler phrases. Format: {"function_name": {"language_code": ["phrase1", "phrase2"]}} |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.set_internal_fillers({ "next_step": { "en-US": ["Moving to the next step...", "Great, let's continue..."], "es": ["Pasando al siguiente paso...", "Excelente, continuemos..."] }, "change_context": { "en-US": ["Let me switch modes...", "One moment..."] }, "check_time": { "en-US": ["Let me check the time...", "Getting the current time..."] } })
add_internal_filler(function_name, language_code, fillers)
¶
Add internal fillers for a single internal function and language.
See set_internal_fillers() for the complete list of supported function_name values and an explanation of what fillers do.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
One of the supported internal function names (see SUPPORTED_INTERNAL_FILLER_NAMES). Common values: 'next_step', 'change_context', 'check_time', 'wait_for_user', 'hangup'. Names outside the supported set log a warning and are ignored by the runtime. |
required |
language_code
|
str
|
Language code (e.g. 'en-US', 'es', 'fr'). |
required |
fillers
|
list[str]
|
List of filler phrases for this function and language. |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.add_internal_filler( "change_context", "en-US", ["Let me switch over to that...", "One moment..."] )
enable_debug_events(level=1)
¶
Enable 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:
| Name | Type | Description | Default |
|---|---|---|---|
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) |
1
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent = AgentBase("my_agent") agent.enable_debug_events(level=1)
@agent.on_debug_event def handle_debug(event_type, data): if event_type == "llm_error": alert_ops_team(data)
add_function_include(url, functions, meta_data=None)
¶
Add a remote function include to the SWAIG configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
URL to fetch remote functions from |
required |
functions
|
list[str]
|
List of function names to include |
required |
meta_data
|
dict[str, Any] | None
|
Optional metadata to include with the function include |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_function_includes(includes)
¶
Set the complete list of function includes
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
includes
|
list[dict[str, Any]]
|
List of include objects, each with url and functions properties |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_mcp_server(url, headers=None, resources=False, resource_vars=None)
¶
Add an external MCP server for tool discovery and invocation.
Tools are discovered via the MCP protocol at session start and registered as SWAIG functions. Resources are optionally fetched into global_data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
MCP server HTTP endpoint URL |
required |
headers
|
dict[str, str] | None
|
Optional HTTP headers (e.g. {"Authorization": "Bearer sk-xxx"}) |
None
|
resources
|
bool
|
Whether to fetch resources into global_data |
False
|
resource_vars
|
dict[str, str] | None
|
Variables for URI template substitution (e.g. {"caller_id": "${caller_id_number}"}) |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
enable_mcp_server()
¶
Expose this agent's @tool functions as an MCP server endpoint.
Adds a /mcp route that speaks JSON-RPC 2.0 (MCP protocol). Other MCP clients (Claude Desktop, other agents, etc.) can connect and use the same tools. The agent's SWML output also references this endpoint for native MCP tool discovery.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_prompt_llm_params(**params)
¶
Set LLM parameters for the main prompt.
Accepts any parameters which will be passed through to the SignalWire server. The server will validate and apply parameters based on the target model's capabilities.
Common parameters include
model: The AI model to use (gpt-4o-mini, gpt-4.1-mini, gpt-4.1-nano, nova-micro, nova-lite, qwen3-235b-A22b-instruct) temperature: Randomness setting. Lower values make output more deterministic. top_p: Alternative to temperature. Controls nucleus sampling. barge_confidence: ASR confidence to interrupt. Higher values make it harder to interrupt. presence_penalty: Topic diversity. Positive values encourage new topics. frequency_penalty: Repetition control. Positive values reduce repetition.
Note: Parameters are model-specific and will be validated by the server. Invalid parameters for the selected model will be handled/ignored by the server.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent.set_prompt_llm_params( model="nova-micro", # Using Amazon's nova-micro model temperature=0.7, top_p=0.9, barge_confidence=0.6 )
set_post_prompt_llm_params(**params)
¶
Set LLM parameters for the post-prompt.
Accepts any parameters which will be passed through to the SignalWire server. The server will validate and apply parameters based on the target model's capabilities.
Common parameters include
model: The AI model to use (gpt-4o-mini, gpt-4.1-mini, gpt-4.1-nano, nova-micro, nova-lite, qwen3-235b-A22b-instruct) temperature: Randomness setting. Lower values make output more deterministic. top_p: Alternative to temperature. Controls nucleus sampling. presence_penalty: Topic diversity. Positive values encourage new topics. frequency_penalty: Repetition control. Positive values reduce repetition.
Note: Parameters are model-specific and will be validated by the server. Invalid parameters for the selected model will be handled/ignored by the server. barge_confidence is not applicable to post-prompt.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent.set_post_prompt_llm_params( model="gpt-4o-mini", temperature=0.5, # More deterministic for post-prompt top_p=0.9 )
ServerlessMixin
¶
Bases: _HostTyped
Mixin class containing all serverless/cloud platform methods for AgentBase
handle_serverless_request(event=None, context=None, mode=None)
¶
Handle serverless environment requests (CGI, Lambda, Cloud Functions)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
Serverless event object (Lambda, Cloud Functions) |
None
|
|
context
|
Serverless context object (Lambda, Cloud Functions) |
None
|
|
mode
|
Override execution mode (from force_mode in run()) |
None
|
Returns:
| Type | Description |
|---|---|
|
Response appropriate for the serverless platform |
StateMixin
¶
Bases: _HostTyped
Mixin class containing all state and session management methods for AgentBase
validate_tool_token(function_name, token, call_id)
¶
Validate a tool token
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the function/tool |
required |
token
|
str
|
Token to validate |
required |
call_id
|
str
|
Call ID for the session |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if token is valid, False otherwise |
ai_config_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
AIConfigMixin
¶
Bases: _HostTyped
Mixin class containing all AI configuration methods for AgentBase
SUPPORTED_INTERNAL_FILLER_NAMES = frozenset({'hangup', 'check_time', 'wait_for_user', 'wait_seconds', 'adjust_response_latency', 'next_step', 'change_context', 'get_visual_input', 'get_ideal_strategy'})
class-attribute
instance-attribute
¶
add_hint(hint)
¶
Add a simple string hint to help the AI agent understand certain words better
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hint
|
str
|
The hint string to add |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_hints(hints)
¶
Add multiple string hints
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hints
|
list[str]
|
List of hint strings |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_pattern_hint(hint, pattern, replace, ignore_case=False)
¶
Add a complex hint with pattern matching
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hint
|
str
|
The hint to match |
required |
pattern
|
str
|
Regular expression pattern |
required |
replace
|
str
|
Text to replace the hint with |
required |
ignore_case
|
bool
|
Whether to ignore case when matching |
False
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_language(name, code, voice, speech_fillers=None, function_fillers=None, engine=None, model=None, params=None)
¶
Add a language configuration to support multilingual conversations
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of the language (e.g., "English", "French") |
required |
code
|
str
|
Language code (e.g., "en-US", "fr-FR") |
required |
voice
|
str
|
TTS voice to use. Can be a simple name (e.g., "en-US-Neural2-F") or a combined format "engine.voice:model" (e.g., "elevenlabs.josh:eleven_turbo_v2_5") |
required |
speech_fillers
|
list[str] | None
|
Optional list of filler phrases for natural speech |
None
|
function_fillers
|
list[str] | None
|
Optional list of filler phrases during function calls |
None
|
engine
|
str | None
|
Optional explicit engine name (e.g., "elevenlabs", "rime") |
None
|
model
|
str | None
|
Optional explicit model name (e.g., "eleven_turbo_v2_5", "arcana") |
None
|
params
|
dict[str, Any] | None
|
Optional per-language params dict (engine-specific tuning,
voice settings, etc.). Emitted as the language object's
|
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Examples:
Simple voice name¶
agent.add_language("English", "en-US", "en-US-Neural2-F")
Explicit parameters¶
agent.add_language("English", "en-US", "josh", engine="elevenlabs", model="eleven_turbo_v2_5")
Combined format¶
agent.add_language("English", "en-US", "elevenlabs.josh:eleven_turbo_v2_5")
Per-language params (engine-specific knobs)¶
agent.add_language("English", "en-US", "josh", engine="elevenlabs", params={"stability": 0.5, "similarity_boost": 0.75})
set_language_params(code, params)
¶
Set (or replace) the per-language params dict on an already-added
language. Useful when language entries are built up via add_language()
first and engine-specific tuning is added later (e.g., from a config
loader). Returns self for chaining.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Language code as previously passed to |
required |
params
|
dict[str, Any]
|
Engine-specific params dict to attach. Empty dict removes the key. |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. No-op if the code isn't found. |
get_language_params(code)
¶
Read the per-language params dict for a previously-added language.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Language code as previously passed to |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
The params dict if set, |
set_languages(languages)
¶
Set all language configurations at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
languages
|
list[dict[str, Any]]
|
List of language configuration dictionaries |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_pronunciation(replace, with_text, ignore_case=False)
¶
Add a pronunciation rule to help the AI speak certain words correctly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
replace
|
str
|
The expression to replace |
required |
with_text
|
str
|
The phonetic spelling to use instead |
required |
ignore_case
|
bool
|
Whether to ignore case when matching |
False
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_pronunciations(pronunciations)
¶
Set all pronunciation rules at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pronunciations
|
list[dict[str, Any]]
|
List of pronunciation rule dictionaries |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_param(key, value)
¶
Set a single AI parameter
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Parameter name |
required |
value
|
Any
|
Parameter value |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_params(params)
¶
Set multiple AI parameters at once
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
params
|
dict[str, Any]
|
Dictionary of parameter name/value pairs |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_global_data(data)
¶
Merge data into the global data available to the AI throughout the conversation.
This merges (not replaces) so that skills and other callers can each contribute keys without clobbering each other.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of global data to merge |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
update_global_data(data)
¶
Update the global data with new values
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary of global data to update |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_native_functions(function_names)
¶
Set the list of native functions to enable
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_names
|
list[str]
|
List of native function names |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_internal_fillers(internal_fillers)
¶
Set internal fillers for native SWAIG functions.
Internal fillers are short phrases the AI agent speaks (via TTS) while an internal/native function is running, so the caller doesn't hear dead air during transitions or background work.
Supported function names (matches the SWAIGInternalFiller schema):
hangup — when the agent is hanging up
check_time — when checking the time
wait_for_user — when waiting for user input
wait_seconds — during deliberate pauses
adjust_response_latency — when adjusting response timing
next_step — transitioning between steps in prompt.contexts
change_context — switching between contexts in prompt.contexts
get_visual_input — processing visual input (enable_vision=True)
get_ideal_strategy — thinking (enable_thinking=True)
Notably NOT supported: change_step, gather_submit, or arbitrary user-defined SWAIG function names. The runtime only honors fillers for the names listed above; everything else is silently ignored at the SWML level. This method warns at registration time if you pass an unknown name so you catch the typo early.
Without internal fillers, callers hear dead air during context switches and other internal transitions — set this on every agent that uses contexts/steps with audio.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
internal_fillers
|
dict[str, dict[str, list[str]]]
|
Dictionary mapping function names to language-specific filler phrases. Format: {"function_name": {"language_code": ["phrase1", "phrase2"]}} |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.set_internal_fillers({ "next_step": { "en-US": ["Moving to the next step...", "Great, let's continue..."], "es": ["Pasando al siguiente paso...", "Excelente, continuemos..."] }, "change_context": { "en-US": ["Let me switch modes...", "One moment..."] }, "check_time": { "en-US": ["Let me check the time...", "Getting the current time..."] } })
add_internal_filler(function_name, language_code, fillers)
¶
Add internal fillers for a single internal function and language.
See set_internal_fillers() for the complete list of supported function_name values and an explanation of what fillers do.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
One of the supported internal function names (see SUPPORTED_INTERNAL_FILLER_NAMES). Common values: 'next_step', 'change_context', 'check_time', 'wait_for_user', 'hangup'. Names outside the supported set log a warning and are ignored by the runtime. |
required |
language_code
|
str
|
Language code (e.g. 'en-US', 'es', 'fr'). |
required |
fillers
|
list[str]
|
List of filler phrases for this function and language. |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.add_internal_filler( "change_context", "en-US", ["Let me switch over to that...", "One moment..."] )
enable_debug_events(level=1)
¶
Enable 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:
| Name | Type | Description | Default |
|---|---|---|---|
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) |
1
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent = AgentBase("my_agent") agent.enable_debug_events(level=1)
@agent.on_debug_event def handle_debug(event_type, data): if event_type == "llm_error": alert_ops_team(data)
add_function_include(url, functions, meta_data=None)
¶
Add a remote function include to the SWAIG configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
URL to fetch remote functions from |
required |
functions
|
list[str]
|
List of function names to include |
required |
meta_data
|
dict[str, Any] | None
|
Optional metadata to include with the function include |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_function_includes(includes)
¶
Set the complete list of function includes
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
includes
|
list[dict[str, Any]]
|
List of include objects, each with url and functions properties |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
add_mcp_server(url, headers=None, resources=False, resource_vars=None)
¶
Add an external MCP server for tool discovery and invocation.
Tools are discovered via the MCP protocol at session start and registered as SWAIG functions. Resources are optionally fetched into global_data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
MCP server HTTP endpoint URL |
required |
headers
|
dict[str, str] | None
|
Optional HTTP headers (e.g. {"Authorization": "Bearer sk-xxx"}) |
None
|
resources
|
bool
|
Whether to fetch resources into global_data |
False
|
resource_vars
|
dict[str, str] | None
|
Variables for URI template substitution (e.g. {"caller_id": "${caller_id_number}"}) |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
enable_mcp_server()
¶
Expose this agent's @tool functions as an MCP server endpoint.
Adds a /mcp route that speaks JSON-RPC 2.0 (MCP protocol). Other MCP clients (Claude Desktop, other agents, etc.) can connect and use the same tools. The agent's SWML output also references this endpoint for native MCP tool discovery.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_prompt_llm_params(**params)
¶
Set LLM parameters for the main prompt.
Accepts any parameters which will be passed through to the SignalWire server. The server will validate and apply parameters based on the target model's capabilities.
Common parameters include
model: The AI model to use (gpt-4o-mini, gpt-4.1-mini, gpt-4.1-nano, nova-micro, nova-lite, qwen3-235b-A22b-instruct) temperature: Randomness setting. Lower values make output more deterministic. top_p: Alternative to temperature. Controls nucleus sampling. barge_confidence: ASR confidence to interrupt. Higher values make it harder to interrupt. presence_penalty: Topic diversity. Positive values encourage new topics. frequency_penalty: Repetition control. Positive values reduce repetition.
Note: Parameters are model-specific and will be validated by the server. Invalid parameters for the selected model will be handled/ignored by the server.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent.set_prompt_llm_params( model="nova-micro", # Using Amazon's nova-micro model temperature=0.7, top_p=0.9, barge_confidence=0.6 )
set_post_prompt_llm_params(**params)
¶
Set LLM parameters for the post-prompt.
Accepts any parameters which will be passed through to the SignalWire server. The server will validate and apply parameters based on the target model's capabilities.
Common parameters include
model: The AI model to use (gpt-4o-mini, gpt-4.1-mini, gpt-4.1-nano, nova-micro, nova-lite, qwen3-235b-A22b-instruct) temperature: Randomness setting. Lower values make output more deterministic. top_p: Alternative to temperature. Controls nucleus sampling. presence_penalty: Topic diversity. Positive values encourage new topics. frequency_penalty: Repetition control. Positive values reduce repetition.
Note: Parameters are model-specific and will be validated by the server. Invalid parameters for the selected model will be handled/ignored by the server. barge_confidence is not applicable to post-prompt.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Example
agent.set_post_prompt_llm_params( model="gpt-4o-mini", temperature=0.5, # More deterministic for post-prompt top_p=0.9 )
auth_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
AuthMixin
¶
Bases: _HostTyped
Mixin class containing all authentication-related methods for AgentBase
validate_basic_auth(username, password)
¶
Validate basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
username
|
str
|
Username from request |
required |
password
|
str
|
Password from request |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if valid, False otherwise |
This method can be overridden by subclasses.
get_basic_auth_credentials(include_source=False)
¶
Get the basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_source
|
bool
|
Whether to include the source of the credentials |
False
|
Returns:
| Type | Description |
|---|---|
tuple[str, str] | tuple[str, str, str]
|
If include_source is False: (username, password) tuple |
tuple[str, str] | tuple[str, str, str]
|
If include_source is True: (username, password, source) tuple, where source is one of: "provided", "environment", or "generated" |
mcp_server_mixin
¶
Copyright (c) 2025 SignalWire
MCP Server Mixin for AgentBase
Exposes @tool decorated functions as an MCP server endpoint at /mcp. Handles the MCP JSON-RPC 2.0 protocol: initialize, tools/list, tools/call.
prompt_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
PromptMixin
¶
Bases: _HostTyped
Mixin class containing all prompt-related methods for AgentBase
contexts
property
¶
Get the ContextBuilder for this agent
Returns:
| Type | Description |
|---|---|
ContextBuilder
|
ContextBuilder instance for defining contexts |
define_contexts(contexts=None)
¶
Define contexts and steps for this agent (alternative to POM/prompt)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
contexts
|
Optional context configuration (dict or ContextBuilder) |
None
|
Returns:
| Type | Description |
|---|---|
Union[AgentBase, ContextBuilder]
|
ContextBuilder for method chaining if no contexts provided |
Note
Contexts can coexist with traditional prompts. The restriction is only that you can't mix POM sections with raw text in the main prompt.
reset_contexts()
¶
Remove all contexts, returning the agent to a no-contexts state.
This is a convenience wrapper around define_contexts().reset().
Use it in a dynamic config callback when you need to rebuild
contexts from scratch for a specific request.
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example::
def on_dynamic_config(query, body, headers, agent):
if query.get("transfer"):
agent.reset_contexts()
ctx = agent.define_contexts().add_context("default")
ctx.add_step("route").set_text("Route the caller.")
set_prompt_text(text)
¶
Set the prompt as raw text instead of using POM
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The raw prompt text |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_post_prompt(text)
¶
Set the post-prompt text for summary generation
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The post-prompt text |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
set_prompt_pom(pom)
¶
Set the prompt as a POM dictionary
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pom
|
list[dict[str, Any]]
|
POM dictionary structure |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_section(title, body='', bullets=None, numbered=False, numbered_bullets=False, subsections=None)
¶
Add a section to the prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Optional section body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
numbered
|
bool
|
Whether this section should be numbered |
False
|
numbered_bullets
|
bool
|
Whether bullets should be numbered |
False
|
subsections
|
list[dict[str, Any]] | None
|
Optional list of subsection objects |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_to_section(title, body=None, bullet=None, bullets=None)
¶
Add content to an existing section (creating it if needed)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str | None
|
Optional text to append to section body |
None
|
bullet
|
str | None
|
Optional single bullet point to add |
None
|
bullets
|
list[str] | None
|
Optional list of bullet points to add |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_add_subsection(parent_title, title, body='', bullets=None)
¶
Add a subsection to an existing section (creating parent if needed)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent_title
|
str
|
Parent section title |
required |
title
|
str
|
Subsection title |
required |
body
|
str
|
Optional subsection body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
prompt_has_section(title)
¶
Check if a section exists in the prompt
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if section exists, False otherwise |
get_prompt()
¶
Get the prompt for the agent
Returns:
| Type | Description |
|---|---|
str | list[dict[str, Any]]
|
Either a string prompt or a POM object as list of dicts |
get_post_prompt()
¶
Get the post-prompt for the agent
Returns:
| Type | Description |
|---|---|
str | None
|
Post-prompt text or None if not set |
serverless_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
MAX_CGI_BODY_SIZE = 10 * 1024 * 1024
module-attribute
¶
ServerlessMixin
¶
Bases: _HostTyped
Mixin class containing all serverless/cloud platform methods for AgentBase
handle_serverless_request(event=None, context=None, mode=None)
¶
Handle serverless environment requests (CGI, Lambda, Cloud Functions)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
Serverless event object (Lambda, Cloud Functions) |
None
|
|
context
|
Serverless context object (Lambda, Cloud Functions) |
None
|
|
mode
|
Override execution mode (from force_mode in run()) |
None
|
Returns:
| Type | Description |
|---|---|
|
Response appropriate for the serverless platform |
skill_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SkillMixin
¶
Bases: _HostTyped
Mixin class containing all skill management methods for AgentBase
add_skill(skill_name, params=None)
¶
Add a skill to this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skill_name
|
str
|
Name of the skill to add |
required |
params
|
dict[str, Any] | None
|
Optional parameters to pass to the skill for configuration |
None
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
Raises:
| Type | Description |
|---|---|
ValueError
|
If skill not found or failed to load with detailed error message |
remove_skill(skill_name)
¶
Remove a skill from this agent
list_skills()
¶
List currently loaded skills
has_skill(skill_name)
¶
Check if skill is loaded
state_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
StateMixin
¶
Bases: _HostTyped
Mixin class containing all state and session management methods for AgentBase
validate_tool_token(function_name, token, call_id)
¶
Validate a tool token
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the function/tool |
required |
token
|
str
|
Token to validate |
required |
call_id
|
str
|
Call ID for the session |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if token is valid, False otherwise |
tool_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
ToolMixin
¶
Bases: _HostTyped
Mixin class containing all tool/function-related methods for AgentBase
define_tool(name, description, parameters, handler, secure=True, fillers=None, webhook_url=None, required=None, is_typed_handler=False, **swaig_fields)
¶
Define a SWAIG function the AI can call.
How this becomes a tool the model sees¶
SWAIG functions are NOT a separate concept from "LLM tools" — they ARE the tools, exposed to the model in exactly the same shape as native OpenAI / Anthropic tool calling. Under the hood, the SDK renders each SWAIG function into an OpenAI-format tool schema:
{"type": "function", "function": {
"name": "<your name>",
"description": "<your description>",
"parameters": {<your JSON schema>}
}}
That schema is sent to the model on every turn. The model reads
the description field and the per-parameter description
strings inside parameters to decide WHEN to call this tool and
HOW to fill in the arguments.
Treat description text as prompt engineering, not as developer documentation:
- Bad: "Lookup function"
- Bad: "Internal helper for account fetching"
-
Good: "Look up a customer's account details by their account number. Use this BEFORE quoting any account-specific information (balance, plan, status, billing date)."
-
Bad parameter: "the id"
- Good parameter: "The customer's account number, exactly 8 digits, no dashes or spaces. Ask the user if they don't provide it."
Vague descriptions are the #1 cause of "the model has the right tool but doesn't call it" failures. Be specific about WHEN to use the tool, what makes it the right choice over alternatives, and any preconditions on the arguments.
Tool count matters too: LLM tool selection accuracy degrades past ~7-8 simultaneously-active tools per call. If you have many tools, partition them across steps using Step.set_functions() so only the relevant subset is active at any moment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name. Must be unique across all SWAIG functions
on this agent. Becomes the |
required |
description
|
str
|
The function description as the LLM will read it. See above — this is prompt-engineered text, not internal docs. Explain what the tool does, when to use it, and what NOT to use it for if there are sibling tools the model might confuse it with. |
required |
parameters
|
dict[str, Any]
|
JSON Schema for the function's arguments. The
per-property |
required |
handler
|
Callable
|
Python callable invoked when the model calls this tool. Receives parsed args and the raw request data. Should return a FunctionResult or dict; anything else logs a warning and is stringified. |
required |
secure
|
bool
|
Whether to require SWAIG token validation on the webhook callback. |
True
|
fillers
|
dict[str, list[str]] | None
|
Optional dict mapping language codes to arrays of filler phrases the AI speaks while the function executes. Format: {"en-US": ["one moment...", "checking..."]} |
None
|
webhook_url
|
str | None
|
Optional external webhook URL. If set, the SDK does not handle the call locally — the model's tool call is forwarded to this URL. |
None
|
required
|
list[str] | None
|
Optional list of required parameter names. May also
be set inside |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters (auto-wrapped from inferred schema). |
False
|
**swaig_fields
|
Additional SWAIG-only fields (e.g. meta_data_token, web_hook_auth_user) included in the generated function definition. |
{}
|
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining. |
Example
agent.define_tool( name="lookup_account", description=( "Look up a customer's account details by their account " "number. Use this BEFORE quoting any account-specific " "information. Do not call it for general questions about " "the product." ), parameters={ "type": "object", "properties": { "account_number": { "type": "string", "description": ( "The customer's 8-digit account number, " "no dashes. Ask the user if not provided." ), } }, "required": ["account_number"], }, handler=self.handle_lookup_account, )
register_swaig_function(function_dict)
¶
Register a raw SWAIG function dictionary (e.g., from DataMap.to_swaig_function())
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_dict
|
dict[str, Any]
|
Complete SWAIG function definition dictionary |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
tool(name=None, **kwargs)
classmethod
¶
Class method decorator for defining SWAIG tools
Used as:
@AgentBase.tool(name="example_function", parameters={...}) def example_function(self, param1): # ...
define_tools()
¶
Define the tools this agent can use
Returns:
| Type | Description |
|---|---|
list[SWAIGFunction | dict[str, Any]]
|
List of SWAIGFunction objects or raw dictionaries (for data_map tools) |
This method can be overridden by subclasses.
on_function_call(name, args, raw_data=None)
¶
Called when a SWAIG function is invoked
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name |
required |
args
|
dict[str, Any]
|
Function arguments |
required |
raw_data
|
dict[str, Any] | None
|
Raw request data |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Function result |
web_mixin
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
MAX_REQUEST_BODY_SIZE = 10 * 1024 * 1024
module-attribute
¶
WebMixin
¶
Bases: _HostTyped
Mixin class containing all web server and routing-related methods for AgentBase
get_app()
¶
Get the FastAPI application instance for deployment adapters like Lambda/Mangum
This method ensures the FastAPI app is properly initialized and configured, then returns it for use with deployment adapters like Mangum for AWS Lambda.
Returns:
| Name | Type | Description |
|---|---|---|
FastAPI |
The configured FastAPI application instance |
as_router()
¶
Get a FastAPI router for this agent
Returns:
| Type | Description |
|---|---|
APIRouter
|
FastAPI router |
serve(host=None, port=None)
¶
Start a web server for this agent
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str | None
|
Optional host to override the default |
None
|
port
|
int | None
|
Optional port to override the default |
None
|
run(event=None, context=None, force_mode=None, host=None, port=None)
¶
Smart run method that automatically detects environment and handles accordingly
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
Serverless event object (Lambda, Cloud Functions) |
None
|
|
context
|
Serverless context object (Lambda, Cloud Functions) |
None
|
|
force_mode
|
Override automatic mode detection for testing |
None
|
|
host
|
str | None
|
Host override for server mode |
None
|
port
|
int | None
|
Port override for server mode |
None
|
Returns:
| Type | Description |
|---|---|
|
Response for serverless modes, None for server mode |
on_request(request_data=None, callback_path=None)
¶
Called when SWML is requested, with request data when available
This method overrides SWMLService's on_request to properly handle SWML generation for AI Agents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
None to use the default SWML rendering (which will call _render_swml) |
on_swml_request(request_data=None, callback_path=None, request=None)
¶
Customization point for subclasses to modify SWML based on request data
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
request
|
Request | None
|
Optional FastAPI Request object for accessing query params, headers, etc. |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
Optional dict with modifications to apply to the SWML document |
register_routing_callback(callback_fn, path='/sip')
¶
Register a callback function that will be called to determine routing based on POST data.
When a routing callback is registered, an endpoint at the specified path is automatically created that will handle requests. This endpoint will use the callback to determine if the request should be processed by this service or redirected.
The callback should take a request object and request body dictionary and return: - A route string if it should be routed to a different endpoint - None if normal processing should continue
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback_fn
|
Callable[[Request, dict[str, Any]], str | None]
|
The callback function to register |
required |
path
|
str
|
The path where this callback should be registered (default: "/sip") |
'/sip'
|
set_dynamic_config_callback(callback)
¶
Set a callback function for dynamic agent configuration
This callback receives the actual agent instance, allowing you to dynamically configure ANY aspect of the agent including adding skills, modifying prompts, changing parameters, etc. based on request data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
Callable[[dict, dict, dict, AgentBase], None]
|
Function that takes (query_params, body_params, headers, agent) and configures the agent using any available methods like: - agent.add_skill(...) - agent.add_language(...) - agent.prompt_add_section(...) - agent.set_params(...) - agent.set_global_data(...) - agent.define_tool(...) |
required |
Example
def my_config(query_params, body_params, headers, agent): if query_params.get('tier') == 'premium': agent.add_skill("advanced_search") agent.add_language("English", "en-US", "premium_voice") agent.set_params({"end_of_speech_timeout": 500}) agent.set_global_data({"tier": query_params.get('tier', 'standard')})
my_agent.set_dynamic_config_callback(my_config)
manual_set_proxy_url(proxy_url)
¶
Manually set the proxy URL base for webhook callbacks
This can be called at runtime to set or update the proxy URL
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
proxy_url
|
str
|
The base URL to use for webhooks (e.g., https://example.ngrok.io) |
required |
Returns:
| Type | Description |
|---|---|
AgentBase
|
Self for method chaining |
setup_graceful_shutdown()
¶
Setup signal handlers for graceful shutdown (useful for Kubernetes)
pom_builder
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
PomBuilder for creating structured POM prompts for SignalWire AI Agents
PomBuilder
¶
Builder class for creating structured prompts using the Prompt Object Model.
This class is a flexible wrapper around the POM API that allows for: - Dynamic creation of sections on demand - Adding content to existing sections - Nesting subsections - Rendering to markdown or XML
Unlike previous implementations, there are no predefined section types - you can create any section structure that fits your needs.
pom = PromptObjectModel()
instance-attribute
¶
__init__()
¶
Initialize a new POM builder with an empty POM
add_section(title, body='', bullets=None, numbered=False, numbered_bullets=False, subsections=None)
¶
Add a new section to the POM
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str
|
Optional body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
numbered
|
bool
|
Whether to number this section |
False
|
numbered_bullets
|
bool
|
Whether to number bullet points |
False
|
subsections
|
list[dict[str, Any]] | None
|
Optional list of subsection objects |
None
|
Returns:
| Type | Description |
|---|---|
PomBuilder
|
Self for method chaining |
add_to_section(title, body=None, bullet=None, bullets=None)
¶
Add content to an existing section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
body
|
str | None
|
Text to append to the section body |
None
|
bullet
|
str | None
|
Single bullet to add |
None
|
bullets
|
list[str] | None
|
List of bullets to add |
None
|
Returns:
| Type | Description |
|---|---|
PomBuilder
|
Self for method chaining |
add_subsection(parent_title, title, body='', bullets=None)
¶
Add a subsection to an existing section, creating the parent if needed
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parent_title
|
str
|
Title of the parent section |
required |
title
|
str
|
Title for the new subsection |
required |
body
|
str
|
Optional body text |
''
|
bullets
|
list[str] | None
|
Optional list of bullet points |
None
|
Returns:
| Type | Description |
|---|---|
PomBuilder
|
Self for method chaining |
has_section(title)
¶
Check if a section with the given title exists
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the section exists, False otherwise |
get_section(title)
¶
Get a section by title
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
title
|
str
|
Section title |
required |
Returns:
| Type | Description |
|---|---|
Section | None
|
Section object or None if not found |
render_markdown()
¶
Render the POM as markdown
render_xml()
¶
Render the POM as XML
to_dict()
¶
Convert the POM to a list of section dictionaries
to_json()
¶
Convert the POM to a JSON string
from_sections(sections)
classmethod
¶
Create a PomBuilder from a list of section dictionaries
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sections
|
list[dict[str, Any]]
|
List of section definition dictionaries |
required |
Returns:
| Type | Description |
|---|---|
PomBuilder
|
A new PomBuilder instance with the sections added |
security
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SIGNALWIRE_SIGNATURE_HEADER = 'x-signalwire-signature'
module-attribute
¶
TWILIO_COMPAT_SIGNATURE_HEADER = 'x-twilio-signature'
module-attribute
¶
SENSITIVE_HEADERS = frozenset({'authorization', 'cookie', 'x-api-key', 'proxy-authorization', 'set-cookie'})
module-attribute
¶
__all__ = ['SENSITIVE_HEADERS', 'SIGNALWIRE_SIGNATURE_HEADER', 'TWILIO_COMPAT_SIGNATURE_HEADER', 'filter_sensitive_headers', 'is_valid_hostname', 'make_webhook_validation_dependency', 'redact_url', 'validate_request', 'validate_webhook_signature']
module-attribute
¶
validate_request(signing_key, signature, url, params_or_raw_body)
¶
Legacy @signalwire/compatibility-api drop-in entry point.
If params_or_raw_body is a string, delegates to
:func:validate_webhook_signature (Scheme A then Scheme B with parsed form).
If it's a mapping or a list of (key, value) tuples, treats it as pre-parsed form params and runs Scheme B directly (with URL port normalization and optional bodySHA256 fallback).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
Customer's Signing Key. Missing raises |
required |
signature
|
str
|
Header value. Missing / empty returns False. |
required |
url
|
str
|
Full URL SignalWire POSTed to. |
required |
params_or_raw_body
|
str | Mapping[str, Any] | list[tuple[str, Any]] | None
|
|
required |
Returns:
| Type | Description |
|---|---|
bool
|
True on match, False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
when |
TypeError
|
when |
validate_webhook_signature(signing_key, signature, url, raw_body)
¶
Validate a SignalWire webhook signature against both schemes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
Customer's Signing Key from the Dashboard. UTF-8 string,
secret. |
required |
signature
|
str
|
The |
required |
url
|
str
|
The full URL SignalWire POSTed to (scheme, host, optional port,
path, query). Must match what the platform saw — see the
|
required |
raw_body
|
str
|
The raw request body bytes as a UTF-8 string, BEFORE any
JSON / form parsing. Must be a |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the signature matches either Scheme A (hex JSON) or Scheme B |
bool
|
(base64 form, with port-normalization variants and optional |
bool
|
bodySHA256 fallback). False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
when |
TypeError
|
when |
make_webhook_validation_dependency(signing_key, *, trust_proxy=False)
¶
Build a FastAPI dependency that enforces signature validation.
The returned coroutine:
- Reads
await request.body()and stashes the bytes onrequest.state.raw_body. - Pulls the
X-SignalWire-Signatureheader (or the Twilio alias). - Reconstructs the public URL (proxy headers / env / fallback).
- Calls :func:
validate_webhook_signature. - On invalid signature: raises
HTTPException(403)to short-circuit the handler. FastAPI'sdependencies=[Depends(...)]only honors short-circuiting via raised exceptions — returning a Response from a dependency does not stop the endpoint. - On valid: returns
Noneso the handler runs as normal.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
The customer's Signing Key. Required, non-empty. |
required |
trust_proxy
|
bool
|
If True, honor |
False
|
Returns:
| Type | Description |
|---|---|
Callable[[Request, Response], Awaitable[Response | None]]
|
Async callable suitable for |
Raises:
| Type | Description |
|---|---|
ValueError
|
at construction time if |
filter_sensitive_headers(headers)
¶
Return a copy of headers with sensitive (credential-bearing) headers
removed, so request headers can be safely passed to user callbacks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
headers
|
Mapping of header name -> value. |
required |
Returns:
| Type | Description |
|---|---|
|
A new dict containing only the non-sensitive headers (keys preserved |
|
|
as given; the sensitivity check is case-insensitive). |
redact_url(url)
¶
Mask the password in a URL's userinfo before logging.
https://user:secret@host/path -> https://user:****@host/path.
A URL with no embedded credentials is returned unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
The URL string (or any value; non-strings are returned as-is). |
required |
Returns:
| Type | Description |
|---|---|
|
The URL with any |
is_valid_hostname(host)
¶
Standalone hostname sanity check: reject empty hosts and any host containing whitespace, slashes, or control characters.
This is the reusable character-level check, independent of the full
:func:signalwire.utils.url_validator.validate_url (which also does scheme
checks, DNS resolution, and private-IP blocking). Callers that only need to
validate a hostname string use this.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
The hostname string. |
required |
Returns:
| Type | Description |
|---|---|
|
True if the hostname is non-empty and contains no whitespace/slashes/ |
|
|
control characters; False otherwise. |
security_utils
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Standalone security hygiene utilities.
These mirror the TypeScript SDK's SecurityUtils (filterSensitiveHeaders,
redactUrl, isValidHostname) so the same protections — keeping credentials out
of user callbacks and logs, reusable hostname validation — are available in the
Python reference and every port.
SENSITIVE_HEADERS = frozenset({'authorization', 'cookie', 'x-api-key', 'proxy-authorization', 'set-cookie'})
module-attribute
¶
filter_sensitive_headers(headers)
¶
Return a copy of headers with sensitive (credential-bearing) headers
removed, so request headers can be safely passed to user callbacks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
headers
|
Mapping of header name -> value. |
required |
Returns:
| Type | Description |
|---|---|
|
A new dict containing only the non-sensitive headers (keys preserved |
|
|
as given; the sensitivity check is case-insensitive). |
redact_url(url)
¶
Mask the password in a URL's userinfo before logging.
https://user:secret@host/path -> https://user:****@host/path.
A URL with no embedded credentials is returned unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
The URL string (or any value; non-strings are returned as-is). |
required |
Returns:
| Type | Description |
|---|---|
|
The URL with any |
is_valid_hostname(host)
¶
Standalone hostname sanity check: reject empty hosts and any host containing whitespace, slashes, or control characters.
This is the reusable character-level check, independent of the full
:func:signalwire.utils.url_validator.validate_url (which also does scheme
checks, DNS resolution, and private-IP blocking). Callers that only need to
validate a hostname string use this.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
The hostname string. |
required |
Returns:
| Type | Description |
|---|---|
|
True if the hostname is non-empty and contains no whitespace/slashes/ |
|
|
control characters; False otherwise. |
session_manager
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Session manager for handling call sessions and security tokens
SessionManager
¶
Manages security tokens for function calls
This implementation is completely stateless - it does not track call sessions or store any information in memory. All validation is done using cryptographic signatures with the tokens containing all necessary information.
token_expiry_secs = token_expiry_secs
instance-attribute
¶
secret_key = secret_key or secrets.token_hex(32)
instance-attribute
¶
__init__(token_expiry_secs=900, secret_key=None)
¶
Initialize the session manager
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_expiry_secs
|
int
|
Seconds until tokens expire (default: 15 minutes) |
900
|
secret_key
|
str | None
|
Secret key for signing tokens (generated if not provided) |
None
|
create_session(call_id=None)
¶
Create a new session ID if one isn't provided
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str | None
|
Optional call ID, generated if not provided |
None
|
Returns:
| Type | Description |
|---|---|
str
|
The call_id for the session |
generate_token(function_name, call_id)
¶
Generate a secure self-contained token for a function call
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the function to generate a token for |
required |
call_id
|
str
|
Call session ID |
required |
Returns:
| Type | Description |
|---|---|
str
|
A secure token |
create_tool_token(function_name, call_id)
¶
Alias for generate_token to maintain backward compatibility
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the function to generate a token for |
required |
call_id
|
str
|
Call session ID |
required |
Returns:
| Type | Description |
|---|---|
str
|
A secure token |
validate_token(call_id, function_name, token)
¶
Validate a function call token
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
call_id
|
str
|
Call session ID |
required |
function_name
|
str
|
Name of the function being called |
required |
token
|
str
|
Token to validate |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if valid, False otherwise |
validate_tool_token(function_name, token, call_id)
¶
Alias for validate_token to maintain backward compatibility
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
function_name
|
str
|
Name of the function being called |
required |
token
|
str
|
Token to validate |
required |
call_id
|
str
|
Call session ID |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if valid, False otherwise |
activate_session(call_id)
¶
Legacy method, does nothing but returns success
end_session(call_id)
¶
Legacy method, does nothing but returns success
get_session_metadata(call_id)
¶
Legacy method, always returns empty metadata
set_session_metadata(call_id, key, value)
¶
Legacy method, does nothing but returns success
debug_token(token)
¶
Debug a token without validating it
This method decodes the token and extracts its components for debugging purposes without performing validation. Requires _debug_mode to be True.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The token to debug |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with token components and analysis |
webhook_middleware
¶
FastAPI middleware / dependency for SignalWire webhook signature validation.
Copyright (c) 2025 SignalWire. Licensed under the MIT License. See LICENSE file in the project root for full license information.
This module ships a small, framework-aware adapter around
:func:signalwire.core.security.webhook_validator.validate_webhook_signature.
Why a custom dependency rather than a vanilla Depends on request.body()?
- We MUST capture the raw bytes BEFORE FastAPI's JSON parser consumes the
stream — re-serialization changes whitespace and key order, which breaks
the Scheme A digest. The dependency stashes the raw body on
request.state.raw_bodyso the downstream handler can re-parse without re-reading the stream. - Reverse-proxy / ngrok deployments need the URL the platform POSTed to,
which differs from the URL the SDK sees. The dependency honors
X-Forwarded-Proto/X-Forwarded-Hostwhentrust_proxy=True, plus theSWML_PROXY_URL_BASEenv var, withrequest.urlas last resort. - The legacy cXML/Compatibility scheme used the
X-Twilio-Signatureheader. We accept it as an alias ofX-SignalWire-Signatureso users migrating from the legacy SDK can keep their callers unchanged.
Usage::
from signalwire.core.security.webhook_middleware import (
make_webhook_validation_dependency,
)
dep = make_webhook_validation_dependency(signing_key="PSK...")
@app.post("/webhook", dependencies=[Depends(dep)])
async def webhook(request: Request):
body = request.state.raw_body # bytes; re-parse if you need JSON
SIGNALWIRE_SIGNATURE_HEADER = 'x-signalwire-signature'
module-attribute
¶
TWILIO_COMPAT_SIGNATURE_HEADER = 'x-twilio-signature'
module-attribute
¶
__all__ = ['SIGNALWIRE_SIGNATURE_HEADER', 'TWILIO_COMPAT_SIGNATURE_HEADER', 'make_webhook_validation_dependency']
module-attribute
¶
make_webhook_validation_dependency(signing_key, *, trust_proxy=False)
¶
Build a FastAPI dependency that enforces signature validation.
The returned coroutine:
- Reads
await request.body()and stashes the bytes onrequest.state.raw_body. - Pulls the
X-SignalWire-Signatureheader (or the Twilio alias). - Reconstructs the public URL (proxy headers / env / fallback).
- Calls :func:
validate_webhook_signature. - On invalid signature: raises
HTTPException(403)to short-circuit the handler. FastAPI'sdependencies=[Depends(...)]only honors short-circuiting via raised exceptions — returning a Response from a dependency does not stop the endpoint. - On valid: returns
Noneso the handler runs as normal.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
The customer's Signing Key. Required, non-empty. |
required |
trust_proxy
|
bool
|
If True, honor |
False
|
Returns:
| Type | Description |
|---|---|
Callable[[Request, Response], Awaitable[Response | None]]
|
Async callable suitable for |
Raises:
| Type | Description |
|---|---|
ValueError
|
at construction time if |
webhook_validator
¶
Webhook signature validation for SignalWire-signed HTTP requests.
Copyright (c) 2025 SignalWire. Licensed under the MIT License. See LICENSE file in the project root for full license information.
Implements both schemes from porting-sdk/webhooks.md:
- Scheme A (RELAY/SWML/JSON): hex(HMAC-SHA1(key, url + raw_body))
- Scheme B (Compat/cXML form): base64(HMAC-SHA1(key, url + sortedFormParams)) with optional bodySHA256 query-param fallback for JSON-on-compat-surface.
Public API
validate_webhook_signature(signing_key, signature, url, raw_body) -> bool validate_request(signing_key, signature, url, params_or_raw_body) -> bool
All comparisons use hmac.compare_digest (constant-time) so the secret
is not leaked over repeated requests.
__all__ = ['validate_request', 'validate_webhook_signature']
module-attribute
¶
validate_webhook_signature(signing_key, signature, url, raw_body)
¶
Validate a SignalWire webhook signature against both schemes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
Customer's Signing Key from the Dashboard. UTF-8 string,
secret. |
required |
signature
|
str
|
The |
required |
url
|
str
|
The full URL SignalWire POSTed to (scheme, host, optional port,
path, query). Must match what the platform saw — see the
|
required |
raw_body
|
str
|
The raw request body bytes as a UTF-8 string, BEFORE any
JSON / form parsing. Must be a |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the signature matches either Scheme A (hex JSON) or Scheme B |
bool
|
(base64 form, with port-normalization variants and optional |
bool
|
bodySHA256 fallback). False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
when |
TypeError
|
when |
validate_request(signing_key, signature, url, params_or_raw_body)
¶
Legacy @signalwire/compatibility-api drop-in entry point.
If params_or_raw_body is a string, delegates to
:func:validate_webhook_signature (Scheme A then Scheme B with parsed form).
If it's a mapping or a list of (key, value) tuples, treats it as pre-parsed form params and runs Scheme B directly (with URL port normalization and optional bodySHA256 fallback).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signing_key
|
str
|
Customer's Signing Key. Missing raises |
required |
signature
|
str
|
Header value. Missing / empty returns False. |
required |
url
|
str
|
Full URL SignalWire POSTed to. |
required |
params_or_raw_body
|
str | Mapping[str, Any] | list[tuple[str, Any]] | None
|
|
required |
Returns:
| Type | Description |
|---|---|
bool
|
True on match, False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
when |
TypeError
|
when |
security_config
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
logger = get_logger('security_config')
module-attribute
¶
security_config = SecurityConfig()
module-attribute
¶
SecurityConfig
¶
Unified security configuration for SignalWire services.
This class provides centralized security settings that can be used by both SWML and Search services, ensuring consistent security behavior.
SSL_ENABLED = 'SWML_SSL_ENABLED'
class-attribute
instance-attribute
¶
SSL_CERT_PATH = 'SWML_SSL_CERT_PATH'
class-attribute
instance-attribute
¶
SSL_KEY_PATH = 'SWML_SSL_KEY_PATH'
class-attribute
instance-attribute
¶
SSL_DOMAIN = 'SWML_DOMAIN'
class-attribute
instance-attribute
¶
SSL_VERIFY_MODE = 'SWML_SSL_VERIFY_MODE'
class-attribute
instance-attribute
¶
ALLOWED_HOSTS = 'SWML_ALLOWED_HOSTS'
class-attribute
instance-attribute
¶
CORS_ORIGINS = 'SWML_CORS_ORIGINS'
class-attribute
instance-attribute
¶
MAX_REQUEST_SIZE = 'SWML_MAX_REQUEST_SIZE'
class-attribute
instance-attribute
¶
RATE_LIMIT = 'SWML_RATE_LIMIT'
class-attribute
instance-attribute
¶
REQUEST_TIMEOUT = 'SWML_REQUEST_TIMEOUT'
class-attribute
instance-attribute
¶
USE_HSTS = 'SWML_USE_HSTS'
class-attribute
instance-attribute
¶
HSTS_MAX_AGE = 'SWML_HSTS_MAX_AGE'
class-attribute
instance-attribute
¶
BASIC_AUTH_USER = 'SWML_BASIC_AUTH_USER'
class-attribute
instance-attribute
¶
BASIC_AUTH_PASSWORD = 'SWML_BASIC_AUTH_PASSWORD'
class-attribute
instance-attribute
¶
DEFAULTS = {SSL_ENABLED: False, SSL_VERIFY_MODE: 'CERT_REQUIRED', ALLOWED_HOSTS: '*', CORS_ORIGINS: '*', MAX_REQUEST_SIZE: 10 * 1024 * 1024, RATE_LIMIT: 60, REQUEST_TIMEOUT: 30, USE_HSTS: True, HSTS_MAX_AGE: 31536000}
class-attribute
¶
__init__(config_file=None, service_name=None)
¶
Initialize security configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config_file
|
str | None
|
Optional path to config file |
None
|
service_name
|
str | None
|
Optional service name for finding service-specific config |
None
|
load_from_env()
¶
Load configuration from environment variables
validate_ssl_config()
¶
Validate SSL configuration.
Returns:
| Type | Description |
|---|---|
tuple[bool, str | None]
|
Tuple of (is_valid, error_message) |
get_ssl_context_kwargs()
¶
Get SSL context kwargs for uvicorn.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary of SSL parameters for uvicorn |
get_basic_auth()
¶
Get basic auth credentials, generating if not set.
If no basic auth password is configured (e.g. SWML_BASIC_AUTH_PASSWORD env var is unset and the SDK caller didn't pass one explicitly), the SDK falls back to a random password generated via secrets.token_urlsafe. That password lives only in memory and is regenerated on every process start, so any external caller (tests, RPC clients, MCP) that doesn't share the auto-generated value will get HTTP 401.
We log a warning the first time the auto-generated fallback fires so the failure mode is visible in logs instead of silently breaking external callers. Set SWML_BASIC_AUTH_USER / SWML_BASIC_AUTH_PASSWORD (or load .env before constructing the agent) to suppress the warning.
Returns:
| Type | Description |
|---|---|
tuple[str, str]
|
Tuple of (username, password) |
get_security_headers(is_https=False)
¶
Get security headers to add to responses.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
is_https
|
bool
|
Whether the connection is over HTTPS |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dictionary of security headers |
should_allow_host(host)
¶
Check if a host is allowed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
The host to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the host is allowed |
get_cors_config()
¶
Get CORS configuration for FastAPI.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary of CORS settings |
get_url_scheme()
¶
Get the URL scheme based on SSL configuration
log_config(service_name)
¶
Log the current security configuration
skill_base
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SkillBase
¶
Bases: ABC
Abstract base class for all agent skills
SKILL_NAME = None
class-attribute
instance-attribute
¶
SKILL_DESCRIPTION = None
class-attribute
instance-attribute
¶
SKILL_VERSION = '1.0.0'
class-attribute
instance-attribute
¶
REQUIRED_PACKAGES = []
class-attribute
¶
REQUIRED_ENV_VARS = []
class-attribute
¶
SUPPORTS_MULTIPLE_INSTANCES = False
class-attribute
instance-attribute
¶
agent = agent
instance-attribute
¶
params = params or {}
instance-attribute
¶
logger = get_logger(f'signalwire.skills.{self.SKILL_NAME}')
instance-attribute
¶
swaig_fields = self.params.pop('swaig_fields', {})
instance-attribute
¶
__init__(agent, params=None)
¶
setup()
abstractmethod
¶
Setup the skill (validate env vars, initialize APIs, etc.) Returns True if setup successful, False otherwise
register_tools()
abstractmethod
¶
Register SWAIG tools with the agent
define_tool(**kwargs)
¶
Wrapper method that automatically includes swaig_fields when defining tools.
This method delegates to self.agent.define_tool() but automatically merges any swaig_fields configured for this skill. Skills should use this method instead of calling self.agent.define_tool() directly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
All arguments supported by agent.define_tool() (name, description, parameters, handler, etc.) |
{}
|
get_hints()
¶
Return speech recognition hints for this skill
get_global_data()
¶
Return data to add to agent's global context
get_prompt_sections()
¶
Return prompt sections to add to agent. Returns empty list if skip_prompt is set to True in params.
cleanup()
¶
Cleanup when skill is removed or agent shuts down
validate_env_vars()
¶
Check if all required environment variables are set
validate_packages()
¶
Check if all required packages are available
get_instance_key()
¶
Get the key used to track this skill instance
For skills that support multiple instances (SUPPORTS_MULTIPLE_INSTANCES = True), this method can be overridden to provide a unique key for each instance.
Default implementation: - If SUPPORTS_MULTIPLE_INSTANCES is False: returns SKILL_NAME - If SUPPORTS_MULTIPLE_INSTANCES is True: returns SKILL_NAME + "_" + tool_name (where tool_name comes from params['tool_name'] or defaults to the skill name)
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
Unique key for this skill instance |
get_skill_data(raw_data)
¶
Read this skill instance's namespaced data from raw_data global_data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
raw_data
|
dict[str, Any]
|
The raw_data dict passed to SWAIG function handlers, expected to contain a 'global_data' key. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[str, Any]
|
The skill's namespaced state, or empty dict if not found. |
update_skill_data(result, data)
¶
Write this skill instance's namespaced data into a FunctionResult.
Wraps the data under the skill's namespace key and calls result.update_global_data().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result
|
FunctionResult
|
The FunctionResult to add the global_data update to. |
required |
data
|
dict[str, Any]
|
The skill state dict to store under the namespace. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
FunctionResult |
FunctionResult
|
The result, for method chaining. |
get_parameter_schema()
classmethod
¶
Get the parameter schema for this skill
This method returns metadata about all parameters the skill accepts, including their types, descriptions, default values, and whether they are required or should be hidden (e.g., API keys).
The base implementation provides common parameters available to all skills. Subclasses should override this method and merge their specific parameters with the base schema.
Returns:
| Type | Description |
|---|---|
dict[str, dict[str, Any]]
|
Dict[str, Dict[str, Any]]: Parameter schema where keys are parameter names |
dict[str, dict[str, Any]]
|
and values are dictionaries containing: - type: Parameter type ("string", "integer", "number", "boolean", "object", "array") - description: Human-readable description - default: Default value if not provided (optional) - required: Whether the parameter is required (default: False) - hidden: Whether to hide this field in UIs (for secrets/keys) - env_var: Environment variable that can provide this value (optional) - enum: List of allowed values (optional) - min/max: Minimum/maximum values for numeric types (optional) |
Example
{ "tool_name": { "type": "string", "description": "Name for the tool when using multiple instances", "default": "my_skill", "required": False }, "api_key": { "type": "string", "description": "API key for the service", "required": True, "hidden": True, "env_var": "MY_API_KEY" } }
skill_manager
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SkillManager
¶
Manages loading and lifecycle of agent skills
agent = agent
instance-attribute
¶
loaded_skills = {}
instance-attribute
¶
logger = get_logger('skill_manager')
instance-attribute
¶
__init__(agent)
¶
load_skill(skill_name, skill_class=None, params=None)
¶
Load and setup a skill by name
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skill_name
|
str
|
Name of the skill to load |
required |
skill_class
|
type[SkillBase] | None
|
Optional skill class (if not provided, will try to find it) |
None
|
params
|
dict[str, Any] | None
|
Optional parameters to pass to the skill |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
tuple |
tuple[bool, str]
|
(success, error_message) - error_message is empty string if successful |
unload_skill(skill_identifier)
¶
Unload a skill and cleanup
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skill_identifier
|
str
|
Either a skill name or an instance key |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if successfully unloaded, False otherwise |
list_loaded_skills()
¶
List instance keys of currently loaded skills
has_skill(skill_identifier)
¶
Check if skill is currently loaded
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skill_identifier
|
str
|
Either a skill name or an instance key |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if loaded, False otherwise |
swaig_function
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SwaigFunction class for defining and managing SWAIG function interfaces
SWAIGFunction
¶
Represents a SWAIG function — i.e., a tool the AI model can call.
A SWAIG function is exactly the same concept as a "tool" in native OpenAI / Anthropic tool calling. Each SWAIGFunction is rendered into the OpenAI tool schema format and sent to the model on every turn:
{"type": "function", "function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}}
The model parses description (and the description inside each
parameter) to decide WHEN to call the tool and HOW to fill in the
arguments. Both fields are prompt-engineered text — the quality of
your descriptions directly drives tool-selection accuracy.
Use AgentBase.define_tool() to create one (it builds the SWAIGFunction for you and registers it). Use the @agent.tool() decorator for the same thing in class-based agents.
name = name
instance-attribute
¶
handler = handler
instance-attribute
¶
description = description
instance-attribute
¶
parameters = parameters or {}
instance-attribute
¶
secure = secure
instance-attribute
¶
fillers = fillers
instance-attribute
¶
wait_file = wait_file
instance-attribute
¶
wait_file_loops = wait_file_loops
instance-attribute
¶
webhook_url = webhook_url
instance-attribute
¶
required = required or []
instance-attribute
¶
is_typed_handler = is_typed_handler
instance-attribute
¶
extra_swaig_fields = extra_swaig_fields
instance-attribute
¶
is_external = webhook_url is not None
instance-attribute
¶
__init__(name, handler, description, parameters=None, secure=False, fillers=None, wait_file=None, wait_file_loops=None, webhook_url=None, required=None, is_typed_handler=False, **extra_swaig_fields)
¶
Initialize a new SWAIG function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Function name. Becomes the |
required |
handler
|
Callable
|
Python callable invoked when the model calls this tool. |
required |
description
|
str
|
LLM-facing description. The model reads this on every turn to decide whether to call the tool. Be specific about WHEN to use it and what makes it the right choice over sibling tools — vague descriptions are the most common cause of "model has the right tool but doesn't call it" failures. |
required |
parameters
|
dict[str, dict] | None
|
JSON Schema for the arguments. Per-property
|
None
|
secure
|
bool
|
Whether this function requires SWAIG token validation. |
False
|
fillers
|
dict[str, list[str]] | None
|
Optional dictionary of filler phrases by language code (deprecated, use wait_file). |
None
|
wait_file
|
str | None
|
Optional URL to audio file to play while function executes. |
None
|
wait_file_loops
|
int | None
|
Optional number of times to loop the wait_file. |
None
|
webhook_url
|
str | None
|
Optional external webhook URL to use instead of local handling. |
None
|
required
|
list[str] | None
|
Optional list of required parameter names. |
None
|
is_typed_handler
|
bool
|
Whether the handler uses type-hinted parameters (auto-wrapped). |
False
|
**extra_swaig_fields
|
Additional SWAIG-only fields (meta_data_token, web_hook_auth_*, etc.) to include in the generated definition. |
{}
|
__call__(*args, **kwargs)
¶
Call the underlying handler function
execute(args, raw_data=None)
¶
Execute the function with the given arguments
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args
|
dict[str, Any]
|
Parsed arguments for the function |
required |
raw_data
|
dict[str, Any] | None
|
Optional raw request data |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Function result as a dictionary (from FunctionResult.to_dict()) |
validate_args(args)
¶
Validate the arguments against the parameter schema.
Uses jsonschema_rs if available for fast validation, falls back to jsonschema if available, otherwise skips validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args
|
dict[str, Any]
|
Arguments to validate |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
Tuple of (is_valid: bool, errors: list[str]) |
tuple
|
If no validation library is available, returns (True, []) |
to_swaig(base_url, token=None, call_id=None, include_auth=True)
¶
Convert this function to a SWAIG-compatible JSON object for SWML
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base_url
|
str
|
Base URL for the webhook |
required |
token
|
str | None
|
Optional auth token to include |
None
|
call_id
|
str | None
|
Optional call ID for session tracking |
None
|
include_auth
|
bool
|
Whether to include auth credentials in URL |
True
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary representation for the SWAIG array in SWML |
swml_builder
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SWML Builder - Fluent API for building SWML documents
This module provides a fluent builder API for creating SWML documents. It allows for chaining method calls to build up a document step by step.
T = TypeVar('T', bound='SWMLBuilder')
module-attribute
¶
SWMLBuilder
¶
Fluent builder for SWML documents
This class provides a fluent interface for building SWML documents by chaining method calls. It delegates to an underlying SWMLService instance for the actual document creation.
service = service
instance-attribute
¶
__init__(service)
¶
Initialize with a SWMLService instance
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service
|
SWMLService
|
The SWMLService to delegate to |
required |
answer(max_duration=None, codecs=None)
¶
Add an 'answer' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_duration
|
int | None
|
Maximum duration in seconds |
None
|
codecs
|
str | None
|
Comma-separated list of codecs |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
hangup(reason=None)
¶
Add a 'hangup' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reason
|
str | None
|
Optional reason for hangup |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
ai(prompt_text=None, prompt_pom=None, post_prompt=None, post_prompt_url=None, swaig=None, **kwargs)
¶
Add an 'ai' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt_text
|
str | None
|
Text prompt for the AI (mutually exclusive with prompt_pom) |
None
|
prompt_pom
|
list[dict[str, Any]] | None
|
POM structure for the AI prompt (mutually exclusive with prompt_text) |
None
|
post_prompt
|
str | None
|
Optional post-prompt text |
None
|
post_prompt_url
|
str | None
|
Optional URL for post-prompt processing |
None
|
swaig
|
dict[str, Any] | None
|
Optional SWAIG configuration |
None
|
**kwargs
|
Additional AI parameters |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
play(url=None, urls=None, volume=None, say_voice=None, say_language=None, say_gender=None, auto_answer=None)
¶
Add a 'play' verb to the main section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str | None
|
Single URL to play (mutually exclusive with urls) |
None
|
urls
|
list[str] | None
|
List of URLs to play (mutually exclusive with url) |
None
|
volume
|
float | None
|
Volume level (-40 to 40) |
None
|
say_voice
|
str | None
|
Voice for text-to-speech |
None
|
say_language
|
str | None
|
Language for text-to-speech |
None
|
say_gender
|
str | None
|
Gender for text-to-speech |
None
|
auto_answer
|
bool | None
|
Whether to auto-answer the call |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
say(text, voice=None, language=None, gender=None, volume=None)
¶
Add a 'play' verb with say: prefix for text-to-speech
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
Text to speak |
required |
voice
|
str | None
|
Voice for text-to-speech |
None
|
language
|
str | None
|
Language for text-to-speech |
None
|
gender
|
str | None
|
Gender for text-to-speech |
None
|
volume
|
float | None
|
Volume level (-40 to 40) |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
add_section(section_name)
¶
Add a new section to the document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add |
required |
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
build()
¶
Build and return the SWML document
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The complete SWML document as a dictionary |
render()
¶
Build and render the SWML document as a JSON string
Returns:
| Type | Description |
|---|---|
str
|
The complete SWML document as a JSON string |
reset()
¶
Reset the document to an empty state
Returns:
| Type | Description |
|---|---|
Self
|
Self for method chaining |
__getattr__(name)
¶
Dynamically generate and return SWML verb methods when accessed
This method is called when an attribute lookup fails through the normal mechanisms. It checks if the attribute name corresponds to a SWML verb defined in the schema, and if so, dynamically creates a method for that verb.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the attribute being accessed |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The dynamically created verb method if name is a valid SWML verb, |
Any
|
otherwise raises AttributeError |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If name is not a valid SWML verb |
swml_handler
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SWML Verb Handlers - Interface and implementations for SWML verb handling
This module defines the base interface for SWML verb handlers and provides implementations for specific verbs that require special handling.
SWMLVerbHandler
¶
Bases: ABC
Base interface for SWML verb handlers
This abstract class defines the interface that all SWML verb handlers must implement. Verb handlers provide specialized logic for complex SWML verbs that cannot be handled generically.
get_verb_name()
abstractmethod
¶
Get the name of the verb this handler handles
Returns:
| Type | Description |
|---|---|
str
|
The verb name as a string |
validate_config(config)
abstractmethod
¶
Validate the configuration for this verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
The configuration dictionary for this verb |
required |
Returns:
| Type | Description |
|---|---|
tuple[bool, list[str]]
|
(is_valid, error_messages) tuple |
build_config(**kwargs)
abstractmethod
¶
Build a configuration for this verb from the provided arguments
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Keyword arguments specific to this verb |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Configuration dictionary |
AIVerbHandler
¶
Bases: SWMLVerbHandler
Handler for the SWML 'ai' verb
The 'ai' verb is complex and requires specialized handling, particularly for managing prompts, SWAIG functions, and AI configurations.
get_verb_name()
¶
Get the name of the verb this handler handles
Returns:
| Type | Description |
|---|---|
str
|
"ai" as the verb name |
validate_config(config)
¶
Validate the configuration for the AI verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
The configuration dictionary for the AI verb |
required |
Returns:
| Type | Description |
|---|---|
tuple[bool, list[str]]
|
(is_valid, error_messages) tuple |
build_config(prompt_text=None, prompt_pom=None, contexts=None, post_prompt=None, post_prompt_url=None, swaig=None, **kwargs)
¶
Build a configuration for the AI verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt_text
|
str | None
|
Text prompt for the AI (mutually exclusive with prompt_pom) |
None
|
prompt_pom
|
list[dict[str, Any]] | None
|
POM structure for the AI prompt (mutually exclusive with prompt_text) |
None
|
contexts
|
dict[str, Any] | None
|
Optional contexts and steps configuration (can be combined with text or pom) |
None
|
post_prompt
|
str | None
|
Optional post-prompt text |
None
|
post_prompt_url
|
str | None
|
Optional URL for post-prompt processing |
None
|
swaig
|
dict[str, Any] | None
|
Optional SWAIG configuration |
None
|
**kwargs
|
Additional AI parameters |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
AI verb configuration dictionary |
VerbHandlerRegistry
¶
Registry for SWML verb handlers
This class maintains a registry of handlers for special SWML verbs and provides methods for accessing and using them.
__init__()
¶
Initialize the registry with default handlers
register_handler(handler)
¶
Register a new verb handler
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
SWMLVerbHandler
|
The handler to register |
required |
get_handler(verb_name)
¶
Get the handler for a specific verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb |
required |
Returns:
| Type | Description |
|---|---|
SWMLVerbHandler | None
|
The handler if found, None otherwise |
has_handler(verb_name)
¶
Check if a handler exists for a specific verb
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if a handler exists, False otherwise |
swml_renderer
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
SWML document rendering utilities for SignalWire AI Agents.
SwmlRenderer
¶
Renders SWML documents for SignalWire AI Agents with AI and SWAIG components
This class provides methods for rendering SWML documents using the SWMLService architecture.
render_swml(prompt, service, post_prompt=None, post_prompt_url=None, swaig_functions=None, startup_hook_url=None, hangup_hook_url=None, prompt_is_pom=False, params=None, add_answer=False, record_call=False, record_format='mp4', record_stereo=True, format='json', default_webhook_url=None)
staticmethod
¶
Generate a complete SWML document with AI configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt
|
str | list[dict[str, Any]]
|
AI prompt text or POM structure |
required |
service
|
SWMLService
|
SWMLService instance to use for document building |
required |
post_prompt
|
str | None
|
Optional post-prompt text |
None
|
post_prompt_url
|
str | None
|
Optional post-prompt URL |
None
|
swaig_functions
|
list[dict[str, Any]] | None
|
List of SWAIG function definitions |
None
|
startup_hook_url
|
str | None
|
Optional startup hook URL |
None
|
hangup_hook_url
|
str | None
|
Optional hangup hook URL |
None
|
prompt_is_pom
|
bool
|
Whether prompt is POM format |
False
|
params
|
dict[str, Any] | None
|
Additional AI verb parameters |
None
|
add_answer
|
bool
|
Whether to add answer verb |
False
|
record_call
|
bool
|
Whether to add record_call verb |
False
|
record_format
|
str
|
Recording format |
'mp4'
|
record_stereo
|
bool
|
Whether to record in stereo |
True
|
format
|
str
|
Output format (json or yaml) |
'json'
|
default_webhook_url
|
str | None
|
Default webhook URL for SWAIG functions |
None
|
Returns:
| Type | Description |
|---|---|
str
|
SWML document as a string |
render_function_response_swml(response_text, service, actions=None, format='json')
staticmethod
¶
Generate a SWML document for a function response
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response_text
|
str
|
Text response to include in the document |
required |
service
|
SWMLService
|
SWMLService instance to use |
required |
actions
|
list[dict[str, Any]] | None
|
Optional list of actions to perform |
None
|
format
|
str
|
Output format (json or yaml) |
'json'
|
Returns:
| Type | Description |
|---|---|
str
|
SWML document as a string |
swml_service
¶
Copyright (c) 2025 SignalWire
This file is part of the SignalWire SDK.
Licensed under the MIT License. See LICENSE file in the project root for full license information.
Base SWML Service for SignalWire Agents
logger = get_logger('swml_service')
module-attribute
¶
MAX_REQUEST_BODY_SIZE = 10 * 1024 * 1024
module-attribute
¶
SWMLService
¶
Bases: ToolMixin
Base class for creating and serving SWML documents.
This class provides core functionality for: - Loading and validating SWML schema - Creating SWML documents - Setting up web endpoints for serving SWML - Managing authentication - Registering SWML functions
It serves as the foundation for more specialized services like AgentBase.
name = name
instance-attribute
¶
route = route.rstrip('/')
instance-attribute
¶
host = host
instance-attribute
¶
port = port if port is not None else int(os.environ.get('PORT', 3000))
instance-attribute
¶
log = logger.bind(service=name)
instance-attribute
¶
security = SecurityConfig(config_file=config_file, service_name=name)
instance-attribute
¶
ssl_enabled = self.security.ssl_enabled
instance-attribute
¶
domain = self.security.domain
instance-attribute
¶
ssl_cert_path = self.security.ssl_cert_path
instance-attribute
¶
ssl_key_path = self.security.ssl_key_path
instance-attribute
¶
schema_utils = SchemaUtils(schema_path, schema_validation=(self._schema_validation))
instance-attribute
¶
verb_registry = VerbHandlerRegistry()
instance-attribute
¶
full_validation_enabled
property
¶
Check if full JSON Schema validation is enabled.
Returns:
| Type | Description |
|---|---|
bool
|
True if schema validator is initialized |
__init__(name, route='/', host='0.0.0.0', port=None, basic_auth=None, schema_path=None, config_file=None, schema_validation=True)
¶
Initialize a new SWML service
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Service name/identifier |
required |
route
|
str
|
HTTP route path for this service |
'/'
|
host
|
str
|
Host to bind the web server to |
'0.0.0.0'
|
port
|
int | None
|
Port to bind the web server to |
None
|
basic_auth
|
tuple[str, str] | None
|
Optional (username, password) tuple for basic auth |
None
|
schema_path
|
str | None
|
Optional path to the schema file |
None
|
config_file
|
str | None
|
Optional path to configuration file |
None
|
schema_validation
|
bool
|
Enable schema validation. Default True. Can also be disabled via SWML_SKIP_SCHEMA_VALIDATION=1 env var. |
True
|
__getattr__(name)
¶
Dynamically generate and return SWML verb methods when accessed
This method is called when an attribute lookup fails through the normal mechanisms. It checks if the attribute name corresponds to a SWML verb defined in the schema, and if so, dynamically creates a method for that verb.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the attribute being accessed |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The dynamically created verb method if name is a valid SWML verb, |
Any
|
otherwise raises AttributeError |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If name is not a valid SWML verb |
reset_document()
¶
Reset the current document to an empty state
add_verb(verb_name, config)
¶
Add a verb to the main section of the current document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verb_name
|
str
|
The name of the verb to add |
required |
config
|
dict[str, Any] | int
|
Configuration for the verb or direct value for certain verbs (e.g., sleep) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the verb was added successfully, False otherwise |
add_section(section_name)
¶
Add a new section to the document
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the section was added, False if it already exists |
add_verb_to_section(section_name, verb_name, config)
¶
Add a verb to a specific section
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
section_name
|
str
|
Name of the section to add to |
required |
verb_name
|
str
|
The name of the verb to add |
required |
config
|
dict[str, Any] | int
|
Configuration for the verb or direct value for certain verbs (e.g., sleep) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the verb was added successfully, False otherwise |
get_document()
¶
Get the current SWML document
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The current SWML document as a dictionary |
render_document()
¶
Render the current SWML document as a JSON string
Returns:
| Type | Description |
|---|---|
str
|
The current SWML document as a JSON string |
register_verb_handler(handler)
¶
Register a custom verb handler
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler
|
SWMLVerbHandler
|
The verb handler to register |
required |
as_router()
¶
Create a FastAPI router for this service
Returns:
| Name | Type | Description |
|---|---|---|
APIRouter |
APIRouter
|
FastAPI router |
register_routing_callback(callback_fn, path='/sip')
¶
Register a callback function that will be called to determine routing based on POST data.
When a routing callback is registered, an endpoint at the specified path is automatically created that will handle requests. This endpoint will use the callback to determine if the request should be processed by this service or redirected.
The callback should take a request object and request body dictionary and return: - A route string if it should be routed to a different endpoint - None if normal processing should continue
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback_fn
|
Callable[[Request, dict[str, Any]], str | None]
|
The callback function to register |
required |
path
|
str
|
The path where this callback should be registered (default: "/sip") |
'/sip'
|
extract_sip_username(request_body)
staticmethod
¶
Extract SIP username from request body
This extracts the username portion of a SIP URI from the 'to' field in the call data of a request body.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_body
|
dict[str, Any]
|
The parsed JSON body of the request |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The extracted SIP username, or None if not found |
on_request(request_data=None, callback_path=None)
¶
Called when SWML is requested, with request data when available
Subclasses can override this to inspect or modify SWML based on the request
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_data
|
dict | None
|
Optional dictionary containing the parsed POST body |
None
|
callback_path
|
str | None
|
Optional callback path |
None
|
Returns:
| Type | Description |
|---|---|
dict | None
|
Optional dict to modify/augment the SWML document |
serve(host=None, port=None, ssl_cert=None, ssl_key=None, ssl_enabled=None, domain=None)
¶
Start a web server for this service
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str | None
|
Host to bind to (defaults to self.host) |
None
|
port
|
int | None
|
Port to bind to (defaults to self.port) |
None
|
ssl_cert
|
str | None
|
Path to SSL certificate file |
None
|
ssl_key
|
str | None
|
Path to SSL key file |
None
|
ssl_enabled
|
bool | None
|
Whether to enable SSL |
None
|
domain
|
str | None
|
Domain name for SSL certificate |
None
|
stop()
¶
Stop the web server
get_basic_auth_credentials(include_source=False)
¶
Get the basic auth credentials
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_source
|
bool
|
Whether to include the source of the credentials |
False
|
Returns:
| Type | Description |
|---|---|
tuple[str, str] | tuple[str, str, str]
|
(username, password) tuple or (username, password, source) tuple if include_source is True |
manual_set_proxy_url(proxy_url)
¶
Manually set the proxy URL base for webhook callbacks
This can be called at runtime to set or update the proxy URL
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
proxy_url
|
str
|
The base URL to use for webhooks (e.g., https://example.ngrok.io) |
required |