Lab L2.9: Multi-Agent Systems

🎯 Assignment: Accept this lab on GitHub Classroom
You’ll get your own repository with starter code, instructions, and automatic grading.

Duration	90 minutes
Prerequisites	Previous module completed

Objectives

• Create multiple agents in one file
• Implement agent-to-agent transfers
• Design agent routing logic

---

How to Complete This Lab

1. Accept the Assignment — Click the GitHub Classroom link above
2. Clone Your Repo — git clone <your-repo-url>
3. Read the README — Your repo has detailed requirements and grading criteria
4. Write Your Code — Implement the solution in solution/agent.py
5. Test Locally — Use swaig-test to verify your agent works
6. Push to Submit — git push triggers auto-grading

---

Key Concepts

The following exercises walk through the concepts you’ll need. Your GitHub Classroom repo README has the specific requirements for grading.

Scenario

Build a customer service system with:

1. General inquiry agent (/)
2. Sales agent (/sales)
3. Support agent (/support)

---

Part 1: Create Individual Agents (30 min)

Task

Define three specialized agent classes.

Starter Code

#!/usr/bin/env python3
"""Lab 2.9: Multi-agent architecture."""

from signalwire_agents import AgentBase, AgentServer, SwaigFunctionResult


class GeneralAgent(AgentBase):
    """General inquiry handler."""

    def __init__(self):
        super().__init__(name="general-agent", route="/")

        self.prompt_add_section(
            "Role",
            "You are the general inquiry agent. Help route callers "
            "or answer basic questions about the company."
        )

        self.prompt_add_section(
            "Routing",
            "For sales inquiries, direct to /sales. "
            "For technical support, direct to /support."
        )

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

    def _setup_functions(self):
        # TODO: Add functions
        pass


class SalesAgent(AgentBase):
    """Sales specialist."""

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

        self.prompt_add_section(
            "Role",
            "You are a sales specialist. Help customers with "
            "product information, pricing, and purchases."
        )

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

    def _setup_functions(self):
        # TODO: Add functions
        pass


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

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

        self.prompt_add_section(
            "Role",
            "You are technical support. Help customers troubleshoot "
            "issues and resolve problems."
        )

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

    def _setup_functions(self):
        # TODO: Add functions
        pass


if __name__ == "__main__":
    # TODO: Create server and register agents
    pass

---

Part 2: Implement Agent Functions (25 min)

General Agent Functions

