Advanced Topics

Multi Agent

View as Markdown

Multi-Agent Servers

Run multiple agents on a single server using AgentServer. Each agent gets its own route, and you can configure SIP-based routing for username-to-agent mapping.

Multi-agent servers let you run several specialized agents from a single process. This simplifies deployment, reduces resource overhead, and provides unified management. Instead of running separate processes for sales, support, and billing agents, you run one server that routes requests to the appropriate agent.

This architecture is especially useful when you have related agents that share infrastructure but have different personas and capabilities.

Single Agent vs Multi-Agent: Decision Guide

Choosing between agent.run() and AgentServer depends on your deployment needs.

Use single agent (agent.run()) when:

  • You have one agent with a single purpose
  • You want the simplest possible deployment
  • Each agent needs isolated resources (memory, CPU)
  • Agents have very different scaling requirements
  • You’re using container orchestration that handles multi-instance deployment

Use AgentServer when:

  • You have multiple related agents (sales, support, billing)
  • Agents share the same deployment environment
  • You want unified health monitoring
  • SIP routing determines which agent handles a call
  • You want to reduce operational overhead of managing multiple processes
  • Agents share common code or resources
Single Agent (agent.run())AgentServer
One agent per processMultiple agents per process
Simple deploymentShared resources
Separate ports per agentSingle port, multiple routes
Independent scalingShared scaling
Isolated failuresUnified monitoring
SIP username routing
Unified health checks

Basic AgentServer

1from signalwire_agents import AgentBase, AgentServer
2
3
4class SalesAgent(AgentBase):
5    def __init__(self):
6        super().__init__(name="sales-agent")
7        self.add_language("English", "en-US", "rime.spore")
8        self.prompt_add_section("Role", "You are a sales representative.")
9
10
11class SupportAgent(AgentBase):
12    def __init__(self):
13        super().__init__(name="support-agent")
14        self.add_language("English", "en-US", "rime.spore")
15        self.prompt_add_section("Role", "You are a support specialist.")
16
17
18if __name__ == "__main__":
19    server = AgentServer(host="0.0.0.0", port=3000)
20
21    server.register(SalesAgent(), "/sales")
22    server.register(SupportAgent(), "/support")
23
24    server.run()

Agents are available at:

EndpointDescription
http://localhost:3000/salesSales agent
http://localhost:3000/supportSupport agent
http://localhost:3000/healthBuilt-in health check

AgentServer Configuration

1server = AgentServer(
2    host="0.0.0.0",     # Bind address
3    port=3000,          # Listen port
4    log_level="info"    # debug, info, warning, error, critical
5)

Registering Agents

With Explicit Route

1server.register(SalesAgent(), "/sales")

Using Agent’s Default Route

1class BillingAgent(AgentBase):
2    def __init__(self):
3        super().__init__(
4            name="billing-agent",
5            route="/billing"  # Default route
6        )
7
8server.register(BillingAgent())  # Uses "/billing"

Server Architecture

Server Architecture.
Server Architecture

Managing Agents

Get All Agents

1agents = server.get_agents()
2for route, agent in agents:
3    print(f"{route}: {agent.get_name()}")

Get Specific Agent

1sales_agent = server.get_agent("/sales")

Unregister Agent

1server.unregister("/sales")

SIP Routing

Route SIP calls to specific agents based on username:

1from signalwire_agents import AgentBase, AgentServer
2
3
4class SalesAgent(AgentBase):
5    def __init__(self):
6        super().__init__(name="sales-agent")
7        self.add_language("English", "en-US", "rime.spore")
8        self.prompt_add_section("Role", "You are a sales representative.")
9
10
11class SupportAgent(AgentBase):
12    def __init__(self):
13        super().__init__(name="support-agent")
14        self.add_language("English", "en-US", "rime.spore")
15        self.prompt_add_section("Role", "You are a support specialist.")
16
17
18if __name__ == "__main__":
19    server = AgentServer()
20
21    server.register(SalesAgent(), "/sales")
22    server.register(SupportAgent(), "/support")
23
24    # Enable SIP routing
25    server.setup_sip_routing("/sip", auto_map=True)
26
27    # Manual SIP username mapping
28    server.register_sip_username("sales-team", "/sales")
29    server.register_sip_username("help-desk", "/support")
30
31    server.run()

