callva-livekit

Drop-in call webhooks and per-call configuration for any LiveKit agent.

Two things most voice agents need and the LiveKit SDK does not provide: a webhook when a call starts and ends, and a way to get the prompt for this call from somewhere other than your source code.

Both are opt-in. Installing the package activates nothing, patches nothing, and changes nothing about how your session behaves.

Install

pip install callva-livekit          # webhooks + config
pip install callva-livekit[s3]      # + recording upload to S3 or R2

Use

from livekit.agents import Agent, AgentServer, AgentSession
from callva.livekit import call as callva_call
from callva.livekit import config as callva_config
from callva.livekit import webhook as callva_webhook

server = AgentServer()

@server.rtc_session(agent_name="my-agent", on_session_end=callva_webhook.on_session_end)
async def entrypoint(ctx):
    await ctx.connect()

    if await callva_call.await_pickup(ctx) is None:      # outbound: rang, nobody answered
        await callva_call.end(ctx, wait=False)
        return

    config = await callva_config.load()

    session = AgentSession(...)
    callva_webhook.attach(session)

    await session.start(agent=Agent(instructions=config.prompt), room=ctx.room, record=True)
    await session.generate_reply(instructions=config.greeting)

Neither call takes a JobContext — it comes from the SDK's own contextvar. load() hands you an object; what you do with the prompt is your decision.

A complete runnable agent is in examples/agent.py, with a local receiver that prints what arrives in examples/receiver.py.

What arrives

call.dialing when the dial goes out, call.started when someone is on the other end, call.ended when it is over — always, however it ended.

There is one terminal event, not two: how a call ended is a value, not a kind, so a consumer has exactly one thing to handle.

The vocabulary is this package's own. LiveKit's sip.callStatus — dialing, ringing, automation, active, hangup — is normalized by LiveKit itself and does not change when a trunk moves between carriers, but it carries no terminal outcome at all: a refused call simply stops updating and the participant vanishes. The outcome therefore comes from the participant's disconnect reason, which cannot tell busy from declined — so this package does not claim to either. rejected means one of them. The raw signals travel untouched in livekit.sip.callStatus and livekit.disconnect_reason.

call.ended_by answers the other question — not how the call came out, but who decided it was over. A completed call the caller rang off from and one a duration limit cut short are the same status and not the same event.

Those five are what the package can see for itself. Anything else that ends a call — a transfer, a supervisor pulling it — is yours to claim, and the word is not interpreted:

callva_call.ended_by("transfer")

First writer wins. An ending has one cause and the first observer is the closest to it, so a later claim over one already made does nothing — which is what makes it safe to claim from two places that both watch the same hangup. The key is on every event and null until something claims it.

{
  "event": "call.ended",
  "id": "8f1c…:call.ended:1757...",
  "timestamp": 1757600000.12,
  "call": {
    "id": "8f1c…",
    "direction": "inbound",
    "from": { "number": "+37255512345", "identity": "sip_+37255512345", "name": null },
    "to":   { "number": "+3726001234", "identity": null, "name": null },
    "started_at": 1757599940.5, "ended_at": 1757600000.1, "duration": 59.6,
    "status": "completed", "ended_by": "user",
    "project_id": "pr_…", "tenant_id": "tn_…", "type": "outbound_campaign"
  },
  "agent": { "id": "ag_…", "name": "Anna", "…": "the agent block exactly as configured" },
  "environment": "production",
  "livekit": {
    "room": { "name": "call-1", "sid": "RM_…", "metadata": null },
    "job":  { "id": "AJ_…", "dispatch_id": "AD_…", "agent_name": "my-agent", "…": "…" },
    "participant": { "identity": "sip_…", "attributes": { "sip.callID": "…" } },
    "sip":  { "callID": "…", "phoneNumber": "…", "twilio": { "callSid": "…" } },
    "session_report": { "chat_history": {}, "usage": [], "options": {}, "…": "…" }
  },
  "recording": { "delivery": "storage", "audio_key": "8f1c….ogg",
                 "session_report_key": "8f1c….session.json" },
  "tags": { "tags": ["lk.success"], "outcome": "success", "reason": null }
}

Everything LiveKit produces is nested under livekit verbatim — including session_report, which is ctx.make_session_report().to_dict() untouched: full chat history with timestamps, per-provider token usage, recorded events, session options. A field the SDK adds tomorrow reaches you without a release here.

The call block is the only thing reshaped, because it is the only thing LiveKit does not model: a stable id across both events, a direction, and a from and a to. Whatever the configuration source filed the call under — project_id, tenant_id, type — travels back in it untouched, and a key that never arrived is absent rather than null.

