pipecat-roark

A Roark analytics observer for Pipecat. Drop one observer into your pipeline — Roark captures call lifecycle, transcripts, tool calls, and a stereo audio recording. No other code changes required.

• Tested with pipecat-ai 0.0.108 (compatible with >= 0.0.104, < 1)
• Python 3.10+
• Runtime-agnostic — same code runs self-hosted and on Pipecat Cloud

Maintained by Roark. File issues at https://github.com/roarkhq/sdk-roark-analytics-python-pipecat/issues.

Contents

• Quick start
• How it works
• Running modes
• Examples
• Advanced
  • Call identity
  • Bring your own AudioBufferProcessor
  • Handling WebRTC disconnects
  • Correlating with OpenTelemetry tracing
• Troubleshooting
• Configuration reference
• Development
• License

Quick start

1. Install

pip install pipecat-roark

2. Create a Pipecat integration & API key

Every Roark API key is bound to a specific integration — the key only works for the integration it was created under. Before you can send calls from this package, create the integration first:

1. In the Roark dashboard, go to Integrations and create a new Pipecat integration.
2. Open that integration and generate an API key for it.
3. Copy the key (represented as rk_live_replace_me below) — this is the value you'll set as ROARK_API_KEY below.

Use the key created under the Pipecat integration. A key from a different integration (or an account-level key not bound to one) will be rejected.

3. Configure

Set one env var:

ROARK_API_KEY=rk_live_replace_me

The Roark API key is all you configure — the observer knows its own service endpoints. ROARK_API_KEY can also be passed as api_key= to RoarkObserver.

4. Wire the observer

Inside your Pipecat bot(runner_args) entry point, drop RoarkObserver into the pipeline's observers=[...] list. Splice the auto-created roark.audio_processor after transport.output() so it sees the bot's audio post-TTS:

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat_roark import RoarkObserver

roark = RoarkObserver(
    api_key="rk_live_replace_me",
    agent_id="support-bot-v3",
    runner_args=runner_args,
    agent_name="Support Bot v3",
    agent_prompt=SYSTEM_PROMPT,
)

pipeline = Pipeline([
    transport.input(), stt, context_aggregator.user(), llm, tts,
    transport.output(),
    roark.audio_processor,          # after transport.output() — L=user, R=bot
    context_aggregator.assistant(),
])

task = PipelineTask(pipeline, params=PipelineParams(observers=[roark]))

That's it — transcripts, tool calls, and the stereo recording flow to Roark automatically.

How it works

The observer subscribes to Pipecat frames and ships a compact event timeline to Roark:

Transcripts and tool calls are forwarded in Pipecat's native shape — Roark maps them to its internal schema on its side.

Audio capture defaults

The observer always creates a sane-default AudioBufferProcessor (stereo, ~256 KB chunks) exposed as roark.audio_processor. The sample rate is adopted from the pipeline's StartFrame, so it tracks whatever the transport/provider negotiated — 8 kHz on Twilio/Telnyx, 16/24/48 kHz on Daily/LiveKit, etc. The rate is forwarded to Roark as the recording sample rate.

Failure mode

Failures are logged and swallowed — the observer never raises into the pipeline. Your call keeps running even if Roark is unreachable.

Running modes

RoarkObserver is runtime-agnostic — the same observer wiring works whether your Pipecat agent runs as a self-hosted process or is deployed to Pipecat Cloud. Write one bot(runner_args) entry point with Pipecat's create_transport helper, and the same file runs in both modes — see examples/bot.py.

Both modes use the same ROARK_API_KEY — the key created under your Pipecat integration (see Create a Pipecat integration & API key). Only where the key is stored differs: a local .env / secrets manager self-hosted, deployment secrets on Pipecat Cloud.

Self-hosted

cp .env.example .env
# fill in ROARK_API_KEY
uv sync --all-extras
uv run python examples/bot.py --transport daily   # or: --transport webrtc

Pipecat Cloud

Set the same vars as deployment secrets, then deploy:

pcc secrets set roark-secrets \
    ROARK_API_KEY=rk_live_replace_me

pcc deploy
pcc agent start <agent-name>

Reference the secrets from your pcc-deploy.toml so the container sees them as os.environ["ROARK_API_KEY"] (etc.) at runtime.

