Duration	2 hours
Day	5 of 5

Learning Objectives

By the end of this module, students will be able to:

• Deploy multiple agents on a single server
• Implement path-based routing
• Share state between agents
• Configure health checks

Topics

1. Why Multi-Agent? (15 min)

The Problem

One server, multiple use cases:

/sales      → Sales Agent
/support    → Support Agent
/billing    → Billing Agent
/spanish    → Spanish Agent

Running separate servers is wasteful and complex.

The Solution: AgentServer

from signalwire_agents import AgentServer

server = AgentServer(host="0.0.0.0", port=3000)
server.register(SalesAgent())
server.register(SupportAgent())
server.register(BillingAgent())
server.run()

One server, multiple agents!

---

2. AgentServer Basics (25 min)

Creating a Server

from signalwire_agents import AgentServer, AgentBase

# Create agents
class SalesAgent(AgentBase):
    def __init__(self):
        super().__init__(name="sales", route="/sales")
        self.prompt_add_section("Role", "You are a sales agent.")
        self.add_language("English", "en-US", "rime.spore")

class SupportAgent(AgentBase):
    def __init__(self):
        super().__init__(name="support", route="/support")
        self.prompt_add_section("Role", "You are a support agent.")
        self.add_language("English", "en-US", "rime.spore")

# Create server and register agents
server = AgentServer(host="0.0.0.0", port=3000)
server.register(SalesAgent())
server.register(SupportAgent())

if __name__ == "__main__":
    server.run()

Endpoints Created

http://localhost:3000/sales       → SalesAgent SWML
http://localhost:3000/sales/swaig → SalesAgent functions
http://localhost:3000/support     → SupportAgent SWML
http://localhost:3000/support/swaig → SupportAgent functions

Server Options

server = AgentServer(
    host="0.0.0.0",       # Bind address
    port=3000,            # Port number
    log_level="info",     # Logging verbosity
)

---

3. Routing Patterns (30 min)

Path-Based Routing

Different paths for different agents:

server.register(SalesAgent())      # /sales
server.register(SupportAgent())    # /support
server.register(BillingAgent())    # /billing

SignalWire Configuration:

• Sales number → https://host.com/sales
• Support number → https://host.com/support
• Billing number → https://host.com/billing

Language-Based Routing

class EnglishAgent(AgentBase):
    def __init__(self):
        super().__init__(name="english", route="/en")
        self.add_language("English", "en-US", "rime.spore")

class SpanishAgent(AgentBase):
    def __init__(self):
        super().__init__(name="spanish", route="/es")
        self.add_language("Spanish", "es-US", "gcloud.es-US-Neural2-A")

Department Router Pattern

Create a router that forwards to specialized agents:

