Sync STT on Pipecat

Overview

Pipecat ships two AssemblyAI speech-to-text services. This guide covers AssemblyAISyncSTTService, which transcribes one VAD-detected speech segment per HTTP request against the Sync API — no WebSocket to hold open, no session to manage. Your local VAD decides where a turn ends; when it does, the segment is POSTed and the finished transcript comes back in the same call.

Available on pipecat-ai 1.9.0+ — from pipecat.services.assemblyai.stt import AssemblyAISyncSTTService.

Choosing sync or streaming

The Sync service is a different service class against a different API — not a mode of the streaming one.

	`AssemblyAISyncSTTService`	`AssemblyAISTTService`
API	Sync — one request per turn	Real-time — persistent WebSocket
Turn detection	Your VAD only	Pipecat VAD or AssemblyAI’s built-in turn detection
Partial transcripts	None — one final transcript per turn	Interim frames throughout the turn
Turn length	120 s maximum	Unbounded
Conversation context	Automatic, client-side rolling buffer	Automatic, server-side carryover
Connection	Opened per turn (pre-warmed by default)	Held open for the session

Reach for the Sync service when your agent already relies on local VAD for turn-taking, when you want per-turn request/response semantics that are simple to retry, log, and reason about, or when you need per-word timestamps on each turn. Reach for the streaming service when you want AssemblyAI to detect end-of-turn for you, when you need interim transcripts to drive UI or speculative inference, or when turns can run longer than 120 seconds.

Pipecat AssemblyAI STT plugin

View Pipecat’s AssemblyAI STT plugin reference.

Streaming on Pipecat

Build the same agent on Universal 3.5 Pro Realtime over a WebSocket.

Quickstart

Install Pipecat

Install Pipecat with the AssemblyAI, LLM, and TTS extras you need:

pip install "pipecat-ai[assemblyai,openai,cartesia,silero]" python-dotenv

What’s included:

assemblyai: AssemblyAI STT services
openai: OpenAI LLM service (used in the example)
cartesia: Cartesia TTS service (used in the example)
silero: Silero VAD — required, since the Sync service segments audio from VAD events

The example uses OpenAI and Cartesia, but you can use any LLM or TTS supported by Pipecat — just swap the extras.

Set your API keys

Set your API keys in a .env file:

ASSEMBLYAI_API_KEY=<YOUR_API_KEY>
OPENAI_API_KEY=<YOUR_OPENAI_KEY>
CARTESIA_API_KEY=<YOUR_CARTESIA_KEY>

Build a minimal agent

Two things differ from a streaming agent: you create and own an aiohttp.ClientSession and pass it to the service, and the assistant aggregator at the end of the pipeline is what feeds the agent’s replies into conversation context.

import os

import aiohttp
from dotenv import load_dotenv
from loguru import logger

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker, ProcessorUnusablePolicy
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.assemblyai.stt import AssemblyAISyncSTTService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.workers.runner import WorkerRunner

load_dotenv(override=True)

transport_params = {
    "daily": lambda: DailyParams(audio_in_enabled=True, audio_out_enabled=True),
    "webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True),
}


async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    # Keep one session for the service: pre-warming only pays off when the warm
    # and transcribe requests share a connection pool.
    async with aiohttp.ClientSession() as session:
        stt = AssemblyAISyncSTTService(
            api_key=os.environ["ASSEMBLYAI_API_KEY"],
            aiohttp_session=session,
            settings=AssemblyAISyncSTTService.Settings(
                language=Language.EN,
                # prompt="Customer support call about order status.",
                # keyterms_prompt=["AssemblyAI", "Pipecat"],
                # timestamps=True,
            ),
            # enable_prewarming=True,   # Default — warms the connection on speech start
            # max_context_turns=5,      # Default — 0 disables automatic context
        )

        llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"])
        tts = CartesiaTTSService(api_key=os.environ["CARTESIA_API_KEY"])

        context = LLMContext()
        user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
            context,
            user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
        )

        pipeline = Pipeline(
            [
                transport.input(),     # Transport user input
                stt,                   # STT
                user_aggregator,       # User responses
                llm,                   # LLM
                tts,                   # TTS
                transport.output(),    # Transport bot output
                assistant_aggregator,  # Assistant responses → automatic conversation context
            ]
        )

        worker = PipelineWorker(
            pipeline,
            params=PipelineParams(enable_metrics=True),
            # A rejected API key leaves the service unusable; end the run instead
            # of continuing silently without transcription.
            processor_unusable_policy=ProcessorUnusablePolicy.END,
        )

        runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
        await runner.add_workers(worker)

        @transport.event_handler("on_client_connected")
        async def on_client_connected(transport, client):
            logger.info("Client connected")
            context.add_message(
                {"role": "system", "content": "You are a helpful voice assistant. Keep replies brief and speakable."}
            )
            await worker.queue_frames([LLMRunFrame()])

        @transport.event_handler("on_client_disconnected")
        async def on_client_disconnected(transport, client):
            logger.info("Client disconnected")
            await runner.cancel()

        await runner.run()