def _setup_functions(self):
    @self.tool(description="Get company information")
    def get_company_info(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        return SwaigFunctionResult(
            "We are TechCorp, providing innovative solutions since 2010. "
            "For sales, I can connect you to our sales team. "
            "For support, I can transfer you to technical support."
        )

    @self.tool(description="Get business hours")
    def get_hours(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        return SwaigFunctionResult(
            "We're open Monday to Friday, 9 AM to 6 PM Eastern. "
            "Our sales team is available during these hours, "
            "and support is available 8 AM to 8 PM."
        )

Sales Agent Functions

def _setup_functions(self):
    PRODUCTS = {
        "basic": {"price": 29.99, "features": ["Email support", "5 users"]},
        "pro": {"price": 79.99, "features": ["Priority support", "25 users", "API access"]},
        "enterprise": {"price": 199.99, "features": ["24/7 support", "Unlimited users", "Custom integration"]}
    }

    @self.tool(description="Get product pricing")
    def get_pricing(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        prices = [f"{name.title()}: ${info['price']}/month"
                  for name, info in PRODUCTS.items()]
        return SwaigFunctionResult(
            "Our plans: " + "; ".join(prices)
        )

    @self.tool(
        description="Get product details",
        parameters={
            "type": "object",
            "properties": {
                "product": {"type": "string", "enum": list(PRODUCTS.keys())}
            },
            "required": ["product"]
        }
    )
    def get_product_details(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        product = args.get("product", "")
        info = PRODUCTS.get(product)
        if not info:
            return SwaigFunctionResult("Unknown product.")
        features = ", ".join(info["features"])
        return SwaigFunctionResult(
            f"{product.title()} plan: ${info['price']}/month. "
            f"Includes: {features}"
        )

    @self.tool(
        description="Start a purchase",
        parameters={
            "type": "object",
            "properties": {
                "product": {"type": "string", "enum": list(PRODUCTS.keys())},
                "customer_email": {"type": "string"}
            },
            "required": ["product", "customer_email"]
        }
    )
    def start_purchase(
        args: dict,
        raw_data: dict = None
    ) -> SwaigFunctionResult:
        product = args.get("product", "")
        customer_email = args.get("customer_email", "")
        info = PRODUCTS.get(product)
        return (
            SwaigFunctionResult(
                f"Great choice! I'm setting up {product.title()} "
                f"for {customer_email}. You'll receive a confirmation email."
            )
            .update_global_data( {
                "purchase_product": product,
                "purchase_email": customer_email,
                "purchase_initiated": True
            })
        )

Support Agent Functions

def _setup_functions(self):
    TROUBLESHOOTING = {
        "login": "Try resetting your password at account.techcorp.com/reset",
        "slow": "Clear your browser cache and try a different browser",
        "error": "Please note the error code and I'll look it up",
        "crash": "Update to the latest version from our downloads page"
    }

    @self.tool(description="Get common solutions")
    def get_common_issues(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        issues = list(TROUBLESHOOTING.keys())
        return SwaigFunctionResult(
            f"Common issues I can help with: {', '.join(issues)}. "
            "What issue are you experiencing?"
        )

    @self.tool(
        description="Troubleshoot an issue",
        parameters={
            "type": "object",
            "properties": {
                "issue_type": {"type": "string"}
            },
            "required": ["issue_type"]
        }
    )
    def troubleshoot(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        issue_type = args.get("issue_type", "")
        issue_lower = issue_type.lower()

        for key, solution in TROUBLESHOOTING.items():
            if key in issue_lower:
                return SwaigFunctionResult(
                    f"For {key} issues: {solution}. "
                    "Did that help resolve your issue?"
                )

        return SwaigFunctionResult(
            "I'll need more details. Can you describe the issue?"
        )

    @self.tool(
        description="Create support ticket",
        parameters={
            "type": "object",
            "properties": {
                "description": {"type": "string"},
                "priority": {"type": "string", "enum": ["low", "medium", "high"]}
            },
            "required": ["description"]
        }
    )
    def create_ticket(
        args: dict,
        raw_data: dict = None
    ) -> SwaigFunctionResult:
        description = args.get("description", "")
        priority = args.get("priority", "medium")
        import datetime
        ticket_id = f"TKT-{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"

        return (
            SwaigFunctionResult(
                f"Created ticket {ticket_id} with {priority} priority. "
                "Our team will contact you within 24 hours."
            )
            .update_global_data( {
                "ticket_id": ticket_id,
                "ticket_description": description,
                "ticket_priority": priority
            })
        )

---

Part 3: Configure AgentServer (20 min)

Task

Set up the server with all agents.

Expected Implementation

if __name__ == "__main__":
    # Create server
    server = AgentServer(host="0.0.0.0", port=3000)

    # Register all agents
    server.register(GeneralAgent())
    server.register(SalesAgent())
    server.register(SupportAgent())

    # Run server
    server.run()

---

Testing

# Run the multi-agent server
python lab2_9_multi_agent.py &

# Test general agent
curl http://localhost:3000/

# Test sales agent
curl http://localhost:3000/sales

# Test support agent
curl http://localhost:3000/support

# Test with swaig-test
swaig-test lab2_9_multi_agent.py --agent-class GeneralAgent --list-tools
swaig-test lab2_9_multi_agent.py --agent-class SalesAgent --list-tools
swaig-test lab2_9_multi_agent.py --agent-class SupportAgent --list-tools

---

Validation Checklist

• Three agents defined with unique routes
• Each agent has specialized functions
• AgentServer registers all agents
• Routes resolve correctly (/, /sales, /support)
• Each agent generates valid SWML

---

Challenge Extension

Add a shared authentication function that all agents can use:

# Create a mixin class
class AuthMixin:
    def add_auth_functions(self):
        @self.tool(description="Verify customer", parameters={
            "type": "object",
            "properties": {"customer_id": {"type": "string"}},
            "required": ["customer_id"]
        })
        def verify_customer(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            customer_id = args.get("customer_id", "")
            # Shared verification logic
            return (
                SwaigFunctionResult(f"Verified customer {customer_id}")
                .update_global_data( {"verified": True, "customer_id": customer_id})
            )

---

Submission

Upload your completed lab2_9_multi_agent.py file.

---

Complete Agent Code

Click to reveal complete solution

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

Lab 2.9 Deliverable: Demonstrates AgentServer with multiple agents
and path-based routing for different specializations.
"""

from signalwire_agents import AgentServer, AgentBase, SwaigFunctionResult


# ============================================================
# General Agent - Main entry point
# ============================================================

class GeneralAgent(AgentBase):
    """General agent that can route to specialists."""

    def __init__(self):
        super().__init__(name="general-agent", route="/general")

        self.prompt_add_section(
            "Role",
            "You are the main receptionist. Help with general questions "
            "or route to sales or support specialists."
        )

        self.prompt_add_section(
            "Routing",
            bullets=[
                "For pricing/purchasing questions: route to sales",
                "For technical issues: route to support",
                "For general questions: answer directly"
            ]
        )

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

    def _setup_functions(self):
        @self.tool(description="Get company information")
        def get_company_info(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return SwaigFunctionResult(
                "Welcome to TechCorp! We offer cloud software solutions. "
                "For sales, pricing, or support, I can connect you to a specialist."
            )

        @self.tool(description="Route to sales specialist")
        def route_to_sales(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return (
                SwaigFunctionResult("Connecting you to our sales team.", post_process=True)
                .swml_transfer("/sales", "Goodbye!", final=True)
            )

        @self.tool(description="Route to support specialist")
        def route_to_support(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return (
                SwaigFunctionResult("Connecting you to technical support.", post_process=True)
                .swml_transfer("/support", "Goodbye!", final=True)
            )


# ============================================================
# Sales Agent - Pricing and purchasing
# ============================================================

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

    PRICING = {
        "starter": {"price": 29, "users": 5, "features": "Basic"},
        "professional": {"price": 79, "users": 25, "features": "Advanced"},
        "enterprise": {"price": 199, "users": "Unlimited", "features": "Premium"}
    }

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

        self.prompt_add_section(
            "Role",
            "You are a sales specialist. Help with pricing, plans, and purchasing."
        )

        self.prompt_add_section(
            "Available Plans",
            bullets=[
                f"{name.title()}: ${info['price']}/mo, {info['users']} users, {info['features']}"
                for name, info in self.PRICING.items()
            ]
        )

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

    def _setup_functions(self):
        @self.tool(description="Get all pricing plans")
        def get_pricing(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            plans = [
                f"{name.title()}: ${info['price']}/month for {info['users']} users"
                for name, info in self.PRICING.items()
            ]
            return SwaigFunctionResult("Our plans: " + "; ".join(plans))

        @self.tool(
            description="Get details for a specific plan",
            parameters={
                "type": "object",
                "properties": {
                    "plan": {
                        "type": "string",
                        "enum": ["starter", "professional", "enterprise"]
                    }
                },
                "required": ["plan"]
            }
        )
        def get_plan_details(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            plan = args.get("plan", "")
            info = self.PRICING.get(plan.lower())
            if not info:
                return SwaigFunctionResult("Plan not found.")
            return SwaigFunctionResult(
                f"{plan.title()} plan: ${info['price']}/month, "
                f"{info['users']} users, {info['features']} features."
            )

        @self.tool(
            description="Start a trial",
            parameters={
                "type": "object",
                "properties": {
                    "email": {"type": "string"},
                    "plan": {"type": "string"}
                },
                "required": ["email"]
            }
        )
        def start_trial(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            email = args.get("email", "")
            plan = args.get("plan", "professional")
            return (
                SwaigFunctionResult(
                    f"Great! I've started a 14-day trial of {plan} for {email}. "
                    "Check your inbox for login details."
                )
                .update_global_data({
                    "trial_email": email,
                    "trial_plan": plan,
                    "lead_captured": True
                })
            )


# ============================================================
# Support Agent - Technical assistance
# ============================================================

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

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

        self.prompt_add_section(
            "Role",
            "You are technical support. Help with issues, troubleshooting, and account problems."
        )

        self.prompt_add_section(
            "Common Issues",
            bullets=[
                "Login problems: Reset password or check email",
                "Performance: Clear cache, check system requirements",
                "Integration: Verify API keys and permissions"
            ]
        )

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

    def _setup_functions(self):
        @self.tool(description="Get troubleshooting steps for login issues")
        def login_help(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return SwaigFunctionResult(
                "For login issues: 1) Check your email is correct, "
                "2) Try password reset, 3) Clear browser cookies, "
                "4) Try incognito mode. Still stuck? I can create a ticket."
            )

        @self.tool(description="Get performance troubleshooting steps")
        def performance_help(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return SwaigFunctionResult(
                "For performance issues: 1) Clear browser cache, "
                "2) Close other tabs, 3) Check internet connection, "
                "4) Try a different browser. Need more help?"
            )

        @self.tool(
            description="Create a support ticket",
            parameters={
                "type": "object",
                "properties": {
                    "issue": {"type": "string"},
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"]
                    }
                },
                "required": ["issue"]
            }
        )
        def create_ticket(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            from datetime import datetime
            issue = args.get("issue", "")
            priority = args.get("priority", "medium")
            ticket_id = f"SUP-{datetime.now().strftime('%Y%m%d%H%M%S')}"
            return (
                SwaigFunctionResult(
                    f"Created ticket {ticket_id}. "
                    "We'll respond within 24 hours for standard issues."
                )
                .update_global_data({
                    "ticket_id": ticket_id,
                    "ticket_issue": issue,
                    "ticket_priority": priority
                })
            )

        @self.tool(description="Check system status")
        def system_status(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            return SwaigFunctionResult(
                "All systems operational. API: 99.9% uptime. "
                "No known issues. Last checked 5 minutes ago."
            )


# ============================================================
# Create and run server
# ============================================================

def create_server():
    """Create multi-agent server."""
    server = AgentServer(host="0.0.0.0", port=3000)

    # Register all agents
    server.register(GeneralAgent())
    server.register(SalesAgent())
    server.register(SupportAgent())

    return server


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