> ## 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).

<Note>
  The `pipecat-pyai` package is maintained by PyAI. It does not replace your
  transport, LLM, context aggregator, or application processors.
</Note>

## 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`

<Tip>
  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.
</Tip>

## Install

Install the verified public release from PyPI:

```bash theme={null}
pip install pipecat-pyai==0.1.0
```

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(language="en")
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(language="en"),
        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="en",
    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 is English-only.

### 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

<Steps>
  <Step title="Start your Pipecat bot">
    Run the same transport and deployment command you use today.
  </Step>

  <Step title="Send a two-turn conversation">
    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.
  </Step>

  <Step title="Inspect pipeline metrics">
    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.
  </Step>

  <Step title="Exercise failure handling">
    Test an invalid key, missing scope, and exhausted concurrency so your
    pipeline presents a safe fallback instead of silently stopping.
  </Step>
</Steps>

## 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. 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.

<CardGroup cols={2}>
  <Card title="Pipecat documentation" href="https://docs.pipecat.ai/overview/pipecat">Review Pipecat's pipeline, transport, and deployment model.</Card>
  <Card title="PyAI Omni overview" href="/guides/omni-overview">Compare the managed speech-to-speech path.</Card>
</CardGroup>