async def bot(runner_args: RunnerArguments):
    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport, runner_args)


if __name__ == "__main__":
    from pipecat.runner.run import main

    main()

The complete runnable example lives in the Pipecat repo: voice-assemblyai-sync.py.

Run and test

Run the agent directly with local audio:

python your_agent.py

Speak into your microphone after hearing the greeting. Because there are no interim transcripts, the first thing you see per turn is the finished transcript, logged once the segment comes back.

How each turn is transcribed

AssemblyAISyncSTTService extends Pipecat’s SegmentedSTTService, so the segmentation is handled by the base class and the AssemblyAI service only transcribes what it’s handed. Per turn:

VAD detects speech start. If pre-warming is enabled, the service fires a warm request in the background (see Connection pre-warming).
Audio buffers into the current segment. A short lead-in is retained, so the delay between actual speech start and VAD detection doesn’t clip the first word.
VAD detects speech end. The segment is closed, padded with a half-second of trailing silence so the model hears the end of speech and finishes the last word, and wrapped in a WAV container.
The segment is POSTed as multipart/form-data — the audio part plus a config part built from your settings — and the transcript returns in the response.
A TranscriptionFrame is pushed with the text, and the turn is appended to the conversation-context buffer for the next request.

Transcription runs off the audio path: segments are queued and transcribed in order by a background task while audio keeps flowing through the service. A graceful stop transcribes what’s queued; a cancel drops it.

Turns are capped at 120 seconds. The Sync API rejects longer audio with a 413. If your callers can monologue past two minutes, use the streaming service instead. See Audio requirements for the full constraints.

This service emits no InterimTranscriptionFrames — a turn produces exactly one final TranscriptionFrame, and empty transcripts are dropped rather than pushed. Anything in your pipeline that reacts to partials won’t fire.

Parameters reference

Constructor arguments

str

required

Your AssemblyAI API key.

aiohttp.ClientSession

required

The HTTP session used for both warm and transcribe requests. Pre-warming only helps when both share this session’s connection pool, so create one session and keep it for the life of the service.

str

default:"https://sync.assemblyai.com"

Base URL for the Sync API. Override for a data-residency endpoint — see Data residency.

int | None

default:"None"

Audio sample rate in Hz. Defaults to the pipeline’s rate.

bool

default:"True"

Open the connection when the user starts speaking so the transcription request skips the handshake. See Connection pre-warming.

int

default:"5"

How many prior turns — user transcripts and agent replies together — are carried as conversation_context on each request. Set to 0 to disable automatic context. Ignored when you set conversation_context yourself.

int

default:"1500"

Character budget for the same buffer. Oldest turns are evicted first once either cap is exceeded.

float

default:"0.65"

P99 latency from speech end to final transcript, in seconds, broadcast at pipeline start for downstream turn timing. Set it to your own measured value.

Settings

Set these inside AssemblyAISyncSTTService.Settings(...).

str

default:"universal-3-5-pro"

The speech model, sent as the X-AAI-Model header.

Language

default:"Language.EN"

The transcription language. Superseded by language_codes when both are set.

list[Language]

default:"None"

Declared audio languages for multilingual or code-switching audio, e.g. [Language.EN, Language.ES]. Regional variants resolve to their base code and duplicates are dropped, preserving declaration order. See Language selection.

str

default:"None"

A natural-language description of what the audio is about — the domain, the scenario, or details of the conversation. Maximum 6000 characters. See Contextual prompting.

list[str]

default:"None"

Key terms or phrases to bias the decoder toward. See Keyterms prompting.

str | list[str]

default:"None"

Prior turns, oldest first. Setting this turns off the service’s automatic context buffer and sends exactly this value. Leave it unset to let the service manage context. See Conversation context.

bool

default:"None"

Compute per-word start/end times, returned on the words of the result, at a small added latency. Unset means the API default (false) applies. See Word timestamps.

language and language_codes are ignored when prompt is set. If you use a custom prompt and need a non-English language, state the language as part of the prompt text — see Specifying the language.

Conversation context

The Sync API is stateless: each request transcribes one clip with no memory of the last. Conversation context is how you give the model the surrounding dialogue anyway, and the Pipecat service assembles it for you. It keeps a rolling buffer of the most recent turns — user transcripts and agent replies together, in the order spoken — and sends them as conversation_context on every request. Agent replies are captured from the pipeline’s assistant-turn frame, so this needs no wiring beyond having the standard context aggregator pair in your pipeline:

context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

pipeline = Pipeline([
    transport.input(),
    stt,
    user_aggregator,
    llm,
    tts,
    transport.output(),
    assistant_aggregator,  # ← without this, only user turns reach the context buffer
])

A turn never appears in its own request’s context: the config is built before the response is recorded.

Tuning the buffer

The buffer is bounded by both caps, and the oldest turns are evicted first when either is exceeded:

	Pipecat default	API maximum
