Understanding SWML, AI Agents, and SWAIG
SignalWire AI represents a paradigm shift in conversational AI infrastructure, arriving at a critical inflection point in communications technology. Built on SignalWire's battle-tested telecommunications platform, it provides the only complete solution that unifies real-time communication protocols with advanced AI capabilities.
We are experiencing the first re-engineering of global voice platforms in over 15 years, driven by several converging forces:
SignalWire AI serves as the universal voice pipes for any form of AI communication, providing:
Key Differentiators:
SignalWire AI operates on a vertically integrated architecture that spans from hardware-level telecom infrastructure to high-level AI abstractions:
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β Developer Layer β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β Python β β SWML β β SWAIG β β
β β SDK β β Documents β β Functions β β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β AI Orchestration β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β Conversationβ β Function β β State β β
β β Manager β β Executor β β Manager β β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β Media Processing β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β Speech β β Text β β Video β β
β β Recognition β β to Speech β β Processing β β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β Protocol & Transport β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β SIP β β WebRTC β β PSTN β β
β β Protocol β β Transport β β Connectivityβ β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β Global Infrastructure β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β Regional β β Carrier β β Edge β β
β β Data β β Relations β β Presence β β
β β Centers β β β β β β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
SignalWire's architecture is built upon FreeSWITCH, the open-source telephony platform that powers 95% of the cloud communications industry. This foundation provides:
SignalWire has created a category in the communications space called PUC (Programmable Unified Communications) - a cloud replacement for local VoIP infrastructure and services that makes communication programmable, composable, and easy to manage.
Call Fabric is SignalWire's implementation of Programmable Unified Communications, providing the foundation for building communications solutions through composable resources.
SignalWire introduces composable telecom infrastructure where every element is modular and reusable:
SWML scripts, Subscribers, AI Agents, and Rooms are resources in SignalWire that enable a resource model similar to familiar Internet components. Clients connect to Resources via Addresses, making applications available at scale.
When resources are composed, you can use APIs to interact with callers and build custom solutions:
SignalWire's infrastructure spans 48 worldwide Points of Presence (POPs) with cloud-agnostic architecture:
SignalWire operates a massive global telecommunications infrastructure that includes:
Audio Codecs:
βββ Opus (48kHz, variable bitrate)
βββ G.722 (16kHz wideband)
βββ G.711 (8kHz, ΞΌ-law/A-law)
βββ G.729 (8kHz, compressed)
βββ AMR-WB (adaptive multi-rate)
Video Codecs:
βββ VP8/VP9 (WebRTC standard)
βββ H.264 (universal compatibility)
βββ H.265/HEVC (next-generation)
βββ AV1 (open standard)
The real-time media processing pipeline operates with strict latency requirements:
SignalWire maintains carrier-grade quality standards:
SWML is a declarative JSON-based markup language designed specifically for defining AI agent behavior in real-time communication scenarios. Unlike traditional APIs that require imperative programming, SWML allows developers to describe what they want the AI to do rather than how to do it.
Every SWML document follows this fundamental structure:
{
"version": "1.0.0",
"sections": {
"main": [
{
"ai": {
"prompt": {
"text": "You are a helpful assistant..."
},
"post_prompt": {
"text": "Summarize the conversation..."
},
"params": {
"temperature": 0.7,
"max_tokens": 1000
}
}
}
]
}
}
The ai verb is the primary building block for conversational AI in SWML. It configures:
Example AI verb with advanced configuration:
{
"ai": {
"prompt": {
"text": "You are a customer service representative for Acme Corp."
},
"params": {
"temperature": 0.3,
"max_tokens": 500,
"end_of_speech_timeout": 300
},
"languages": [
{
"name": "English",
"code": "en-US",
"voice": "elevenlabs.rachel"
}
],
"SWAIG": {
"functions": [
{
"function": "check_order_status",
"description": "Check the status of a customer order",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID to check"
}
},
"required": ["order_id"]
}
}
]
}
}
}
SWML documents are executed by the SignalWire AI engine using a sophisticated state machine:
{
"switch": {
"variable": "customer_tier",
"case": {
"premium": [
{"ai": {"prompt": "VIP greeting for premium customers"}}
],
"standard": [
{"ai": {"prompt": "Standard greeting"}}
]
}
}
}
{
"loop": {
"count": 3,
"do": [
{"ai": {"prompt": "Please provide your account number"}},
{"input": {"max_digits": 10, "timeout": 30}}
]
}
}
{
"set": {
"customer_id": "12345",
"call_timestamp": "${now()}"
}
}
SignalWire's AI Agent system is fundamentally different from traditional AI platforms. Instead of bolting AI capabilities onto existing communication infrastructure, SignalWire embeds the AI kernel directly into the media stack.
Voice, transcription, memory, and LLM reasoning run inside the media layer - no detours.
Traditional AI communication systems suffer from architectural inefficiencies:
βββββββββββββββ ββββββββββββββββ βββββββββββββββ ββββββββββββββββ
β Audio βββββΆβ TranscriptionβββββΆβ AI API βββββΆβ Text-to- β
β Stream β β Service β β Call β β Speech β
βββββββββββββββ ββββββββββββββββ βββββββββββββββ ββββββββββββββββ
Network hop Network hop Network hop Network hop
SignalWire's embedded approach eliminates these hops:
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β SignalWire Media Stack β
β βββββββββββββββ ββββββββββββββββ βββββββββββββββ ββββββββββββββββ β
β β Audio ββββ€ Transcriptionββββ€ AI Reasoningββββ€ Text-to-Speechβ β
β β Stream β β Engine β β Engine β β Engine β β
β βββββββββββββββ ββββββββββββββββ βββββββββββββββ ββββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
Single memory space
The AI Agent system is built around a conversation-centric architecture that manages:
The conversation manager maintains sophisticated state across multiple dimensions:
conversation_context = {
"messages": [
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "What's the weather like?"},
{"role": "assistant", "content": "I'll check the weather for you"},
{"role": "function", "name": "get_weather", "content": "72Β°F, sunny"}
],
"metadata": {
"turn_count": 2,
"function_calls": 1,
"start_time": "2024-01-15T10:30:00Z"
}
}
The system automatically manages context windows, employing strategies like:
The AI system prevents common issues like:
The AI system supports multiple input and output modalities:
SWAIG (SignalWire AI Gateway) is SignalWire's production-ready equivalent of the Model Context Protocol (MCP). It serves as a secure, high-performance interface that allows AI agents to call external functions and services, and has been in stable production use for over a year.
SWAIG provides:
Functions are defined using a JSON Schema-based format:
{
"function": "get_customer_balance",
"description": "Retrieve current account balance for a customer",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "Unique customer identifier"
},
"account_type": {
"type": "string",
"enum": ["checking", "savings", "credit"],
"description": "Type of account to check"
}
},
"required": ["customer_id"]
},
"web_hook_url": "https://mybank.com/api/balance",
"web_hook_auth_user": "api_user",
"web_hook_auth_password": "secure_password"
}
When an AI agent needs to call a function:
{
"function": "get_customer_balance",
"argument": {
"parsed": [{"customer_id": "12345", "account_type": "checking"}],
"raw": "{\"customer_id\":\"12345\",\"account_type\":\"checking\"}"
},
"call_id": "abc-123-def",
"ai_session_id": "session-456",
"conversation_id": "conv-789",
"caller_id_name": "John Doe",
"caller_id_num": "+15551234567",
"project_id": "proj-123",
"space_id": "space-456"
}
{
"response": "The checking account balance is $1,247.83",
"action": [
{
"say": "Your current checking account balance is $1,247.83"
}
]
}
{
"response": "Found customer information",
"action": [
{
"toggle_functions": [
{"function": "basic_lookup", "active": false},
{"function": "detailed_lookup", "active": true}
]
}
]
}
{
"response": "Customer verified",
"action": [
{
"set": {
"customer_verified": true,
"verification_timestamp": "${now()}"
}
}
]
}
{
"response": "Transferring to specialist",
"action": [
{
"SWML": {
"version": "1.0.0",
"sections": {
"main": [
{"transfer": {"to": "+15551234567"}}
]
}
}
}
]
}
The Prompt Object Model is SignalWire's structured approach to AI prompt management. Instead of simple text prompts, POM allows complex, multi-section prompts that can include:
{
"prompt": {
"pom": [
{
"role": "system",
"content": {
"type": "text",
"text": "You are a ${role} for ${company}"
}
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "Help me with ${task}"
},
{
"type": "audio",
"url": "https://example.com/context.wav"
}
]
}
]
}
}
POM supports real-time variable substitution:
{
"prompt": "You are helping ${customer_name} (ID: ${customer_id}) who called at ${call_time}. Their account status is ${account_status}."
}
Variables can be: - Static: Set at agent initialization - Dynamic: Updated during conversation - Computed: Calculated from functions or system state - Contextual: Derived from conversation history
POM supports different prompt formats for different AI models:
{
"prompt": {
"text": "You are a helpful assistant",
"messages": [
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello"}
],
"vision": {
"system": "You can see and describe images",
"image_url": "https://example.com/image.jpg"
}
}
}
SignalWire operates a massive global telecommunications infrastructure that includes:
Audio Codecs:
βββ Opus (48kHz, variable bitrate)
βββ G.722 (16kHz wideband)
βββ G.711 (8kHz, ΞΌ-law/A-law)
βββ G.729 (8kHz, compressed)
βββ AMR-WB (adaptive multi-rate)
Video Codecs:
βββ VP8/VP9 (WebRTC standard)
βββ H.264 (universal compatibility)
βββ H.265/HEVC (next-generation)
βββ AV1 (open standard)
The real-time media processing pipeline operates with strict latency requirements:
SignalWire maintains carrier-grade quality standards:
The SignalWire AI Python SDK fundamentally reimagines how developers control real-time communication. Instead of maintaining complex stateful connections, the SDK operates on a stateless client/server model where your Python code can dynamically control live calls through HTTP webhooks.
from signalwire_agents import AgentBase, SwaigFunctionResult
class CallControlAgent(AgentBase):
def __init__(self):
super().__init__(name="CallControl")
@tool(name="transfer_to_specialist")
def transfer_to_specialist(self, args, raw_data):
department = args.get("department", "sales")
# Dynamically control the live call
if department == "billing":
phone_number = "+15551234567"
elif department == "technical":
phone_number = "+15559876543"
else:
phone_number = "+15555555555"
# Return SWML to execute call transfer
return SwaigFunctionResult(
f"Transferring you to {department}",
action=[{
"transfer": {
"to": phone_number,
"timeout": 30
}
}]
)
@tool(name="start_conference")
def start_conference(self, args, raw_data):
# Create conference and invite participants
return SwaigFunctionResult(
"Starting conference call",
action=[{
"conference": {
"name": f"meeting-{raw_data.get('call_id')}",
"participants": args.get("participants", [])
}
}]
)
The SDK's Skills System transforms complex AI capabilities into reusable, composable building blocks:
class SuperAgent(AgentBase):
def __init__(self):
super().__init__(name="SuperAgent")
# Add web search capability instantly
self.add_skill("web_search", {
"num_results": 5,
"delay": 0.5 # Rate limiting
})
# Add datetime awareness
self.add_skill("datetime")
# Add mathematical computation
self.add_skill("math")
# Now your agent can search the web, handle dates, and do math
# without writing any additional code
def get_prompt(self):
return """You are a research assistant with web search,
date/time awareness, and mathematical capabilities."""
from signalwire_agents.core.skill_base import SkillBase
class DatabaseSkill(SkillBase):
SKILL_NAME = "database_lookup"
def execute(self, query: str, table: str):
# Your custom database logic
results = self.db.query(table, query)
return f"Found {len(results)} results"
# Register and use your custom skill
agent.register_skill(DatabaseSkill())
agent.add_skill("database_lookup")
The SDK completely abstracts away SWML complexity, letting you control sophisticated call flows with simple Python:
class DynamicFlowAgent(AgentBase):
@tool(name="route_customer")
def route_customer(self, args, raw_data):
customer_tier = args.get("tier")
issue_type = args.get("issue_type")
# Complex routing logic in simple Python
if customer_tier == "premium" and issue_type == "technical":
return SwaigFunctionResult(
"Connecting you to our senior technical team",
action=[{
"connect": {
"to": "premium-tech-queue",
"music": "premium_hold_music"
}
}]
)
elif issue_type == "billing":
return SwaigFunctionResult(
"Let me get your account details first",
action=[{
"gather": {
"input": "speech",
"timeout": 10,
"hints": ["account number", "phone number"]
}
}]
)
else:
# Default to AI conversation
return SwaigFunctionResult("How can I help you today?")
from signalwire_agents.core.data_map import DataMap
class CRMAgent(AgentBase):
def __init__(self):
super().__init__(name="CRMAgent")
# Define complex API integration in a few lines
customer_lookup = (DataMap('get_customer')
.description('Look up customer information')
.parameter('phone', 'string', 'Customer phone number', required=True)
.webhook('GET', 'https://crm.company.com/api/customers/phone/${args.phone}')
.output(SwaigFunctionResult('Customer: ${response.name} - Account: ${response.account_id}'))
.error_keys(['error', 'not_found'])
)
# Register as a callable function during conversations
self.register_swaig_function(customer_lookup.to_swaig_function())
class SessionAwareAgent(AgentBase):
def __init__(self):
super().__init__(name="SessionAgent")
self.sessions = {} # In-memory session store
def on_call_start(self, call_id: str, caller_data: dict):
# Initialize session state
self.sessions[call_id] = {
"start_time": datetime.now(),
"caller_id": caller_data.get("caller_id_num"),
"interaction_count": 0
}
def on_call_end(self, call_id: str):
# Clean up session and log analytics
session = self.sessions.pop(call_id, {})
duration = datetime.now() - session.get("start_time", datetime.now())
self.log.info("call_completed",
call_id=call_id,
duration=duration.seconds,
interactions=session.get("interaction_count", 0))
@tool(name="track_interaction")
def track_interaction(self, args, raw_data):
call_id = raw_data.get("call_id")
if call_id in self.sessions:
self.sessions[call_id]["interaction_count"] += 1
return SwaigFunctionResult("Interaction tracked")
# This is ALL you need for a production-ready AI agent
agent = MyAgent()
agent.run() # Auto-detects environment (server/CGI/Lambda)
# Mix and match capabilities like building blocks
agent.add_skill("web_search")
agent.add_skill("database_lookup")
agent.add_skill("payment_processing")
# Agent now has web search, database access, and payment capabilities
# Generate complex SWML dynamically based on business logic
def generate_menu_swml(self, customer_tier):
if customer_tier == "premium":
return {
"play": {"url": "https://assets.com/premium_greeting.mp3"},
"gather": {
"input": "dtmf speech",
"choices": ["sales", "support", "account_manager"]
}
}
else:
return {
"play": {"text": "Thank you for calling"},
"gather": {
"input": "dtmf",
"choices": ["1", "2", "3"]
}
}
/health and /ready endpointsSignalWire AI implements security at multiple layers:
Function calls use a sophisticated token-based security system:
# Token generation (handled automatically by SDK)
tool_token = session_manager.create_tool_token(
tool_name="get_customer_data",
call_id="abc-123",
permissions=["read:customer", "read:account"],
expires_in=3600 # 1 hour
)
# Token validation during function calls
def validate_function_call(request):
token = request.headers.get("Authorization")
function_name = request.json.get("function")
call_id = request.json.get("call_id")
if not session_manager.validate_tool_token(
function_name, token, call_id
):
raise UnauthorizedError("Invalid or expired token")
SignalWire AI is designed to support enterprise compliance requirements and has demonstrated compliance capabilities across multiple standards:
SignalWire's architecture provides the foundation for meeting various industry compliance requirements, with proven implementations for healthcare (HIPAA), payment processing (PCI), and telecommunications (data sovereignty) use cases.
SignalWire AI is engineered for real-time performance:
Component | Typical Latency | Max Latency
----------------------------|-----------------|-------------
Speech Recognition | 50-100ms | 200ms
AI Model Response | 200-500ms | 1000ms
Text-to-Speech Synthesis | 100-200ms | 400ms
SWAIG Function Call | 100-300ms | 1000ms
End-to-End Turn Latency | 500-1000ms | 2000ms
SignalWire AI provides built-in structured logging capabilities using structlog:
# Built-in structured logging (actually available)
from signalwire_agents import AgentBase
class CustomerServiceAgent(AgentBase):
def __init__(self):
super().__init__(name="CustomerService")
@tool(name="process_payment")
def process_payment(self, args, raw_data):
# Use built-in structured logging
self.log.info("Processing payment", customer_id=args.get("customer_id"))
try:
result = self._process_payment_internal(args)
self.log.info("Payment successful", transaction_id=result.get("id"))
return SwaigFunctionResult(f"Payment processed: {result.get('id')}")
except Exception as e:
self.log.error("Payment failed", error=str(e))
return SwaigFunctionResult("Payment processing failed")
def _process_payment_internal(self, args):
# Integration with external systems
pass
SignalWire AI enables a wide range of applications across industries. The platform is trusted by customers of all sizes, from startups to large enterprises, powering everything from consumer device messaging and contact center solutions to enterprise AI voice systems and next-generation telecommunications infrastructure.
class CRMIntegratedAgent(AgentBase):
def __init__(self, crm_config):
super().__init__(name="CRMAgent")
self.crm = CRMClient(crm_config)
@tool(name="lookup_customer")
def lookup_customer(self, args, raw_data):
phone = raw_data.get("caller_id_num")
customer = self.crm.find_by_phone(phone)
if customer:
return SwaigFunctionResult(
f"Hello {customer.name}, I see you're calling about account {customer.account_id}"
)
else:
return SwaigFunctionResult("I don't find an account with this number")
@tool(name="create_ticket")
def create_ticket(self, args, raw_data):
ticket = self.crm.create_ticket(
customer_id=args.get("customer_id"),
issue=args.get("issue_description"),
priority=args.get("priority", "medium")
)
return SwaigFunctionResult(f"Created ticket #{ticket.id}")
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult
class DatabaseAgent(AgentBase):
def __init__(self):
super().__init__(name="DatabaseAgent")
# Create DataMap tool for database operations
user_lookup = (DataMap('get_user_info')
.description('Get user information from database')
.parameter('user_id', 'string', 'User ID to look up', required=True)
.webhook('GET', 'https://your-api.com/users/${args.user_id}')
.output(SwaigFunctionResult('User: ${response.name} (${response.email}) - Status: ${response.status}'))
.error_keys(['error'])
)
# Register the DataMap tool as a SWAIG function
self.register_swaig_function(user_lookup.to_swaig_function())
class PaymentAgent(AgentBase):
def __init__(self, stripe_key):
super().__init__(name="PaymentAgent")
self.stripe = stripe.StripeClient(stripe_key)
@tool(name="process_payment")
def process_payment(self, args, raw_data):
try:
charge = self.stripe.charges.create(
amount=int(args.get("amount") * 100), # Stripe uses cents
currency="usd",
source=args.get("payment_token"),
description=args.get("description")
)
return SwaigFunctionResult(
f"Payment processed successfully. Transaction ID: {charge.id}"
)
except stripe.error.CardError as e:
return SwaigFunctionResult(
f"Payment failed: {e.user_message}"
)
# Use SignalWire AI behind an enterprise API gateway
class EnterpriseAgent(AgentBase):
def __init__(self):
super().__init__(name="Enterprise")
self.configure_auth(
method="oauth2",
client_id="enterprise_client",
client_secret="secure_secret"
)
def get_auth_headers(self):
token = self.oauth_client.get_access_token()
return {"Authorization": f"Bearer {token}"}
# Agent as part of a microservices ecosystem
class OrderServiceAgent(AgentBase):
def __init__(self, service_discovery):
super().__init__(name="OrderService")
self.services = service_discovery
@tool(name="create_order")
def create_order(self, args, raw_data):
# Call order service
order_service = self.services.get_service("order-service")
order = order_service.create_order(args)
# Call inventory service
inventory_service = self.services.get_service("inventory-service")
inventory_service.reserve_items(order.items)
# Call payment service
payment_service = self.services.get_service("payment-service")
payment = payment_service.process_payment(order.total)
return SwaigFunctionResult(f"Order {order.id} created successfully")
Install the SDK:
bash
pip install signalwire-agents
Create Your First Agent: ```python from signalwire_agents import AgentBase
class HelloWorldAgent(AgentBase): def get_prompt(self): return "You are a friendly AI assistant"
agent = HelloWorldAgent() agent.run() ```
For production deployments:
class ProductionAgent(AgentBase):
def __init__(self):
super().__init__(
name="production-agent",
port=8080,
basic_auth=("user", "secure_password")
)
# Configure logging
self.setup_logging(level="INFO")
# Add custom middleware
self.add_middleware(CORSMiddleware)
self.add_middleware(SecurityMiddleware)
def get_prompt(self):
return self.load_prompt_from_file("prompts/customer_service.txt")
def on_call_start(self, call_id: str):
# Custom call initialization
self.log.info(f"Call started: {call_id}")
def on_call_end(self, call_id: str):
# Cleanup and analytics
self.analytics.track_call_completion(call_id)
SignalWire AI represents the convergence of enterprise-grade telecommunications infrastructure and cutting-edge AI technology. By providing native protocol support, real-time processing capabilities, and a developer-friendly SDK, it enables the creation of sophisticated conversational AI applications that can scale to millions of users.
The combination of SWML's declarative approach, SWAIG's powerful function integration, and the platform's global telecommunications infrastructure makes SignalWire AI the definitive platform for building production-ready AI agents.
Whether you're building customer service bots, sales automation tools, or complex multi-modal AI applications, SignalWire AI provides the foundation for reliable, scalable, and secure conversational AI solutions.