agent is the configuration response's own agent block, echoed back on every event exactly as it arrived, including whatever a per-call override changed and whatever this schema does not name. Nothing in it is interpreted here; the sender reads its own values back, which is what a platform deciding per call needs. environment is the same passthrough for the response's top-level environment, and is null when the source named none — it is never read from this process's environment.

Requests carry X-Webhook-Idempotency-Key, and X-Webhook-Signature when a secret is set — sha256 HMAC over {timestamp}.{body}, with X-Webhook-Timestamp alongside. Delivery retries on 5xx and network errors and fails fast on 4xx.

Being answered, and hanging up

participant = await callva_call.await_pickup(ctx)   # None if nobody picked up
await callva_call.end(reason="the agent said goodbye")
callva_call.leave_console_when_done(ctx)            # console only; a real worker stays up

An inbound call is answered by the time a participant exists. An outbound one is not: the participant appears while the phone is still ringing, and sip.callStatus is what says otherwise. Waiting for a participant alone reports a ringing call as live, and reports one that was never picked up as live too. await_pickup waits for the real thing, and a trunk that answers and hangs up inside a second does not slip past it.

Nothing is reported from there and nothing is torn down — the reason is left on the call's state, where webhook turns it into an outcome, and hanging up stays your decision.

end releases the caller before it ends the job. Shutting the job down only takes the agent out of the room; whoever is on the other end stays connected to a room with nobody in it until the server's empty_timeout expires, which on a telephone call means it has not ended. It waits for the agent to stop speaking first — pass wait=False for a call being abandoned rather than finished — and the report and the recording still go out, because they belong to the shutdown sequence and a closed room does not interrupt it.

A call nobody is ending

callva_call.supervise(
    session,
    max_duration=config.agent.max_duration,
    silence_timeout=config.agent.user_silence_timeout,     # None: the quiet is not watched
    prompt_phrases=config.agent.prompt_phrases,
    max_prompt_attempts=config.agent.max_prompt_attempts,
    call_silence_timeout=config.agent.call_silence_timeout,
    utter=say_this,                                        # async def say_this(phrase: str | None)
)

A call ends on its own terms or it does not end at all: the caller hangs up and the job stays in the room, the line runs on past what anyone meant to pay for, or the other end simply stops answering. supervise watches for all three and hangs up through end, so the caller is released and the report still goes out.

The quiet is measured here rather than by the framework's user_away_timeout, which is a single edge at a fixed timeout that nothing re-arms — no use for counting. After silence_timeout of nothing said, the caller is reminded with one of prompt_phrases, up to max_prompt_attempts times; when those are spent the call ends call_silence_timeout later, as ended_by: silence, or the conversation simply goes on where you set none.

Two measurements, two kinds of evidence. The clock restarts on any sign of life — a voice-detected edge either way, any transcript, the agent speaking, a tool landing — because what it prevents is talking over somebody. The count goes back to zero only on a transcript carrying words, interim or final: a cough is not an answer, and one must not buy back a reminder. A reminder is held back for a moment in case the caller was only pausing, a tool in flight is not quiet, and none of it runs before the caller is on the call or after it has ended.

The reminders belong to an episode of quiet rather than to the call: a caller who answers and goes quiet again is reminded again.

This package never speaks. utter is your async callable, handed the phrase and returning nothing — how a line reaches a particular model is that stack's business. It is waited on for at most PROMPT_GRACE's sibling UTTERANCE_TIMEOUT, because it is yours and a coroutine that never returns would park the watch for the life of the process; giving up on one costs that reminder and nothing else. Without a callable at all the timing, the counting and the ending still happen; nothing is said, and a line is logged saying so.

It is called with None where no phrase was written, and that is an ask like any other: nothing was written for this reminder, so say something suitable yourself. Handle both cases — an operator who switched reminders on and never composed a sentence still meant the caller to be checked on, and this package will not answer that by inventing one, in a language of its own choosing, for somebody else's agent to say. Every number is yours too: no timeout, phrase or attempt count is invented here, and the whole of it is inert until silence_timeout is set.

end_when_away=True is the short version for anyone who wants it — hang up on the framework's own away edge, at whatever timeout the session was built with. It is inert on a session built with user_away_timeout=None, which is what a deployment measuring its own quiet passes, and asking for both on one call is warned about: two clocks on one silence, and the framework's fixed one wins.

When a call goes wrong

callva_webhook.collect_errors()     # once, where the worker starts up

Every error logged anywhere in the process is kept and delivered inside call.ended, as errors. There is no separate event for a call that fell apart: that call still ends and still reports — what was missing was ever saying why.

It attaches a handler to the root logger, which is a process-wide thing to do and so is asked for rather than assumed. Our own errors are never collected, because a failing delivery would report itself forever.

