Custom FrameProcessor

Pipecat’s architecture is made up of a Pipeline, FrameProcessors, and Frames. See the Core Concepts for a full review. From that architecture, recall that FrameProcessors are the workers in the pipeline that receive frames and complete actions based on the frames received. Pipecat comes with many FrameProcessors built in. These consist of services, like OpenAILLMService or CartesiaTTSService, utilities, like LLMTextProcessor, and other things. Largely, you can build most of your application with these built-in FrameProcessors, but commonly, your application code may require custom frame processing logic. For example, you may want to perform an action as a result of a frame that’s pushed in the pipeline.

Example: MetricsFrame logger

This custom FrameProcessor format and logs MetricsFrames:

class MetricsFrameLogger(FrameProcessor):
    """MetricsFrameLogger formats and logs all MetericsFrames"""

    def __init__(self):
        super().__init__()

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        if isinstance(frame, MetricsFrame):
            logger.info(f"{frame.name}\n    {format_metrics(frame.data)}")
            await self.push_frame(frame, direction)

        # ALWAYS push all frames
        else:
            # SUPER IMPORTANT: always push every frame!
            await self.push_frame(frame, direction)

This frame processor looks for MetricsFrames. When it sees one, it formats the data and logs it. It uses this format_metrics function:

def format_metrics(metrics, indent=0):
    lines = []
    tab = "\t" * indent

    for metric in metrics:
        lines.append(tab + type(metric).__name__)
        for field, value in vars(metric).items():
            if hasattr(value, "__dict__") and not isinstance(
                value, (str, int, float, bool, type(None))
            ):
                lines.append(f"{tab}\t{field}={type(value).__name__}")
                for k, v in vars(value).items():
                    lines.append(f"{tab}\t\t{k}={repr(v)}")
            else:
                lines.append(f"{tab}\t{field}={repr(value)}")

    return "\n".join(lines)

See this working example using the MetricsFrameLogger FrameProcessor

Add to a Pipeline

# Create and initialize the custom FrameProcessor
metrics_frame_processor = MetricsFrameLogger()

pipeline = Pipeline(
    [
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
        metrics_frame_processor,  # Our custom FrameProcessor that pretty prints metrics frames
    ]
)

With this positioning, the MetricsFrameLogger FrameProcessor will receive every MetericsFrame in the pipeline.

Key Requirements

FrameProcessors must inherit from the base FrameProcessor class. This ensures that your custom FrameProcessor will correctly handle frames like StartFrame, EndFrame, InterruptionFrame without having to write custom logic for those frames. This inheritance also provides it with the ability to process_frame() and push_frame():

process_frame() is what allows the FrameProcessor to receive frames and add custom conditional logic based on the frames that are received.
push_frame() allows the FrameProcessor to push frames to the pipeline. Normally, frames are pushed DOWNSTREAM, but based on which processors need the output, you can also push UPSTREAM or in both directions.

Essential Implementation Details

To ensure proper base class inheritance, it’s critical to include:

super().__init__() in your __init__ method
await super().process_frame(frame, direction) in your process_frame() method

class MyCustomProcessor(FrameProcessor):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)  # ✅ Required
        # Your initialization code here

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)  # ✅ Required

        # Your custom frame processing logic here
        if isinstance(frame, SomeSpecificFrame):
            # Handle the frame
            pass

        await self.push_frame(frame, direction)  # ✅ Required - pass frame through

Setup and cleanup

A processor is set up before any frame reaches it, and cleaned up when the pipeline tears down. This is where a processor that needs a connection, a client, or the pipeline’s audio configuration gets it:

from pipecat.processors.frame_processor import FrameProcessor, FrameProcessorSetup

class MyCustomProcessor(FrameProcessor):
    async def setup(self, setup: FrameProcessorSetup):
        await super().setup(setup)          # ✅ Required
        self._sample_rate = setup.audio_out_sample_rate
        self._client = await connect()

    async def cleanup(self):
        await super().cleanup()             # ✅ Required
        await self._client.close()

FrameProcessorSetup

