attach() (observe)

attach() is VoiceGateway’s observe seam: a passive meter for cost and latency. Call it once with your AgentSession or PipelineTask and every STT, LLM, and TTS call flowing through it is priced and recorded. It never reroutes, throttles, or blocks a call: control lives in guard(). Because attach() is the only source of metrics, pairing it with guard() never double-counts; guard() writes no metrics of its own.

Signature

voicegateway.attach(
    session,                          # LiveKit AgentSession OR Pipecat PipelineTask
    *,
    project: str | None = None,       # env: VOICEGW_PROJECT, else "default"
    agent_id: str | None = None,      # env: VOICEGW_AGENT_ID, else hostname
    tenant_id: str | None = None,     # optional per-call tenant attribution
    channel: str | None = None,       # "telephony" | "web"; auto-detected
    collector_url: str | None = None, # fleet push target (env: VOICEGW_COLLECTOR_URL)
    api_key: str | None = None,       # collector key (env: VOICEGW_API_KEY)
    sink: Sink | None = None,         # advanced/testing override
    room: str | None = None,          # LiveKit room name for probe correlation; auto-resolved
    heartbeat: bool = False,          # register + heartbeat this process in the fleet roster
    transcript: bool = True,          # capture the call transcript (LiveKit only for now)
    snapshots: bool = False,          # capture conversation-state snapshots for replay (opt-in)
    turns: bool = True,               # capture per-exchange turn timing (LiveKit only)
    dead_air: bool = True,            # watch for silence longer than the threshold (LiveKit only)
) -> str                              # correlation session id stamped on every row

attach() picks its framework by inspecting the session’s type. An object it does not recognize as a LiveKit or Pipecat target falls through the LiveKit path silently: no error, no rows written. Pass a real AgentSession or PipelineTask.

Wiring

LiveKit
Pipecat

from livekit.agents import Agent, AgentSession
from livekit.plugins import deepgram, openai, cartesia

import voicegateway


async def entrypoint(ctx):
    await ctx.connect()

    session = AgentSession(
        stt=deepgram.STT(model="nova-3"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=cartesia.TTS(model="sonic-3"),
    )

    voicegateway.attach(session, project="my-agent")

    await session.start(agent=Agent(instructions="Be helpful."), room=ctx.room)

attach() subscribes to the per-component metrics_collected events (works with any plugin, no wrapping) and finalizes on the session’s close event: in-flight writes drain, then the sink flushes.

Pipecat has no cumulative usage aggregate, so attach() sums the metrics it observes. enable_metrics and enable_usage_metrics must both be on, or Pipecat emits no usage frames and there is nothing to record:

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.cartesia.tts import CartesiaTTSService

import voicegateway

transport = ...  # your Pipecat transport: DailyTransport, a SIP serializer, LocalAudioTransport, etc.

pipeline = Pipeline([
    transport.input(),
    DeepgramSTTService(api_key=DEEPGRAM_API_KEY),
    OpenAILLMService(api_key=OPENAI_API_KEY, model="gpt-4o-mini"),
    CartesiaTTSService(api_key=CARTESIA_API_KEY, voice_id=VOICE_ID),
    transport.output(),
])
task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
)

voicegateway.attach(task, project="my-agent")

voicegateway.Observer(...) takes the same core keyword arguments and can be passed to PipelineTask(observers=[...]) instead, as an equivalent to attach(task). It finalizes on the pipeline’s EndFrame.

What it records

One row per request, through a Sink (local SQLite by default, or a collector when collector_url / VOICEGW_COLLECTOR_URL is set):

Field group	What it captures
modality + provider + model	`stt` / `llm` / `tts`, provider name, model id
usage units	STT audio minutes, LLM prompt/completion/cached tokens, TTS characters
cost	priced through `voice-prices` from the usage units
latency	time to first byte (`ttfb_ms`) and total latency
correlation	session id, `project`, `agent_id`, `tenant_id`, `channel`
routing	`fallback_from` / `status` when a `guard()` fell back to this provider

LLM and TTS usage come straight from the framework’s usage metric. STT is derived from audio duration: on Pipecat, the observer accumulates each STT service’s AudioRawFrame bytes and converts 16-bit mono PCM to seconds.

Channel, session, and tenant

Channel ("telephony" / "web") auto-detects from the transport when omitted: a LiveKit SIP participant or a Pipecat Twilio/Telnyx/Plivo serializer means telephony; anything else means web. Session correlation is automatic and identical on both frameworks: the id is created (or reused) via Python contextvars the moment attach() runs. Every row from that session shares it, and a task spawned inside the same async context inherits it too, with no argument needed. Tenant attribution is opt-in: pass tenant_id= when one deployment serves several customers. See Tenant attribution.

policy: naming the set instead of spelling it out

