Recording a call

This page assumes you have an observer. See Overview if not.

The shape of a recorded call

session = vaani.start_session(agent_id="support-bot", metadata={"env": "prod"})

turn = session.start_turn()
op = turn.start_operation(type="llm", provider="openai", model="gpt-4o-mini")
op.event("first_token")
op.end(status="ok", response={"tokens": 128})
turn.end()

finalized = await session.end(outcome="completed")
await vaani.upload_package(finalized)

Everything below is detail on those seven lines.

Sessions

session = vaani.start_session(
    session_id="call-8f21",              # optional; a UUID is generated
    agent_id="support-bot",
    metadata={"env": "prod", "version": "2026.08.1", "region": "eu-west-1"},
)

metadata becomes the filter and group-by dimensions in the dashboard. Establish a convention early — it is much harder to backfill than to set.

end() is idempotent. It closes open turns and operations, composes the stereo call.audio, and writes manifest.json last.

Turns

turn = session.start_turn()          # generated id
turn = session.start_turn("turn-7")  # your id
turn.id                              # always a string
turn.ended                           # bool
turn.end()

turn.start_operation(**kwargs) is session.start_operation with turn_id prefilled. turn.context() scopes ambient instrumentation to this turn.

A turn has no timing of its own — it is a grouping key. Its latency is derived from the milestones of the operations inside it.

Operations

op = session.start_operation(
    type="stt",                # stt | llm | tts | tool
    turn_id=turn.id,
    scope="turn",              # or "connection"
    endpoint_id="stt",
    provider="deepgram",
    model="nova-3",
    transport="manual",
    started_at_ms=None,        # defaults to session.now()
    request={"language": "en"},
)

Calls on an already-ended operation are silently ignored — event(), sample(), set_turn(), set_request() and a second end() are all no-ops. That keeps a late callback from throwing inside your agent.

set_turn for late attribution

Streaming frameworks often start work before they know which utterance it belongs to:

op = session.start_operation(type="stt", provider="deepgram")
# ... the framework resolves the utterance
op.set_turn(turn.id)
op.end(status="ok")

Milestones

op.event("speech_started")
op.event("first_partial", {"text_length": 12})
op.event("first_byte", occurred_at_ms=session.now())

Data can be passed as a mapping, as keyword arguments, or both. occurred_at_ms is pulled out of the payload if present; otherwise the session clock is stamped at the moment of the call — so record milestones when they happen.

Repeated names accumulate rather than overwrite:

for frame in stream:
    op.event("received_frame", total_bytes=running_total)

produces one entry with the first occurred_at_ms, the latest last_at_ms, a count, and the merged payload. See Milestones and samples for the names the dashboard recognises.

Samples

op.sample("partial_transcript", {"text": partial}, limit=100)

Bounded at limit per name (default 100). At the cap, further samples are dropped and the bucket is marked truncated: true. Payloads pass through the payload_max_bytes bound.

Audio

FORMAT = {"encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1}

session.record_inbound_audio(caller_pcm, FORMAT)    # right channel
session.record_outbound_audio(agent_pcm, FORMAT)    # left channel

Both return True if the chunk was recorded, False if capture is off or the session has ended.

Requirements, all enforced:

• The chunk must be bytes, bytearray or memoryview — otherwise TypeError.
• encoding must be pcm_s16le — otherwise ValueError.
• sample_rate_hz and channels must be positive integers — otherwise ValueError.
• The format cannot change within a session, per track.

An optional timestamp_ms in the format overrides the arrival time:

session.record_inbound_audio(chunk, {**FORMAT, "timestamp_ms": frame.captured_at_ms})

Websockets

from vaani_observer import observe_websocket

handle = observe_websocket(vaani, socket, session=session, url=url, endpoint_id="stt")
...
handle.detach()

Or via the observer, which classifies the URL for you:

handle = vaani.observe_websocket(socket, session=session, url=url)

The span is created with scope="connection" and records connected, sent_frame, received_frame (accumulating byte counts) and the close code. Frame contents and authentication headers are never recorded.

For transports the SDK cannot patch, feed it neutral lifecycle data yourself:

session.record_websocket_event(
    event="closed", url=url, code=1000, occurred_at_ms=session.now()
)

Ambient context

with session.context():
    await client.post("https://api.openai.com/v1/chat/completions", json=body)

with session.with_turn(turn.id):
    ...

with session.with_endpoint("llm", turn_id=turn.id):
    await provider_sdk.complete(prompt)   # bypasses URL classification

Instrumented httpx and aiohttp calls are recorded only when there is an ambient session and the URL matches an endpoint rule. current_context() reads the active ObserverContext.

For callbacks:

socket.on_message(session.bind(handle_message))

bind() handles both sync and async callables, and an async handler inherits an enclosing endpoint or turn.

Deferring capture past the response

async def handler(request):
    response = await downstream()
    session.defer_capture(record_body(response))
    return response

session.end() waits for deferred work, so a body read that outlives the response still lands in the package.

Ending and uploading

finalized = await session.end(outcome="completed")
finalized.session_id
finalized.directory

await vaani.upload_package(finalized)     # or upload_package_sync(finalized)

At shutdown:

await vaani.flush()

flush() finalizes every open session and waits for writes to land. Finalization errors propagate — a package that could not be written should be visible, not swallowed.

Next