Pipecat Plugin

Overview

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

Prerequisites

A Munsit AI account
Python 3.9 or higher

API Key

Go to Munsit - 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

Parameter	Type	Default	Description
`api_key`	str	env `MUNSIT_API_KEY`	Munsit API key
`voice_id`	str	`ar-hijazi-female-2`	Voice identifier
`model`	str	`faseeh-v1-preview`	TTS model
`stability`	float	0.5	Voice consistency (0.0-1.0)
`speed`	float	1.0	Speech rate (0.7-1.2)
`base_url`	str	`https://api.munsit.com/api/v1`	API base URL

Available Voice IDs

To choose a voice for your agent:

Visit the Munsit Voice Library
Listen to different voices to find the one that best fits your use case
Click “Copy Voice ID” button next to your chosen voice
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 → Munsit TTS → Transport → Speaker

User speaks into the microphone
Transport (e.g., Daily WebRTC) captures audio
STT (e.g., Deepgram) converts speech to text
LLM (e.g., OpenAI GPT-4o) generates a response
Munsit TTS converts the response to Arabic audio (streaming PCM16 at 24kHz)
Transport sends audio back to the user

Pipecat’s TTSService base class aggregates LLM tokens into complete sentences before calling Munsit. Each sentence triggers one HTTP streaming request to the Munsit 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,
            "Munsit 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("MUNSIT_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 Status	Meaning	Action
401	Invalid API key	Check `MUNSIT_API_KEY`
402	Insufficient balance	Add credits at app.munsit.com
404	Voice or model not found	Check `voice_id` and `model`
429	Rate limit exceeded	Reduce 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 MUNSIT_API_KEY is set and valid
Ensure voice_id exists in the Munsit 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.munsit.com
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

Quickstart

Text To Speech

Speech to Text

Capabilities

Integrations

Overview

Prerequisites

API Key

Installation

Quick Start

Configuration Reference

Available Voice IDs

Pipeline Architecture

Full Example

Runtime Configuration

Error Handling

Troubleshooting

​Overview

​Prerequisites

​API Key

​Installation

​Quick Start

​Configuration Reference

​Available Voice IDs

​Pipeline Architecture

​Full Example

​Runtime Configuration

​Error Handling

​Troubleshooting

Overview

Prerequisites

API Key

Installation

Quick Start

Configuration Reference

Available Voice IDs

Pipeline Architecture

Full Example

Runtime Configuration

Error Handling

Troubleshooting