Use PyAI with Pipecat

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.

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

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:

pip install pipecat-pyai==0.1.0

Keep any Pipecat transport, VAD, and LLM extras your application already installs. Then set the PyAI key:

export PYAI_API_KEY=pyai_test_...

Add PyAI to the pipeline

Put Hear after the transport input and Speak before the transport output:

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

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

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

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. For applications that already manage an aiohttp.ClientSession, pass it to the TTS service:

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

Start your Pipecat bot

Run the same transport and deployment command you use today.

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.

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.

Exercise failure handling

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

Pipecat documentation

Review Pipecat’s pipeline, transport, and deployment model.

PyAI Omni overview

Compare the managed speech-to-speech path.

Get started

Omni

Guides

Frameworks

Integrations

Use cases

Reference

What stays under your control

Prerequisites

Install

Add PyAI to the pipeline

Complete task shape

Configuration

Hear speech-to-text

Speak text-to-speech

Run and verify

Production notes

Troubleshooting

Which path should I choose?

Pipecat documentation

PyAI Omni overview

​What stays under your control

​Prerequisites

​Install

​Add PyAI to the pipeline

​Complete task shape

​Configuration

​Hear speech-to-text

​Speak text-to-speech

​Share a TTS HTTP session

​Run and verify

​Production notes

​Troubleshooting

​Which path should I choose?

Pipecat documentation

PyAI Omni overview

What stays under your control

Prerequisites

Install

Add PyAI to the pipeline

Complete task shape

Configuration

Hear speech-to-text

Speak text-to-speech

Share a TTS HTTP session

Run and verify

Production notes

Troubleshooting

Which path should I choose?