LiveKit Agents + ContextDB

Add customer memory to LiveKit Agents.

LiveKit Agents runs the realtime session, voice pipeline, and function tools. Fetch a caller snapshot before AgentSession.start, then evaluate memory policy inside tools that authenticate, authorize, and recheck state before they book, refund, or change an account.

Prefetch recall into the system prompt when the participant joins. Call evaluate_action inside consequential tools and close every outcome with report_execution. recall_for_action returns memories for compatibility. It is not authorization. Examples use the public Cloud client against ContextDB Cloud. The local engine uses pycontextdb. This independent guide has no vendor partnership or endorsement.

Runnable starter

Run the LiveKit starter.

The starter pins its LiveKit and ContextDB dependencies, recalls before each reply, and keeps every key in the worker environment.

npx degit atomsai/contextdb-clients/starters contextdb-starters
cd contextdb-starters/livekit-agents-python
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Add your LiveKit, PyAI, OpenAI, and ContextDB server keys.
set -a; source .env; set +a
python agent.py dev

Inspect the starter and pinned requirements → Try a memory scenario in Playground →

Session start

Start every call with the customer's history.

Fetch the caller's context while the session is being established, compress it into the instructions, and the agent opens the call already knowing the history instead of interrogating the caller.

# livekit-agents entrypoint
from livekit.agents import Agent, AgentSession, function_tool
from contextdb_cloud_client import CloudClient

async def entrypoint(ctx):
    identity = await authenticate_room_participant(ctx.room)
    caller = identity.customer_id  # derived from the authenticated room session

    async with CloudClient(BASE_URL, api_key=KEY) as cdb:
        snapshot = await cdb.recall(caller, "caller context", top_k=5)

    agent = Agent(
        instructions=(
            "You are the scheduling assistant.\n"
            f"Known caller context:\n{snapshot.context}"
        ),
        tools=[build_booking_tool(identity)],
    )
    await AgentSession().start(agent=agent, room=ctx.room)

Inside the tool

The booking tool checks the customer's confirmed day.

The model decides to call book_visit because something in the conversation sounded like an instruction. The model supplies only the day. The tool captures the customer partition from the authenticated room, checks host authorization and current slot state, then branches on act, ask, or abstain.

def build_booking_tool(identity):
    caller = identity.customer_id  # closed over, never supplied by the model

    @function_tool
    async def book_visit(day: str) -> str:
        """Book one service visit for the authenticated caller."""
        slot = await calendar.get_slot(day)
        authorized = await calendar.can_book(
            actor_id=identity.actor_id,
            customer_id=caller,
            slot=slot,
        )
        if not authorized:
            return "This session is not authorized to book that visit."

        async with CloudClient(BASE_URL, api_key=KEY) as cdb:
            decision = await cdb.evaluate_action(
                caller, f"book service visit on {day}"
            )

            if decision.outcome == "act":
                try:
                    booking = await calendar.book(
                        caller, day, expected_version=slot.version
                    )
                except SlotChanged:
                    await cdb.report_execution(
                        caller, decision.decision_id, "appointment.book", "failed",
                        idempotency_key=f"livekit-receipt-{decision.decision_id}",
                        error_code="current_state_changed",
                    )
                    return "That slot just changed, so I did not book it."
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "succeeded",
                    idempotency_key=f"livekit-receipt-{decision.decision_id}",
                    external_ref=booking.ref,
                )
                return f"Booked for {day}. Reference {booking.ref}."

            if decision.outcome == "ask":
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "skipped",
                    idempotency_key=f"livekit-receipt-{decision.decision_id}",
                )
                return f"Please confirm that I should book {day}."

            if decision.outcome == "abstain":
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "skipped",
                    idempotency_key=f"livekit-receipt-{decision.decision_id}",
                )
                return "I cannot book from the available evidence."

            raise RuntimeError("unknown ContextDB action outcome")

    return book_visit

LiveKit call order

Add memory at four points in the LiveKit call.

Moment	LiveKit surface	ContextDB call
Participant joins	Entrypoint, before session start	`recall` → snapshot into instructions
Consequential tool call	Inside `@function_tool`	`evaluate_action` → branch, then `report_execution`
Caller confirms out loud	Confirmation tool	Host authenticates and retains the end-user attestation, then `confirm` records the scoped memory update
Session ends	Shutdown callback	`remember` the durable facts with source and confidence

LiveKit implementation

How memory fits into a LiveKit session.

Does the snapshot fetch slow down session start?

Run it concurrently with session setup. It is one HTTPS call for a handful of compressed memories, not a transcript dump. For inbound telephony, resolve the caller and prefetch during ring time so the context is ready before the agent speaks.

Can I use this with LiveKit's realtime model integrations?

Yes. The pattern is model-agnostic: context goes into instructions, and the gate lives inside your function tools, which realtime models call the same way.

What if my agent runs multiple tools per call?

Gate the consequential ones only. Weather lookups and knowledge answers can stay ungated. Each consequential tool should derive identity from the room session and apply its own host authorization.

Start from the LiveKit Agents starter.

The linked starter covers conversational recall and sourced writes. The authenticated, host-owned evaluate_action and report_execution boundary is shown on this page for you to add.

Copy the LiveKit Agents starter