Field	Type	Default	Description
`clock`	`BaseClock`		The pipeline clock
`task_manager`	`BaseTaskManager`		Creates and tracks the processor’s asyncio tasks
`pipeline_worker`	`PipelineWorker`		The worker running this pipeline
`audio_in_sample_rate`	`int`	`16000`	Input audio sample rate in Hz
`audio_out_sample_rate`	`int`	`24000`	Output audio sample rate in Hz
`enable_metrics`	`bool`	`False`	Whether to collect performance metrics
`enable_tracing`	`bool`	`False`	Whether tracing is enabled
`enable_usage_metrics`	`bool`	`False`	Whether to collect usage metrics
`observer`	`BaseObserver \| None`	`None`	The pipeline observer, if one is attached
`report_only_initial_ttfb`	`bool`	`False`	Whether to report only the first TTFB per service
`tracing_context`	`TracingContext \| None`	`None`	Tracing context for distributed tracing

metrics_enabled, usage_metrics_enabled, and report_only_initial_ttfb are also available as properties on the processor itself, and processor_setup returns the whole object once setup has run.

Don’t read this configuration off the StartFrame in process_frame(). StartFrame.audio_in_sample_rate and its siblings are deprecated since v1.8.0 and removed in 2.0.0.

Processors are set up concurrently, so a resource two of them share — the client an input and output transport hold between them — needs guarding. acquires and releases from pipecat.utils.shared do that: the first owner to acquire runs the decorated method while the rest wait for it, and releases runs only for the last owner to let go.

A processor that raises while setting up pushes an ErrorFrame upstream, so the application learns its pipeline came up degraded. A processor that raises while being cleaned up no longer costs the rest of the pipeline its teardown — the failure is logged and every other processor is still released.

Holding frames until a condition is met

A processor that establishes a connection during setup may receive frames before it is ready for them. pause_processing_all_frames_until() holds everything arriving at the processor until a condition resolves, then delivers it in order:

async def setup(self, setup: FrameProcessorSetup):
    await super().setup(setup)
    self._ready = asyncio.Event()
    await self.pause_processing_all_frames_until(self._ready.wait, timeout=10)

The pause takes effect from the frame after the one being processed, so a StartFrame that triggers it still travels downstream. Both queues are held while it is in force, so give it a timeout; the pause is always lifted at cleanup.

Critical Responsibility: Frame Forwarding

FrameProcessors receive all frames that are pushed through the pipeline. This gives them a lot of power, but also a great responsibility. Critically, they must push all frames through the pipeline; if they don’t, they block frames from moving through the Pipeline, which will cause issues in how your application functions. As well as formatting and logging MetricsFrames, MetricsFrameLogger also has an await self.push_frame(frame, direction) which pushes the frame through to the next processor in the pipeline.

Frame Direction

When pushing frames, you can specify the direction:

# Push downstream (default)
await self.push_frame(frame, FrameDirection.DOWNSTREAM)

# Push upstream
await self.push_frame(frame, FrameDirection.UPSTREAM)

Most custom FrameProcessors will push frames downstream, but upstream can be useful for sending control frames or error notifications back up the pipeline.

Best Practices

Always call the parent methods: Use super().__init__() and await super().process_frame()
Forward all frames: Make sure every frame is pushed through with await self.push_frame(frame, direction)
Handle frames conditionally: Use isinstance() checks to handle specific frame types
Use proper error handling: Wrap risky operations in try/catch blocks
Position carefully in pipeline: Consider where in the pipeline your processor needs to be to receive the right frames

With these patterns, you can create powerful custom FrameProcessors that extend Pipecat’s capabilities for your specific use case.

Get Started

Migration

Learning Pipecat

Fundamentals

Flows

Evals

Features

Telephony

Deployment

Examples & Recipes

Custom FrameProcessor

Example: MetricsFrame logger

Add to a Pipeline

Key Requirements

Essential Implementation Details

Setup and cleanup

FrameProcessorSetup

Holding frames until a condition is met

Critical Responsibility: Frame Forwarding

Frame Direction

Best Practices

​Example: MetricsFrame logger

​Add to a Pipeline

​Key Requirements

​Essential Implementation Details

​Setup and cleanup

​FrameProcessorSetup

​Holding frames until a condition is met

​Critical Responsibility: Frame Forwarding

​Frame Direction

​Best Practices

Example: MetricsFrame logger

Add to a Pipeline

Key Requirements

Essential Implementation Details

Setup and cleanup

FrameProcessorSetup

Holding frames until a condition is met

Critical Responsibility: Frame Forwarding

Frame Direction

Best Practices