Lab L2.7: Call Recording

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

Duration	45 minutes
Prerequisites	Previous module completed

Objectives

• Enable call recording
• Configure stereo recording
• Handle recording events

---

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 payment processing agent that:

1. Discloses recording at call start
2. Pauses recording during payment collection
3. Resumes recording after sensitive data

---

Part 1: Recording Configuration (10 min)

Task

Set up basic recording with stereo and MP3 format.

Starter Code

#!/usr/bin/env python3
"""Lab 2.7: Call recording."""

from signalwire_agents import AgentBase, SwaigFunctionResult


class PaymentAgent(AgentBase):
    def __init__(self):
        super().__init__(name="payment-agent")

        self.prompt_add_section(
            "Role",
            "You process payments for customers. "
            "Always disclose recording and pause for card details."
        )

        self.prompt_add_section(
            "Recording Policy",
            bullets=[
                "Disclose recording at call start",
                "Pause recording before collecting card number",
                "Resume recording after payment processed",
                "Never read back full card numbers"
            ]
        )

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

        self._configure_recording()
        self._setup_functions()

    def _configure_recording(self):
        # TODO: Configure recording
        pass

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


if __name__ == "__main__":
    agent = PaymentAgent()
    agent.run()

Expected Implementation

def _configure_recording(self):
    self.set_params({
        "record_call": True,
        "record_format": "mp3",
        "record_stereo": True  # Separate caller/agent channels
    })

---

Part 2: Consent Collection (15 min)

Task

Create a function to get recording consent from the caller.

Expected Implementation

