A Pipecat agent is built as a Pipeline of processors that frames of data move through in sequence. For a voice agent, that sequence is: microphone → transport → Munsit STT → LLM → Munsit TTS → transport → speaker. The caller speaks; the transport (this guide uses Daily’s WebRTC transport, the one documented in Munsit’s own Pipecat integration guide) captures the audio; Munsit’s STT service converts it to text; an LLM generates a reply; Munsit’s TTS service converts that reply to streaming Arabic audio PCM16 at 24 kHz and the transport plays it back. Pipecat’s TTSService base class aggregates LLM tokens into complete sentences before calling Munsit, so each sentence becomes one HTTP streaming request, with PCM16 chunks yielded back to the pipeline as they arrive.
Prerequisites
• Python 3.9 or later
• A Munsit account and an API key (generated from Munsit’s API Keys dashboard)
• A Daily account and API key, if you use Daily as your WebRTC transport (Pipecat supports other transports too)
• An LLM provider key (this guide uses OpenAI’s GPT-4o, but any Pipecat-supported LLM service will work)
Step 1: Install and Authenticate
pip install pipecat-plugins-munsit
Your Munsit API key is shown only once at creation save it securely as the MUNSIT_API_KEY environment variable rather than hardcoding it. The plugin’s TTS service also expects a shared aiohttp session rather than opening its own connection per request:
import aiohttp
from pipecat_plugins_faseeh import FaseehTTSService
async with aiohttp.ClientSession() as session:
tts = FaseehTTSService(
api_key="your-api-key",
aiohttp_session=session,
)
Keep MUNSIT_API_KEY (and DAILY_API_KEY, OPENAI_API_KEY) out of version control; load them from environment variables or a .env file via python-dotenv in production.
Step 2: Write the Agent
A complete Daily-transported agent auto-creates a Daily room, wires Munsit STT → GPT-4o → Munsit TTS, and greets the caller as soon as they join:
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_plugins_munsit import MunsitSTTService
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=48000,
),
)
stt = MunsitSTTService(api_key=os.getenv("MUNSIT_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())
Note the import pattern: the single pipecat-plugins-munsit package ships two importable modules pipecat_plugins_munsit for the STT service (MunsitSTTService) and pipecat_plugins_faseeh for the TTS service (FaseehTTSService). This is exactly as documented in Munsit’s own Pipecat integration guide at the time of writing; if you hit an ImportError on either name, confirm you’re importing from the correct module rather than assuming both classes live in one.
Run it like any Python script:
python agent.py
Since the example auto-creates a Daily room via DailyRESTHelper, the terminal output (or Daily’s dashboard) will give you the room URL to join and test from a browser or the Daily Prebuilt UI.
.webp)
.webp)
.webp)
%20for%20Arabic%20Conversational%20AI%20%20%20.png)
.svg)
.svg)