A record carries the logger, the level, the line, and — where something was raised — what was raised and where it passed:

{ "logger": "callva.livekit", "level": "ERROR",
  "message": "the google stack was asked to open the call and could not",
  "exception": "google.api_core.exceptions.PermissionDenied\n  at callva.livekit.internal.call.opening:191 in open\n  at google.api_core.grpc_helpers:76 in error_remapped_callable" }

Not the exception's own message and not a rendered traceback. A client library's sentence is where an organisation id, a project, a quota or a key it read back ends up, and a traceback carries this deployment's absolute container paths and its source lines, read off its disk at the moment of the failure. The class and the frames say what failed and roughly where, which is the part a reader acts on. The log line itself is kept, because a sentence somebody wrote about what went wrong is the reason this exists at all — and cut at 500 characters, because it is not ours. Anything a library attaches to a record structurally, through extra=, is not read here at all.

The trade that leaves is worth saying plainly: the lines other libraries write about the call do travel. If you point WEBHOOK_URL at a party you do not control, that is what you are forwarding.

What this package writes into errors about a failed configuration request is its own account of the failure — configuration request to https://platform.test/v1/config failed: HTTP 500 — and not the page the endpoint answered with. The endpoint that serves a configuration and the endpoint that receives a report belong to two parties as often as to one, and a framework's stack trace is not the second one's to keep. The body is written to this container's log in full, and carried on the FetchError for a caller still holding it. The endpoint is named by scheme, host, port and path; userinfo, query and fragment are where a key rides, so they are dropped from what is written down and not from what is sent. The path is kept, because without it a deployment serving several endpoints from one host cannot tell which failed — so do not put a secret in one.

Configuration for a call

Configuration reaches the agent through agent dispatch metadata, read as ctx.job.metadata:

{ "callva": { "call_id": "…", "direction": "outbound", "to": "+372…",
              "config": { "agent": { "prompt": "…" } } } }

Put a config_url there instead of a config, or set CONFIG_URL, and the agent follows that instead. The request is the question — it carries who is calling, which number they reached and the whole SIP envelope — so the endpoint can answer "this number belongs to that customer, here is their prompt". That is the inbound case in one hop.

A pointer can also be a local file — file:///etc/agent.json or a plain ./agent.json. It is read as it is, with no request and no waiting for anyone to join, which makes it the shortest development loop there is. It cannot answer per caller, so it is not the production channel.

The response:

{
  "call":  { "id": "019f0c4e-1f3a-7a55-9d21-2b0e5f77a1c3", "direction": "inbound",
             "project_id": "pr_…", "tenant_id": "tn_…", "type": "outbound_campaign" },
  "environment": "production",
  "agent": {
    "id": "…", "name": "Anna",
    "prompt":   "You are speaking with {{ name }}.",
    "greeting": "Hi {{ name }}, how can I help?",
    "greeting_type": "message",
    "agent_waits_for_user": false,
    "prompt_variables": { "name": "Anna", "attempt": 2, "vip": true },
    "max_duration_seconds": 600,
    "user_silence_timeout_seconds": 15
  },
  "preset":   { "name": "gemini_vertex", "config": { "voice": "Aoede" } },
  "tools":    { "endCall": { "type": "end_call", "enabled": true } },
  "services": { "webhook": { "url": "https://tenant.example/hook", "secret": "…" } }
}

Every value has one home. The prompt, the greeting and the variables belong to the agent; the id belongs to the call; the webhook belongs to the services. Nothing is repeated at the root for convenience, so nothing can disagree with itself.

preset is the only block that varies with the speech stack. An agent that builds its own pipeline ignores it and reads the rest; an agent that is assembled from configuration reads all of it. tools describes what the agent may call, not what it did.

{{ name }} is substituted into both prompt and greeting. A placeholder with no variable is left exactly as it was and logged — one missing key must not take down a call that is already ringing. JSON types survive: config.variables.get_int("attempt") is 2.

Opening the call is three states, not two, and config.agent carries all three: speaks_first says whether the agent opens at all, and greeting_type says whether the greeting is a line to speak (message) or an instruction to compose one from (prompt).

services.webhook overrides the environment, which is what lets one worker serve many tenants.

call.id files the call under an id you already hold. A responder that creates a record for the call before answering can name it here, and every event afterwards carries that id — so both sides address one record, and neither has to store a field holding the other's identifier. It is the only part of the call's identity configuration may decide. A dispatcher that named the call outranks it, and an id offered after the first event has gone out is refused with a warning.

extra is there for what this schema does not describe, and is never interpreted.

What the responder sends about itself comes back in the report: the whole agent block, the environment, and the project_id, tenant_id and type it filed the call under. A responder that decides something per call therefore reads back the value that was in force for that call, not the one it has stored.