@self.tool(
    description="Get recording consent from caller",
    parameters={
        "type": "object",
        "properties": {
            "consent_given": {
                "type": "string",
                "enum": ["yes", "no"],
                "description": "Whether caller consents to recording"
            }
        },
        "required": ["consent_given"]
    }
)
def handle_consent(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    consent_given = args.get("consent_given", "")
    if consent_given == "yes":
        return (
            SwaigFunctionResult(
                "Thank you for your consent. How can I help you today?"
            )
            .update_global_data( {"recording_consent": True})
            .record_call(control_id="main")
        )
    else:
        return (
            SwaigFunctionResult(
                "No problem, this call will not be recorded. "
                "How can I help you today?"
            )
            .update_global_data( {"recording_consent": False})
        )

---

Part 3: Secure Payment Collection with pay() (20 min)

Task

Implement PCI-compliant payment collection using the SWML pay method. This keeps card data away from the LLM entirely - the data is collected via IVR and sent directly to your payment gateway.

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Payment Flow                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   1. Agent calls process_payment()                          │
│   2. SWML pay verb triggered                                │
│   3. Card collected via IVR (DTMF keypad)                   │
│   4. Card data sent directly to payment gateway             │
│   5. Gateway returns success/failure                        │
│   6. Agent receives only pay_result (success/failure)       │
│                                                             │
│   Card data NEVER passes through the LLM!                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Expected Implementation

@self.tool(
    description="Process payment for customer",
    parameters={
        "type": "object",
        "properties": {
            "amount": {
                "type": "string",
                "description": "Amount to charge (e.g., '49.99')"
            }
        },
        "required": ["amount"]
    }
)
def process_payment(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    amount = args.get("amount", "0.00")
    raw_data = raw_data or {}
    global_data = raw_data.get("global_data", {})

    # Get public URL from SDK (auto-detected from ngrok/proxy headers)
    base_url = self.get_full_url().rstrip('/')
    payment_url = f"{base_url}/payment"

    return (
        SwaigFunctionResult(
            "I'll collect your payment information securely. "
            "Please enter your card number using your phone keypad.",
            post_process=True
        )
        .pay(
            payment_connector_url=payment_url,
            charge_amount=amount,
            input_method="dtmf",
            security_code=True,
            postal_code=True,
            max_attempts=3,
            timeout=10,
            ai_response=(
                "The payment result is ${pay_result}. "
                "If successful, confirm the payment. "
                "If failed, apologize and offer to try another card."
            )
        )
    )

Mock Payment Gateway

Add this to your agent to handle payment gateway requests:

# In create_server() function:
@server.app.post("/payment")
async def payment_gateway(request: Request):
    """Mock payment gateway endpoint."""
    data = await request.json()

    card_number = data.get("payment_card_number", "")
    charge_amount = data.get("charge_amount", "0.00")

    # Test card numbers
    TEST_CARDS = {
        "4111111111111111": "success",
        "4000000000000002": "declined",
    }

    last_four = card_number[-4:] if len(card_number) >= 4 else "****"
    print(f"Payment: card=****{last_four}, amount=${charge_amount}")

    scenario = TEST_CARDS.get(card_number, "success")

    if scenario == "success":
        return JSONResponse({
            "charge_id": f"ch_{uuid.uuid4().hex[:12]}",
            "error_code": None,
            "error_message": None
        })
    else:
        return JSONResponse({
            "charge_id": None,
            "error_code": "card_declined",
            "error_message": "Your card was declined"
        })

pay_result Values

| Value | Description | |——-|————-| | success | Payment completed | | payment-connector-error | Gateway error | | too-many-failed-attempts | Max retries exceeded | | caller-interrupted-with-star | User pressed * | | caller-hung-up | Call ended | —

Testing

# Test agent
swaig-test lab2_7_recording.py --dump-swml

# Verify recording settings
swaig-test lab2_7_recording.py --dump-swml | grep -A3 "record"

# Test consent flow
swaig-test lab2_7_recording.py --exec handle_consent \
  --consent_given "yes"

# Test payment function (card data handled via IVR, not SWAIG)
swaig-test lab2_7_recording.py --exec process_payment \
  --amount "49.99"

# Test mock payment gateway directly
curl -X POST http://localhost:3000/payment \
  -H "Content-Type: application/json" \
  -d '{"payment_card_number": "4111111111111111", "charge_amount": "49.99"}'
# Expected: {"charge_id": "ch_...", "error_code": null, "error_message": null}

# Test declined card
curl -X POST http://localhost:3000/payment \
  -H "Content-Type: application/json" \
  -d '{"payment_card_number": "4000000000000002", "charge_amount": "49.99"}'
# Expected: {"charge_id": null, "error_code": "card_declined", ...}

---

Validation Checklist

• Recording enabled with stereo MP3
• Consent flow updates metadata and starts recording
• Payment uses pay() method (card data never touches LLM)
• Mock payment gateway endpoint works
• Test card 4111111111111111 returns success
• Test card 4000000000000002 returns declined

---

Security Notes

This lab demonstrates PCI-compliant payment collection:

• Card data is collected via IVR (DTMF keypad), not spoken to the AI
• Card data is sent directly to the payment gateway
• Only pay_result (success/failure) is returned to the agent
• The LLM never sees or processes card numbers

Production considerations:

• Replace mock gateway with real payment processor (Stripe, etc.)
• Use HTTPS for all payment endpoints
• Implement proper error handling and retry logic
• Add audit logging for payment attempts

---

Submission

Upload your completed lab2_7_recording.py file.

---

Complete Agent Code

Click to reveal complete solution

#!/usr/bin/env python3
"""Payment agent with call recording and PCI-compliant payment collection.

Lab 2.7 Deliverable: Demonstrates call recording configuration,
consent flow, and secure payment collection using the SWML pay method.
Card data is collected via IVR and never passes through the LLM.
"""

import uuid
from signalwire_agents import AgentBase, AgentServer, SwaigFunctionResult
from fastapi import Request
from fastapi.responses import JSONResponse


# Test card numbers for different scenarios
TEST_CARDS = {
    "4111111111111111": "success",      # Visa - success
    "4000000000000002": "declined",     # Visa - declined
    "5555555555554444": "success",      # Mastercard - success
}


class PaymentAgent(AgentBase):
    """Payment processing agent with recording controls and PCI-compliant payments."""

    def __init__(self):
        super().__init__(name="payment-agent")

        # Recording configuration
        self.set_params({
            "record_call": True,
            "record_format": "mp3",
            "record_stereo": True  # Separate caller/agent channels
        })

        self.prompt_add_section(
            "Role",
            "You process payments for customers. "
            "Always disclose recording at call start."
        )

        self.prompt_add_section(
            "Recording Policy",
            bullets=[
                "Disclose recording at call start",
                "Use process_payment function for card collection",
                "Card data is collected securely via IVR",
                "Never ask customer to speak card numbers"
            ]
        )

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

        self._setup_functions()

    def _setup_functions(self):
        """Define payment and recording control functions."""

        @self.tool(
            description="Get recording consent from caller",
            parameters={
                "type": "object",
                "properties": {
                    "consent_given": {
                        "type": "string",
                        "enum": ["yes", "no"],
                        "description": "Whether caller consents to recording"
                    }
                },
                "required": ["consent_given"]
            }
        )
        def handle_consent(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            consent_given = args.get("consent_given", "")
            if consent_given == "yes":
                return (
                    SwaigFunctionResult(
                        "Thank you for your consent. How can I help you today?"
                    )
                    .update_global_data({"recording_consent": True})
                    .record_call(control_id="main", stereo=True, format="mp3")
                )
            else:
                return (
                    SwaigFunctionResult(
                        "No problem, this call will not be recorded. "
                        "How can I help you today?"
                    )
                    .update_global_data({"recording_consent": False})
                )

        @self.tool(
            description="Process payment for customer",
            parameters={
                "type": "object",
                "properties": {
                    "amount": {
                        "type": "string",
                        "description": "Amount to charge (e.g., '49.99')"
                    }
                },
                "required": ["amount"]
            }
        )
        def process_payment(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            amount = args.get("amount", "0.00")

            # Get public URL from SDK (auto-detected from ngrok/proxy headers)
            base_url = self.get_full_url().rstrip('/')
            payment_url = f"{base_url}/payment"

            # Card data collected via IVR - never touches the LLM
            return (
                SwaigFunctionResult(
                    "I'll collect your payment securely. "
                    "Please enter your card number using your phone keypad.",
                    post_process=True
                )
                .pay(
                    payment_connector_url=payment_url,
                    charge_amount=amount,
                    input_method="dtmf",
                    security_code=True,
                    postal_code=True,
                    max_attempts=3,
                    ai_response=(
                        "The payment result is ${pay_result}. "
                        "If successful, confirm the payment. "
                        "If failed, apologize and offer to try another card."
                    )
                )
            )


def create_server():
    """Create AgentServer with payment gateway endpoint."""
    server = AgentServer(host="0.0.0.0", port=3000)

    # Register the payment agent
    agent = PaymentAgent()
    server.register(agent, "/")

    # Add the mock payment gateway endpoint
    @server.app.post("/payment")
    async def payment_gateway(request: Request):
        """Mock payment gateway endpoint.

        In production, this would connect to a real payment processor
        like Stripe, Square, or Braintree.
        """
        data = await request.json()

        card_number = data.get("payment_card_number", "")
        charge_amount = data.get("charge_amount", "0.00")

        last_four = card_number[-4:] if len(card_number) >= 4 else "****"
        print(f"Payment: card=****{last_four}, amount=${charge_amount}")

        scenario = TEST_CARDS.get(card_number, "success")

        if scenario == "success":
            return JSONResponse({
                "charge_id": f"ch_{uuid.uuid4().hex[:12]}",
                "error_code": None,
                "error_message": None
            })
        else:
            return JSONResponse({
                "charge_id": None,
                "error_code": "card_declined",
                "error_message": "Your card was declined"
            })

    return server


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