Skip to main content

​
Overview

Pipecat is an open-source Python framework for building real-time voice and multimodal conversational AI agents. The pipecat-plugins-faseeh package bridges Faseeh TTS into Pipecat pipelines, letting you build Arabic voice agents with low-latency streaming audio.

​
Prerequisites

​
API Key

Go to Faseeh - API Keys, Generate an API key, then save it securely.
Your API key is only shown once. Save it securely.

​
Installation

pip install pipecat-plugins-faseeh

​
Quick Start

import aiohttp
from pipecat_plugins_faseeh import FaseehTTSService

async with aiohttp.ClientSession() as session:
    tts = FaseehTTSService(
        api_key="your-api-key",
        aiohttp_session=session,
    )

​
Configuration Reference

ParameterTypeDefaultDescription
api_keystrenv FASEEH_API_KEYFaseeh API key
voice_idstrar-hijazi-female-2Voice identifier
modelstrfaseeh-v1-previewTTS model
stabilityfloat0.5Voice consistency (0.0-1.0)
speedfloat1.0Speech rate (0.7-1.2)
base_urlstrhttps://api.faseeh.ai/api/v1API base URL

​
Available Voice IDs

To choose a voice for your agent:
  1. Visit the Faseeh Voice Library
  2. Listen to different voices to find the one that best fits your use case
  3. Click “Copy Voice ID” button next to your chosen voice
  4. Use that voice ID in your code:
tts = FaseehTTSService(
    api_key="your-api-key",
    voice_id="ar-uae-male-1",  # Paste the copied voice ID here
    aiohttp_session=session,
)

​
Pipeline Architecture

In a typical Pipecat pipeline, data flows through processors in sequence: 
Microphone → Transport → STT → LLM → Faseeh TTS → Transport → Speaker
  1. User speaks into the microphone
  2. Transport (e.g., Daily WebRTC) captures audio
  3. STT (e.g., Deepgram) converts speech to text
  4. LLM (e.g., OpenAI GPT-4o) generates a response
  5. Faseeh TTS converts the response to Arabic audio (streaming PCM16 at 24kHz)
  6. Transport sends audio back to the user
Pipecat’s TTSService base class aggregates LLM tokens into complete sentences before calling Faseeh. Each sentence triggers one HTTP streaming request to the Faseeh API, and PCM16 audio chunks are yielded back to the pipeline as they arrive.

​
Full Example

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import EndFrame, LLMMessagesUpdateFrame
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,
    LLMUserAggregatorParams,
)
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.daily.transport import DailyParams, DailyTransport

from pipecat_plugins_faseeh import FaseehTTSService

load_dotenv(override=True)


async def main():
    async with aiohttp.ClientSession() as session:
        # Auto-create a Daily room
        from pipecat.transports.daily.utils import DailyRESTHelper, DailyRoomParams

        daily_helper = DailyRESTHelper(
            daily_api_key=os.getenv("DAILY_API_KEY", ""),
            aiohttp_session=session,
        )
        room = await daily_helper.create_room(DailyRoomParams())
        token = await daily_helper.get_token(room.url)

        transport = DailyTransport(
            room.url,
            token,
            "Faseeh Bot",
            DailyParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                audio_out_sample_rate=24000,
            ),
        )

        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY", ""))
        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY", ""), model="gpt-4o")
        tts = FaseehTTSService(
            api_key=os.getenv("FASEEH_API_KEY"),
            aiohttp_session=session,
        )

        messages = [
            {
                "role": "system",
                "content": "You are a helpful Arabic-speaking assistant. Respond in Arabic.",
            }
        ]
        context = LLMContext(messages=messages)
        context_aggregator = LLMContextAggregatorPair(
            context,
            user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
        )

        pipeline = Pipeline([
            transport.input(),
            stt,
            context_aggregator.user(),
            llm,
            tts,
            transport.output(),
            context_aggregator.assistant(),
        ])

        task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))

        @transport.event_handler("on_first_participant_joined")
        async def on_joined(transport, participant):
            await task.queue_frames([LLMMessagesUpdateFrame(messages, run_llm=True)])

        @transport.event_handler("on_participant_left")
        async def on_left(transport, participant, reason):
            await task.queue_frame(EndFrame())

        runner = PipelineRunner()
        await runner.run(task)


if __name__ == "__main__":
    asyncio.run(main())

​
Runtime Configuration

Change voice, model, speed, or stability during an active conversation: 
from pipecat.frames.frames import TTSUpdateSettingsFrame

# Switch voice
await task.queue_frame(TTSUpdateSettingsFrame(settings={"voice_id": "ar-emirati-male-1"}))

# Adjust speed and stability
await task.queue_frame(TTSUpdateSettingsFrame(settings={"speed": 1.1, "stability": 0.8}))

# Switch model
await task.queue_frame(TTSUpdateSettingsFrame(settings={"model": "faseeh-v2"}))

​
Error Handling

The plugin yields non-fatal ErrorFrame objects instead of raising exceptions. This means your pipeline continues running even if a single TTS request fails. Common errors:
HTTP StatusMeaningAction
401Invalid API keyCheck FASEEH_API_KEY
402Insufficient balanceAdd credits at app.faseeh.ai
404Voice or model not foundCheck voice_id and model
429Rate limit exceededReduce request frequency
Handle errors in your application: 
from pipecat.frames.frames import ErrorFrame

@task.event_handler("on_error")
async def on_error(task, error: ErrorFrame):
    logger.error(f"TTS error: {error.error}")

​
Troubleshooting

No audio output?
  • Verify FASEEH_API_KEY is set and valid
  • Ensure voice_id exists in the Faseeh voice library
  • Check that the pipeline sample rate matches (24000 Hz default)
High latency?
  • The plugin uses HTTP streaming by default for lowest latency
  • Check network connectivity to api.faseeh.ai
  • Consider using a voice with faster generation characteristics
Import errors?
  • Ensure both pipecat-ai and pipecat-plugins-faseeh are installed
  • Minimum Pipecat version: 0.0.100