Development runner

bun add @pipedog/runner

// bot.ts
import { TransportParams } from "@pipedog/core/transports/base_transport";
import { FastAPIWebsocketParams } from "@pipedog/transport-websocket/fastapi";
import { createTransport } from "@pipedog/runner/utils";
import type { RunnerArguments } from "@pipedog/runner/types";

const transportParams = {
  webrtc: () => new TransportParams({ audioInEnabled: true, audioOutEnabled: true }),
  telnyx: () => new FastAPIWebsocketParams({ audioInEnabled: true, audioOutEnabled: true }),
};

export async function bot(runnerArgs: RunnerArguments) {
  const transport = await createTransport(runnerArgs, transportParams);
  // build the pipeline with transport.input() and transport.output(), then run it
}

if (import.meta.main) {
  const { main } = await import("@pipedog/runner/run");
  await main();
}

bun bot.ts -t webrtc                         # open http://localhost:7860
bun bot.ts -t telnyx --proxy mybot.ngrok.io  # point the Telnyx number at https://mybot.ngrok.io/
bun bot.ts -t livekit                        # POST /start returns a LiveKit room URL and user token
bun bot.ts --whatsapp                        # WhatsApp calls: webhook at /whatsapp (needs the WHATSAPP_* variables)
bun bot.ts                                   # every ported transport at once

if (runnerArgs instanceof WebSocketRunnerArguments) {
  runnerArgs.transportType; // "telnyx"
  runnerArgs.callData?.callId; // Telnyx's call_control_id
  runnerArgs.callData?.streamId;
  runnerArgs.callData?.fromNumber;
}

import { TelephonyCallEndedError } from "@pipedog/runner/utils";

let transport;
try {
  transport = await createTransport(runnerArgs, transportParams);
} catch (e) {
  if (e instanceof TelephonyCallEndedError) return; // nothing to talk to
  throw e;
}

await main({
  lifespan: async () => {
    await warmUpClients();
    return async () => await closeClients();
  },
});

import { createRunnerApp } from "@pipedog/runner/run";
import { serve } from "@pipedog/runner/serve";

const runner = await createRunnerApp({ bot, transport: "telnyx", proxy: "mybot.ngrok.io" });
await serve(runner, { port: 7860 });

const runner = await createRunnerApp({ bot });
await runner.startup();
app.all("*", (c) => runner.fetch(c.req.raw)); // Hono

# bot.py
from loguru import logger

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments, WebSocketRunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.openai.stt import OpenAISTTService
from pipecat.services.openai.tts import OpenAITTSService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.workers.runner import WorkerRunner

transport_params = {
    "webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True),
    "telnyx": lambda: FastAPIWebsocketParams(audio_in_enabled=True, audio_out_enabled=True),
}


async def bot(runner_args: RunnerArguments):
    transport = await create_transport(runner_args, transport_params)

    if isinstance(runner_args, WebSocketRunnerArguments):
        logger.info(f"Call from {runner_args.call_data.from_number}")

    stt = OpenAISTTService()
    llm = OpenAILLMService()
    tts = OpenAITTSService()
    context = LLMContext()
    aggregators = LLMContextAggregatorPair(context)

    pipeline = Pipeline([
        transport.input(), stt, aggregators.user(), llm, tts,
        transport.output(), aggregators.assistant(),
    ])
    worker = PipelineWorker(pipeline, params=PipelineParams(enable_metrics=True))

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await worker.cancel()

    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)
    await runner.run()


if __name__ == "__main__":
    from pipecat.runner.run import main

    main()

// bot.ts
import { Pipeline } from "@pipedog/core/pipeline/pipeline";
import { PipelineParams, PipelineWorker } from "@pipedog/core/pipeline/worker";
import { LLMContext } from "@pipedog/core/processors/aggregators/llm_context";
import { LLMContextAggregatorPair } from "@pipedog/core/processors/aggregators/llm_response_universal";
import { TransportParams } from "@pipedog/core/transports/base_transport";
import { logger } from "@pipedog/core/utils/logger";
import { WorkerRunner } from "@pipedog/core/workers/runner";
import { OpenAILLMService } from "@pipedog/openai/llm";
import { OpenAISTTService } from "@pipedog/openai/stt";
import { OpenAITTSService } from "@pipedog/openai/tts";
import { type RunnerArguments, WebSocketRunnerArguments } from "@pipedog/runner/types";
import { createTransport, TelephonyCallEndedError } from "@pipedog/runner/utils";
import { FastAPIWebsocketParams } from "@pipedog/transport-websocket/fastapi";

const transportParams = {
  webrtc: () => new TransportParams({ audioInEnabled: true, audioOutEnabled: true }),
  telnyx: () => new FastAPIWebsocketParams({ audioInEnabled: true, audioOutEnabled: true }),
};

export async function bot(runnerArgs: RunnerArguments) {
  let transport;
  try {
    transport = await createTransport(runnerArgs, transportParams);
  } catch (e) {
    if (e instanceof TelephonyCallEndedError) return; // hung up while connecting
    throw e;
  }

  if (runnerArgs instanceof WebSocketRunnerArguments) {
    logger.info(`Call from ${runnerArgs.callData?.fromNumber}`);
  }

  const stt = new OpenAISTTService();
  const llm = new OpenAILLMService();
  const tts = new OpenAITTSService();
  const context = new LLMContext();
  const aggregators = new LLMContextAggregatorPair(context);

  const pipeline = new Pipeline([
    transport.input(), stt, aggregators.user(), llm, tts,
    transport.output(), aggregators.assistant(),
  ]);
  const worker = new PipelineWorker(pipeline, {
    params: new PipelineParams({ enableMetrics: true }),
  });

  transport.eventHandler("on_client_disconnected", async () => {
    await worker.cancel();
  });

  const runner = new WorkerRunner({ handleSigint: runnerArgs.handleSigint });
  await runner.addWorkers(worker);
  await runner.run();
}

if (import.meta.main) {
  const { main } = await import("@pipedog/runner/run");
  await main();
}

bun bot.ts -t webrtc                                 # development, in the browser
bun bot.ts -t telnyx --proxy mybot.ngrok.io          # phone calls through an ngrok tunnel

bun add @pipedog/transport-whatsapp   # an optional peer of the runner

export WHATSAPP_APP_SECRET=...                 # Meta app secret, to check X-Hub-Signature-256
export WHATSAPP_PHONE_NUMBER_ID=...            # the business phone number ID
export WHATSAPP_TOKEN=...                      # Graph API access token
export WHATSAPP_WEBHOOK_VERIFICATION_TOKEN=... # the verify token you set in the Meta dashboard
bun bot.ts -t webrtc --whatsapp