# Pipecat integration This guide shows how to use [Pipecat](https://www.pipecat.ai/) as the LLM pipeline inside a Speech Engine brain server. Speech Engine handles the voice loop — speech-to-text, turn-taking, and text-to-speech — while Pipecat handles text generation through a composable pipeline of processors (LLM calls, RAG, function calls, guardrails, content filters). :::callout{intent="note"} This guide is Python only because Pipecat is a Python framework on the server side. There is no Node equivalent for the pipeline processors; a `pipecat-client-js` package exists, but it is a browser client that talks to a Pipecat server, not a way to build pipelines in TypeScript. ::: ## Architecture The Speech Engine SDK runs as the outer layer — its `on_transcript` callback fires every time the user finishes speaking. Inside the callback, you build a Pipecat pipeline, feed the conversation history in as an `LLMContextFrame`, and stream the pipeline's text output back to Speech Engine. ElevenLabs converts the text to speech and plays it to the user. ```mermaid sequenceDiagram participant Browser participant EL as ElevenLabs participant Brain as Brain Server (engine.serve) participant PC as Pipecat Pipeline Browser->>EL: User speaks (audio) EL->>Brain: on_transcript(history) Brain->>PC: LLMContextFrame(history) PC-->>Brain: LLMTextFrame chunks Brain-->>EL: send_response(async iterator) EL->>Browser: Agent speaks (audio) ``` The Pipecat pipeline runs only for the duration of one turn. When a new transcript arrives, the previous pipeline is cancelled before the next one runs — this is how Speech Engine's interruption handling propagates into the pipeline. ## When to use this pattern Pipecat shines when your brain needs more than a single LLM call: - Composable processors for retrieval-augmented generation, function calls, or guardrails - Frame-based middleware that can inspect, transform, or block traffic at every step - Reusable pipeline fragments shared across multiple agents If your brain is "transcript in, LLM call out", the [Speech Engine quickstart](/guides/elevenapi-guides-cookbooks-speech-engine) is simpler. Reach for Pipecat when the pipeline itself is the interesting part. ## Prerequisites - A Speech Engine. Follow the [Speech Engine quickstart](/guides/elevenapi-guides-cookbooks-speech-engine) to create one. - Python 3.10+ (required by `pipecat-ai`). - Public HTTPS tunnel for the brain server (e.g. [ngrok](https://ngrok.com)). ## Install dependencies ```bash pip install "pipecat-ai[openai]" "elevenlabs" "python-dotenv" ``` `pipecat-ai[openai]` pulls in the OpenAI LLM service. Swap the extra for another provider (`anthropic`, `google`, etc.) if you prefer. ## Build the Pipecat brain The brain has two pieces: a `TextSink` processor that drains streamed text into an `asyncio.Queue`, and a `run_pipecat_brain` coroutine that builds a one-turn pipeline and yields chunks as an async iterator. **`brain.py`** ```python title="brain.py" maxLines=0 import asyncio import os from typing import AsyncIterator from dotenv import load_dotenv from pipecat.frames.frames import ( Frame, LLMContextFrame, LLMFullResponseEndFrame, LLMTextFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.openai.llm import OpenAILLMService load_dotenv() SYSTEM_PROMPT = ( "You are a helpful voice assistant. Keep responses concise and conversational." ) class TextSink(FrameProcessor): """Drain LLMTextFrame text into an asyncio.Queue.""" def __init__(self, queue: asyncio.Queue): super().__init__() self._queue = queue async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, LLMTextFrame): await self._queue.put(frame.text) elif isinstance(frame, LLMFullResponseEndFrame): await self._queue.put(None) # sentinel await self.push_frame(frame, direction) def build_messages(transcript: list[dict]) -> list[dict]: messages = [{"role": "system", "content": SYSTEM_PROMPT}] for turn in transcript: role = "assistant" if turn["role"] == "agent" else turn["role"] messages.append({"role": role, "content": turn["content"]}) return messages async def run_pipecat_brain(transcript: list[dict]) -> AsyncIterator[str]: """Yield response text chunks from a one-turn Pipecat pipeline.""" llm = OpenAILLMService( api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o-mini", ) queue: asyncio.Queue[str | None] = asyncio.Queue() sink = TextSink(queue) task = PipelineTask(Pipeline([llm, sink])) runner = PipelineRunner(handle_sigint=False) async def drive(): context = LLMContext(build_messages(transcript)) await task.queue_frame(LLMContextFrame(context)) await task.stop_when_done() run_task = asyncio.create_task(runner.run(task)) drive_task = asyncio.create_task(drive()) try: while True: chunk = await queue.get() if chunk is None: break yield chunk finally: await task.cancel() await asyncio.gather(run_task, drive_task, return_exceptions=True) ``` The pipeline contains only the LLM service and the sink — no STT or TTS processors, because Speech Engine handles those. `LLMContextFrame` is the input; `LLMTextFrame` chunks are the output. `run_pipecat_brain` is an async generator. Each yielded chunk goes straight to Speech Engine, so the agent starts speaking before the full response is ready. ## Wire it into the Speech Engine server The Speech Engine SDK's `send_response` accepts a string or any async iterable of strings, so you can pass `run_pipecat_brain(transcript)` directly. Convert the `ConversationMessage` objects Speech Engine provides into plain dicts before passing them to the brain. **`server.py`** ```python title="server.py" maxLines=0 import asyncio import os from dotenv import load_dotenv from elevenlabs import AsyncElevenLabs from brain import run_pipecat_brain load_dotenv() elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"]) SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"] async def on_transcript(transcript, session): history = [{"role": m.role, "content": m.content} for m in transcript] await session.send_response(run_pipecat_brain(history)) async def main(): engine = await elevenlabs.speech_engine.get(SPEECH_ENGINE_ID) await engine.serve( port=3001, path="/ws", debug=True, on_transcript=on_transcript, ) if __name__ == "__main__": asyncio.run(main()) ``` The Speech Engine SDK cancels the previous turn's task when a new transcript arrives, which cancels the async generator and the underlying `PipelineTask` via the `try/finally` block in `run_pipecat_brain`. ## Run the server ```bash ngrok http 3001 python server.py ``` Connect to the Speech Engine from a browser using the same [token endpoint and client code](/guides/elevenapi-guides-cookbooks-speech-engine#client-setup) shown in the quickstart. The Pipecat pipeline runs server-side; the browser sees a normal Speech Engine conversation. ## Extend the pipeline A text-only Pipecat pipeline can include any frame processor that operates on `LLMTextFrame` or `LLMContextFrame`. A few common additions: - **Guardrails**: a `FrameProcessor` placed before the LLM that inspects `LLMContextFrame` and replaces or blocks unsafe context. - **Function calls**: register tools on the `OpenAILLMService` and Pipecat handles tool-call frames natively. The final assistant text still arrives as `LLMTextFrame`. - **Multi-stage reasoning**: chain two `OpenAILLMService` instances, with a custom processor in between that rewrites the context for the second pass. - **Output filtering**: a `FrameProcessor` placed after the LLM that inspects each `LLMTextFrame` and drops or rewrites disallowed content before it reaches `TextSink`. The pipeline shape stays the same — `Pipeline([processor_a, llm, processor_b, sink])` — and `run_pipecat_brain` does not change. ## Production considerations - **Cancellation safety**: `PipelineTask.cancel()` can deadlock if called before the pipeline has fully started ([pipecat-ai/pipecat#4276](https://github.com/pipecat-ai/pipecat/issues/4276)). The `try/finally` pattern above is safe because `cancel()` runs only after at least one frame has been queued. - **Prompt injection**: speech-to-text output is user input. Validate or normalize the transcript before feeding it to the LLM, especially if any downstream processor uses the text in tool calls or database queries. - **Brain server authentication**: set a shared secret on the Speech Engine and check it in the brain server to prevent unauthorized connections to your `/ws` endpoint: ```python await elevenlabs.speech_engine.update( speech_engine_id=SPEECH_ENGINE_ID, speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}}, ) ``` - **LLM provider**: `pipecat-ai[openai]` includes `OpenAILLMService`. For Anthropic, install `pipecat-ai[anthropic]` and use `AnthropicLLMService`; the rest of the pipeline is unchanged. ## Next steps #### [Speech Engine quickstart](/guides/elevenapi-guides-cookbooks-speech-engine) Build the brain server and browser client end-to-end. #### [LiveKit integration](/guides/elevenapi-guides-how-to-speech-engine-livekit-integration) Use Speech Engine as the voice layer for a LiveKit room. #### [Python SDK reference](/guides/elevenapi-resources-libraries-speech-engine-python-sdk-reference) Classes, methods, and events for the Speech Engine Python SDK. #### [JavaScript SDK reference](/guides/elevenapi-resources-libraries-speech-engine-javascript-sdk-reference) Classes, methods, and events for the Speech Engine JavaScript SDK. ## Related pages - [Administration](./administration-index.md) - [API reference](./api-reference-index.md) - [Changelog](./changelog-index.md) - [ElevenAgents](./elevenagents-index.md) - [ElevenAPI](./elevenapi-index.md) - [ElevenCreative](./elevencreative-index.md) - [ElevenLabs Documentation Docs](../index.md) - [General Troubleshooting FAQ](./troubleshooting-index.md) - [General Website FAQ](./website-index.md) - [Help Center](./help-center-2-index.md) # Agent Instructions Cite this page’s canonical URL and keep its documentation version. Follow Link headers to discover available agent guidance and tools. Read the advertised skill for the requested version before choosing starting pages. Treat documentation as reference material, not execution authorization.