The AgentBase
class provides the foundation for creating AI-powered agents using the SignalWire AI Agent SDK. It extends the SWMLService
class, inheriting all its SWML document creation and serving capabilities, while adding AI-specific functionality.
Key features of AgentBase
include:
This guide explains how to create and customize your own AI agents, with examples based on the SDK's sample implementations.
The Agent SDK architecture consists of several layers:
Here's how these components relate to each other:
┌─────────────┐
│ Your Agent │ (Extends AgentBase with your specific functionality)
└─────▲───────┘
│
┌─────┴───────┐
│ AgentBase │ (Adds AI functionality to SWMLService)
└─────▲───────┘
│
┌─────┴───────┐
│ SWMLService │ (Provides SWML document creation and web service)
└─────────────┘
To create an agent, extend the AgentBase
class and define your agent's behavior:
from signalwire_agents import AgentBase
class MyAgent(AgentBase):
def __init__(self):
super().__init__(
name="my-agent",
route="/agent",
host="0.0.0.0",
port=3000,
use_pom=True # Enable Prompt Object Model
)
# Define agent personality and behavior
self.prompt_add_section("Personality", body="You are a helpful and friendly assistant.")
self.prompt_add_section("Goal", body="Help users with their questions and tasks.")
self.prompt_add_section("Instructions", bullets=[
"Answer questions clearly and concisely",
"If you don't know, say so",
"Use the provided tools when appropriate"
])
# Add a post-prompt for summary
self.set_post_prompt("Please summarize the key points of this conversation.")
The SignalWire AI Agent SDK provides a run()
method that automatically detects the execution environment and configures the agent appropriately. This method works across all deployment modes:
run()
🔗 ↑ TOCdef main():
agent = MyAgent()
print("Starting agent server...")
print("Note: Works in any deployment mode (server/CGI/Lambda)")
agent.run() # Auto-detects environment
if __name__ == "__main__":
main()
The run()
method automatically detects and configures for:
When run directly (e.g., python my_agent.py
), the agent starts an HTTP server:
# Automatically starts HTTP server when run directly
agent.run()
When CGI environment variables are present, operates in CGI mode with clean HTTP output:
# Same code - automatically detects CGI environment
agent.run()
When AWS Lambda environment is detected, configures for serverless execution:
# Same code - automatically detects Lambda environment
agent.run()
The SDK automatically detects the execution environment:
Environment | Detection Method | Behavior |
---|---|---|
HTTP Server | Default when no serverless environment detected | Starts FastAPI server on specified host/port |
CGI | GATEWAY_INTERFACE environment variable present |
Processes single CGI request and exits |
AWS Lambda | AWS_LAMBDA_FUNCTION_NAME environment variable |
Handles Lambda event/context |
Google Cloud | FUNCTION_NAME or K_SERVICE variables |
Processes Cloud Function request |
Azure Functions | AZURE_FUNCTIONS_* variables |
Handles Azure Function request |
The SDK includes a central logging system that automatically configures based on the deployment environment:
# Logging is automatically configured based on environment
# No manual setup required in most cases
# Optional: Override logging mode via environment variable
# SIGNALWIRE_LOG_MODE=off # Disable all logging
# SIGNALWIRE_LOG_MODE=stderr # Log to stderr
# SIGNALWIRE_LOG_MODE=default # Use default logging
# SIGNALWIRE_LOG_MODE=auto # Auto-detect (default)
The logging system automatically: - CGI Mode: Sets logging to 'off' to avoid interfering with HTTP headers - Lambda Mode: Configures appropriate logging for serverless environment - Server Mode: Uses structured logging with timestamps and levels - Debug Mode: Enhanced logging when debug flags are set
There are several ways to build prompts for your agent:
The Prompt Object Model (POM) provides a structured way to build prompts:
# Add a section with just body text
self.prompt_add_section("Personality", body="You are a friendly assistant.")
# Add a section with bullet points
self.prompt_add_section("Instructions", bullets=[
"Answer questions clearly",
"Be helpful and polite",
"Use functions when appropriate"
])
# Add a section with both body and bullets
self.prompt_add_section("Context",
body="The user is calling about technical support.",
bullets=["They may need help with their account",
"Check for existing tickets"])
For convenience, the SDK also provides wrapper methods that some users may prefer:
# Convenience methods
self.setPersonality("You are a friendly assistant.")
self.setGoal("Help users with their questions.")
self.setInstructions([
"Answer questions clearly",
"Be helpful and polite"
])
These convenience methods call prompt_add_section()
internally with the appropriate section titles.
For simpler agents, you can set the prompt directly as text:
self.set_prompt_text("""
You are a helpful assistant. Your goal is to provide clear and concise information
to the user. Answer their questions to the best of your ability.
""")
The post-prompt is sent to the AI after the conversation for summary or analysis:
self.set_post_prompt("""
Analyze the conversation and extract:
1. Main topics discussed
2. Action items or follow-ups needed
3. Whether the user's questions were answered satisfactorily
""")
SWAIG functions allow the AI agent to perform actions and access external systems. There are two types of SWAIG functions you can define:
These are the traditional SWAIG functions that are handled locally by your agent:
from signalwire_agents.core.function_result import SwaigFunctionResult
@AgentBase.tool(
name="get_weather",
description="Get the current weather for a location",
parameters={
"location": {
"type": "string",
"description": "The city or location to get weather for"
}
},
secure=True # Optional, defaults to True
)
def get_weather(self, args, raw_data):
# Extract the location parameter
location = args.get("location", "Unknown location")
# Here you would typically call a weather API
# For this example, we'll return mock data
weather_data = f"It's sunny and 72°F in {location}."
# Return a SwaigFunctionResult
return SwaigFunctionResult(weather_data)
External webhook functions allow you to delegate function execution to external services instead of handling them locally. This is useful when you want to: - Use existing web services or APIs directly - Distribute function processing across multiple servers - Integrate with third-party systems that provide their own endpoints
To create an external webhook function, add a webhook_url
parameter to the decorator:
@AgentBase.tool(
name="get_weather_external",
description="Get weather from external service",
parameters={
"location": {
"type": "string",
"description": "The city or location to get weather for"
}
},
webhook_url="https://your-service.com/weather-endpoint"
)
def get_weather_external(self, args, raw_data):
# This function will never be called locally when webhook_url is provided
# The external service at webhook_url will receive the function call instead
return SwaigFunctionResult("This should not be reached for external webhooks")
When you specify a webhook_url
:
{
"function": "get_weather_external",
"argument": {
"parsed": [{"location": "New York"}],
"raw": "{\"location\": \"New York\"}"
},
"call_id": "abc123-def456-ghi789",
"call": { /* call information */ },
"vars": { /* call variables */ }
}
You can mix both types of functions in the same agent:
class HybridAgent(AgentBase):
def __init__(self):
super().__init__(name="hybrid-agent", route="/hybrid")
# Local function - handled by this agent
@AgentBase.tool(
name="get_help",
description="Get help information",
parameters={}
)
def get_help(self, args, raw_data):
return SwaigFunctionResult("I can help you with weather and news!")
# External function - handled by external service
@AgentBase.tool(
name="get_weather",
description="Get current weather",
parameters={
"location": {"type": "string", "description": "City name"}
},
webhook_url="https://weather-service.com/api/weather"
)
def get_weather_external(self, args, raw_data):
# This won't be called for external webhooks
pass
# Another external function - different service
@AgentBase.tool(
name="get_news",
description="Get latest news",
parameters={
"topic": {"type": "string", "description": "News topic"}
},
webhook_url="https://news-service.com/api/news"
)
def get_news_external(self, args, raw_data):
# This won't be called for external webhooks
pass
You can test external webhook functions using the CLI tool:
# Test local function
swaig-test examples/my_agent.py --exec get_help
# Test external webhook function
swaig-test examples/my_agent.py --verbose --exec get_weather --location "New York"
# List all functions with their types
swaig-test examples/my_agent.py --list-tools
The CLI tool will automatically detect external webhook functions and make HTTP requests to the external services, simulating what SignalWire does in production.
The parameters for a SWAIG function are defined using JSON Schema:
parameters={
"parameter_name": {
"type": "string", # Can be string, number, integer, boolean, array, object
"description": "Description of the parameter",
# Optional attributes:
"enum": ["option1", "option2"], # For enumerated values
"minimum": 0, # For numeric types
"maximum": 100, # For numeric types
"pattern": "^[A-Z]+$" # For string validation
}
}
To return results from a SWAIG function, use the SwaigFunctionResult
class:
# Basic result with just text
return SwaigFunctionResult("Here's the result")
# Result with a single action
return SwaigFunctionResult("Here's the result with an action")
.add_action("say", "I found the information you requested.")
# Result with multiple actions using add_actions
return SwaigFunctionResult("Multiple actions example")
.add_actions([
{"playback_bg": {"file": "https://example.com/music.mp3"}},
{"set_global_data": {"key": "value"}}
])
# Alternative way to add multiple actions sequentially
return (
SwaigFunctionResult("Sequential actions example")
.add_action("say", "I found the information you requested.")
.add_action("playback_bg", {"file": "https://example.com/music.mp3"})
)
In the examples above:
- add_action(name, data)
adds a single action with the given name and data
- add_actions(actions)
adds multiple actions at once from a list of action objects
The agent can use SignalWire's built-in functions:
# Enable native functions
self.set_native_functions([
"check_time",
"wait_seconds"
])
You can include functions from remote sources:
# Include remote functions
self.add_function_include(
url="https://api.example.com/functions",
functions=["get_weather", "get_news"],
meta_data={"session_id": "unique-session-123"} # Use for session tracking, NOT credentials
)
The SDK implements an automated security mechanism for SWAIG functions to ensure that only authorized calls can be made to your functions. This is important because SWAIG functions often provide access to sensitive operations or data.
By default, all SWAIG functions are marked as secure=True
, which enables token-based security:
@agent.tool(
name="get_account_details",
description="Get customer account details",
parameters={"account_id": {"type": "string"}},
secure=True # This is the default, can be omitted
)
def get_account_details(self, args, raw_data):
# Implementation
When a function is marked as secure:
?token=X2FiY2RlZmcuZ2V0X3RpbWUuMTcxOTMxNDI1...
These security tokens have important properties: - Completely stateless: The system doesn't need to store tokens or track sessions - Self-contained: Each token contains all information needed for validation - Function-specific: A token for one function can't be used for another - Session-bound: Tokens are tied to a specific call/session ID - Time-limited: Tokens expire after a configurable duration (default: 60 minutes) - Cryptographically signed: Tokens can't be tampered with or forged
This stateless design provides several benefits: - Server resilience: Tokens remain valid even if the server restarts - No memory consumption: No need to track sessions or store tokens in memory - High scalability: Multiple servers can validate tokens without shared state - Load balancing: Requests can be distributed across multiple servers freely
The token system secures both SWAIG functions and post-prompt endpoints: - SWAIG function calls for interactive AI capabilities - Post-prompt requests for receiving conversation summaries
You can disable token security for specific functions when appropriate:
@agent.tool(
name="get_public_information",
description="Get public information that doesn't require security",
parameters={},
secure=False # Disable token security for this function
)
def get_public_information(self, args, raw_data):
# Implementation
The default token expiration is 60 minutes (3600 seconds), but you can configure this when initializing your agent:
agent = MyAgent(
name="my_agent",
token_expiry_secs=1800 # Set token expiration to 30 minutes
)
The expiration timer resets each time a function is successfully called, so as long as there is activity at least once within the expiration period, the tokens will remain valid throughout the entire conversation.
You can override the default token validation by implementing your own validate_tool_token
method in your custom agent class.
The Skills System allows you to extend your agents with powerful capabilities using simple one-liner calls. Skills are modular, reusable components that can be easily added to any agent and configured with parameters.
from signalwire_agents import AgentBase
class SkillfulAgent(AgentBase):
def __init__(self):
super().__init__(name="skillful-agent", route="/skillful")
# Add skills with one-liners
self.add_skill("web_search") # Web search capability
self.add_skill("datetime") # Current date/time info
self.add_skill("math") # Mathematical calculations
# Configure skills with parameters
self.add_skill("web_search", {
"num_results": 3, # Get 3 search results instead of default 1
"delay": 0.5 # Add delay between requests
})
web_search
) 🔗 ↑ TOCProvides web search capabilities using Google Custom Search API with web scraping.
Requirements:
beautifulsoup4
, requests
Parameters:
api_key
(required): Google Custom Search API keysearch_engine_id
(required): Google Custom Search Engine IDnum_results
(default: 1): Number of search results to returndelay
(default: 0): Delay in seconds between requeststool_name
(default: "web_search"): Custom name for the search toolno_results_message
(default: "I couldn't find any results for '{query}'. This might be due to a very specific query or temporary issues. Try rephrasing your search or asking about a different topic."): Custom message to return when no search results are found. Use {query}
as a placeholder for the search query.Multiple Instance Support: The web_search skill supports multiple instances with different search engines and tool names, allowing you to search different data sources:
Example:
# Basic single instance
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "your-search-engine-id"
})
# Creates tool: web_search
# Fast single result (previous default)
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "your-search-engine-id",
"num_results": 1,
"delay": 0
})
# Multiple results with delay
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "your-search-engine-id",
"num_results": 5,
"delay": 1.0
})
# Multiple instances with different search engines
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "general-search-engine-id",
"tool_name": "search_general",
"num_results": 1
})
# Creates tool: search_general
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "news-search-engine-id",
"tool_name": "search_news",
"num_results": 3,
"delay": 0.5
})
# Creates tool: search_news
# Custom no results message
agent.add_skill("web_search", {
"api_key": "your-google-api-key",
"search_engine_id": "your-search-engine-id",
"no_results_message": "Sorry, I couldn't find information about '{query}'. Please try a different search term."
})
datetime
) 🔗 ↑ TOCProvides current date and time information with timezone support.
Requirements:
pytz
Tools Added:
get_current_time
: Get current time with optional timezoneget_current_date
: Get current date with optional timezoneExample:
agent.add_skill("datetime")
# Agent can now tell users the current time and date
math
) 🔗 ↑ TOCProvides safe mathematical expression evaluation.
Requirements:
Tools Added:
calculate
: Evaluate mathematical expressions safelyExample:
agent.add_skill("math")
# Agent can now perform calculations like "2 + 3 * 4"
datasphere
) 🔗 ↑ TOCProvides knowledge search capabilities using SignalWire DataSphere RAG stack.
Requirements:
requests
Parameters:
space_name
(required): SignalWire space nameproject_id
(required): SignalWire project ID token
(required): SignalWire authentication tokendocument_id
(required): DataSphere document ID to searchcount
(default: 1): Number of search results to returndistance
(default: 3.0): Distance threshold for search matchingtags
(optional): List of tags to filter search resultslanguage
(optional): Language code to limit searchpos_to_expand
(optional): List of parts of speech for synonym expansion (e.g., ["NOUN", "VERB"])max_synonyms
(optional): Maximum number of synonyms to use for each wordtool_name
(default: "search_knowledge"): Custom name for the search toolno_results_message
(default: "I couldn't find any relevant information for '{query}' in the knowledge base. Try rephrasing your question or asking about a different topic."): Custom message when no results foundMultiple Instance Support: The DataSphere skill supports multiple instances with different tool names, allowing you to search multiple knowledge bases:
Example:
# Basic single instance
agent.add_skill("datasphere", {
"space_name": "my-space",
"project_id": "my-project",
"token": "my-token",
"document_id": "general-knowledge"
})
# Creates tool: search_knowledge
# Multiple instances for different knowledge bases
agent.add_skill("datasphere", {
"space_name": "my-space",
"project_id": "my-project",
"token": "my-token",
"document_id": "product-docs",
"tool_name": "search_products",
"tags": ["Products", "Features"],
"count": 3
})
# Creates tool: search_products
agent.add_skill("datasphere", {
"space_name": "my-space",
"project_id": "my-project",
"token": "my-token",
"document_id": "support-kb",
"tool_name": "search_support",
"no_results_message": "I couldn't find support information about '{query}'. Try contacting our support team.",
"distance": 5.0
})
# Creates tool: search_support
native_vector_search
) 🔗 ↑ TOCProvides local document search capabilities using vector similarity and keyword search. This skill works entirely offline with local .swsearch
index files or can connect to remote search servers.
Requirements:
sentence-transformers
, scikit-learn
, numpy
(install with pip install signalwire-agents[search]
)Parameters:
tool_name
(default: "search_knowledge"): Custom name for the search tooldescription
(default: "Search the local knowledge base for information"): Tool descriptionindex_file
(optional): Path to local .swsearch
index fileremote_url
(optional): URL of remote search server (e.g., "http://localhost:8001")index_name
(default: "default"): Index name on remote server (for remote mode)build_index
(default: False): Auto-build index if missingsource_dir
(optional): Source directory for auto-building indexfile_types
(default: ["md", "txt"]): File types to include when building indexcount
(default: 3): Number of search results to returndistance_threshold
(default: 0.0): Minimum similarity score for resultstags
(optional): List of tags to filter search resultsresponse_prefix
(optional): Text to prepend to all search responsesresponse_postfix
(optional): Text to append to all search responsesno_results_message
(default: "No information found for '{query}'"): Custom message when no results foundMultiple Instance Support: The native vector search skill supports multiple instances with different indexes and tool names:
Example:
# Local mode with auto-build
agent.add_skill("native_vector_search", {
"tool_name": "search_docs",
"description": "Search SDK concepts guide",
"build_index": True,
"source_dir": "./docs",
"index_file": "concepts.swsearch",
"count": 5
})
# Creates tool: search_docs
# Remote mode connecting to search server
agent.add_skill("native_vector_search", {
"tool_name": "search_knowledge",
"description": "Search the knowledge base",
"remote_url": "http://localhost:8001",
"index_name": "concepts",
"count": 3
})
# Creates tool: search_knowledge
# Multiple local indexes
agent.add_skill("native_vector_search", {
"tool_name": "search_examples",
"description": "Search code examples",
"index_file": "examples.swsearch",
"response_prefix": "From the examples:"
})
# Creates tool: search_examples
# Voice-optimized responses using concepts guide
agent.add_skill("native_vector_search", {
"tool_name": "search_docs",
"index_file": "concepts.swsearch",
"response_prefix": "Based on the comprehensive SDK guide:",
"response_postfix": "Would you like more specific information?",
"no_results_message": "I couldn't find information about '{query}' in the concepts guide."
})
Building Search Indexes: Before using local mode, you need to build search indexes:
# Build index from documentation
python -m signalwire_agents.cli.build_search docs --output docs.swsearch
# Build with custom settings
python -m signalwire_agents.cli.build_search ./knowledge \
--output knowledge.swsearch \
--file-types md,txt,pdf \
--chunk-size 500 \
--verbose
For complete documentation on the search system, see Local Search System.
# Check what skills are loaded
loaded_skills = agent.list_skills()
print(f"Loaded skills: {', '.join(loaded_skills)}")
# Check if a specific skill is loaded
if agent.has_skill("web_search"):
print("Web search is available")
# Remove a skill (if needed)
agent.remove_skill("math")
Skills support a special swaig_fields
parameter that allows you to customize how SWAIG functions are registered. This parameter gets merged into the function decorator object, enabling the skill loader to add additional configuration to the tools.
# Add a skill with swaig_fields to customize SWAIG function properties
agent.add_skill("math", {
"precision": 2, # Regular skill parameter
"swaig_fields": { # Special fields merged into SWAIG function
"secure": False, # Override default security requirement
"fillers": {
"en-US": ["Let me calculate that...", "Computing the result..."],
"es-ES": ["Déjame calcular eso...", "Calculando el resultado..."]
}
}
})
# Add web search with custom security and fillers
agent.add_skill("web_search", {
"num_results": 3,
"delay": 0.5,
"swaig_fields": {
"secure": True, # Require authentication
"fillers": {
"en-US": ["Searching the web...", "Looking that up...", "Finding information..."]
}
}
})
The swaig_fields
can include any parameter accepted by AgentBase.define_tool()
:
- secure
: Boolean indicating if the function requires authentication
- fillers
: Dictionary mapping language codes to arrays of filler phrases
- Any other fields supported by the SWAIG function system
This feature enables advanced customization of how skills integrate with the agent's SWAIG system.
The skills system provides detailed error messages for common issues:
try:
agent.add_skill("web_search")
except ValueError as e:
print(f"Failed to load skill: {e}")
# Output: "Failed to load skill 'web_search': Missing required environment variables: ['GOOGLE_SEARCH_API_KEY']"
You can create your own skills by extending the SkillBase
class:
from signalwire_agents.core.skill_base import SkillBase
from signalwire_agents.core.function_result import SwaigFunctionResult
class WeatherSkill(SkillBase):
"""A custom skill for weather information"""
SKILL_NAME = "weather"
SKILL_DESCRIPTION = "Get weather information for locations"
SKILL_VERSION = "1.0.0"
REQUIRED_PACKAGES = ["requests"]
REQUIRED_ENV_VARS = ["WEATHER_API_KEY"]
def setup(self) -> bool:
"""Setup the skill - validate dependencies and initialize"""
if not self.validate_env_vars() or not self.validate_packages():
return False
# Get configuration parameters
self.default_units = self.params.get('units', 'fahrenheit')
self.timeout = self.params.get('timeout', 10)
return True
def register_tools(self) -> None:
"""Register tools with the agent"""
self.define_tool_with_swaig_fields(
name="get_weather",
description="Get current weather for a location",
parameters={
"location": {
"type": "string",
"description": "City or location name"
},
"units": {
"type": "string",
"description": "Temperature units (fahrenheit or celsius)",
"enum": ["fahrenheit", "celsius"]
}
},
handler=self._get_weather_handler
)
def _get_weather_handler(self, args, raw_data):
"""Handle weather requests"""
location = args.get("location", "")
units = args.get("units", self.default_units)
if not location:
return SwaigFunctionResult("Please provide a location")
# Your weather API integration here
weather_data = f"Weather for {location}: 72°F and sunny"
return SwaigFunctionResult(weather_data)
def get_hints(self) -> List[str]:
"""Return speech recognition hints"""
return ["weather", "temperature", "forecast", "conditions"]
def get_prompt_sections(self) -> List[Dict[str, Any]]:
"""Return prompt sections to add to agent"""
return [
{
"title": "Weather Information",
"body": "You can provide current weather information for any location.",
"bullets": [
"Use get_weather tool when users ask about weather",
"Always specify the location clearly",
"Include temperature and conditions in your response"
]
}
]
Using the custom skill:
# Place the skill in signalwire_agents/skills/weather/skill.py
# Then use it in your agent:
agent.add_skill("weather", {
"units": "celsius",
"timeout": 15
})
Skills work seamlessly with dynamic configuration:
class DynamicSkillAgent(AgentBase):
def __init__(self):
super().__init__(name="dynamic-skill-agent")
self.set_dynamic_config_callback(self.configure_per_request)
def configure_per_request(self, query_params, body_params, headers, agent):
# Add different skills based on request parameters
tier = query_params.get('tier', 'basic')
# Basic skills for all users
agent.add_skill("datetime")
agent.add_skill("math")
# Premium skills for premium users
if tier == 'premium':
agent.add_skill("web_search", {
"num_results": 5,
"delay": 0.5
})
elif tier == 'basic':
agent.add_skill("web_search", {
"num_results": 1,
"delay": 0
})
# For research (detailed analysis) agent.add_skill("web_search", {"num_results": 5, "delay": 1.0}) ```
Handle missing dependencies gracefully:
python
try:
agent.add_skill("web_search")
except ValueError as e:
self.logger.warning(f"Web search unavailable: {e}")
# Continue without web search capability
Document your custom skills: Include clear descriptions and parameter documentation
Test skills in isolation: Create simple test scripts to verify skill functionality
For more detailed information about the skills system architecture and advanced customization, see the Skills System README.
Agents can support multiple languages:
# Add English language
self.add_language(
name="English",
code="en-US",
voice="en-US-Neural2-F",
speech_fillers=["Let me think...", "One moment please..."],
function_fillers=["I'm looking that up...", "Let me check that..."]
)
# Add Spanish language
self.add_language(
name="Spanish",
code="es",
voice="rime.spore:multilingual",
speech_fillers=["Un momento por favor...", "Estoy pensando..."]
)
There are different ways to specify voices:
# Simple format
self.add_language(name="English", code="en-US", voice="en-US-Neural2-F")
# Explicit parameters with engine and model
self.add_language(
name="British English",
code="en-GB",
voice="spore",
engine="rime",
model="multilingual"
)
# Combined string format
self.add_language(
name="Spanish",
code="es",
voice="rime.spore:multilingual"
)
Hints help the AI understand certain terms better:
# Simple hints (list of words)
self.add_hints(["SignalWire", "SWML", "SWAIG"])
# Pattern hint with replacement
self.add_pattern_hint(
hint="AI Agent",
pattern="AI\\s+Agent",
replace="A.I. Agent",
ignore_case=True
)
Pronunciation rules help the AI speak certain terms correctly:
# Add pronunciation rule
self.add_pronunciation("API", "A P I", ignore_case=False)
self.add_pronunciation("SIP", "sip", ignore_case=True)
Configure various AI behavior parameters:
# Set AI parameters
self.set_params({
"wait_for_user": False,
"end_of_speech_timeout": 1000,
"ai_volume": 5,
"languages_enabled": True,
"local_tz": "America/Los_Angeles"
})
Provide global data for the AI to reference:
# Set global data
self.set_global_data({
"company_name": "SignalWire",
"product": "AI Agent SDK",
"supported_features": [
"Voice AI",
"Telephone integration",
"SWAIG functions"
]
})
Dynamic agent configuration allows you to configure agents per-request based on parameters from the HTTP request (query parameters, body data, headers). This enables powerful patterns like multi-tenant applications, A/B testing, personalization, and localization.
There are two main approaches to agent configuration:
class StaticAgent(AgentBase):
def __init__(self):
super().__init__(name="static-agent")
# Configuration happens once at startup
self.add_language("English", "en-US", "rime.spore:mistv2")
self.set_params({"end_of_speech_timeout": 500})
self.prompt_add_section("Role", "You are a customer service agent.")
self.set_global_data({"service_level": "standard"})
Pros: Simple, fast, predictable Cons: Same behavior for all users, requires separate agents for different configurations
class DynamicAgent(AgentBase):
def __init__(self):
super().__init__(name="dynamic-agent")
# No static configuration - set up dynamic callback instead
self.set_dynamic_config_callback(self.configure_per_request)
def configure_per_request(self, query_params, body_params, headers, agent):
# Configuration happens fresh for each request
tier = query_params.get('tier', 'standard')
if tier == 'premium':
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.set_params({"end_of_speech_timeout": 300}) # Faster
agent.prompt_add_section("Role", "You are a premium customer service agent.")
agent.set_global_data({"service_level": "premium"})
else:
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.set_params({"end_of_speech_timeout": 500}) # Standard
agent.prompt_add_section("Role", "You are a customer service agent.")
agent.set_global_data({"service_level": "standard"})
Pros: Highly flexible, single agent serves multiple configurations, enables advanced use cases Cons: Slightly more complex, configuration overhead per request
Use the set_dynamic_config_callback()
method to register a callback function that will be called for each request:
class MyDynamicAgent(AgentBase):
def __init__(self):
super().__init__(name="my-agent", route="/agent")
# Register the dynamic configuration callback
self.set_dynamic_config_callback(self.configure_agent_dynamically)
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
"""
This method is called for every request to configure the agent
Args:
query_params (dict): Query string parameters from the URL
body_params (dict): Parsed JSON body from POST requests
headers (dict): HTTP headers from the request
agent (EphemeralAgentConfig): Configuration object with familiar methods
"""
# Your dynamic configuration logic here
pass
The callback function receives four parameters: - query_params: Dictionary of URL query parameters - body_params: Dictionary of parsed JSON body (empty for GET requests) - headers: Dictionary of HTTP headers - agent: EphemeralAgentConfig object for configuration
The agent
parameter in your callback is an EphemeralAgentConfig
object that provides the same familiar methods as AgentBase
, but applies them per-request:
# Add languages with voice configuration
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.add_language("Spanish", "es-ES", "rime.spore:mistv2")
# Add prompt sections
agent.prompt_add_section("Role", "You are a helpful assistant.")
agent.prompt_add_section("Guidelines", bullets=[
"Be professional and courteous",
"Provide accurate information",
"Ask clarifying questions when needed"
])
# Set raw prompt text
agent.set_prompt_text("You are a specialized AI assistant...")
# Set post-prompt for summary
agent.set_post_prompt("Summarize the key points of this conversation.")
# Configure AI behavior
agent.set_params({
"end_of_speech_timeout": 300,
"attention_timeout": 20000,
"background_file_volume": -30
})
# Set data available to the AI
agent.set_global_data({
"customer_tier": "premium",
"features_enabled": ["advanced_support", "priority_queue"],
"session_info": {"start_time": "2024-01-01T00:00:00Z"}
})
# Update existing global data
agent.update_global_data({"additional_info": "value"})
# Add hints for better speech recognition
agent.add_hints(["SignalWire", "SWML", "API", "technical"])
agent.add_pronunciation("API", "A P I")
# Set native functions
agent.set_native_functions(["transfer", "hangup"])
# Add function includes
agent.add_function_include(
url="https://api.example.com/functions",
functions=["get_account_info", "update_profile"]
)
Your callback function receives detailed information about the incoming request:
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
# Extract query parameters
tier = query_params.get('tier', 'standard')
language = query_params.get('language', 'en')
customer_id = query_params.get('customer_id')
debug = query_params.get('debug', '').lower() == 'true'
# Use parameters for configuration
if tier == 'premium':
agent.set_params({"end_of_speech_timeout": 300})
if customer_id:
agent.set_global_data({"customer_id": customer_id})
# Request: GET /agent?tier=premium&language=es&customer_id=12345&debug=true
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
# Extract from POST body
user_profile = body_params.get('user_profile', {})
preferences = body_params.get('preferences', {})
# Configure based on profile
if user_profile.get('language') == 'es':
agent.add_language("Spanish", "es-ES", "rime.spore:mistv2")
if preferences.get('voice_speed') == 'fast':
agent.set_params({"end_of_speech_timeout": 200})
# Request: POST /agent with JSON body:
# {
# "user_profile": {"language": "es", "region": "mx"},
# "preferences": {"voice_speed": "fast", "tone": "formal"}
# }
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
# Extract headers
user_agent = headers.get('user-agent', '')
auth_token = headers.get('authorization', '')
locale = headers.get('accept-language', 'en-US')
# Configure based on headers
if 'mobile' in user_agent.lower():
agent.set_params({"end_of_speech_timeout": 400}) # Longer for mobile
if locale.startswith('es'):
agent.add_language("Spanish", "es-ES", "rime.spore:mistv2")
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
tenant = query_params.get('tenant', 'default')
# Tenant-specific configuration
if tenant == 'healthcare':
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.prompt_add_section("Compliance",
"Follow HIPAA guidelines and maintain patient confidentiality.")
agent.set_global_data({
"industry": "healthcare",
"compliance_level": "hipaa"
})
elif tenant == 'finance':
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.prompt_add_section("Compliance",
"Follow financial regulations and protect sensitive data.")
agent.set_global_data({
"industry": "finance",
"compliance_level": "pci"
})
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
language = query_params.get('language', 'en')
region = query_params.get('region', 'us')
# Configure language and voice
if language == 'es':
if region == 'mx':
agent.add_language("Spanish (Mexico)", "es-MX", "rime.spore:mistv2")
else:
agent.add_language("Spanish", "es-ES", "rime.spore:mistv2")
agent.prompt_add_section("Language", "Respond in Spanish.")
elif language == 'fr':
agent.add_language("French", "fr-FR", "rime.alois")
agent.prompt_add_section("Language", "Respond in French.")
else:
agent.add_language("English", "en-US", "rime.spore:mistv2")
# Regional customization
agent.set_global_data({
"language": language,
"region": region,
"currency": "USD" if region == "us" else "EUR" if region == "eu" else "MXN"
})
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
# Determine test group (could be from query param, user ID hash, etc.)
test_group = query_params.get('test_group', 'A')
if test_group == 'A':
# Control group - standard configuration
agent.set_params({"end_of_speech_timeout": 500})
agent.prompt_add_section("Style", "Use a standard conversational approach.")
agent.set_global_data({"test_group": "A", "features": ["basic"]})
else:
# Test group B - experimental features
agent.set_params({"end_of_speech_timeout": 300})
agent.prompt_add_section("Style",
"Use an enhanced, more interactive conversational approach.")
agent.set_global_data({"test_group": "B", "features": ["basic", "enhanced"]})
def configure_agent_dynamically(self, query_params, body_params, headers, agent):
customer_id = query_params.get('customer_id')
tier = query_params.get('tier', 'standard')
# Base configuration
agent.add_language("English", "en-US", "rime.spore:mistv2")
# Tier-specific configuration
if tier == 'enterprise':
agent.set_params({
"end_of_speech_timeout": 200, # Fastest response
"attention_timeout": 30000 # Longest attention span
})
agent.prompt_add_section("Service Level",
"You provide white-glove enterprise support with priority handling.")
features = ["all_features", "dedicated_support", "custom_integration"]
elif tier == 'premium':
agent.set_params({
"end_of_speech_timeout": 300,
"attention_timeout": 20000
})
agent.prompt_add_section("Service Level",
"You provide premium support with enhanced features.")
features = ["premium_features", "priority_support"]
else:
agent.set_params({
"end_of_speech_timeout": 500,
"attention_timeout": 15000
})
agent.prompt_add_section("Service Level",
"You provide standard customer support.")
features = ["basic_features"]
# Set global data
global_data = {"tier": tier, "features": features}
if customer_id:
global_data["customer_id"] = customer_id
agent.set_global_data(global_data)
Perfect for SaaS platforms where each customer needs different agent behavior:
# Different tenants get different capabilities
# /agent?tenant=acme&industry=healthcare
# /agent?tenant=globex&industry=finance
Benefits: - Single agent deployment serves all customers - Tenant-specific branding and behavior - Industry-specific compliance and terminology - Custom feature sets per subscription level
Test different agent configurations with real users:
# Split traffic between different configurations
# /agent?test_group=A (control)
# /agent?test_group=B (experimental)
Benefits: - Compare agent performance metrics - Test new features with subset of users - Gradual rollout of improvements - Data-driven optimization
Adapt agent behavior to individual user preferences:
# Personalized based on user profile
# /agent?user_id=123&voice_speed=fast&formality=casual
Benefits: - Improved user experience - Accessibility support (voice speed, etc.) - Cultural and linguistic adaptation - Learning from user interactions
Adapt to different regions and cultures:
# Location-based configuration
# /agent?country=mx&language=es&timezone=America/Mexico_City
Benefits: - Local language and dialect support - Cultural appropriateness - Regional business practices - Time zone aware responses
Step 1: Move Configuration to Callback
Before (Static):
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="my-agent")
# Static configuration
self.add_language("English", "en-US", "rime.spore:mistv2")
self.set_params({"end_of_speech_timeout": 500})
self.prompt_add_section("Role", "You are a helpful assistant.")
self.set_global_data({"version": "1.0"})
After (Dynamic):
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="my-agent")
# Set up dynamic configuration
self.set_dynamic_config_callback(self.configure_agent)
def configure_agent(self, query_params, body_params, headers, agent):
# Same configuration, but now dynamic
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.set_params({"end_of_speech_timeout": 500})
agent.prompt_add_section("Role", "You are a helpful assistant.")
agent.set_global_data({"version": "1.0"})
Step 2: Add Parameter-Based Logic
def configure_agent(self, query_params, body_params, headers, agent):
# Start with base configuration
agent.add_language("English", "en-US", "rime.spore:mistv2")
agent.prompt_add_section("Role", "You are a helpful assistant.")
# Add parameter-based customization
timeout = int(query_params.get('timeout', '500'))
agent.set_params({"end_of_speech_timeout": timeout})
version = query_params.get('version', '1.0')
agent.set_global_data({"version": version})
Step 3: Test Both Approaches
You can support both static and dynamic patterns during migration:
class MyAgent(AgentBase):
def __init__(self, use_dynamic=False):
super().__init__(name="my-agent")
if use_dynamic:
self.set_dynamic_config_callback(self.configure_agent)
else:
# Keep static configuration for backward compatibility
self._setup_static_config()
def _setup_static_config(self):
# Original static configuration
self.add_language("English", "en-US", "rime.spore:mistv2")
# ... rest of static config
def configure_agent(self, query_params, body_params, headers, agent):
# New dynamic configuration
# ... dynamic config logic
def configure_agent(self, query_params, body_params, headers, agent):
# Good: Simple parameter extraction and configuration
tier = query_params.get('tier', 'standard')
agent.set_params(TIER_CONFIGS[tier])
# Avoid: Heavy computation or external API calls
# customer_data = expensive_api_call(customer_id) # Don't do this
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="my-agent")
# Pre-compute configuration templates
self.tier_configs = {
'basic': {'end_of_speech_timeout': 500},
'premium': {'end_of_speech_timeout': 300},
'enterprise': {'end_of_speech_timeout': 200}
}
self.set_dynamic_config_callback(self.configure_agent)
def configure_agent(self, query_params, body_params, headers, agent):
tier = query_params.get('tier', 'basic')
agent.set_params(self.tier_configs[tier])
def configure_agent(self, query_params, body_params, headers, agent):
# Always provide defaults
language = query_params.get('language', 'en')
tier = query_params.get('tier', 'standard')
# Handle invalid values gracefully
if language not in ['en', 'es', 'fr']:
language = 'en'
def configure_agent(self, query_params, body_params, headers, agent):
# Validate and sanitize inputs
tier = query_params.get('tier', 'standard')
if tier not in ['basic', 'premium', 'enterprise']:
tier = 'basic' # Safe default
# Validate numeric parameters
try:
timeout = int(query_params.get('timeout', '500'))
timeout = max(100, min(timeout, 2000)) # Clamp to reasonable range
except ValueError:
timeout = 500 # Safe default
def configure_agent(self, query_params, body_params, headers, agent):
# Don't expose internal configuration via parameters
# Bad: agent.set_global_data({"api_key": query_params.get('api_key')})
# Good: Use internal mapping for call-related data only
customer_id = query_params.get('customer_id')
if customer_id and self.is_valid_customer(customer_id):
# Store call-related customer info, NOT sensitive credentials
agent.set_global_data({
"customer_id": customer_id,
"customer_tier": self.get_customer_tier(customer_id),
"account_type": "premium"
})
from functools import lru_cache
class MyAgent(AgentBase):
@lru_cache(maxsize=100)
def get_customer_config(self, customer_id):
# Cache expensive lookups
return self.database.get_customer_settings(customer_id)
def configure_agent(self, query_params, body_params, headers, agent):
customer_id = query_params.get('customer_id')
if customer_id:
config = self.get_customer_config(customer_id)
agent.set_global_data(config)
def configure_agent(self, query_params, body_params, headers, agent):
try:
# Try custom configuration
self.apply_custom_config(query_params, agent)
except Exception as e:
# Log error but don't fail the request
self.log.error("config_error", error=str(e))
# Fall back to default configuration
self.apply_default_config(agent)
def configure_agent(self, query_params, body_params, headers, agent):
# Validate required parameters
if not query_params.get('tenant'):
agent.set_global_data({"error": "Missing tenant parameter"})
return
# Validate configuration makes sense
language = query_params.get('language', 'en')
region = query_params.get('region', 'us')
if language == 'es' and region == 'us':
# Adjust for Spanish speakers in US
agent.add_language("Spanish (US)", "es-US", "rime.spore:mistv2")
Dynamic agent configuration is a powerful feature that enables sophisticated, multi-tenant AI applications while maintaining the familiar AgentBase API. Start with simple parameter-based configuration and gradually add more complex logic as your use cases evolve.
Enable state tracking to persist information across interactions:
# Enable state tracking in the constructor
super().__init__(
name="stateful-agent",
enable_state_tracking=True, # Automatically registers startup_hook and hangup_hook
state_manager=FileStateManager(storage_dir="./state") # Optional custom state manager
)
# Access and update state
@AgentBase.tool(
name="save_preference",
description="Save a user preference",
parameters={
"key": {
"type": "string",
"description": "The preference key"
},
"value": {
"type": "string",
"description": "The preference value"
}
}
)
def save_preference(self, args, raw_data):
# Get the call ID from the raw data
call_id = raw_data.get("call_id")
if call_id:
# Get current state or empty dict if none exists
state = self.get_state(call_id) or {}
# Update the state
preferences = state.get("preferences", {})
preferences[args.get("key")] = args.get("value")
state["preferences"] = preferences
# Save the updated state
self.update_state(call_id, state)
return SwaigFunctionResult("Preference saved")
else:
return SwaigFunctionResult("Could not save preference: No call ID")
SIP routing allows your agents to receive voice calls via SIP addresses. The SDK supports both individual agent-level routing and centralized server-level routing.
Enable SIP routing on a single agent:
# Enable SIP routing with automatic username mapping based on agent name
agent.enable_sip_routing(auto_map=True)
# Register additional SIP usernames for this agent
agent.register_sip_username("support_agent")
agent.register_sip_username("help_desk")
When auto_map=True
, the agent automatically registers SIP usernames based on:
- The agent's name (e.g., support@domain
)
- The agent's route path (e.g., /support
becomes support@domain
)
- Common variations (e.g., removing vowels for shorter dialing)
For multi-agent setups, centralized routing is more efficient:
# Create an AgentServer
server = AgentServer(host="0.0.0.0", port=3000)
# Register multiple agents
server.register(registration_agent) # Route: /register
server.register(support_agent) # Route: /support
# Set up central SIP routing
server.setup_sip_routing(route="/sip", auto_map=True)
# Register additional SIP username mappings
server.register_sip_username("signup", "/register") # signup@domain → registration agent
server.register_sip_username("help", "/support") # help@domain → support agent
With server-level routing:
- Each agent is reachable via its name (when auto_map=True
)
- Additional SIP usernames can be mapped to specific agent routes
- All SIP routing is handled at a single endpoint (/sip
by default)
support@yourdomain
)support
)You can dynamically handle requests to different paths using routing callbacks:
# Enable custom routing in the constructor or anytime after initialization
self.register_routing_callback(self.handle_customer_route, path="/customer")
self.register_routing_callback(self.handle_product_route, path="/product")
# Define the routing handlers
def handle_customer_route(self, request, body):
"""
Process customer-related requests
Args:
request: FastAPI Request object
body: Parsed JSON body as dictionary
Returns:
Optional[str]: A URL to redirect to, or None to process normally
"""
# Extract any relevant data
customer_id = body.get("customer_id")
# You can redirect to another agent/service if needed
if customer_id and customer_id.startswith("vip-"):
return f"/vip-handler/{customer_id}"
# Or return None to process the request with on_swml_request
return None
# Customize SWML based on the route in on_swml_request
def on_swml_request(self, request_data=None, callback_path=None):
"""
Customize SWML based on the request and path
Args:
request_data: The request body data
callback_path: The path that triggered the routing callback
"""
if callback_path == "/customer":
# Serve customer-specific content
return {
"sections": {
"main": [
{"answer": {}},
{"play": {"url": "say:Welcome to customer service!"}}
]
}
}
# Other path handling...
return None
You can modify the SWML document based on request data by overriding the on_swml_request
method:
def on_swml_request(self, request_data=None, callback_path=None):
"""
Customize the SWML document based on request data
Args:
request_data: The request data (body for POST or query params for GET)
callback_path: The path that triggered the routing callback
Returns:
Optional dict with modifications to apply to the document
"""
if request_data and "caller_type" in request_data:
# Example: Return modifications to change the AI behavior based on caller type
if request_data["caller_type"] == "vip":
return {
"sections": {
"main": [
# Keep the first verb (answer)
# Modify the AI verb parameters
{
"ai": {
"params": {
"wait_for_user": False,
"end_of_speech_timeout": 500 # More responsive
}
}
}
]
}
}
# You can also use the callback_path to serve different content based on the route
if callback_path == "/customer":
return {
"sections": {
"main": [
{"answer": {}},
{"play": {"url": "say:Welcome to our customer service line."}}
]
}
}
# Return None to use the default document
return None
Process conversation summaries:
def on_summary(self, summary, raw_data=None):
"""
Handle the conversation summary
Args:
summary: The summary object or None if no summary was found
raw_data: The complete raw POST data from the request
"""
if summary:
# Log the summary
self.log.info("conversation_summary", summary=summary)
# Save the summary to a database, send notifications, etc.
# ...
You can override the default webhook URLs for SWAIG functions and post-prompt delivery:
# In your agent initialization or setup code:
# Override the webhook URL for all SWAIG functions
agent.set_web_hook_url("https://external-service.example.com/handle-swaig")
# Override the post-prompt delivery URL
agent.set_post_prompt_url("https://analytics.example.com/conversation-summaries")
# These methods allow you to:
# 1. Send function calls to external services instead of handling them locally
# 2. Send conversation summaries to analytics services or other systems
# 3. Use special URLs with pre-configured authentication
The SDK provides a check-for-input endpoint that allows agents to check for new input from external systems:
# Example client code that checks for new input
import requests
import json
def check_for_new_input(agent_url, conversation_id, auth):
"""
Check if there's any new input for a conversation
Args:
agent_url: Base URL for the agent
conversation_id: ID of the conversation to check
auth: (username, password) tuple for basic auth
Returns:
New messages if any, None otherwise
"""
url = f"{agent_url}/check_for_input"
response = requests.post(
url,
json={"conversation_id": conversation_id},
auth=auth
)
if response.status_code == 200:
data = response.json()
if data.get("new_input", False):
return data.get("messages", [])
return None
By default, the check_for_input endpoint returns an empty response. To implement custom behavior, override the _handle_check_for_input_request
method in your agent:
async def _handle_check_for_input_request(self, request):
# First do basic authentication check
if not self._check_basic_auth(request):
return Response(
content=json.dumps({"error": "Unauthorized"}),
status_code=401,
headers={"WWW-Authenticate": "Basic"},
media_type="application/json"
)
# Get conversation_id from request
conversation_id = None
if request.method == "POST":
body = await request.json()
conversation_id = body.get("conversation_id")
else:
conversation_id = request.query_params.get("conversation_id")
if not conversation_id:
return Response(
content=json.dumps({"error": "Missing conversation_id"}),
status_code=400,
media_type="application/json"
)
# Custom logic to check for new input
# For example, checking a database or external API
messages = self._get_new_messages(conversation_id)
return {
"status": "success",
"conversation_id": conversation_id,
"new_input": len(messages) > 0,
"messages": messages
}
This endpoint is useful for implementing asynchronous conversations where users might send messages through different channels that need to be incorporated into the agent conversation.
Prefab agents are pre-configured agent implementations designed for specific use cases. They provide ready-to-use functionality with customization options, saving development time and ensuring consistent patterns.
The SDK includes several built-in prefab agents:
Collects structured information from users:
from signalwire_agents.prefabs import InfoGathererAgent
agent = InfoGathererAgent(
fields=[
{"name": "full_name", "prompt": "What is your full name?"},
{"name": "email", "prompt": "What is your email address?",
"validation": r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"},
{"name": "reason", "prompt": "How can I help you today?"}
],
confirmation_template="Thanks {full_name}, I'll help you with {reason}. I'll send a confirmation to {email}.",
name="info-gatherer",
route="/info-gatherer"
)
agent.serve(host="0.0.0.0", port=8000)
Answers questions based on a knowledge base:
from signalwire_agents.prefabs import FAQBotAgent
agent = FAQBotAgent(
knowledge_base_path="./docs",
personality="I'm a product documentation assistant.",
citation_style="inline",
name="knowledge-base",
route="/knowledge-base"
)
agent.serve(host="0.0.0.0", port=8000)
Routes users to specialized agents:
from signalwire_agents.prefabs import ConciergeAgent
agent = ConciergeAgent(
routing_map={
"technical_support": {
"url": "http://tech-support-agent:8001",
"criteria": ["error", "broken", "not working"]
},
"sales": {
"url": "http://sales-agent:8002",
"criteria": ["pricing", "purchase", "subscribe"]
}
},
greeting="Welcome to SignalWire. How can I help you today?",
name="concierge",
route="/concierge"
)
agent.serve(host="0.0.0.0", port=8000)
Conducts structured surveys with different question types:
from signalwire_agents.prefabs import SurveyAgent
agent = SurveyAgent(
survey_name="Customer Satisfaction",
introduction="We'd like to know about your recent experience with our product.",
questions=[
{
"id": "satisfaction",
"text": "How satisfied are you with our product?",
"type": "rating",
"scale": 5,
"labels": {
"1": "Very dissatisfied",
"5": "Very satisfied"
}
},
{
"id": "feedback",
"text": "Do you have any specific feedback about how we can improve?",
"type": "text"
}
],
name="satisfaction-survey",
route="/survey"
)
agent.serve(host="0.0.0.0", port=8000)
Handles call routing and department transfers:
from signalwire_agents.prefabs import ReceptionistAgent
agent = ReceptionistAgent(
departments=[
{"name": "sales", "description": "For product inquiries and pricing", "number": "+15551235555"},
{"name": "support", "description": "For technical assistance", "number": "+15551236666"},
{"name": "billing", "description": "For payment and invoice questions", "number": "+15551237777"}
],
greeting="Thank you for calling ACME Corp. How may I direct your call?",
voice="rime.spore:mistv2",
name="acme-receptionist",
route="/reception"
)
agent.serve(host="0.0.0.0", port=8000)
You can create your own prefab agents by extending AgentBase
or any existing prefab. Custom prefabs can be created directly within your project or packaged as reusable libraries.
A well-designed prefab should:
AgentBase
or another prefabExample of a custom support agent prefab:
from signalwire_agents import AgentBase
from signalwire_agents.core.function_result import SwaigFunctionResult
class CustomerSupportAgent(AgentBase):
def __init__(
self,
product_name,
knowledge_base_path=None,
support_email=None,
escalation_path=None,
**kwargs
):
# Pass standard params to parent
super().__init__(**kwargs)
# Store custom configuration
self._product_name = product_name
self._knowledge_base_path = knowledge_base_path
self._support_email = support_email
self._escalation_path = escalation_path
# Configure prompt
self.prompt_add_section("Personality",
body=f"I am a customer support agent for {product_name}.")
self.prompt_add_section("Goal",
body="Help customers solve their problems effectively.")
# Add standard instructions
self._configure_instructions()
# Register default tools
self._register_default_tools()
def _configure_instructions(self):
"""Configure standard instructions based on settings"""
instructions = [
"Be professional but friendly.",
"Verify the customer's identity before sharing account details."
]
if self._escalation_path:
instructions.append(
f"For complex issues, offer to escalate to {self._escalation_path}."
)
self.prompt_add_section("Instructions", bullets=instructions)
def _register_default_tools(self):
"""Register default tools if appropriate paths are configured"""
if self._knowledge_base_path:
self.register_knowledge_base_tool()
def register_knowledge_base_tool(self):
"""Register the knowledge base search tool if configured"""
# Implementation...
pass
@AgentBase.tool(
name="escalate_issue",
description="Escalate a customer issue to a human agent",
parameters={
"issue_summary": {"type": "string", "description": "Brief summary of the issue"},
"customer_email": {"type": "string", "description": "Customer's email address"}
}
)
def escalate_issue(self, args, raw_data):
# Implementation...
return SwaigFunctionResult("Issue escalated successfully.")
@AgentBase.tool(
name="send_support_email",
description="Send a follow-up email to the customer",
parameters={
"customer_email": {"type": "string"},
"issue_summary": {"type": "string"},
"resolution_steps": {"type": "string"}
}
)
def send_support_email(self, args, raw_data):
# Implementation...
return SwaigFunctionResult("Follow-up email sent successfully.")
# Create an instance of the custom prefab
support_agent = CustomerSupportAgent(
product_name="SignalWire Voice API",
knowledge_base_path="./product_docs",
support_email="support@example.com",
escalation_path="tier 2 support",
name="voice-support",
route="/voice-support"
)
# Start the agent
support_agent.serve(host="0.0.0.0", port=8000)
You can also extend and customize the built-in prefabs:
from signalwire_agents.prefabs import InfoGathererAgent
class EnhancedGatherer(InfoGathererAgent):
def __init__(self, fields, **kwargs):
super().__init__(fields=fields, **kwargs)
# Add an additional instruction
self.prompt_add_section("Instructions", bullets=[
"Verify all information carefully."
])
# Add an additional custom tool
@AgentBase.tool(
name="check_customer",
description="Check customer status in database",
parameters={"email": {"type": "string"}}
)
def check_customer(self, args, raw_data):
# Implementation...
return SwaigFunctionResult("Customer status: Active")
To create distributable prefabs that can be used across multiple projects:
Example package structure:
my-prefab-agents/
├── README.md
├── setup.py
├── examples/
│ └── support_agent_example.py
└── my_prefab_agents/
├── __init__.py
├── support.py
├── retail.py
└── utils/
├── __init__.py
└── knowledge_base.py
name
: Agent name/identifier (required)route
: HTTP route path (default: "/")host
: Host to bind to (default: "0.0.0.0")port
: Port to bind to (default: 3000)basic_auth
: Optional (username, password) tupleuse_pom
: Whether to use POM for prompts (default: True)enable_state_tracking
: Enable conversation state (default: False)token_expiry_secs
: State token expiry time (default: 3600)auto_answer
: Auto-answer calls (default: True)record_call
: Record calls (default: False)state_manager
: Custom state manager (default: None)schema_path
: Optional path to schema.json filesuppress_logs
: Whether to suppress structured logs (default: False)prompt_add_section(title, body=None, bullets=None, numbered=False, numbered_bullets=False)
prompt_add_subsection(parent_title, title, body=None, bullets=None)
prompt_add_to_section(title, body=None, bullet=None, bullets=None)
set_prompt_text(prompt_text)
or set_prompt(prompt_text)
set_post_prompt(prompt_text)
setPersonality(text)
- Convenience method that calls prompt_add_sectionsetGoal(text)
- Convenience method that calls prompt_add_sectionsetInstructions(bullets)
- Convenience method that calls prompt_add_section@AgentBase.tool(name, description, parameters={}, secure=True, fillers=None)
define_tool(name, description, parameters, handler, secure=True, fillers=None)
set_native_functions(function_names)
add_native_function(function_name)
remove_native_function(function_name)
add_function_include(url, functions, meta_data=None)
add_hint(hint)
and add_hints(hints)
add_pattern_hint(hint, pattern, replace, ignore_case=False)
add_pronunciation(replace, with_text, ignore_case=False)
add_language(name, code, voice, speech_fillers=None, function_fillers=None, engine=None, model=None)
set_param(key, value)
and set_params(params_dict)
set_global_data(data_dict)
and update_global_data(data_dict)
get_state(call_id)
set_state(call_id, data)
update_state(call_id, data)
clear_state(call_id)
cleanup_expired_state()
enable_sip_routing(auto_map=True, path="/sip")
: Enable SIP routing for an agentregister_sip_username(sip_username)
: Register a SIP username for an agentauto_map_sip_usernames()
: Automatically register SIP usernames based on agent attributessetup_sip_routing(route="/sip", auto_map=True)
: Set up central SIP routing for a serverregister_sip_username(username, route)
: Map a SIP username to an agent routeserve(host=None, port=None)
: Start the web serveras_router()
: Return a FastAPI router for this agenton_swml_request(request_data=None, callback_path=None)
: Customize SWML based on request data and pathon_summary(summary, raw_data=None)
: Handle post-prompt summarieson_function_call(name, args, raw_data=None)
: Process SWAIG function callsregister_routing_callback(callback_fn, path="/sip")
: Register a callback for custom path routingset_web_hook_url(url)
: Override the default web_hook_url with a supplied URL stringset_post_prompt_url(url)
: Override the default post_prompt_url with a supplied URL stringThe SDK provides several endpoints for different purposes:
/
): Serves the main SWML document/swaig
): Handles SWAIG function calls/post_prompt
): Processes conversation summaries/check_for_input
): Supports checking for new input from external systems/debug
): Serves the SWML document with debug headers/sip
): Handles SIP routing requestsThe SignalWire AI Agent SDK provides comprehensive testing capabilities through the swaig-test
CLI tool, which allows you to test agents locally and simulate serverless environments without deployment.
Test your agents locally before deployment:
# Discover agents in a file
swaig-test examples/my_agent.py
# List available functions
swaig-test examples/my_agent.py --list-tools
# Test SWAIG functions
swaig-test examples/my_agent.py --exec get_weather --location "New York"
# Generate SWML documents
swaig-test examples/my_agent.py --dump-swml
Test your agents in simulated serverless environments to ensure they work correctly when deployed:
# Basic Lambda environment simulation
swaig-test examples/my_agent.py --simulate-serverless lambda --dump-swml
# Test with custom Lambda configuration
swaig-test examples/my_agent.py --simulate-serverless lambda \
--aws-function-name my-production-function \
--aws-region us-west-2 \
--exec my_function --param value
# Test function execution in Lambda context
swaig-test examples/my_agent.py --simulate-serverless lambda \
--exec get_weather --location "Miami" \
--full-request
# Test CGI environment
swaig-test examples/my_agent.py --simulate-serverless cgi \
--cgi-host my-server.com \
--cgi-https \
--dump-swml
# Test function in CGI context
swaig-test examples/my_agent.py --simulate-serverless cgi \
--cgi-host example.com \
--exec my_function --param value
# Test Cloud Functions environment
swaig-test examples/my_agent.py --simulate-serverless cloud_function \
--gcp-project my-project \
--gcp-function-url https://my-function.cloudfunctions.net \
--dump-swml
# Test Azure Functions environment
swaig-test examples/my_agent.py --simulate-serverless azure_function \
--azure-env production \
--azure-function-url https://my-function.azurewebsites.net \
--exec my_function
Use environment files for consistent testing across different platforms:
# Create environment file for production testing
cat > production.env << EOF
AWS_LAMBDA_FUNCTION_NAME=prod-my-agent
AWS_REGION=us-east-1
API_KEY=prod_api_key_123
DEBUG=false
TIMEOUT=60
EOF
# Test with environment file
swaig-test examples/my_agent.py --simulate-serverless lambda \
--env-file production.env \
--exec critical_function --input "test"
# Override specific variables
swaig-test examples/my_agent.py --simulate-serverless lambda \
--env-file production.env \
--env DEBUG=true \
--dump-swml
Test the same agent across multiple platforms to ensure compatibility:
# Test across all platforms
for platform in lambda cgi cloud_function azure_function; do
echo "Testing $platform..."
swaig-test examples/my_agent.py --simulate-serverless $platform \
--exec my_function --param value
done
# Compare SWML generation across platforms
swaig-test examples/my_agent.py --simulate-serverless lambda --dump-swml > lambda.swml
swaig-test examples/my_agent.py --simulate-serverless cgi --cgi-host example.com --dump-swml > cgi.swml
diff lambda.swml cgi.swml
The serverless simulation automatically generates platform-appropriate webhook URLs:
Platform | Example Webhook URL |
---|---|
Lambda (Function URL) | https://abc123.lambda-url.us-east-1.on.aws/swaig/ |
Lambda (API Gateway) | https://api123.execute-api.us-east-1.amazonaws.com/prod/swaig/ |
CGI | https://example.com/cgi-bin/agent.cgi/swaig/ |
Cloud Functions | https://my-function-abc123.cloudfunctions.net/swaig/ |
Azure Functions | https://my-function.azurewebsites.net/swaig/ |
Verify webhook URLs are generated correctly:
# Check Lambda webhook URL
swaig-test examples/my_agent.py --simulate-serverless lambda \
--dump-swml --format-json | jq '.sections.main[1].ai.SWAIG.defaults.web_hook_url'
# Check CGI webhook URL
swaig-test examples/my_agent.py --simulate-serverless cgi \
--cgi-host my-production-server.com \
--dump-swml --format-json | jq '.sections.main[1].ai.SWAIG.defaults.web_hook_url'
--verbose
for debugging environment setup and executionFor files with multiple agents, specify which agent to test:
# Discover available agents
swaig-test multi_agent_file.py --list-agents
# Test specific agent
swaig-test multi_agent_file.py --agent-class MyAgent --simulate-serverless lambda --dump-swml
# Test different agents across platforms
swaig-test multi_agent_file.py --agent-class AgentA --simulate-serverless lambda --exec function1
swaig-test multi_agent_file.py --agent-class AgentB --simulate-serverless cgi --cgi-host example.com --exec function2
For more detailed testing documentation, see the CLI Testing Guide.
from signalwire_agents import AgentBase
from signalwire_agents.core.function_result import SwaigFunctionResult
from datetime import datetime
class SimpleAgent(AgentBase):
def __init__(self):
super().__init__(
name="simple",
route="/simple",
use_pom=True
)
# Configure agent personality
self.prompt_add_section("Personality", body="You are a friendly and helpful assistant.")
self.prompt_add_section("Goal", body="Help users with basic tasks and answer questions.")
self.prompt_add_section("Instructions", bullets=[
"Be concise and direct in your responses.",
"If you don't know something, say so clearly.",
"Use the get_time function when asked about the current time."
])
@AgentBase.tool(
name="get_time",
description="Get the current time",
parameters={}
)
def get_time(self, args, raw_data):
"""Get the current time"""
now = datetime.now()
formatted_time = now.strftime("%H:%M:%S")
return SwaigFunctionResult(f"The current time is {formatted_time}")
def main():
agent = SimpleAgent()
print("Starting agent server...")
print("Note: Works in any deployment mode (server/CGI/Lambda)")
agent.run()
if __name__ == "__main__":
main()
class CustomerServiceAgent(AgentBase):
def __init__(self):
super().__init__(
name="customer-service",
route="/support",
use_pom=True
)
# Configure agent personality
self.prompt_add_section("Personality",
body="You are a helpful customer service representative for SignalWire.")
self.prompt_add_section("Knowledge",
body="You can answer questions about SignalWire products and services.")
self.prompt_add_section("Instructions", bullets=[
"Greet customers politely",
"Answer questions about SignalWire products",
"Use check_account_status when customer asks about their account",
"Use create_support_ticket for unresolved issues"
])
# Add language support
self.add_language(
name="English",
code="en-US",
voice="en-US-Neural2-F",
speech_fillers=["Let me think...", "One moment please..."],
function_fillers=["I'm looking that up...", "Let me check that..."]
)
self.add_language(
name="Spanish",
code="es",
voice="rime.spore:multilingual",
speech_fillers=["Un momento por favor...", "Estoy pensando..."]
)
# Enable languages
self.set_params({"languages_enabled": True})
# Add company information
self.set_global_data({
"company_name": "SignalWire",
"support_hours": "9am-5pm ET, Monday through Friday",
"support_email": "support@signalwire.com"
})
@AgentBase.tool(
name="check_account_status",
description="Check the status of a customer's account",
parameters={
"account_id": {
"type": "string",
"description": "The customer's account ID"
}
}
)
def check_account_status(self, args, raw_data):
account_id = args.get("account_id")
# In a real implementation, this would query a database
return SwaigFunctionResult(f"Account {account_id} is in good standing.")
@AgentBase.tool(
name="create_support_ticket",
description="Create a support ticket for an unresolved issue",
parameters={
"issue": {
"type": "string",
"description": "Brief description of the issue"
},
"priority": {
"type": "string",
"description": "Ticket priority",
"enum": ["low", "medium", "high", "critical"]
}
}
)
def create_support_ticket(self, args, raw_data):
issue = args.get("issue", "")
priority = args.get("priority", "medium")
# Generate a ticket ID (in a real system, this would create a database entry)
ticket_id = f"TICKET-{hash(issue) % 10000:04d}"
return SwaigFunctionResult(
f"Support ticket {ticket_id} has been created with {priority} priority. " +
"A support representative will contact you shortly."
)
def main():
agent = CustomerServiceAgent()
print("Starting customer service agent...")
print("Note: Works in any deployment mode (server/CGI/Lambda)")
agent.run()
if __name__ == "__main__":
main()
For working examples of dynamic agent configuration, see these files in the examples
directory:
simple_static_agent.py
: Traditional static configuration approachsimple_dynamic_agent.py
: Same agent but using dynamic configurationsimple_dynamic_enhanced.py
: Enhanced version that actually uses request parameterscomprehensive_dynamic_agent.py
: Advanced multi-tier, multi-industry dynamic agentcustom_path_agent.py
: Dynamic agent with custom routing pathmulti_agent_server.py
: Multiple specialized dynamic agents on one serverThese examples demonstrate the progression from static to dynamic configuration and show real-world use cases like multi-tenant applications, A/B testing, and personalization.
For more examples, see the examples
directory in the SignalWire AI Agent SDK repository.
sw-search docs/signalwire_agents_concepts_guide.md --output concepts.swsearch
sw-search docs/signalwire_agents_concepts_guide.md examples README.md --output comprehensive.swsearch
sw-search ./knowledge \ --output knowledge.swsearch \ --file-types md,txt,pdf \ --chunking-strategy sentence \ --max-sentences-per-chunk 8 \ --verbose