When configuration cannot be resolved the call is terminated and the reason logged. An agent without its prompt is a broken call either way. Pass on_error="continue" if you would rather carry on.

When the endpoint says no

An endpoint can also refuse the call — the number is disabled, the balance is spent, too many calls are already up. That is not a failed request, and it does not arrive as one:

{ "error": "Balance exhausted.", "action": "terminate",
  "reason_code": "insufficient_balance",
  "caller_message": "Sorry, this service is unavailable right now." }

Flat and top-level, with no wrapper around it — a wrapper key would be one vendor's envelope, and this package knows none. action is what decides: a body asking for the call to end is a refusal whatever HTTP status carried it, and it is raised on the first response and never retried, because a refusal spelled as a 5xx is still an answer and retrying it only hammers an endpoint that is already struggling.

try:
    config = await callva_config.load()
except callva_config.ConfigRefused as refused:
    if refused.caller_message:
        await session.generate_reply(instructions=f"Say: {refused.caller_message}")
    await callva_call.end(reason=refused.reason_code or "refused")
    return

reason_code is carried and never read here — which codes mean what is the responder's vocabulary, the same way environment is passed through untouched. caller_message is what the responder wrote for the person on the phone, and is the reason this is an object and not a log line.

ConfigRefused is a ConfigError, so an agent that already handles configuration failing needs no new handler: an agent that was refused is as short of a prompt as one that could not ask. Catching the specific case first, as above, is all new code does differently. on_error="continue" covers it like anything else — ask to carry on without configuration and you carry on, with the refusal logged.

Nothing is hung up for it either: the message is there to be said first, and ending the call stays your decision.

Room metadata is deliberately not used as a channel: it is broadcast to every participant in the room, so a prompt placed there is readable by any connected client.

Environment

Every value has a constructor argument that takes precedence.

Recording

With a bucket configured, the recording and the session report are written under the same call id — <call_id>.ogg and <call_id>.session.json — and call.ended carries recording.delivery: "storage" with recording.audio_key and recording.session_report_key, which are known before the bytes move. The session report is not a transcript, and it deliberately does not take the plain <call_id>.json name: a platform that stores a transcript of its own is likely to have claimed it, and this would land on top of it. Needs record=True on session.start() and the codecs extra (pip install "livekit-agents[codecs]").

What travels is the key, never a URL to fetch it with. A link that plays a recording to whoever holds it does not belong in a webhook body, and the bucket is this deployment's own configuration rather than something a consumer should act on. Whoever holds the credentials reads the object.

Without a bucket there is nowhere to put the audio and none is kept. That is logged as an error naming RECORDING_S3_BUCKET on every call that recorded something, because a missing recording is otherwise discovered weeks later from an empty field.

The call.ended webhook is always sent before the upload, so a call is closed out with a terminal status even if the process does not survive the transfer. That makes delivery: "storage" a statement of intent, so a call.recording event follows a successful upload carrying the same keys plus "stored": true — the statement of fact. A failed upload sends nothing, and the call is still closed out.

call.recording is deliberately thin: event, id, timestamp, call, agent, environment, tags and recording, and nothing else. Everything LiveKit produced reached the same consumer under the same call id in call.ended minutes earlier, and on a measured two-minute call the session report alone was 62 KB of an 81 KB envelope — a figure that grows with the call. The agent block stays, so a consumer reading only this event can still tell which configuration was in force.

Versioning

Semantic versioning, and the compatibility promise is about what a receiver has to parse, not about the size of the diff:

• the patch position — fixes, and fields added to a payload. Adding is not breaking: a receiver ignores keys it does not know, and because everything LiveKit produces is nested verbatim, fields the SDK adds arrive without a release here at all. A new exception type belongs here too when it lands under one a caller already catches.
• the minor position — anything a receiver could choke on: a field renamed or removed, an event name changed, a header changed, an environment variable renamed, a public function changed — an exception that escapes every handler written against the last release included.
• 1.0.0 — when the contract is worth freezing.

In 0.x the digits are shifted one place: the middle number is the breaking one, which is what ^0.1.0 means to every resolver. So a 0.2.0 here is not a large release — it is a release that someone's receiver has to be told about.

Two things worth knowing

Your agent needs agent_name set and explicit dispatch. With automatic dispatch job.metadata arrives empty, and nothing can be addressed to this call.

If you cannot use on_session_end — you are still on WorkerOptions — attach() falls back to a shutdown callback and says so in the log. That path is bounded by shutdown_process_timeout, 10 seconds by default, after which the worker kills the process mid-upload. Raise it, or move to AgentServer.

License

Apache-2.0