class RouterAgent(AgentBase):
    def __init__(self):
        super().__init__(name="router", route="/main")

        self.prompt_add_section(
            "Role",
            "You route callers to the right department. "
            "Ask what they need help with."
        )

    @AgentBase.tool(
        description="Route to appropriate department",
        parameters={
            "type": "object",
            "properties": {
                "department": {
                    "type": "string",
                    "enum": ["sales", "support", "billing"]
                }
            },
            "required": ["department"]
        }
    )
    def route_call(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        department = args.get("department", "")
        routes = {
            "sales": "+15551111111",
            "support": "+15552222222",
            "billing": "+15553333333",
        }
        return (
            SwaigFunctionResult(f"Connecting you to {department}.")
            .connect(routes[department], final=True)
        )

---

4. Shared State (25 min)

The Challenge

Agents on same server may need to share:

• Customer data
• Session information
• Configuration

Shared Dictionary

from signalwire_agents import AgentServer, AgentBase

# Shared state
shared_data = {
    "config": {
        "business_hours": "9 AM to 5 PM",
        "support_email": "support@example.com"
    },
    "sessions": {}
}

class SalesAgent(AgentBase):
    def __init__(self, shared):
        super().__init__(name="sales", route="/sales")
        self.shared = shared

    @AgentBase.tool(description="Get business hours")
    def get_hours(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        hours = self.shared["config"]["business_hours"]
        return SwaigFunctionResult(f"We're open {hours}.")

class SupportAgent(AgentBase):
    def __init__(self, shared):
        super().__init__(name="support", route="/support")
        self.shared = shared

    @AgentBase.tool(description="Get support email")
    def get_email(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        email = self.shared["config"]["support_email"]
        return SwaigFunctionResult(f"Email us at {email}.")

# Create with shared state
server = AgentServer()
server.register(SalesAgent(shared_data))
server.register(SupportAgent(shared_data))

Session Sharing

class Agent1(AgentBase):
    def store_customer(self, call_id: str, name: str):
        self.shared["sessions"][call_id] = {"name": name}

class Agent2(AgentBase):
    def get_customer(self, call_id: str):
        return self.shared["sessions"].get(call_id, {})

---

5. Health Checks (15 min)

Built-in Health Endpoint

AgentServer includes /health:

curl http://localhost:3000/health

Response:

{
  "status": "healthy",
  "agents": ["sales", "support", "billing"],
  "timestamp": "2024-01-15T10:30:00Z"
}

Custom Health Checks

@server.app.get("/ready")
async def readiness():
    # Check dependencies
    db_ok = check_database()
    api_ok = check_external_api()

    if db_ok and api_ok:
        return {"status": "ready"}
    else:
        return JSONResponse(
            status_code=503,
            content={"status": "not ready"}
        )

Load Balancer Configuration

For production with multiple instances:

┌─────────────────┐
│  Load Balancer  │
│   /health check │
└────────┬────────┘
         │
    ┌────┴────┐
    │         │
┌───▼───┐ ┌───▼───┐
│ Inst1 │ │ Inst2 │
│ :3000 │ │ :3000 │
└───────┘ └───────┘

---

6. Complete Example (20 min)

Click to reveal complete solution

Multi-Department Support Server

#!/usr/bin/env python3
"""Multi-agent support server."""

import os
from signalwire_agents import AgentServer, AgentBase, SwaigFunctionResult


# Shared configuration
CONFIG = {
    "company": "TechCorp",
    "hours": "9 AM to 6 PM EST",
    "support_email": "help@techcorp.com",
}


class MainAgent(AgentBase):
    """Front door agent that routes calls."""

    def __init__(self):
        super().__init__(name="main", route="/")

        self.prompt_add_section(
            "Role",
            f"You are the receptionist for {CONFIG['company']}. "
            "Help callers reach the right department."
        )

        self.prompt_add_section(
            "Departments",
            bullets=[
                "Sales - for new purchases and pricing",
                "Support - for technical help",
                "Billing - for payments and invoices"
            ]
        )

        self.add_language("English", "en-US", "rime.spore")

    @AgentBase.tool(description="Transfer to a department")
    def transfer_department(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        department = args.get("department", "")
        routes = {
            "sales": "/sales",
            "support": "/support",
            "billing": "/billing"
        }
        if department.lower() not in routes:
            return SwaigFunctionResult(
                "I can transfer you to sales, support, or billing."
            )
        return (
            SwaigFunctionResult(f"Transferring to {department}.")
            .swml_change_context(routes[department.lower()])
        )


class SalesAgent(AgentBase):
    """Sales department agent."""

    def __init__(self):
        super().__init__(name="sales", route="/sales")

        self.prompt_add_section(
            "Role",
            f"You are a sales representative for {CONFIG['company']}. "
            "Help customers with product information and pricing."
        )

        self.add_language("English", "en-US", "rime.marsh")

    @AgentBase.tool(description="Get pricing information")
    def get_pricing(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        product = args.get("product", "")
        prices = {
            "basic": "$99/month",
            "pro": "$199/month",
            "enterprise": "Custom pricing"
        }
        price = prices.get(product.lower(), "Please specify: basic, pro, or enterprise")
        return SwaigFunctionResult(f"{product} is {price}.")


class SupportAgent(AgentBase):
    """Technical support agent."""

    def __init__(self):
        super().__init__(name="support", route="/support")

        self.prompt_add_section(
            "Role",
            f"You are a technical support agent for {CONFIG['company']}. "
            "Help customers resolve technical issues."
        )

        self.add_language("English", "en-US", "rime.spore")

    @AgentBase.tool(description="Create support ticket")
    def create_ticket(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        issue = args.get("issue", "")
        ticket_id = "T-" + str(hash(issue))[:6]
        return SwaigFunctionResult(
            f"I've created ticket {ticket_id} for your issue. "
            f"You'll receive updates at {CONFIG['support_email']}."
        )


class BillingAgent(AgentBase):
    """Billing department agent."""

    def __init__(self):
        super().__init__(name="billing", route="/billing")

        self.prompt_add_section(
            "Role",
            f"You are a billing specialist for {CONFIG['company']}. "
            "Help with payments, invoices, and account questions."
        )

        self.add_language("English", "en-US", "rime.cove")

    @AgentBase.tool(description="Get account balance")
    def get_balance(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        account_id = args.get("account_id", "")
        # Mock balance lookup
        return SwaigFunctionResult(
            f"Account {account_id} has a balance of $0.00. You're all paid up!"
        )


def main():
    """Start the multi-agent server."""
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", "3000"))

    server = AgentServer(host=host, port=port)

    # Register all agents
    server.register(MainAgent())
    server.register(SalesAgent())
    server.register(SupportAgent())
    server.register(BillingAgent())

    print(f"Multi-agent server starting on {host}:{port}")
    print("Endpoints:")
    print(f"  Main:    http://{host}:{port}/")
    print(f"  Sales:   http://{host}:{port}/sales")
    print(f"  Support: http://{host}:{port}/support")
    print(f"  Billing: http://{host}:{port}/billing")
    print(f"  Health:  http://{host}:{port}/health")

    server.run()


if __name__ == "__main__":
    main()

---

Key Takeaways

1. AgentServer hosts multiple agents - One server, many endpoints
2. Route-based organization - Different paths for different agents
3. Shared state is possible - Pass data between agents
4. Health checks are essential - Monitor server status
5. Scales horizontally - Run multiple instances behind load balancer

Preparation for Lab 2.9

• Have 2-3 agent concepts ready
• Think about what they might share
• Consider routing logic

Lab Preview

In Lab 2.9, you will:

1. Create a multi-agent server
2. Implement department routing
3. Add shared configuration
4. Test health endpoints