Complete API Reference

{ .api }
import pipecat

# Package version
__version__: str  # "0.0.100"

def version() -> str:
    """Returns the Pipecat version string."""
    pass

{ .api }
from pipecat.frames.frames import Frame, SystemFrame, DataFrame, ControlFrame

class Frame:
    """Base frame with id, name, pts, metadata."""
    id: int
    name: str
    pts: Optional[int]  # Presentation timestamp (nanoseconds)
    metadata: Dict[str, Any]
    transport_source: Optional[str]
    transport_destination: Optional[str]

class SystemFrame(Frame):
    """High-priority, immediate processing, not interrupted."""
    pass

class DataFrame(Frame):
    """Normal priority, ordered processing, can be interrupted."""
    pass

class ControlFrame(Frame):
    """Control signals, ordered processing, can be interrupted."""
    pass

{ .api }
from pipecat.processors.frame_processor import (
    FrameProcessor,
    FrameDirection
)

class FrameProcessor:
    """Base class for all frame processors."""
    
    def __init__(self, name: Optional[str] = None, **kwargs):
        """Initialize processor."""
        pass
    
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Process a frame. Override in subclasses."""
        await self.push_frame(frame, direction)
    
    async def push_frame(self, frame: Frame, direction: FrameDirection):
        """Push frame to next processor."""
        pass
    
    async def queue_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
        """Queue frame for processing."""
        pass
    
    async def start(self):
        """Initialize when pipeline starts."""
        pass
    
    async def stop(self):
        """Cleanup when pipeline stops."""
        pass
    
    async def cancel(self):
        """Immediate cleanup on cancellation."""
        pass
    
    async def interrupt(self):
        """Handle interruption event."""
        pass
    
    def link(self, processor: "FrameProcessor"):
        """Link to downstream processor."""
        pass
    
    @property
    def interruptions_enabled(self) -> bool:
        """Whether processor can be interrupted."""
        return True

{ .api }
from pipecat.pipeline.pipeline import Pipeline

class Pipeline:
    """Sequential processor chain."""
    
    def __init__(
        self,
        processors: List[FrameProcessor],
        name: Optional[str] = None,
        enable_direct_mode: bool = False
    ):
        """Initialize pipeline."""
        pass
    
    async def run(self):
        """Run pipeline until completion."""
        pass
    
    async def queue_frames(self, frames: List[Frame]):
        """Queue frames for processing."""
        pass
    
    def get_processors(self) -> List[FrameProcessor]:
        """Get all processors."""
        pass

{ .api }
from pipecat.pipeline.task import PipelineTask, PipelineParams

class PipelineParams:
    """Pipeline task configuration."""
    allow_interruptions: bool = True
    enable_metrics: bool = False
    enable_usage_metrics: bool = False
    report_only_initial_ttfb: bool = False

class PipelineTask:
    """Pipeline execution task."""
    
    def __init__(
        self,
        pipeline: Pipeline,
        params: Optional[PipelineParams] = None,
        clock: Optional[BaseClock] = None
    ):
        """Initialize task."""
        pass
    
    async def run(self):
        """Run pipeline until EndFrame."""
        pass
    
    async def queue_frame(self, frame: Frame):
        """Queue single frame."""
        pass
    
    async def queue_frames(self, frames: List[Frame]):
        """Queue multiple frames."""
        pass
    
    async def cancel(self):
        """Cancel task immediately."""
        pass
    
    def has_finished(self) -> bool:
        """Check if task completed."""
        pass

{ .api }
from pipecat.services.llm_service import LLMService

class LLMService(AIService):
    """Base LLM service."""
    
    def set_context(self, context: LLMContext):
        """Set LLM context."""
        pass
    
    def register_function(
        self,
        name: str,
        handler: Callable,
        description: str = "",
        properties: Optional[Dict] = None,
        required: Optional[List[str]] = None
    ):
        """Register function for calling."""
        pass
    
    def unregister_function(self, name: str):
        """Unregister function."""
        pass

{ .api }
from pipecat.services.tts_service import TTSService

class TTSService(AIService):
    """Base TTS service."""
    
    async def run_tts(self, text: str) -> AsyncGenerator[AudioRawFrame, None]:
        """Synthesize text to audio."""
        pass
    
    def set_voice(self, voice_id: str):
        """Set voice."""
        pass
    
    def set_model(self, model: str):
        """Set TTS model."""
        pass

{ .api }
from pipecat.services.stt_service import STTService

class STTService(AIService):
    """Base STT service."""
    
    async def run_stt(self, audio: bytes) -> Optional[str]:
        """Transcribe audio."""
        pass
    
    def set_model(self, model: str):
        """Set STT model."""
        pass
    
    def set_language(self, language: str):
        """Set language."""
        pass
    
    @property
    def is_muted(self) -> bool:
        """Whether STT is muted."""
        pass

{ .api }
from pipecat.transports.base_transport import BaseTransport, TransportParams

class BaseTransport(BaseObject):
    """Base transport."""
    
    def __init__(self, params: TransportParams, **kwargs):
        """Initialize transport."""
        pass
    
    async def start(self):
        """Start transport."""
        pass
    
    async def stop(self):
        """Stop transport."""
        pass
    
    async def send_audio(self, audio: bytes):
        """Send audio data."""
        pass
    
    async def send_image(self, image: bytes):
        """Send image/video."""
        pass
    
    async def send_message(self, message: dict):
        """Send data message."""
        pass
    
    def input(self) -> FrameProcessor:
        """Get input processor."""
        pass
    
    def output(self) -> FrameProcessor:
        """Get output processor."""
        pass

{ .api }
from pipecat.processors.aggregators.llm_context import LLMContext

class LLMContext:
    """Universal LLM context."""
    
    def __init__(
        self,
        messages: Optional[List[Dict]] = None,
        tools: Optional[List[Dict]] = None,
        settings: Optional[Dict[str, Any]] = None
    ):
        """Initialize context."""
        pass
    
    def add_message(self, message: Dict):
        """Add message."""
        pass
    
    def add_messages(self, messages: List[Dict]):
        """Add multiple messages."""
        pass
    
    def set_tool_choice(self, tool_choice: str | Dict):
        """Set tool choice strategy."""
        pass
    
    def get_messages(self) -> List[Dict]:
        """Get all messages."""
        pass
    
    def set_tools(self, tools: List[Dict]):
        """Set available tools."""
        pass
    
    def get_tools(self) -> List[Dict]:
        """Get tools."""
        pass
    
    def update_settings(self, settings: Dict[str, Any]):
        """Update settings."""
        pass
    
    def get_settings(self) -> Dict[str, Any]:
        """Get settings."""
        pass