LiveKit Agents + ContextDB
Give every LiveKit caller an agent that remembers them.
LiveKit Agents gives you the realtime session, the voice pipeline, and function tools. ContextDB slots in as two things: a caller snapshot loaded at session start, and a trust check inside the tools that change the real world.
recall into the system prompt when the participant joins,
and call recall_for_action inside any function tool that
books, refunds, or mutates state.
Session start
Start every call with the customer's history.
The cheapest latency win in voice memory: 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): caller = resolve_caller(ctx.room) # your identity mapping 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=[book_visit], ) 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 tool's first job is
to check whether trusted memory agrees. If the only evidence is a
maybe, the tool returns a question instead of a booking.
@function_tool
async def book_visit(day: str, caller_id: str) -> str:
"""Book a service visit. Only books against
confirmed caller preferences."""
async with CloudClient(BASE_URL, api_key=KEY) as cdb:
evidence = await cdb.recall_for_action(
caller_id, f"book service visit on {day}"
)
if not evidence:
pending = await cdb.pending_confirmations(caller_id)
if pending:
# surface the ask; confirm on an explicit yes
return (f"Please confirm: {pending[0].content}. "
"Shall I book that?")
return "No confirmed day on file. Which day works?"
booking = await calendar.book(caller_id, day)
return f"Booked for {day}. Reference {booking.ref}."
Action map
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 |
recall_for_action → act, ask, or abstain |
| Caller confirms out loud | Confirmation tool | confirm → pending fact graduates |
| Session ends | Shutdown callback | remember the durable facts with source and confidence |
Questions
LiveKit-specific questions.
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 should stay fast and ungated. The action map above is the split we recommend.
Add customer memory to your LiveKit agent.
The quickstart walks the exact recall and gate calls above.