Coming from Pipecat (Python)

pipedog is designed so that someone who knows Pipecat can read pipedog code without a dictionary. Class names, frame names, event names, module paths and wire formats are the same; what changes is what TypeScript forces to change. This page is the short version of the porting rulebook, docs/CONVENTIONS.md in the repository.

A side-by-side exampleCopy Anchor Link

from pipecat.frames.frames import Frame, TextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


class Shout(FrameProcessor):
    def __init__(self, *, suffix: str = "!", **kwargs):
        super().__init__(**kwargs)
        self._suffix = suffix

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, TextFrame):
            frame = TextFrame(text=frame.text.upper() + self._suffix)
        await self.push_frame(frame, direction)


shout = Shout(suffix="!!", name="shout")


@shout.event_handler("on_error")
async def on_error(processor, error):
    print(error.error)

import { type ErrorFrame, type Frame, TextFrame } from "@pipedog/core/frames/frames";
import {
  type FrameDirection,
  FrameProcessor,
  type FrameProcessorOptions,
} from "@pipedog/core/processors/frame_processor";

export interface ShoutOptions extends FrameProcessorOptions {
  suffix?: string;
}

class Shout extends FrameProcessor {
  protected _suffix: string;

  constructor(options: ShoutOptions = {}) {
    super(options);
    this._suffix = options.suffix ?? "!";
  }

  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);
    if (frame instanceof TextFrame) {
      frame = new TextFrame({ text: frame.text.toUpperCase() + this._suffix });
    }
    await this.pushFrame(frame, direction);
  }
}

const shout = new Shout({ suffix: "!!", name: "shout" });

shout.eventHandler("on_error", async (_processor: FrameProcessor, error: ErrorFrame) => {
  console.log(error.error);
});

Modules and importsCopy Anchor Link

One Python module is one TypeScript file at the same path, with the same snake_case name:

The core import closure (frames, processors base, pipeline, workers, bus, observers, RTVI, base transports and services, turns, utils) is one package, @pipedog/core. Each provider, transport, serializer and audio model is its own package, like Pipecat's extras.

NamingCopy Anchor Link

Constructors and parametersCopy Anchor Link

• Keyword arguments become one options object, named <Class>Options and extending the parent's. The whole object is passed to super(options): that is how **kwargs flows.
• Required positional parameters of functions stay positional: push_frame(frame, direction=DOWNSTREAM) is pushFrame(frame, direction = FrameDirection.DOWNSTREAM).
• Keyword-only or optional groups become a trailing options object: push_error(msg, exception=e) is pushError(msg, { exception: e }).
• None becomes null, and defaults use ?? (never ||, which swallows 0, "" and false).
• Pydantic params classes (PipelineParams, InputParams) become classes with an options constructor and defaults: new PipelineParams({ enableMetrics: true }).
• Dataclasses become classes with fields and an options constructor.

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(enable_metrics=True, audio_out_sample_rate=24000),
    idle_timeout_secs=300,
)

const worker = new PipelineWorker(pipeline, {
  params: new PipelineParams({ enableMetrics: true, audioOutSampleRate: 24000 }),
  idleTimeoutSecs: 300,
});

Time, bytes and unitsCopy Anchor Link

• Durations and timeouts stay in seconds (floats), as upstream, so configs and docs carry over: sleep(0.5), idleTimeoutSecs: 300, waitFor(fn, 5).
• Clock values stay in nanoseconds as number (frame.pts).
• bytes is Uint8Array, never Buffer. PCM is 16-bit little-endian; read samples with new Int16Array(u8.buffer, u8.byteOffset, u8.length / 2) (copy first if byteOffset is odd).
• time.time(), time.monotonic(), time.monotonic_ns() are time(), monotonic() and monotonicNs() from @pipedog/asyncio.

EnumsCopy Anchor Link

Plain enums become an as const object plus a type of the same name, keeping upstream's values. X.FOO.value is just X.FOO:

