feat(observer): add optional roark_integration_id for call attribution

Adds a keyword-only `roark_integration_id` to `RoarkObserver`, defaulting from
the `ROARK_INTEGRATION_ID` env var (self-hosted) / `pcc secrets` (Pipecat
Cloud); the constructor arg wins when both are present. When set it ships as
`roarkIntegrationId` on both the `call-started` and `call-ended` webhook bodies
so app-roark-analytics can attribute each call to a specific Roark Pipecat
integration. Fully backward compatible — when unset the field is omitted and the
server falls back to the project's self-hosted integration. The observer stays
transport-agnostic; the id is the only new field.

Bumps version to 0.1.5 and documents ROARK_INTEGRATION_ID in the README setup /
running-modes sections, .env.example, and the example bot.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: munkhorgil

.env.example

@@ -6,6 +6,10 @@
 # This is the only Roark setting you need; the observer knows its own endpoints.
 ROARK_API_KEY=rk_live_replace_me
 
+# Roark integration id — copy it from the Roark dashboard when you create a
+# Pipecat integration. Optional; omit to fall back to the project's self-hosted integration.
+ROARK_INTEGRATION_ID=
+
 # ---------------------------------------------------------------------------
 # Services — only needed to run examples/bot.py (not by the library itself).
 # Swap providers freely in your own pipeline; RoarkObserver is provider-agnostic.

CHANGELOG.md

@@ -6,6 +6,16 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Added
+
+- **Optional `roark_integration_id`** to attribute each call to a specific Roark
+  Pipecat integration. Pass it to `RoarkObserver(roark_integration_id=...)` or
+  set the `ROARK_INTEGRATION_ID` environment variable (the constructor arg wins);
+  copy the id from the Roark dashboard when you create a Pipecat integration. When
+  set, it ships as `roarkIntegrationId` on both the `call-started` and
+  `call-ended` webhook bodies. Fully backward compatible — when unset the field is
+  omitted and Roark falls back to the project's self-hosted integration.
+
 ## [0.1.4] - 2026-05-28
 
 ### Added

README.md

@@ -45,10 +45,17 @@ Set one env var:
 
 ```bash
 ROARK_API_KEY=rk_live_...
+
+# Optional — copy from the Roark dashboard when you create a Pipecat integration.
+ROARK_INTEGRATION_ID=...
 ```
 
 > 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`.
+>
+> `ROARK_INTEGRATION_ID` is optional: set it to attribute calls to a specific
+> Roark Pipecat integration (also accepted as `roark_integration_id=`). Omit it
+> and Roark falls back to the project's self-hosted integration.
 
 ### 3. Wire the observer
 
@@ -130,15 +137,15 @@ and the same file runs in both modes — see `examples/bot.py`.
 |---|---|---|
 | Entry point | `python bot.py` → `pipecat.runner.run.main()` dispatches to `bot()` | Platform invokes `bot(runner_args)` per session |
 | Room/token | You provision (Daily REST, `pipecat.runner.daily.configure`, …) | Injected via `DailyRunnerArguments` |
-| Env vars | `.env` / your secrets manager | `pcc secrets set <name> KEY=value …` |
+| Env vars | `ROARK_API_KEY` (+ optional `ROARK_INTEGRATION_ID`) in `.env` / your secrets manager | same vars via `pcc secrets set <name> KEY=value …` |
 | Teardown | `EndFrame` is reliable | Sessions can vanish — wire [`aflush()` on disconnect](#handling-webrtc-disconnects) |
 | Observer wiring | ← identical → | ← identical → |
 
 ### Self-hosted
 
 ```bash
 cp .env.example .env
-# fill in ROARK_API_KEY
+# fill in ROARK_API_KEY (and optionally ROARK_INTEGRATION_ID)
 uv sync --all-extras
 uv run python examples/bot.py --transport daily   # or: --transport webrtc
 ```
@@ -149,7 +156,8 @@ Set the same vars as deployment secrets, then deploy:
 
 ```bash
 pcc secrets set roark-secrets \
