Consent and Escalations

Your agent now has tools, handles failures gracefully, and you can measure its performance. For production deployments, you'll also need to handle operational workflows like collecting consent and escalating to human agents when needed.

Why these workflows matter

Real-world voice agents need to:

• Collect recording consent before capturing calls (often legally required)
• Escalate to humans when the AI can't help or the user requests it
• Hand off between specialized agents based on conversation context
• Preserve context across these transitions

Collecting recording consent with tasks

Tasks are focused units of work that complete and return a result. They're perfect for structured interactions like consent collection.

Define a consent task:

from livekit.agents import AgentTask, function_tool

class CollectConsent(AgentTask[bool]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="""
            Ask for recording consent and get a clear yes or no answer.
            Be polite and professional.
            """,
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="""
            Briefly introduce yourself, then ask for permission to record 
            the call for quality assurance and training purposes.
            Make it clear that they can decline.
            """
        )

    @function_tool()
    async def consent_given(self) -> None:
        """Use this when the user gives consent to record."""
        self.complete(True)

    @function_tool()
    async def consent_denied(self) -> None:
        """Use this when the user denies consent to record."""
        self.complete(False)

When you await a task inside an agent method like on_enter, the task takes control of the session and runs until it calls self.complete(). The task handles all user interaction during this time, then returns its result and gives control back to the agent. No explicit .run() call is needed.

Use the task in your agent:

from livekit.agents import Agent

class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a friendly customer service representative."
        )

    async def on_enter(self) -> None:
        consent = await CollectConsent(chat_ctx=self.chat_ctx)
        
        if consent:
            await self.session.generate_reply(
                instructions="Thank them and offer your assistance."
            )
        else:
            await self.session.generate_reply(
                instructions="Let them know you understand and will proceed without recording."
            )

If you're using LiveKit's agent observability features and want session recordings to respect the user's consent choice, you can start a new agent session with record set to False when consent is denied. This gives you fine-grained control over which sessions get recorded. Check the observability docs for more details on session-level recording settings.

Building a manager escalation

When users ask for a supervisor or the AI can't help, you need a clean handoff. Create a separate manager agent:

class ManagerAgent(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="""
            You are a customer service manager. You handle escalated issues 
            that frontline agents couldn't resolve. Be empathetic and 
            solution-focused. You have authority to offer refunds, credits,
            or other accommodations.
            """,
            chat_ctx=chat_ctx,
            tts="cartesia/sonic-3:6f84f4b8-58a2-430c-8c79-688dad597532",  # Different voice
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="""
            Introduce yourself as a manager. Acknowledge that the customer 
            asked to speak with someone senior. Ask how you can help resolve 
            their concern.
            """
        )

Add an escalation tool to your main agent:

from livekit.agents import Agent, function_tool, RunContext

class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="""
            You are a friendly customer service representative. Help customers 
            with general inquiries. If they ask for a manager or you can't 
            resolve their issue, use the escalate_to_manager tool.
            """
        )

    @function_tool()
    async def escalate_to_manager(self, context: RunContext) -> ManagerAgent:
        """Transfer the customer to a manager when requested or when you cannot resolve their issue."""
        return ManagerAgent(chat_ctx=self.chat_ctx), "Transferring you to a manager now."

The LLM decides when to call this tool based on two things: the tool's docstring and the agent's instructions. The docstring tells the LLM what the tool does ("Transfer the customer to a manager when requested or when you cannot resolve their issue"). The agent's instructions reinforce this by mentioning "If they ask for a manager or you can't resolve their issue, use the escalate_to_manager tool."

When a tool returns a new Agent as the first value in a tuple, the session automatically hands off control to that agent. The second value in the tuple is a message the LLM will speak before the handoff completes.

Preserving conversation context

By default, each agent starts fresh. To carry conversation history across handoffs, pass the chat context:

@function_tool()
async def escalate_to_manager(self, context: RunContext) -> ManagerAgent:
    """Transfer to a manager."""
    # Pass chat_ctx to preserve conversation history
    return ManagerAgent(chat_ctx=self.chat_ctx), "Transferring you to a manager now."

The manager now has full context of what's already been discussed.

Multi-step workflows with task groups

For complex flows with multiple steps (like collecting shipping info), use task groups. A TaskGroup lets you define multiple tasks that run in sequence. Each task has an ID and description. When the group completes, you can access results by ID.

from livekit.agents.beta.workflows import TaskGroup
from livekit.agents import AgentTask, function_tool, RunContext
from dataclasses import dataclass

