saa-pipecat-client

Pipecat / Daily client for SAA.

Tells your voice agent which speech is actually for it. SAA is the addressee layer: one decision per utterance about whether speech was meant for your agent, before STT, LLM, or TTS. The mental model is simple: audio in -> the SAA addressee gate -> only addressed audio out. No wake word, model-agnostic, drop-in for any Pipecat pipeline.

It adds attention-aware gating, barge-in, and proactive interjection to any Pipecat voice agent running on Daily, including bots deployed to Daily Bots and Pipecat Cloud.

The attention model runs on attention labs' service, so this is a thin Apache-2.0 client: it mints a Daily meeting token, starts a session, and listens for typed events. All inference runs server-side.

Install

pip install saa-pipecat-client

Requirements: Python 3.11+ (pipecat-ai 1.x dropped 3.10). macOS / Linux / WSL2 only

Quickstart: existing Pipecat bot

import os, asyncio
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.transports.daily.transport import DailyTransport, DailyParams

from saa_pipecat_client import (
    AttentionEngine, attention_agent_token, start_attention_session,
)


async def main() -> None:
    # 1. Mint a hidden-bot Daily meeting token using YOUR Daily API key.
    #    We never see it.
    agent_token = attention_agent_token(
        daily_api_key=os.environ["DAILY_API_KEY"],
        room_name="sess-xyz",
    )

    # 2. Summon the saa hosted bot into the room.
    session = await start_attention_session(
        api_key=os.environ["SAA_API_KEY"],
        room_url="https://your-org.daily.co/sess-xyz",
        agent_token=agent_token,
        participant_identity="user-omar",
        attention_config={"frames_per_turn": 3, "vad_threshold": 0.5},
    )

    # 3. Stand up your Pipecat pipeline, unchanged from your existing setup.
    transport = DailyTransport(
        "https://your-org.daily.co/sess-xyz",
        your_user_token,
        "Voice Agent",
        DailyParams(
            audio_in_enabled=True,
            video_in_enabled=True,
            audio_in_sample_rate=16000,
            audio_out_sample_rate=16000,
        ),
    )

    # 4. Attach the engine. Pass the PipelineTask so upstream actions
    #    (mute, set_threshold, ...) can be queued back to the SAA agent.
    engine = AttentionEngine(transport, agent_identity=session.agent_identity)

    @engine.on_prediction
    def _(p):
        # Gate your STT, only class 2 (talking-to-device) gets through.
        your_llm_gate.set_enabled(p.aligned_class == 2)

    @engine.on_interrupt
    async def _(ev):
        await your_tts.cancel()
        await engine.responding_stop()

    @engine.on_interjection
    async def _(ev):
        await your_tts.say("Want me to help with something?")

    pipeline = Pipeline([transport.input(), stt, llm, tts, transport.output()])
    task = PipelineTask(pipeline)
    engine.bind_task(task)
    await engine.start()

    runner = PipelineRunner()
    try:
        await runner.run(task)
    finally:
        await engine.stop()
        await session.stop()


if __name__ == "__main__":
    asyncio.run(main())

That's the full integration. Works with any Pipecat pipeline.

Greenfield: build_attention_runner

For new voice agents:

from saa_pipecat_client import build_attention_runner, TurnReadyEvent

async def handle_turn(event: TurnReadyEvent, transport):
    response_pcm = await my_llm.respond(event.audio_pcm16, frames=event.frames)
    await publish_response_audio(transport, response_pcm)

run = build_attention_runner(on_turn=handle_turn)
# pass `transport` and `task` you built; the factory mints the token,
# starts the session, and wires the engine.
engine, session = await run(room_url, room_name, human_identity, transport, task)

runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    await engine.stop()
    await session.stop()

Environment: SAA_API_KEY, DAILY_API_KEY.

Event types

Classes: 0=silent, 1=human-to-human, 2=human-to-device. responding is True while the AI is mid-playback.

Each is delivered through an @engine.on_* callback: on_prediction, on_vad, on_warmup, on_listening_start, on_listening_cancelled, on_turn_ready, on_interrupt, on_interjection, on_error, on_utterance_ended, on_utterance_config.

Upstream actions

await engine.mute()                  # stop feeding mic into the hosted processor
await engine.unmute()
await engine.responding_start()      # AI is now speaking
await engine.responding_stop()
await engine.set_threshold(0.65)     # model class-2 confidence threshold
await engine.add_assistant_turn(text)     # utterance handling: what the assistant said, as spoken
await engine.set_utterance_threshold(0.97) # utterance handling: one-sided class-1 decision threshold
await engine.clear_utterance_history()    # utterance handling: forget the dialogue history

Each call constructs a DailyOutputTransportMessageUrgentFrame addressed to the SAA agent's participant_id and queues it onto the bound PipelineTask. Pipecat's DailyTransport does not expose a public send_app_message(); the frame-queue path is the only supported send mechanism. Calls issued before the bot has joined are buffered and flushed once its participant id resolves.

Data plane

JSON envelopes on the Daily app-message topic "saa", same shapes as saa-livekit-client, so a single consumer-side event handler can serve both transports:

Binary turn payload (PCM + JPEGs) uses the same layout as saa-livekit-client, see _wire.py. Chunk reassembly is handled inside AttentionEngine; consumers see typed events only.

Utterance handling

Opt in with start_attention_session(..., utterance_handling=True). The hosted agent then segments speech into utterances on voice activity, transcribes each one when it ends, and scores whether it was aimed at the device (prediction == 2) or at a person (prediction == 1). One UtteranceEndedEvent per utterance carries the text, the verdict and the utterance audio; UtteranceConfigEvent arrives after started and says whether the feature is on for this session.

The classifier is conditioned on the preceding dialogue and is unreliable without it, so feed back every assistant response, as spoken, with await engine.add_assistant_turn(text). assistant_turns on the event shows how many assistant lines were in the scored history; 0 means you have not fed any.

session = await start_attention_session(..., utterance_handling=True)
engine = AttentionEngine(transport, agent_identity=session.agent_identity)

@engine.on_utterance_ended
def _(u):
    print(u.text, u.prediction, u.confidence, u.decision, u.preview)

# after each assistant response has been spoken
await engine.add_assistant_turn(assistant_transcript)

This stream is independent of TurnReadyEvent: the two will not line up one to one. Drive your LLM from one or the other. While the feature is in preview, preview is True and the verdict is preview grade; the server delivers transcripts and does not retain them.

Daily Bots compatibility

saa-pipecat-client is a pure pip dependency, so any Pipecat pipeline that runs locally also runs on Daily Bots and Pipecat Cloud. No extra deployment knobs.

Requirements

• Python 3.11+ (pipecat-ai 1.x dropped 3.10 support)
• pipecat-ai[daily] >= 1.0.0
• daily-python >= 0.19.0
• macOS / Linux / WSL2 only, daily-python ships no Windows wheels.
• Daily room URL must be publicly reachable from our cloud (no private VPC)
• Audio + video tracks must both be available (the model is multimodal)
• Customer voice agent and hosted attention bot share the same Daily room

Docs

• Integration guide: attentionlabs.ai/docs/integrations/pipecat
• Examples: examples/pipecat/

License

Apache-2.0. © Socero Inc.