> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pyai.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Use PyAI with Pipecat
> Keep Pipecat's explicit frame pipeline and custom transports while using PyAI Hear for streaming speech-to-text and PyAI Speak for agent audio.
Use **Pipecat** when you want an explicit Python pipeline where your
application controls transports, processors, context, the LLM, tools, and turn
policy. Add **PyAI Hear** and **PyAI Speak** as two services in that pipeline.
If you want PyAI to operate the complete conversation loop, use
[Omni](/guides/omni-overview).
The `pipecat-pyai` package is maintained by PyAI. It does not replace your
transport, LLM, context aggregator, or application processors.
## What stays under your control
```mermaid theme={null}
flowchart LR
caller[Caller] --> input[Pipecat transport input]
input --> hear[PyAI Hear STT]
hear --> processors[Your context and processors]
processors --> brain[Your LLM and tools]
brain --> speak[PyAI Speak TTS]
speak --> output[Pipecat transport output]
output --> caller
```
Pipecat still owns frame order and routing. You can place custom processors
before, between, or after the PyAI services and keep the deployment model you
already use.
## Prerequisites
* Python 3.11 or newer
* A Pipecat application with a transport and LLM service
* A PyAI key with `hear:stream` and `speak:synthesize` scopes
* A voice id from `GET https://api.pyai.com/v1/voices`
A `pyai_test_` sandbox key includes the required Hear and Speak scopes,
starts working immediately, and does not require billing. Keep the key on the
server that runs the Pipecat pipeline.
## Install
Install the verified public release from PyPI:
```bash theme={null}
pip install pipecat-pyai==0.1.2
```
Keep any Pipecat transport, VAD, and LLM extras your application already
installs. Then set the PyAI key:
```bash theme={null}
export PYAI_API_KEY=pyai_test_...
```
## Add PyAI to the pipeline
Put Hear after the transport input and Speak before the transport output:
```python theme={null}
from pipecat.pipeline.pipeline import Pipeline
from pipecat_pyai import PyAISTTService, PyAITTSService
stt = PyAISTTService()
tts = PyAITTSService(voice="stock_emma_en_gb")
pipeline = Pipeline([
transport.input(),
stt,
user_aggregator,
llm,
tts,
transport.output(),
assistant_aggregator,
])
```
The names `transport`, `user_aggregator`, `assistant_aggregator`, and `llm`
refer to the objects your Pipecat application already creates. Your VAD and
turn configuration remain in effect.
## Complete task shape
```python theme={null}
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
)
from pipecat_pyai import PyAISTTService, PyAITTSService
async def run(transport, llm):
context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline([
transport.input(),
PyAISTTService(),
user_aggregator,
llm,
PyAITTSService(voice="stock_emma_en_gb"),
transport.output(),
assistant_aggregator,
])
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
)
runner = PipelineRunner()
await runner.run(task)
```
The transport remains responsible for joining the room or call and queuing the
initial frames required by your application.
## Configuration
### Hear speech-to-text
```python theme={null}
PyAISTTService(
api_key=None, # defaults to PYAI_API_KEY
base_url="https://api.pyai.com",
model="pyai-hear",
language="", # automatic detection
sample_rate=16000,
)
```
The service opens the canonical PyAI Hear WebSocket, sends raw PCM audio frames,
sends a commit when Pipecat VAD reports the end of speech, and emits
`InterimTranscriptionFrame` plus finalized `TranscriptionFrame` objects. Hear
streaming detects the language automatically when you omit `language`.
To pin recognition, pass `en`, `es`, `fr`, `de`, `hi`, `it`, `pt`, or `nl`.
### Speak text-to-speech
```python theme={null}
PyAITTSService(
voice="stock_emma_en_gb",
api_key=None, # defaults to PYAI_API_KEY
base_url="https://api.pyai.com",
model="pyai-speak",
sample_rate=24000,
)
```
The service sends Pipecat text frames to Speak and yields raw mono
`TTSAudioRawFrame` objects at the configured rate. Use a streaming-capable
stock, cloned, or designed voice id available to the same PyAI organization.
### Share a TTS HTTP session
For applications that already manage an `aiohttp.ClientSession`, pass it to
the TTS service:
```python theme={null}
tts = PyAITTSService(
voice="stock_emma_en_gb",
aiohttp_session=session,
)
```
The TTS service will reuse that session and leave its lifecycle under your
application's control.
## Run and verify
Run the same transport and deployment command you use today.
Speak one short request, wait for agent audio, then interrupt or ask a
follow-up. This verifies transcript finalization, context flow, TTS, and
interruption behavior.
Confirm the Pipecat task receives transcript frames and TTS usage metrics.
Log the transport region, worker region, LLM, VAD settings, voice, and PyAI
region with any latency result.
Test an invalid key, missing scope, and exhausted concurrency so your
pipeline presents a safe fallback instead of silently stopping.
## Production notes
* Keep your Pipecat worker, transport media server, and PyAI speech endpoints in
nearby regions.
* A running Hear service counts against the PyAI key's realtime concurrency.
* Hear and Speak meter against your PyAI account; see the
[pricing page](https://pyai.com/pricing) for current terms. Pipecat Cloud,
transport, your LLM, and worker hosting remain separate.
* Keep a bounded queue and timeout around custom processors so one integration
cannot stall caller audio indefinitely.
* Use Omni instead when you want PyAI to own turn-taking, reasoning, tools,
caller continuity when an Agent profile has continuity enabled and a caller
key is available, and optional managed telephony as one system.
## Troubleshooting
| Symptom | Check |
| - | - |
| `ValueError: A PyAI API key is required` | Set `PYAI_API_KEY` on the worker or pass `api_key=`. |
| `403 forbidden` | Add `hear:stream` or `speak:synthesize` to the key. |
| No caller transcript frames | Confirm audio frames reach the STT processor and the transport rate is available on the frame. |
| Agent text appears but no audio plays | Confirm the voice id belongs to the key's organization and the TTS service precedes transport output. |
| New sessions receive `429` | Close idle Hear sockets or wait for realtime concurrency to free up. |
## Which path should I choose?
Choose **Pipecat + PyAI** for custom frame processors, transport flexibility,
branching, and per-stage observability. Choose **Omni** when one managed
speech-to-speech contract is more valuable than operating the pipeline.
Review Pipecat's pipeline, transport, and deployment model.
Compare the managed speech-to-speech path.
Streaming behavior, language support, and transcript formatting.
Voice selection, streaming modes, and PCM output.
Phone codecs, PCM rates, and resampling boundaries.
Enroll and use a consented custom voice.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.