LiveKit — Tools, Handoffs & Workflows¶

from livekit.agents import Agent, RunContext, ToolError, function_tool


class SupportAgent(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a support agent. Keep answers under two sentences.")

    @function_tool()
    async def lookup_order(self, context: RunContext, order_id: str) -> dict:
        """Look up an order's status.

        Args:
            order_id: The order number, for example ord-42.
        """
        if not order_id.startswith("ord-"):
            raise ToolError(f"Unknown order id format: {order_id}")     # the model sees this message and can tell the user
        return {"order_id": order_id, "status": "shipped"}

from livekit.agents.beta.tools import EndCallTool

end_call = EndCallTool(extra_description="Only end after the issue is resolved.", end_instructions="Thank the caller and say goodbye.")
agent = Agent(instructions="...", tools=end_call.tools)        # adds an `end_call` tool

from livekit.agents import AgentSession, mcp          # uv add "livekit-agents[mcp]"

session = AgentSession(
    # stt=..., llm=..., tts=...
    tools=[
        mcp.MCPToolset(
            id="orders",
            mcp_server=mcp.MCPServerHTTP(
                "https://mcp.internal.example.com/mcp",           # /mcp = streamable HTTP, /sse = SSE; or transport_type=...
                headers={"Authorization": f"Bearer {token}"},
                allowed_tools=["order_status", "refund_quote"],   # everything else is excluded
            ),
        ),
        mcp.MCPToolset(id="files", mcp_server=mcp.MCPServerStdio(command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/data"])),
    ],
)

from dataclasses import dataclass, field

from livekit.agents import Agent, AgentSession, RunContext, function_tool


@dataclass
class CallState:
    customer: str | None = None
    handled: list[str] = field(default_factory=list)


class SupportAgent(Agent):                                   # the class from "Function Tools" with one more tool
    # __init__ and lookup_order as above

    @function_tool()
    async def transfer_to_billing(self, context: RunContext[CallState]) -> "BillingAgent":
        """Transfer the caller to the billing team."""
        return BillingAgent(chat_ctx=self.chat_ctx.copy(exclude_instructions=True))


class BillingAgent(Agent):
    def __init__(self, chat_ctx=None) -> None:
        super().__init__(instructions="You handle billing questions only.", chat_ctx=chat_ctx)

    async def on_enter(self) -> None:                       # runs on every entry: greet or continue
        self.session.generate_reply(instructions="Say you are the billing team.")


session = AgentSession[CallState](userdata=CallState())      # llm=... in real use

from dataclasses import dataclass

from livekit.agents import Agent, AgentTask, RunContext, function_tool


@dataclass
class ConsentResult:
    agreed: bool


class CollectConsent(AgentTask[ConsentResult]):
    def __init__(self, chat_ctx=None) -> None:
        super().__init__(
            instructions="Ask whether the call may be recorded. Call record_consent with the answer.",
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(instructions="Ask for recording consent.")

    @function_tool()
    async def record_consent(self, context: RunContext, agreed: bool) -> None:
        """Record the caller's answer.

        Args:
            agreed: True if the caller agrees to recording.
        """
        self.complete(ConsentResult(agreed=agreed))


class Intake(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are an intake agent.")

    @function_tool()
    async def start_recording_flow(self, context: RunContext) -> str:
        """Ask the caller for recording consent and report the outcome."""
        result = await CollectConsent(chat_ctx=self.chat_ctx)      # awaiting runs the task
        return "consent granted" if result.agreed else "consent refused"

session = AgentSession[CallState](llm=llm, userdata=CallState())
# in a tool:   context.userdata.handled.append(order_id)
# anywhere:    session.userdata.customer = "ord-42"

from livekit.agents import Agent, ChatContext, ChatMessage, StopResponse


class Guarded(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="Answer from the provided facts.")

    async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage) -> None:
        text = (new_message.text_content or "").strip()
        if not text:
            raise StopResponse()                              # no reply for an empty turn
        facts = await search_kb(text)                         # your retriever
        turn_ctx.add_message(role="assistant", content=f"Facts: {facts}")

from pydantic import BaseModel
from livekit.agents import ChatContext
from livekit.plugins import openai


class Ticket(BaseModel):
    category: str
    urgent: bool


async def classify(llm: openai.LLM, text: str) -> Ticket:
    ctx = ChatContext()
    ctx.add_message(role="user", content=text)
    stream = llm.chat(chat_ctx=ctx, response_format=Ticket)
    out = "".join([c.delta.content async for c in stream if c.delta and c.delta.content])
    return Ticket.model_validate_json(out)

from livekit.agents import room_io

def on_text(session, event: room_io.TextInputEvent) -> None:
    if event.text == "/status":
        session.say("The agent is running normally.")
        return
    session.interrupt()
    session.generate_reply(user_input=event.text)         # default behaviour

options = room_io.RoomOptions(text_input=room_io.TextInputOptions(text_input_cb=on_text))

import json

from livekit import rtc
from livekit.agents import Agent, RunContext, ToolError, function_tool, get_job_context


class Assistant(Agent):
    @function_tool()
    async def get_user_location(self, context: RunContext, high_accuracy: bool = False) -> str:
        """Ask the user's browser for its location.

        Args:
            high_accuracy: Whether to ask for a precise fix.
        """
        room = get_job_context().room
        user = next(iter(room.remote_participants.values()), None)
        if user is None:
            raise ToolError("The user is not connected.")
        try:
            return await room.local_participant.perform_rpc(
                destination_identity=user.identity,
                method="getUserLocation",
                payload=json.dumps({"highAccuracy": high_accuracy}),
                response_timeout=10.0 if high_accuracy else 5.0,
            )
        except rtc.RpcError as exc:
            raise ToolError(f"Unable to retrieve user location ({exc.code})") from exc


async def entrypoint(ctx):                                       # after the session started:
    @ctx.room.local_participant.register_rpc_method("setLanguage")
    async def set_language(data: rtc.RpcInvocationData) -> str:   # data.caller_identity, data.payload
        session.userdata = {"lang": data.payload}
        return "ok"