> ## Documentation Index
> Fetch the complete documentation index at: https://assemblyai.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Sync STT on Pipecat
export const ModelBadges = ({models}) => {
return
{models.map(model =>
{model}
)}
;
};
## Overview
[Pipecat](https://docs.pipecat.ai/) 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](/docs/sync-stt/getting-started/quickstart) — 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.
```mermaid theme={null}
flowchart LR
U["User audio"] --> VAD["VAD
(Silero)"]
VAD -->|"one segment
per turn"| STT["AssemblyAI Sync STT
POST /transcribe"]
STT --> LLM["LLM"]
LLM --> TTS["TTS"]
TTS --> U
```
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](/docs/voice-agents/pipecat-universal-3-5-pro#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](/docs/sync-stt/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.
}
href="https://docs.pipecat.ai/server/services/stt/assemblyai"
>
View Pipecat's AssemblyAI STT plugin reference.
Build the same agent on Universal 3.5 Pro Realtime over a WebSocket.
## Quickstart
Install Pipecat with the AssemblyAI, LLM, and TTS extras you need:
```bash theme={null}
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 in a `.env` file:
```env theme={null}
ASSEMBLYAI_API_KEY=
OPENAI_API_KEY=
CARTESIA_API_KEY=
```
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](#conversation-context).
```python expandable theme={null}
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](https://github.com/pipecat-ai/pipecat/blob/main/examples/voice/voice-assemblyai-sync.py).
Run the agent directly with local audio:
```bash theme={null}
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:
1. **VAD detects speech start.** If pre-warming is enabled, the service fires a warm request in the background (see [Connection pre-warming](#connection-pre-warming)).
2. **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.
3. **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.
4. **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.
5. **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](/docs/voice-agents/pipecat-universal-3-5-pro) instead. See
[Audio requirements](/docs/sync-stt/audio-requirements) for the full constraints.
This service emits **no `InterimTranscriptionFrame`s** — 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
Your AssemblyAI API key.
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.
Base URL for the Sync API. Override for a data-residency endpoint — see
[Data residency](#data-residency).
Audio sample rate in Hz. Defaults to the pipeline's rate.
Open the connection when the user starts speaking so the transcription request
skips the handshake. See [Connection pre-warming](#connection-pre-warming).
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.
Character budget for the same buffer. Oldest turns are evicted first once either
cap is exceeded.
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(...)`.
The speech model, sent as the `X-AAI-Model` header.
The transcription language. Superseded by `language_codes` when both are set.
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](/docs/sync-stt/language-selection).
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](/docs/sync-stt/prompting-and-keyterms#contextual-prompting).
Key terms or phrases to bias the decoder toward. See
[Keyterms prompting](/docs/sync-stt/prompting-and-keyterms#keyterms-prompting).
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](#conversation-context).
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](/docs/sync-stt/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](/docs/sync-stt/prompting-and-keyterms#specifying-the-language).
## Conversation context
The Sync API is stateless: each request transcribes one clip with no memory of the last. [Conversation context](/docs/sync-stt/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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
await stt.warm()
```
Set `enable_prewarming=False` to disable the automatic warm on speech start. See [Connection pre-warming](/docs/sync-stt/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:
```python theme={null}
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](/docs/sync-stt/endpoints-and-data-zones).
## 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:
```python theme={null}
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](/docs/sync-stt/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.
## Related
The WebSocket service, with AssemblyAI's built-in turn detection.
Use the Sync API directly, without Pipecat.
Improve accuracy with contextual prompts and key terms.
Duration, size, format, and sample-rate constraints.