-    ROARK_API_KEY=rk_live_...
+    ROARK_API_KEY=rk_live_... \
+    ROARK_INTEGRATION_ID=...        # optional — from the Roark dashboard
 
 pcc deploy
 pcc agent start <agent-name>

examples/bot.py

@@ -107,6 +107,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments) -> Non
         # Tie the Roark record to the platform-provided session id so you can
         # correlate Roark calls with Pipecat Cloud / Daily session logs.
         pipecat_call_id=getattr(runner_args, "session_id", None),
+        # Which Roark Pipecat integration this deployment belongs to (from the
+        # Roark dashboard). Optional — omit to fall back to the project's
+        # self-hosted integration.
+        roark_integration_id=os.environ.get("ROARK_INTEGRATION_ID"),
     )
 
     pipeline = Pipeline(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "pipecat-roark"
-version = "0.1.4"
+version = "0.1.5"
 description = "Roark analytics observer for Pipecat — capture call lifecycle, transcripts, tool calls, and recordings from any Pipecat pipeline."
 readme = "README.md"
 license = { file = "LICENSE" }

src/pipecat_roark/_types.py

@@ -56,6 +56,7 @@ class CallStartedPayload(TypedDict, total=False):
     agentId: str
     agentName: str
     agentPrompt: str
+    roarkIntegrationId: str  # Roark integration id; omitted falls back to self-hosted
 
 
 class CallEndedPayload(TypedDict, total=False):
@@ -78,6 +79,7 @@ class CallEndedPayload(TypedDict, total=False):
     recordingNumChannels: int
     transcript: list[TranscriptMessage]
     toolCalls: list[ToolCallMessage | ToolResultMessage]
+    roarkIntegrationId: str  # Roark integration id; omitted falls back to self-hosted
 
 
 class ChunkUploadUrlResponse(TypedDict, total=False):

src/pipecat_roark/observer.py

@@ -65,6 +65,7 @@
 import asyncio
 import json
 import logging
+import os
 import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Literal, cast
@@ -194,6 +195,7 @@ def __init__(
         agent_prompt: str | None = None,
         audio_buffer_processor: AudioBufferProcessor | None = None,
         pipecat_call_id: str | None = None,
+        roark_integration_id: str | None = None,
     ) -> None:
         """Construct a ``RoarkObserver`` for a single Pipecat call.
 
@@ -217,6 +219,12 @@ def __init__(
                 the same value to ``PipelineTask(conversation_id=...)`` when
                 OpenTelemetry tracing is enabled so each Roark call can be
                 looked up by ``conversation.id`` in your tracing backend.
+            roark_integration_id: The Roark integration this deployment
+                belongs to, copied from the Roark dashboard when you create a
+                Pipecat integration. Optional — defaults to the
+                ``ROARK_INTEGRATION_ID`` environment variable, and when neither
+                is set nothing is sent and Roark falls back to the project's
+                self-hosted integration.
         """
         super().__init__()
 
@@ -226,6 +234,7 @@ def __init__(
         self._agent_name = agent_name
         self._agent_prompt = agent_prompt
         self._pipecat_call_id = pipecat_call_id or str(uuid.uuid4())
+        self._roark_integration_id = roark_integration_id or os.environ.get("ROARK_INTEGRATION_ID")
 
         self._transcript: list[TranscriptMessage] = []
         self._tool_calls: list[ToolCallMessage | ToolResultMessage] = []
@@ -518,8 +527,15 @@ async def _post_call_started(self) -> None:
             payload["agentName"] = self._agent_name
         if self._agent_prompt:
             payload["agentPrompt"] = self._agent_prompt
+        if self._roark_integration_id:
+            payload["roarkIntegrationId"] = self._roark_integration_id
 
-        log.info("call-started: pipecatCallId=%s agentId=%s", self._pipecat_call_id, self._agent_id)
+        log.info(
+            "call-started: pipecatCallId=%s agentId=%s roarkIntegrationId=%s",
+            self._pipecat_call_id,
+            self._agent_id,
+            self._roark_integration_id,
+        )
         await self._client.post_call_started(payload)
 
     # ------------------------------------------------------------------ timing
@@ -631,14 +647,18 @@ async def _flush_call_ended(self, *, reason: str) -> None:
             )
         if self._tool_calls:
             payload["toolCalls"] = list(self._tool_calls)
+        if self._roark_integration_id:
+            payload["roarkIntegrationId"] = self._roark_integration_id
 
         log.info(
-            "call-ended: pipecatCallId=%s reason=%s transcript=%d toolCalls=%d chunks=%d",
+            "call-ended: pipecatCallId=%s reason=%s transcript=%d toolCalls=%d chunks=%d "
+            "roarkIntegrationId=%s",
             self._pipecat_call_id,
             reason,
             len(self._transcript),
             len(self._tool_calls),
             self._chunk_index,
+            self._roark_integration_id,
         )
         await self._client.post_call_ended(payload)
         await self._client.aclose()

tests/test_observer.py

@@ -152,6 +152,71 @@ async def test_on_pipeline_started_posts_call_started_with_required_fields() ->
     assert "eventTimestamp" in payload
 
 
+@pytest.mark.asyncio
+async def test_roark_integration_id_from_constructor_on_both_payloads() -> None:
+    obs = RoarkObserver(
+        api_key="rk_test", agent_id="agent-1", roark_integration_id="int_abc123"
+    )
+    fake = _FakeClient()
+    obs._client = fake  # type: ignore[assignment]
+
+    await obs.on_pipeline_started()
+    await obs.on_push_frame(_push(EndFrame()))
+
+    assert fake.started[0]["roarkIntegrationId"] == "int_abc123"
+    assert fake.ended[0]["roarkIntegrationId"] == "int_abc123"
+
+
+@pytest.mark.asyncio
+async def test_roark_integration_id_from_env_on_both_payloads(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ROARK_INTEGRATION_ID", "int_from_env")
+    obs = RoarkObserver(api_key="rk_test", agent_id="agent-1")
+    fake = _FakeClient()
+    obs._client = fake  # type: ignore[assignment]
+
+    await obs.on_pipeline_started()
+    await obs.on_push_frame(_push(EndFrame()))
+
+    assert fake.started[0]["roarkIntegrationId"] == "int_from_env"
+    assert fake.ended[0]["roarkIntegrationId"] == "int_from_env"
+
+
+@pytest.mark.asyncio
+async def test_constructor_roark_integration_id_overrides_env(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ROARK_INTEGRATION_ID", "int_from_env")
+    obs = RoarkObserver(
+        api_key="rk_test", agent_id="agent-1", roark_integration_id="int_explicit"
+    )
+    fake = _FakeClient()
+    obs._client = fake  # type: ignore[assignment]
+
+    await obs.on_pipeline_started()
+    await obs.on_push_frame(_push(EndFrame()))
+
+    assert fake.started[0]["roarkIntegrationId"] == "int_explicit"
+    assert fake.ended[0]["roarkIntegrationId"] == "int_explicit"
+
+
+@pytest.mark.asyncio
+async def test_roark_integration_id_absent_when_unset(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("ROARK_INTEGRATION_ID", raising=False)
+    obs = RoarkObserver(api_key="rk_test", agent_id="agent-1")
+    fake = _FakeClient()
+    obs._client = fake  # type: ignore[assignment]
+
+    await obs.on_pipeline_started()
+    await obs.on_push_frame(_push(EndFrame()))
+
+    assert "roarkIntegrationId" not in fake.started[0]
+    assert "roarkIntegrationId" not in fake.ended[0]
+
+
 @pytest.mark.asyncio
 async def test_multiple_on_pipeline_started_calls_only_post_once() -> None:
     obs = RoarkObserver(api_key="rk_test", agent_id="agent-1")