# Define result types for each task
@dataclass
class EmailResult:
    email_address: str

@dataclass 
class AddressResult:
    address: str

# Task to collect email
class GetEmailTask(AgentTask[EmailResult]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Collect the user's email address."
        )
    
    @function_tool()
    async def record_email(self, context: RunContext, email: str) -> None:
        """Record the user's email address"""
        self.complete(EmailResult(email_address=email))

# Task to collect shipping address
class GetAddressTask(AgentTask[AddressResult]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Collect the user's shipping address."
        )
    
    @function_tool()
    async def record_address(self, context: RunContext, address: str) -> None:
        """Record the user's shipping address"""
        self.complete(AddressResult(address=address))

# Agent that uses the task group
class CheckoutAgent(Agent):
    async def on_enter(self) -> None:
        task_group = TaskGroup()
        
        # Each task wrapped in lambda so it can be reinitialized if user goes back
        task_group.add(
            lambda: GetEmailTask(), 
            id="email", 
            description="Collect email address"
        )
        task_group.add(
            lambda: GetAddressTask(), 
            id="address", 
            description="Collect shipping address"
        )
        
        # Run tasks in sequence
        results = await task_group
        
        # Access results by task ID
        email = results.task_results["email"].email_address
        address = results.task_results["address"].address
        
        await self.session.generate_reply(
            instructions=f"Confirm the order will be sent to {email} at {address}."
        )

Task groups support going back to previous steps if the user needs to make corrections.

Warm handoff to a human

For ultimate escalation, you might need to transfer to an actual human agent. This typically involves SIP for telephony.

SIP and telephony configuration is its own topic with more setup involved. LiveKit has separate resources that go deeper into telephony, so don't worry if this part seems complex. Here's the pattern for how human handoff works:

from livekit.agents import function_tool, RunContext, get_job_context

@function_tool()
async def transfer_to_human(self, context: RunContext) -> None:
    """Transfer the call to a human agent."""
    context.disallow_interruptions()
    
    await context.session.say(
        "I'm transferring you to a human agent now. Please hold."
    )
    
    room = get_job_context().room
    await room.local_participant.publish_sip_participant(
        sip_trunk_id="your-trunk-id",
        dial_to="sip:support@your-pbx.com",
    )

Putting it all together

Your production agent flow might look like this:

1. Consent task collects recording permission
2. Triage agent determines the user's need
3. Specialized agent handles the specific request (billing, support, sales)
4. Escalation to a manager agent if needed
5. Human handoff as a last resort

Each transition preserves relevant context while allowing agents to have different personalities, tools, and permissions.

Testing escalation paths

Run your agent and test:

• Decline consent and verify the agent handles it gracefully
• Ask to "speak to a manager" and verify the handoff
• Interrupt during an escalation to test resilience
• Try edge cases like asking for escalation immediately

With consent and escalation workflows in place, your agent is ready for real customer interactions.

Workshop complete

You've built a production-quality voice agent from scratch. Here's what you covered:

1. Voice agent foundations - The STT-LLM-TTS pipeline and why WebRTC matters for real-time audio
2. Semantic turn detection - Preventing awkward interruptions by understanding when users finish speaking
3. Personality and fallbacks - Crafting system prompts, selecting voices, and surviving provider outages
4. Metrics and preemptive generation - Measuring performance and reducing response latency
5. Tools and MCP integration - Connecting your agent to external APIs and services
6. Consent and escalations - Building production workflows with tasks and agent handoffs

Your agent can now listen in real-time, respond naturally, call external services, collect structured information, and hand off to specialized agents or humans when needed.

Next steps

• Browse the LiveKit Agents documentation for advanced topics like telephony, video processing, and realtime models
• Explore the Python agents examples repository for more complex workflows
• Join the LiveKit community Slack to ask questions and share what you're building
• Check out LiveKit Cloud for production hosting with built-in observability

Alternative: Agent Builder

LiveKit Cloud includes Agent Builder, a visual tool that generates a basic voice agent in minutes. You configure providers through a web interface, set a system prompt, and it generates Python code you can download.

Agent Builder features:

• Configure tools visually
• Built-in test interface
• One-click deploy to LiveKit Cloud
• Download and customize code

The code Agent Builder generates is yours. Download it, open it in your editor, and customize it. Add the tools, workflows, and custom logic you learned in this workshop. Agent Builder gives you a starting point, and everything you've learned here applies to extending that code.

You can find Agent Builder in your LiveKit Cloud dashboard.