When auto_map=True, the server automatically creates mappings:

  • Agent name → route (e.g., “salesagent” → “/sales”)
  • Route path → route (e.g., “sales” → “/sales”)

SIP Routing Flow

SIP Routing Flow.
SIP Routing Flow

Health Check Endpoint

AgentServer provides a built-in health check:

$curl http://localhost:3000/health

Response:

1{
2    "status": "ok",
3    "agents": 2,
4    "routes": ["/sales", "/support"]
5}

Serverless Deployment

AgentServer supports serverless environments automatically:

1from signalwire_agents import AgentBase, AgentServer
2
3
4class MyAgent(AgentBase):
5    def __init__(self):
6        super().__init__(name="my-agent")
7        self.add_language("English", "en-US", "rime.spore")
8
9
10server = AgentServer()
11server.register(MyAgent(), "/agent")
12
13
14## AWS Lambda handler
15def lambda_handler(event, context):
16    return server.run(event, context)
17
18
19## CGI mode (auto-detected)
20if __name__ == "__main__":
21    server.run()

Complete Example

1#!/usr/bin/env python3
2## multi_agent_server.py - Server with multiple specialized agents
3from signalwire_agents import AgentBase, AgentServer
4from signalwire_agents.core.function_result import SwaigFunctionResult
5
6
7class SalesAgent(AgentBase):
8    def __init__(self):
9        super().__init__(name="sales-agent")
10        self.add_language("English", "en-US", "rime.spore")
11
12        self.prompt_add_section(
13            "Role",
14            "You are a sales representative for Acme Corp."
15        )
16
17        self.define_tool(
18            name="get_pricing",
19            description="Get product pricing",
20            parameters={
21                "type": "object",
22                "properties": {
23                    "product": {"type": "string", "description": "Product name"}
24                },
25                "required": ["product"]
26            },
27            handler=self.get_pricing
28        )
29
30    def get_pricing(self, args, raw_data):
31        product = args.get("product", "")
32        # Pricing lookup logic
33        return SwaigFunctionResult(f"The price for {product} is $99.99")
34
35
36class SupportAgent(AgentBase):
37    def __init__(self):
38        super().__init__(name="support-agent")
39        self.add_language("English", "en-US", "rime.spore")
40
41        self.prompt_add_section(
42            "Role",
43            "You are a technical support specialist."
44        )
45
46        self.define_tool(
47            name="create_ticket",
48            description="Create a support ticket",
49            parameters={
50                "type": "object",
51                "properties": {
52                    "issue": {"type": "string", "description": "Issue description"}
53                },
54                "required": ["issue"]
55            },
56            handler=self.create_ticket
57        )
58
59    def create_ticket(self, args, raw_data):
60        issue = args.get("issue", "")
61        # Ticket creation logic
62        return SwaigFunctionResult(f"Created ticket #12345 for: {issue}")
63
64
65class BillingAgent(AgentBase):
66    def __init__(self):
67        super().__init__(name="billing-agent")
68        self.add_language("English", "en-US", "rime.spore")
69
70        self.prompt_add_section(
71            "Role",
72            "You help customers with billing questions."
73        )
74
75
76if __name__ == "__main__":
77    # Create server
78    server = AgentServer(host="0.0.0.0", port=3000)
79
80    # Register agents
81    server.register(SalesAgent(), "/sales")
82    server.register(SupportAgent(), "/support")
83    server.register(BillingAgent(), "/billing")
84
85    # Enable SIP routing
86    server.setup_sip_routing("/sip", auto_map=True)
87
88    # Custom SIP mappings
89    server.register_sip_username("sales", "/sales")
90    server.register_sip_username("help", "/support")
91    server.register_sip_username("accounts", "/billing")
92
93    print("Agents available:")
94    for route, agent in server.get_agents():
95        print(f"  {route}: {agent.get_name()}")
96
97    server.run()

AgentServer Methods Summary

MethodPurpose
register(agent, route)Register an agent at a route
unregister(route)Remove an agent
get_agents()Get all registered agents
get_agent(route)Get agent by route
setup_sip_routing(route, auto_map)Enable SIP-based routing
register_sip_username(username, route)Map SIP username to route
run()Start the server

Performance Considerations

Running multiple agents in a single process has implications:

Memory: Each agent maintains its own state, but they share the Python interpreter. For most deployments, this reduces overall memory compared to separate processes.

CPU: Agents share CPU resources. A heavy-load agent can affect others. Monitor and adjust if needed.

