Pipecat Voice Agents with Inworld TTS: The Complete Integration Guide

pip install "pipecat-ai[deepgram,silero,webrtc,runner]"

# Inworld API key from platform.inworld.ai (used as Basic auth by Pipecat)
export INWORLD_API_KEY="your-key"
export DEEPGRAM_API_KEY="your-key"   # or any STT provider Pipecat supports

from pipecat.services.inworld.tts import InworldTTSService

tts = InworldTTSService(
    api_key=os.environ["INWORLD_API_KEY"],
    settings=InworldTTSService.Settings(
        voice="Ashley",             # any library voice ID or your cloned voice
        model="inworld-tts-2",      # default; also inworld-tts-1.5-max / -mini
        temperature=1.1,             # expressiveness steering
        delivery_mode="BALANCED",   # STABLE | BALANCED | CREATIVE
    ),
)

# bot.py - minimal Pipecat voice agent with Inworld TTS
# Tested against Pipecat 1.5.0 (released July 4, 2026)

import os

from dotenv import load_dotenv

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
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.deepgram.stt import DeepgramSTTService
from pipecat.services.inworld.tts import InworldTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.workers.runner import WorkerRunner

load_dotenv(override=True)

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


async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])

    tts = InworldTTSService(
        api_key=os.environ["INWORLD_API_KEY"],
        settings=InworldTTSService.Settings(
            voice="Ashley",
            model="inworld-tts-2",
            temperature=1.1,
        ),
    )

    # Inworld's LLM Router is OpenAI-compatible: same key as TTS,
    # 220+ models served at cost.
    llm = OpenAILLMService(
        api_key=os.environ["INWORLD_API_KEY"],
        base_url="https://api.inworld.ai/v1",
        settings=OpenAILLMService.Settings(
            model="openai/gpt-5.5",
            system_instruction=(
                "You are a helpful voice assistant. Your replies are spoken"
                " aloud, so keep them short and avoid symbols and lists."
            ),
        ),
    )

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

    pipeline = Pipeline(
        [
            transport.input(),      # audio in from the browser
            stt,                    # speech to text
            user_aggregator,        # user turns + VAD
            llm,                    # streamed LLM tokens
            tts,                    # Inworld TTS, streamed audio
            transport.output(),     # audio out to the browser
            assistant_aggregator,   # assistant history
        ]
    )

    worker = PipelineWorker(pipeline, params=PipelineParams(enable_metrics=True))

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        context.add_message({"role": "developer", "content": "Say a short hello."})
        await worker.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await worker.cancel()

    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)
    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()

import base64, json, os, time, urllib.request

body = {
    "text": "Hi there! I'm a Pipecat voice agent speaking with an Inworld voice.",
    "voiceId": "Ashley",
    "modelId": "inworld-tts-2",
    "audioConfig": {"audioEncoding": "LINEAR16", "sampleRateHertz": 24000},
}
req = urllib.request.Request(
    "https://api.inworld.ai/tts/v1/voice:stream",
    data=json.dumps(body).encode(),
    headers={
        "Authorization": f"Basic {os.environ['INWORLD_API_KEY']}",
        "Content-Type": "application/json",
    },
)
t0 = time.time()
first = None
with urllib.request.urlopen(req) as resp:
    for line in resp:
        if not line.strip():
            continue
        chunk = json.loads(line)
        audio = chunk.get("result", {}).get("audioContent")
        if audio and first is None:
            first = time.time() - t0
            print(f"first audio chunk after {first * 1000:.0f} ms")
        # base64.b64decode(audio) -> audio bytes for the pipeline

first audio chunk after 395 ms