# Frame processors > Write a custom FrameProcessor - handle frames, push them on, pick a direction, set up resources and listen to events. Source: https://polvallverdu.github.io/pipedog/docs/concepts/frame-processors/ Frame processors are the workers of a pipeline. Each has a specific job (transcribe audio, generate text, play audio, log metrics...). They: * **receive frames** from the previous processor, * **process the data**, * **create new frames** with their output, * **pass frames along** to the next processor. pipedog ports Pipecat's built-in processors (aggregators, filters, observers, and services as their packages land). Most applications also need some custom frame processing logic; this page shows how to write it. ## Example: MetricsFrame logger [#example-metricsframe-logger] This custom processor formats and logs every `MetricsFrame`: ```ts function formatMetrics(metrics: MetricsData[]): string { return metrics .map((metric) => { const fields = Object.entries(metric) .map(([key, value]) => `\t${key}=${JSON.stringify(value)}`) .join("\n"); return `${metric.constructor.name}\n${fields}`; }) .join("\n"); } /** Formats and logs every MetricsFrame. */ export class MetricsFrameLogger extends FrameProcessor { override async processFrame(frame: Frame, direction: FrameDirection): Promise { await super.processFrame(frame, direction); if (frame instanceof MetricsFrame) { logger.info(`${frame.name}\n${formatMetrics(frame.data)}`); } // SUPER IMPORTANT: always push every frame! await this.pushFrame(frame, direction); } } ``` Add it to a pipeline wherever it should see the frames, for example at the very end so it receives every `MetricsFrame` the services produce: ```ts const pipeline = new Pipeline([ transportInput, stt, userAggregator, llm, tts, transportOutput, assistantAggregator, new MetricsFrameLogger(), // our custom processor ]); ``` ## Key requirements [#key-requirements] Custom processors extend `FrameProcessor` from `@pipedog/core/processors/frame_processor`. The base class handles `StartFrame`, `CancelFrame`, `InterruptionFrame` and pause / resume frames for you, and provides `processFrame()` and `pushFrame()`: * **`processFrame(frame, direction)`** receives every frame and is where your conditional logic goes. * **`pushFrame(frame, direction?)`** sends a frame on. Normally downstream, but you can push upstream, or both ways. Two calls are essential: 1. **`super(options)`** in the constructor, passing the whole options object (that is how Python's `**kwargs` flows to the base class). 2. **`await super.processFrame(frame, direction)`** first thing in `processFrame()`. ```ts export interface MyCustomProcessorOptions extends FrameProcessorOptions { greeting?: string; } export class MyCustomProcessor extends FrameProcessor { protected _greeting: string; constructor(options: MyCustomProcessorOptions = {}) { super(options); // ✅ Required: forwards name, enableDirectMode, metrics... this._greeting = options.greeting ?? "hello"; } override async processFrame(frame: Frame, direction: FrameDirection): Promise { await super.processFrame(frame, direction); // ✅ Required if (frame instanceof TranscriptionFrame) { console.log(`${this._greeting}, user ${frame.userId} said: ${frame.text}`); } await this.pushFrame(frame, direction); // ✅ Required: pass the frame through } } ``` Constructor options extend `FrameProcessorOptions` (see the [FrameProcessor API](https://polvallverdu.github.io/pipedog/docs/api/frame-processor.md)): `name`, `enableDirectMode` (skip the internal queues and process frames right away) and `metrics`. ## Critical responsibility: frame forwarding [#critical-responsibility-frame-forwarding] Processors receive **all** frames pushed through the pipeline, which gives them a lot of power and a great responsibility: they must push every frame on. A processor that drops frames it doesn't care about blocks the pipeline, and your application stops working in confusing ways. Note the unconditional `await this.pushFrame(frame, direction)` at the end of every example here. ## Frame direction [#frame-direction] Frames flow in one of two directions: ```ts export const FrameDirection = { DOWNSTREAM: 1, UPSTREAM: 2 } as const; ``` **Downstream** (the default) goes from input to output: audio enters at the transport input, gets transcribed, runs through the LLM, turns into speech and reaches the transport output. **Upstream** sends information back towards the start: errors travel upstream, and the assistant context aggregator at the end pushes context frames upstream to the LLM. ```ts // Push downstream (the default) await this.pushFrame(frame, FrameDirection.DOWNSTREAM); // Push upstream, back towards the start of the pipeline await this.pushFrame(new TextFrame({ text: "hi" }), FrameDirection.UPSTREAM); // Create two instances, push one each way, linked by `broadcastSiblingId` await this.broadcastFrame(UserStartedSpeakingFrame); // Copy a frame you didn't create and push it both ways await this.broadcastFrameInstance(frame); ``` Prefer `broadcastFrame(FrameClass, options)` when you create the frame; use `broadcastFrameInstance(frame)` for a frame you received, which copies every field except `id` and `name`. ## Setup and cleanup [#setup-and-cleanup] A processor is set up before any frame reaches it, and cleaned up when the pipeline tears down. That is where a processor that needs a connection, a client, or the pipeline's audio configuration gets it: ```ts export class ConnectedProcessor extends FrameProcessor { protected _sampleRate = 0; protected _client: Client | null = null; override async setup(setup: FrameProcessorSetup): Promise { await super.setup(setup); // ✅ Required this._sampleRate = setup.audioOutSampleRate; this._client = await connect(); } override async cleanup(): Promise { await super.cleanup(); // ✅ Required await this._client?.close(); } } ``` `FrameProcessorSetup` carries `clock`, `taskManager`, `pipelineWorker`, `audioInSampleRate` (default 16000), `audioOutSampleRate` (default 24000), `enableMetrics`, `enableTracing`, `enableUsageMetrics`, `observer`, `reportOnlyInitialTtfb` and `tracingContext`. The processor also exposes `metricsEnabled`, `usageMetricsEnabled`, `reportOnlyInitialTtfb` and, once set up, `processorSetup`. Don't read this configuration off the `StartFrame`: upstream deprecated those `StartFrame` fields and pipedog doesn't have them. Processors are set up **concurrently**. A processor that throws while setting up pushes an `ErrorFrame` upstream, so the application learns its pipeline came up degraded; one that throws while being cleaned up is logged and doesn't stop the others from being released. ### Holding frames until a condition is met [#holding-frames-until-a-condition-is-met] A processor that connects during setup may receive frames before it is ready. `pauseProcessingAllFramesUntil(ready, { timeout })` holds everything arriving until `ready()` resolves, then delivers it in order. The pause takes hold from the frame *after* the one being processed, and is always lifted: when `ready` resolves, when `timeout` seconds (default 5) elapse, or at cleanup. ```ts export class SlowToConnect extends FrameProcessor { protected _ready = new Event(); override async setup(setup: FrameProcessorSetup): Promise { await super.setup(setup); // Hold every frame until connected, for at most 10 seconds. await this.pauseProcessingAllFramesUntil(() => this._ready.wait(), { timeout: 10 }); this.createTask(async () => { await connect(); this._ready.set(); }, "connect"); } } ``` ## Errors [#errors] Report errors with `pushError()`, which builds an `ErrorFrame`, classifies it, fires `on_error` and pushes it upstream. An error that can't be recovered from (a rejected API key, say) makes the processor unusable (`isUsable` becomes `false`) and the worker applies its `processorUnusablePolicy`. ```ts try { await riskyOperation(); } catch (e) { // A recoverable error: report it and keep going. await this.pushError("Failed to process audio chunk, skipping", { exception: e }); } // An error that leaves this processor unable to do any more work. await this.pushError("Critical operation failed", { forceTreatAsPermanent: true }); ``` Every broad `catch` in a processor should start with `rethrowIfCancelled(e)`; see [Interruptions & cancellation](https://polvallverdu.github.io/pipedog/docs/concepts/interruptions-and-cancellation/#rethrowifcancelled). ## Event handlers [#event-handlers] Every frame processor, including services and transport inputs/outputs, emits these events. Event names are the same strings as in Pipecat. Python's `@obj.event_handler("on_x")` decorator becomes `obj.eventHandler("on_x", handler)` (which returns the handler) or `obj.addEventHandler("on_x", handler)`. The first argument of every handler is the processor. | Event | Arguments | Description | | ------------------------- | -------------------------------- | -------------------------------------------------------------------------------- | | `on_error` | `(processor, error: ErrorFrame)` | An error occurred in this processor, before its `ErrorFrame` is pushed upstream. | | `on_usable_changed` | `(processor, isUsable: boolean)` | The processor stopped (or started again) being able to do its job. | | `on_before_process_frame` | `(processor, frame)` | A frame is about to be processed. | | `on_after_process_frame` | `(processor, frame)` | A frame has just been processed. | | `on_before_push_frame` | `(processor, frame)` | A frame is about to be pushed. | | `on_after_push_frame` | `(processor, frame)` | A frame has just been pushed. | ```ts const tts = new MyCustomProcessor({ name: "tts" }); // Like Python's `@tts.event_handler("on_error")` decorator. tts.eventHandler("on_error", async (processor: FrameProcessor, error: ErrorFrame) => { console.log(`Error in ${processor}: ${error.error}`); if (error.exception) { console.log("Exception:", error.exception); } if (error.processor && !error.processor.isUsable) { console.log("This processor can no longer do its job"); } }); tts.eventHandler("on_usable_changed", async (processor: FrameProcessor, isUsable: boolean) => { if (!isUsable) console.warn(`${processor} is down`); }); // The same handler on several processors. async function handleServiceError(processor: FrameProcessor, error: ErrorFrame): Promise { logger.error(`Service error in ${processor.name}: ${error.error}`); } const stt = new MyCustomProcessor({ name: "stt" }); const llm = new MyCustomProcessor({ name: "llm" }); for (const service of [stt, llm, tts]) { service.addEventHandler("on_error", handleServiceError); } // Frame hooks run inline: keep them fast. llm.eventHandler("on_before_push_frame", async (processor: FrameProcessor, frame: Frame) => { console.log(`${processor} is about to push: ${frame}`); }); ``` All of these except `on_usable_changed` are **synchronous**: handlers run inline and block the pipeline until they complete. Use them for fast work like logging, counting or setting flags, not for I/O. `on_usable_changed` runs in its own task. ## Example: response logger [#example-response-logger] Watching for frames that bracket a piece of work is a common pattern. `LLMFullResponseStartFrame` and `LLMFullResponseEndFrame` bracket every LLM response: ```ts export class ResponseLogger extends FrameProcessor { override async processFrame(frame: Frame, direction: FrameDirection): Promise { await super.processFrame(frame, direction); if (frame instanceof LLMFullResponseStartFrame) { logger.info("LLM response started"); } else if (frame instanceof LLMFullResponseEndFrame) { logger.info("LLM response finished"); } await this.pushFrame(frame, direction); } } ``` ## Testing a processor [#testing-a-processor] `runTest()` from `@pipedog/core/tests/utils` (the port of `pipecat.tests.utils.run_test`) runs a processor in a real pipeline, sends frames, and checks the frame types that come out: ```ts import { expect, test } from "bun:test"; import { type Frame, TextFrame } from "@pipedog/core/frames/frames"; import { type FrameDirection, FrameProcessor } from "@pipedog/core/processors/frame_processor"; import { runTest } from "@pipedog/core/tests/utils"; class UpperCaseProcessor extends FrameProcessor { override async processFrame(frame: Frame, direction: FrameDirection): Promise { await super.processFrame(frame, direction); if (frame instanceof TextFrame) { await this.pushFrame(new TextFrame({ text: frame.text.toUpperCase() }), direction); } else { await this.pushFrame(frame, direction); } } } test("test_upper_case", async () => { const [down] = await runTest(new UpperCaseProcessor(), { framesToSend: [new TextFrame({ text: "hello" })], expectedDownFrames: [TextFrame], }); expect((down[0] as TextFrame).text).toEqual("HELLO"); }); ``` ## Best practices [#best-practices] 1. **Always call the parent**: `super(options)` and `await super.processFrame(frame, direction)`. 2. **Forward all frames** with `await this.pushFrame(frame, direction)`. 3. **Handle frames conditionally** with `instanceof` checks. 4. **Handle errors properly**: `rethrowIfCancelled(e)` first in every `catch`, then `pushError()`. 5. **Position carefully**: put the processor where it receives the frames it needs. 6. **Track your tasks**: start background work with `this.createTask()`, never a floating promise.