Startup time: All agents initialize when the server starts. More agents = longer startup.

Isolation: A crash in one agent’s handler can affect the entire server. Implement proper error handling in your handlers.

Scaling: You scale the entire server, not individual agents. If one agent needs more capacity, you scale everything. For very different scaling needs, consider separate deployments.

Shared State Between Agents

Agents in an AgentServer are independent instances—they don’t share state by default. Each agent has its own prompts, functions, and configuration.

If you need shared state:

Use external storage (Redis, database) rather than Python globals:

1import redis
2
3class SharedStateAgent(AgentBase):
4    def __init__(self, redis_client):
5        super().__init__(name="shared-state-agent")
6        self.redis = redis_client
7        # ... setup
8
9    def some_handler(self, args, raw_data):
10        # Read shared state
11        shared_value = self.redis.get("shared_key")
12        # Update shared state
13        self.redis.set("shared_key", "new_value")
14        return SwaigFunctionResult("Done")
15
16# In main
17redis_client = redis.Redis(host='localhost', port=6379)
18server = AgentServer()
19server.register(SharedStateAgent(redis_client), "/agent1")
20server.register(AnotherAgent(redis_client), "/agent2")

Sharing configuration:

For shared configuration like API keys or business rules, use a shared module:

1# config.py
2SHARED_CONFIG = {
3    "company_name": "Acme Corp",
4    "support_hours": "9 AM - 5 PM",
5    "api_key": os.environ.get("API_KEY")
6}
7
8# agents.py
9from config import SHARED_CONFIG
10
11class SalesAgent(AgentBase):
12    def __init__(self):
13        super().__init__(name="sales-agent")
14        self.prompt_add_section(
15            "Company",
16            f"You work for {SHARED_CONFIG['company_name']}"
17        )

Routing Logic

AgentServer routes requests based on URL path and SIP username. Understanding this routing helps you design your agent structure.

Path-based routing is straightforward:

  • Request to /sales → Sales agent
  • Request to /support → Support agent

SIP routing extracts the username from the SIP address:

  • sip:sales@example.com → looks up “sales” → routes to /sales
  • sip:help-desk@example.com → looks up “help-desk” → routes based on mapping

Auto-mapping creates automatic mappings from agent names and route paths:

1server.setup_sip_routing("/sip", auto_map=True)
2# Creates mappings like:
3# "salesagent" → "/sales" (from agent name, normalized)
4# "sales" → "/sales" (from route path without leading /)

Manual mapping gives explicit control:

1server.register_sip_username("sales-team", "/sales")
2server.register_sip_username("tech-support", "/support")

Common Patterns

Department-Based Routing

Route calls to different departments based on phone number or SIP username:

1server = AgentServer()
2
3server.register(SalesAgent(), "/sales")
4server.register(SupportAgent(), "/support")
5server.register(BillingAgent(), "/billing")
6server.register(ReceptionistAgent(), "/main")  # Default/main line
7
8server.setup_sip_routing("/sip", auto_map=True)

Time-Based Routing

Route to different agents based on business hours (implement in a custom router):

1class TimeSensitiveServer:
2    def __init__(self):
3        self.server = AgentServer()
4        self.server.register(LiveAgent(), "/live")
5        self.server.register(AfterHoursAgent(), "/afterhours")
6
7    def get_current_agent_route(self):
8        from datetime import datetime
9        hour = datetime.now().hour
10        if 9 <= hour < 17:  # Business hours
11            return "/live"
12        return "/afterhours"

Feature-Based Agents

Different agents for different capabilities:

1server.register(GeneralAgent(), "/general")        # Basic Q&A
2server.register(OrderAgent(), "/orders")           # Order management
3server.register(TechnicalAgent(), "/technical")    # Technical support
4server.register(EscalationAgent(), "/escalation")  # Human escalation

Best Practices

DO:

  • Use meaningful route names (/sales, /support, /billing)
  • Enable SIP routing for SIP-based deployments
  • Monitor /health endpoint for availability
  • Use consistent naming between routes and SIP usernames
  • Implement proper error handling in all agent handlers
  • Use external storage for shared state
  • Log which agent handles each request for debugging

DON’T:

  • Register duplicate routes
  • Forget to handle routing conflicts
  • Mix agent.run() and AgentServer for the same agent
  • Store shared state in Python globals (use external storage)
  • Put agents with very different scaling needs in the same server