The four capture flags below are the mechanism. policy= is the way to say what you actually want, because the interesting thing is the set and four independent booleans cannot name one.

vg.attach(session, policy="timing_only")   # nothing the caller said

Policy	transcript	snapshots	turns	dead_air
`standard` (default)	on	off	on	on
`timing_only`	off	off	on	on
`lean`	off	off	on	off
`debug`	on	on	on	on
`off`	off	off	off	off

The flags mix two unrelated questions, and the policies separate them:

What does it cost to run? Dead air polls once a second for the life of every session. Turn capture costs nothing between events. lean is timing_only without that standing cost.
What does it disclose? A transcript is what the caller said. A snapshot is your own system prompt and every tool payload. debug is the only policy that carries snapshots, and it is named for when that trade is worth making.

Tool-call rows ride turns: they correlate to a turn and their turn_index is meaningless without one. Both are timing-only and neither carries a payload. An explicit argument overrides the policy, so attach(session, policy="timing_only", dead_air=False) is lean. An unknown policy name is refused rather than defaulting, because a typo that silently selected standard would turn “nothing the caller said” into transcript capture.

The environment kill-switches still beat everything, including a policy. A fleet-wide override that a policy could cancel would not be a kill-switch.

room, heartbeat, transcript, snapshots, turns, dead_air (LiveKit)

Param	Behavior
`room`	LiveKit room name stamped on each row, so `voicegw livekit latency` can read the STT/LLM/TTS split back by room. Auto-resolved from the running job context.
`heartbeat=True`	Registers this process in the fleet roster and heartbeats its presence (the dashboard’s Fleet/Agents view). Best for single-process agents where `attach()` is the sole writer. In LiveKit’s per-call subprocess model (`agent dev`), call `register_worker(agent_id, local=True)` at your `__main__` boot instead, and skip `heartbeat=True` there, because the subprocess would become a second writer of the same roster row.
`transcript=True` (default)	On close, user/agent turns are read from the framework’s conversation history and written to local storage for the Calls page. `transcript=False` disables it per attach; `VOICEGW_TRANSCRIPTS=0` kills it fleet-wide (the env var wins over the argument). Pipecat accepts the flag but does not capture transcripts yet.
`turns=True` (default)	Captures one row per caller/agent exchange: turn index and four speech timestamps, from which `response_speed_ms` is derived. `VOICEGW_TURNS=0` kills it fleet-wide. Pipecat accepts the flag and captures nothing. See Turns and transcripts.
`dead_air=True` (default)	Runs a per-session watchdog that writes an event when neither party speaks for longer than the threshold. `VOICEGW_DEAD_AIR=0` kills it fleet-wide. A separate flag from `turns` because it polls once a second for the life of the call. See Dead air.
`snapshots=True`	Captures conversation-state snapshots for session replay. Off by default, the one place this differs from `transcript`. `VOICEGW_SNAPSHOTS=0` kills it fleet-wide and beats the argument. Pipecat accepts the flag and does not capture yet.

snapshots defaults off because it is a strictly larger disclosure than a transcript. A transcript is what the caller said. A snapshot carries your system prompt, the full message history, and every tool call’s arguments and result, so it captures your own prompt and whatever payloads your tools handle. That should be asked for, not assumed.

Snapshots need a local sink. When VOICEGW_COLLECTOR_URL is set, capture is skipped: a collector has no replay tables, and the dashboard reads replay from the local store, so capturing there would buffer rows nothing could flush.

Config: what actually gates writes

attach() always writes cost and latency; no config toggle turns it off. It resolves storage from environment variables only, never from voicegw.yaml: VOICEGW_DB_PATH for local SQLite (default ~/.config/voicegateway/voicegw.db), or VOICEGW_COLLECTOR_URL + VOICEGW_API_KEY for fleet mode. voicegw.yaml’s observability: block (latency_tracking, cost_tracking, request_logging, all default true) configures the Gateway behind voicegw serve / voicegw status / the dashboard: a separate reader-side process, not attach()’s own write path.

Instrument

Cost tracking

Conversations

Examples

Signature

Wiring

What it records

Channel, session, and tenant

policy: naming the set instead of spelling it out

room, heartbeat, transcript, snapshots, turns, dead_air (LiveKit)

Config: what actually gates writes

See also

​Signature

​Wiring

​What it records

​Channel, session, and tenant

​policy: naming the set instead of spelling it out

​room, heartbeat, transcript, snapshots, turns, dead_air (LiveKit)

​Config: what actually gates writes

​See also

Signature

Wiring

What it records

Channel, session, and tenant

policy: naming the set instead of spelling it out

room, heartbeat, transcript, snapshots, turns, dead_air (LiveKit)

Config: what actually gates writes

See also