In this lesson, you'll master advanced features of the SignalWire Agents SDK and learn production deployment best practices. We'll cover custom SWAIG functions, error handling, debugging techniques, and deployment strategies.
SWAIG (SignalWire AI Gateway) functions allow your agent to perform actions beyond conversation. These can integrate with APIs, databases, or perform calculations.
from signalwire_agents import AgentBase
from signalwire_agents.core.function_result import SwaigFunctionResult
class MyAgent(AgentBase):
def __init__(self):
super().__init__(name="My Agent", route="/")
# Define a function using the decorator
@self.tool("calculate_price", description="Calculate total price with tax")
async def calculate_price(amount: float, tax_rate: float = 0.08):
"""
Calculate total price including tax
Args:
amount: Base price
tax_rate: Tax rate (default 8%)
Returns:
SwaigFunctionResult with calculation
"""
tax = amount * tax_rate
total = amount + tax
return SwaigFunctionResult(
f"The total price is ${total:.2f} "
f"(${amount:.2f} + ${tax:.2f} tax)"
)
Required Parameters:
@self.tool(
"create_order",
description="Create a new order",
required=["customer_name", "items"] # These params are required
)
async def create_order(
customer_name: str,
items: str,
priority: str = "normal" # Optional with default
):
# Implementation
Parameter Types:
# Supported types
async def example_function(
text: str, # String
number: int, # Integer
decimal: float, # Float
flag: bool, # Boolean
optional: str = None # Optional
):
pass
The SwaigFunctionResult class provides rich responses:
@self.tool("check_inventory", description="Check product availability")
async def check_inventory(product_id: str):
# Simulate inventory check
in_stock = 5
if in_stock > 0:
result = SwaigFunctionResult(f"Product {product_id} is in stock ({in_stock} units)")
# Add structured data
result.add_data({
"product_id": product_id,
"quantity": in_stock,
"warehouse": "main"
})
# Add action for the agent
result.add_action("set_global_data", {
"last_checked_product": product_id,
"stock_level": in_stock
})
return result
else:
# Return error state
return SwaigFunctionResult(
f"Product {product_id} is out of stock",
error=True
)
Both patterns are supported:
# Async function (recommended for I/O operations)
@self.tool("fetch_data", description="Fetch data from API")
async def fetch_data(query: str):
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.example.com/{query}") as resp:
data = await resp.json()
return SwaigFunctionResult(f"Found {len(data)} results")
# Sync function (for quick calculations)
@self.tool("calculate", description="Perform calculation")
def calculate(x: int, y: int):
return SwaigFunctionResult(f"Result: {x + y}")
Proper error handling ensures your agent gracefully handles failures.
@self.tool("process_order", description="Process customer order")
async def process_order(order_id: str):
try:
# Validate input
if not order_id or len(order_id) < 5:
return SwaigFunctionResult(
"Invalid order ID format",
error=True
)
# Simulate processing
if order_id.startswith("TEST"):
raise ValueError("Test orders cannot be processed")
# Success case
return SwaigFunctionResult(f"Order {order_id} processed successfully")
except ValueError as e:
return SwaigFunctionResult(
f"Order processing failed: {str(e)}",
error=True
)
except Exception as e:
# Log unexpected errors
logger.error(f"Unexpected error processing order: {e}")
return SwaigFunctionResult(
"An unexpected error occurred. Please try again.",
error=True
)
class RobustAgent(AgentBase):
def __init__(self):
super().__init__(name="Robust Agent", route="/")
# Add error handling instructions to prompt
self.prompt_add_section(
"Error Handling",
body="How to handle errors gracefully:",
bullets=[
"If a function returns an error, acknowledge it politely",
"Offer alternative solutions when possible",
"Never expose technical error details to customers",
"Always maintain a helpful, professional tone"
]
)
@self.tool("update_customer", description="Update customer information")
async def update_customer(customer_id: str, email: str = None, phone: str = None):
# Input validation
errors = []
if not customer_id:
errors.append("Customer ID is required")
if email and "@" not in email:
errors.append("Invalid email format")
if phone and len(phone) < 10:
errors.append("Phone number must be at least 10 digits")
if errors:
return SwaigFunctionResult(
f"Validation failed: {', '.join(errors)}",
error=True
)
# Process valid input
return SwaigFunctionResult("Customer updated successfully")
Effective logging is crucial for troubleshooting and monitoring.
from signalwire_agents.core.logging_config import get_logger
logger = get_logger(__name__)
class DebugAgent(AgentBase):
def __init__(self):
super().__init__(name="Debug Agent", route="/")
logger.info("Initializing Debug Agent")
@self.tool("debug_function", description="Test function with logging")
async def debug_function(param: str):
logger.debug(f"Function called with param: {param}")
try:
# Some operation
result = param.upper()
logger.info(f"Operation successful: {result}")
return SwaigFunctionResult(result)
except Exception as e:
logger.error(f"Operation failed: {e}", exc_info=True)
return SwaigFunctionResult("Operation failed", error=True)
# Set log level when creating server
server = AgentServer(log_level="debug")
# Or via environment variable
export SWML_LOG_LEVEL=debug
# Available levels:
# - debug: Detailed information for debugging
# - info: General information (default)
# - warning: Warning messages
# - error: Error messages only
# - critical: Critical issues only
1. Request Logging:
def configure_dynamic(self, query_params, body_params, headers, agent):
logger.debug(f"Query params: {query_params}")
logger.debug(f"Body params: {body_params}")
logger.debug(f"Headers: {headers}")
2. SWML Inspection:
# Dump SWML without running
swaig-test agent.py --dump-swml
# Test specific functions
swaig-test agent.py --exec function_name --param value
3. Interactive Debugging:
@self.tool("debug_state", description="Debug agent state")
async def debug_state():
import json
state = {
"agent_name": self.get_name(),
"functions": list(self._tools.keys()),
"languages": self._languages
}
logger.info(f"Agent state: {json.dumps(state, indent=2)}")
return SwaigFunctionResult("State logged to console")
# Core configuration
export SWML_AUTH_USER=produser
export SWML_AUTH_PASS=strongpassword
export SWML_LOG_LEVEL=info
# SSL configuration
export SWML_SSL_ENABLED=true
export SWML_SSL_CERT_PATH=/etc/ssl/certs/agent.crt
export SWML_SSL_KEY_PATH=/etc/ssl/private/agent.key
export SWML_DOMAIN=agents.example.com
# Performance tuning
export PYTORCH_DISABLE_AVX512=1 # For compatibility
export WORKERS=4 # Number of worker processes
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY . .
# Create non-root user
RUN useradd -m -u 1000 agent && chown -R agent:agent /app
USER agent
# Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
CMD curl -f http://localhost:3000/health || exit 1
# Run agent
CMD ["python", "agent.py"]
# /etc/systemd/system/signalwire-agent.service
[Unit]
Description=SignalWire AI Agent
After=network.target
[Service]
Type=simple
User=agent
WorkingDirectory=/opt/signalwire-agent
Environment="SWML_LOG_LEVEL=info"
Environment="SWML_SSL_ENABLED=true"
ExecStart=/usr/bin/python3 /opt/signalwire-agent/agent.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
# Add custom health checks
@server.app.get("/health/detailed")
async def detailed_health():
return {
"status": "healthy",
"timestamp": datetime.now().isoformat(),
"agents": {
"triage": "active",
"sales": "active",
"support": "active"
},
"checks": {
"database": check_database_connection(),
"search_index": check_search_index(),
"memory_usage": get_memory_usage()
}
}
import pytest
from signalwire_agents.core.function_result import SwaigFunctionResult
@pytest.mark.asyncio
async def test_calculate_price():
# Create agent instance
agent = MyAgent()
# Get the function
calc_func = agent._tools["calculate_price"]["function"]
# Test normal case
result = await calc_func(amount=100.0, tax_rate=0.08)
assert isinstance(result, SwaigFunctionResult)
assert "108.00" in result.message
# Test edge cases
result = await calc_func(amount=0, tax_rate=0)
assert "0.00" in result.message
# test_integration.py
import requests
import json
def test_agent_swml_generation():
"""Test that agent generates valid SWML"""
response = requests.get("http://localhost:3000/")
assert response.status_code == 200
swml = response.json()
assert "ai" in swml
assert "prompt" in swml["ai"]
assert "voice" in swml["ai"]
def test_function_execution():
"""Test function execution via swaig-test"""
import subprocess
result = subprocess.run(
["swaig-test", "agent.py", "--exec", "calculate_price", "--amount", "100"],
capture_output=True,
text=True
)
assert "108.00" in result.stdout
# Using Apache Bench
ab -n 1000 -c 10 http://localhost:3000/
# Using curl in a loop
for i in {1..100}; do
curl -s http://localhost:3000/ > /dev/null &
done
wait
from functools import lru_cache
import asyncio
class OptimizedAgent(AgentBase):
def __init__(self):
super().__init__(name="Optimized Agent", route="/")
self._cache = {}
@self.tool("get_product_info", description="Get product information")
async def get_product_info(product_id: str):
# Check cache first
if product_id in self._cache:
logger.debug(f"Cache hit for {product_id}")
return SwaigFunctionResult(self._cache[product_id])
# Expensive operation
info = await fetch_from_database(product_id)
# Cache for 5 minutes
self._cache[product_id] = info
asyncio.create_task(self._expire_cache(product_id, 300))
return SwaigFunctionResult(info)
async def _expire_cache(self, key: str, seconds: int):
await asyncio.sleep(seconds)
self._cache.pop(key, None)
# Good: Concurrent operations
@self.tool("get_full_info", description="Get complete information")
async def get_full_info(customer_id: str):
# Run multiple queries concurrently
orders, profile, preferences = await asyncio.gather(
get_orders(customer_id),
get_profile(customer_id),
get_preferences(customer_id)
)
return SwaigFunctionResult(f"Found {len(orders)} orders")
# Bad: Sequential operations
async def get_full_info_slow(customer_id: str):
orders = await get_orders(customer_id) # Waits
profile = await get_profile(customer_id) # Then waits
preferences = await get_preferences(customer_id) # Then waits
# Limit search results
self.add_skill("native_vector_search", {
"tool_name": "search_knowledge",
"index_file": "knowledge.swsearch",
"count": 3 # Limit results to reduce memory
})
# Clean up large objects
@self.tool("process_large_data", description="Process large dataset")
async def process_large_data(dataset_id: str):
data = await load_large_dataset(dataset_id)
result = process_data(data)
# Explicitly clean up
del data
return SwaigFunctionResult(f"Processed {result['count']} items")
import re
@self.tool("safe_search", description="Search with sanitized input")
async def safe_search(query: str):
# Sanitize input
safe_query = re.sub(r'[^\w\s-]', '', query)
safe_query = safe_query.strip()[:100] # Limit length
if not safe_query:
return SwaigFunctionResult("Invalid search query", error=True)
# Safe to use
results = await search_database(safe_query)
return SwaigFunctionResult(f"Found {len(results)} results")
import os
from typing import Optional
class SecureAgent(AgentBase):
def __init__(self):
super().__init__(name="Secure Agent", route="/")
# Load secrets from environment
self._api_key = os.environ.get("API_KEY")
if not self._api_key:
logger.warning("API_KEY not set")
@self.tool("secure_api_call", description="Make secure API call")
async def secure_api_call(endpoint: str):
if not self._api_key:
return SwaigFunctionResult("API not configured", error=True)
# Never log secrets
logger.info(f"Calling API endpoint: {endpoint}")
# logger.info(f"Using key: {self._api_key}") # NEVER DO THIS
headers = {"Authorization": f"Bearer {self._api_key}"}
# Make API call...
from datetime import datetime, timedelta
from collections import defaultdict
class RateLimitedAgent(AgentBase):
def __init__(self):
super().__init__(name="Rate Limited Agent", route="/")
self._call_counts = defaultdict(list)
@self.tool("limited_function", description="Rate limited function")
async def limited_function(user_id: str):
# Check rate limit (10 calls per minute)
now = datetime.now()
minute_ago = now - timedelta(minutes=1)
# Clean old entries
self._call_counts[user_id] = [
t for t in self._call_counts[user_id]
if t > minute_ago
]
# Check limit
if len(self._call_counts[user_id]) >= 10:
return SwaigFunctionResult(
"Rate limit exceeded. Please try again later.",
error=True
)
# Record call
self._call_counts[user_id].append(now)
# Process normally
return SwaigFunctionResult("Function executed successfully")
You've mastered advanced SignalWire Agents features! You've learned:
Technical Skills:
Best Practices:
What's Next?
In the final lesson, you'll learn how to extend agents with custom skills, create complex conversation flows, and integrate with external services.
Before deploying to production:
← Lesson 3: Building Multi-Agent Systems | Tutorial Overview | Lesson 5: Extending Your Agents →