Turns	`max_context_turns=5`	500
Characters	`max_context_chars=1500`	16000

The defaults are deliberately conservative — every carried turn is uploaded again on the next request. Raise them when your conversations hinge on detail established several turns back:

stt = AssemblyAISyncSTTService(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    aiohttp_session=session,
    max_context_turns=12,
    max_context_chars=4000,
)

Set max_context_turns=0 to turn automatic context off entirely.

Supplying context yourself

Setting conversation_context in Settings disables the automatic buffer and sends exactly your value — useful when your application already tracks the dialogue, or when you want to seed the model with context from before the call:

stt = AssemblyAISyncSTTService(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    aiohttp_session=session,
    settings=AssemblyAISyncSTTService.Settings(
        conversation_context=[
            "Hi, thank you for calling. How can I help?",
            "I'd like to check on the status of my order.",
        ],
    ),
)

This is a static value — the service won’t append to it. To change it mid-conversation, push an update; because every request rebuilds its config from the current settings, the new value applies to the next turn with no reconnect:

from pipecat.frames.frames import STTUpdateSettingsFrame

await worker.queue_frame(
    STTUpdateSettingsFrame(
        delta=AssemblyAISyncSTTService.Settings(conversation_context=[...]),
    )
)

Connection pre-warming

Because each turn is its own HTTP request, connection setup would otherwise sit in the latency budget of every turn. Pre-warming takes it off the critical path: the service sends a warm request the moment VAD reports speech start, so DNS, TCP, and TLS complete while the user is still talking, and the transcribe request that follows starts uploading immediately. This is on by default. Two things are worth knowing:

The session must be shared. The warmed connection lives in your aiohttp.ClientSession pool. Passing a different session — or letting one be created per request — forfeits the saving entirely.
Warming is best-effort. Failures are logged at debug level and swallowed, since a failed warm-up only costs you the latency saving, never the transcription.

To warm at some other moment — say, when a call connects, before anyone speaks:

await stt.warm()

Set enable_prewarming=False to disable the automatic warm on speech start. See Connection pre-warming for what the handshake actually costs.

Data residency

Point base_url at a regional endpoint to keep audio and transcripts inside a zone:

stt = AssemblyAISyncSTTService(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    aiohttp_session=session,
    base_url="https://sync.eu.assemblyai.com",  # or https://sync.us.assemblyai.com
)

The default (https://sync.assemblyai.com) routes to the nearest available region, which may be in the US or the EU. See Cloud endpoints & data residency.

Error handling

A failed request is logged and pushed downstream as an ErrorFrame rather than raised — the pipeline keeps running and that turn simply produces no transcript. The HTTP status rides on the underlying exception so Pipecat can classify the failure: a rejected key (401) marks the service unusable, while a rate limit (429) or a server error (5xx) does not. That distinction is what processor_unusable_policy acts on:

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    processor_unusable_policy=ProcessorUnusablePolicy.END,
)

ProcessorUnusablePolicy.END ends the run when a processor becomes unusable — better than an agent that keeps listening and never hears anything. See Error handling for the full status and error_code table.

Metrics

can_generate_metrics() returns True: each turn is a discrete request, so its duration is measured and reported through Pipecat’s usual metrics with enable_metrics=True. For downstream turn timing, the service broadcasts ttfs_p99_latency at pipeline start — 0.65 seconds by default. Measure your own P99 from speech end to final transcript and set it explicitly; the default is a general figure and your network distance to the endpoint moves it.

Streaming on Pipecat

The WebSocket service, with AssemblyAI’s built-in turn detection.

Sync STT quickstart

Use the Sync API directly, without Pipecat.

Prompting and keyterms

Improve accuracy with contextual prompts and key terms.

Audio requirements

Duration, size, format, and sample-rate constraints.

Getting started

Features

API reference

Advanced

Integrations

Overview

Choosing sync or streaming

Pipecat AssemblyAI STT plugin

Streaming on Pipecat

Quickstart

How each turn is transcribed

Parameters reference

Constructor arguments

Settings

Conversation context

Tuning the buffer

Supplying context yourself

Connection pre-warming

Data residency

Error handling

Metrics

Streaming on Pipecat

Sync STT quickstart

Prompting and keyterms

Audio requirements

​Overview

​Choosing sync or streaming

Pipecat AssemblyAI STT plugin

Streaming on Pipecat

​Quickstart

​How each turn is transcribed

​Parameters reference

​Constructor arguments

​Settings

​Conversation context

​Tuning the buffer

​Supplying context yourself

​Connection pre-warming

​Data residency

​Error handling

​Metrics

​Related

Streaming on Pipecat

Sync STT quickstart

Prompting and keyterms

Audio requirements

Overview

Choosing sync or streaming

Quickstart

How each turn is transcribed

Parameters reference

Constructor arguments

Settings

Conversation context

Tuning the buffer

Supplying context yourself

Connection pre-warming

Data residency

Error handling

Metrics

Related