Examples

Two example files ship with the package:

• examples/basic_observer.py — minimal transport-agnostic wiring sketch. Shows where RoarkObserver and roark.audio_processor slot into a Pipeline / PipelineTask. STT / LLM / TTS stages are omitted — copy them into your own pipeline.
• examples/bot.py — runnable foundational voice assistant (Deepgram STT → OpenAI LLM → Cartesia TTS) with RoarkObserver wired in. Same file runs self-hosted (--transport webrtc / --transport daily) and deploys to Pipecat Cloud unchanged.

cp .env.example .env
# fill in:
#   ROARK_API_KEY
#   DEEPGRAM_API_KEY, OPENAI_API_KEY, CARTESIA_API_KEY

uv sync --all-extras
uv pip install "pipecat-ai[silero,deepgram,openai,cartesia,webrtc,daily]"

# Local browser via Pipecat's built-in WebRTC (no third-party transport account):
uv run python examples/bot.py --transport webrtc
# Or Daily (see Pipecat runner docs for transport-specific setup):
uv run python examples/bot.py --transport daily

Advanced

Call identity

Pass the runner_args received by your Pipecat bot entry point directly to the observer. It resolves one call ID internally, in this order:

1. SmallWebRTC runner_args.webrtc_connection.pc_id.
2. Pipecat Cloud runner_args.session_id.
3. A random UUID, generated once for this observer.

The native SmallWebRTC and Pipecat Cloud identifiers let Roark correlate both sides of a supported simulation automatically. Other runner and transport types still ingest normally, but their UUID fallback is local to the observer and does not provide native-ID simulation merging.

The resolved value is available through the read-only observer.pipecat_call_id property. Applications do not need to generate or forward an identifier themselves.

Bring your own AudioBufferProcessor

If you need to tune sample rate, channel count, or buffer size, instantiate AudioBufferProcessor yourself and pass it via audio_buffer_processor=:

from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor

audio_buffer = AudioBufferProcessor(sample_rate=16000, num_channels=1, buffer_size=128 * 1024)

pipeline = Pipeline([..., transport.output(), audio_buffer, ...])

RoarkObserver(
    api_key="rk_live_replace_me",
    agent_id="support-bot-v3",
    runner_args=runner_args,
    audio_buffer_processor=audio_buffer,
)

Handling WebRTC disconnects

Pipecat's WebRTC transports (notably SmallWebRTC) sometimes tear down without pushing EndFrame through observers. Call aflush() from the disconnect handler to guarantee the call is finalized on Roark:

@transport.event_handler("on_client_disconnected")
async def _on_disconnect(_, __):
    await roark_observer.aflush(reason="client-disconnected")

aflush() is idempotent — the regular EndFrame path will no-op on the next call.

Correlating with OpenTelemetry tracing

If you also enable Pipecat's OpenTelemetry tracing (PipelineTask(enable_tracing=True)), use the observer's resolved, read-only call ID as PipelineTask.conversation_id. Each Roark call can then be looked up by the same value in your tracing backend:

from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat_roark import RoarkObserver

roark = RoarkObserver(
    api_key="rk_live_replace_me",
    agent_id="support-bot-v3",
    runner_args=runner_args,
)

task = PipelineTask(
    pipeline,
    params=PipelineParams(observers=[roark]),
    enable_tracing=True,
    conversation_id=roark.pipecat_call_id,
)

Pipecat sets conversation.id as a span attribute on a root "conversation" span (and propagates it to every child span). The OTel traceId itself is auto-generated and unrelated to your call ID; correlation happens by attribute value. To find the trace for a Roark call, query your backend by conversation.id = <pipecatCallId> (e.g., Honeycomb: where conversation.id = "...", Jaeger: tag filter, Datadog: @conversation.id:...).

Because roark.pipecat_call_id is read-only, the lifecycle payload and tracing attribute cannot drift after observer construction.

Troubleshooting

Configuration reference

observer.pipecat_call_id is the read-only resolved call ID. Use it as PipelineTask(conversation_id=...) when OpenTelemetry tracing is enabled.

Development

uv sync --all-extras
uv run pytest
uv run ruff check .

License

MIT — see LICENSE.