Multi-Agent Servers

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

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

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

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

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

1
server.unregister("/sales")

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

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

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

1
import redis
2
3
class 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 FunctionResult("Done")
15
16
# In main
17
redis_client = redis.Redis(host='localhost', port=6379)
18
server = AgentServer()
19
server.register(SharedStateAgent(redis_client), "/agent1")
20
server.register(AnotherAgent(redis_client), "/agent2")

1
# config.py
2
SHARED_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
9
from config import SHARED_CONFIG
10
11
class 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
        )

1
server.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 /)

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

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

1
class 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"

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