export const FrameDirection = { DOWNSTREAM: 1, UPSTREAM: 2 } as const;
export type FrameDirection = (typeof FrameDirection)[keyof typeof FrameDirection];

Enums with methods or properties (like ErrorCategory.is_permanent) become classes with static instances, a value field and getters (ErrorCategory.AUTHENTICATION.isPermanent), plus values() and fromValue(v). StrEnums are as const objects with string values. There is no TypeScript enum or namespace anywhere: the code is erasable so Node can run it directly.

Async and cancellationCopy Anchor Link

Python's asyncio semantics come from @pipedog/asyncio. The rules that matter most are in Interruptions & cancellation.

And three rules:

1. except Exception: does not catch CancelledError, but catch (e) catches everything. Every broad catch starts with rethrowIfCancelled(e).
2. Foreign awaits (fetch, an SDK, a WebSocket) are not cancellation points. Pass currentSignal() to APIs that take an AbortSignal, and/or wrap the await in abortable().
3. Never leave a promise floating. Where upstream fires and forgets a task, use this.createTask(...) so it is tracked and cancelled at cleanup.

Classes and inheritanceCopy Anchor Link

Multiple inheritance: a second base that is a mixin becomes a class-factory mixin (class C(A, M) is class C extends M(A), and instanceof M still works). Two concrete bases in the same hierarchy (such as a frame that is both a system frame and a text frame) extend the first and register the second as a virtual base, so instanceof works for both.

FramesCopy Anchor Link

• Frame constructors always take one options object: TextFrame("hi") is new TextFrame({ text: "hi" }), even where upstream tests pass positional arguments.
• init=True dataclass fields come from the options; init=False fields are initializers.
• Frames that are uninterruptible by default declare override interruptible = false.
• broadcast_frame(FrameCls, **kwargs) is broadcastFrame(FrameCls, options).
• Field order follows upstream, and __post_init__ logic runs in the constructor after super().

LoggingCopy Anchor Link

from loguru import logger becomes import { logger } from "@pipedog/core/utils/logger", with the same level methods (trace, debug, info, warning, error, critical). logger.exception(msg) is logger.error(msg, e).

The development runnerCopy Anchor Link

Pipecat's pipecat.runner is @pipedog/runner, with the same entry point: export a bot(runnerArgs) function and call main(). The flags (-t webrtc, -t twilio, --proxy, --host, --port, ...) are the same.

The FastAPI app is a fetch handler, so the runner mounts in any framework. Daily, MOQ and the Vonage Video Connector are recognized but fail with a clear message: their transports aren't ported yet. See Development runner for the full mapping.

What isn't ported yetCopy Anchor Link

The core, Pipecat Flows, evals, the CLI and most providers are ported; Daily and the other room transports, the Krisp, Koala and AIC filters, the Redis and pgmq buses and a few providers are not. The roadmap lists the gaps with their upstream module counts.

Deprecated APIs are not portedCopy Anchor Link

Anything upstream marks deprecated (with @deprecated, warn_deprecated, or "will be removed in 2.0.0") is left out of pipedog, including:

• the PipelineTask / PipelineTaskParams / PipelineRunner aliases (use PipelineWorker, PipelineParams and WorkerRunner);
• EndTaskFrame / CancelTaskFrame (use EndWorkerFrame / CancelWorkerFrame);
• the UninterruptibleFrame mixin (use interruptible = false);
• deprecated parameters such as push_error(fatal=) and ErrorFrame.fatal (check processor.isUsable, or pass forceTreatAsPermanent);
• the deprecated StartFrame configuration fields (read FrameProcessorSetup in setup()).

What was dropped from each module is recorded in its notes in porting/manifest.json.

TestsCopy Anchor Link

Upstream tests are ported 1:1 with bun:test, keeping their names so the two suites can be diffed: class TestX(unittest.IsolatedAsyncioTestCase) is describe("TestX", ...), async def test_y is test("test_y", ...), and run_test(...) is runTest(...) from @pipedog/core/tests/utils.