Voice agent fully on Cloudflare

client ──WebSocket──► Worker ──getByName(id)──► VoiceSession (Durable Object)
                                                   │ FastAPIWebsocketTransport
                                                   │ Flux (env.AI, WebSocket): transcripts and turns
                                                   │ Workers AI LLM (OpenAI-compatible REST)
                                                   │ Aura (env.AI, WebSocket): speech

bun add @pipedog/core @pipedog/asyncio @pipedog/cloudflare \
  @pipedog/transport-websocket @pipedog/serializer-protobuf

{
  "name": "voice-agent",
  "main": "src/worker.ts",
  "compatibility_date": "2026-10-01",
  "compatibility_flags": ["nodejs_compat"],
  "ai": { "binding": "AI" },
  "durable_objects": {
    "bindings": [{ "name": "VOICE_SESSION", "class_name": "VoiceSession" }]
  },
  "migrations": [{ "tag": "v1", "new_classes": ["VoiceSession"] }]
}

bunx wrangler secret put CLOUDFLARE_ACCOUNT_ID
bunx wrangler secret put CLOUDFLARE_API_TOKEN

import type { WorkersAIBinding } from "@pipedog/cloudflare/client";

export interface Env {
  /** The Workers AI binding: `"ai": { "binding": "AI" }` in wrangler.jsonc. */
  AI: WorkersAIBinding;
  /** For the LLM, which uses Workers AI's OpenAI-compatible REST endpoint. */
  CLOUDFLARE_ACCOUNT_ID: string;
  CLOUDFLARE_API_TOKEN: string;
}

import { WorkersAIFluxSTTService } from "@pipedog/cloudflare/flux_stt";
import { WorkersAILLMService } from "@pipedog/cloudflare/llm";
import { WorkersAIAuraTTSService } from "@pipedog/cloudflare/tts";
import { LLMRunFrame } from "@pipedog/core/frames/frames";
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 { WorkerRunner } from "@pipedog/core/workers/runner";
import { ProtobufFrameSerializer } from "@pipedog/serializer-protobuf/protobuf";
import {
  FastAPIWebsocketParams,
  FastAPIWebsocketTransport,
} from "@pipedog/transport-websocket/fastapi";
import type { WebsocketConnection } from "@pipedog/transport-websocket/socket";

/** Flux and the client both use 16 kHz audio; Aura answers at 24 kHz. */
const INPUT_SAMPLE_RATE = 16000;
const OUTPUT_SAMPLE_RATE = 24000;

export async function runVoiceAgent(connection: WebsocketConnection, env: Env): Promise<void> {
  const transport = new FastAPIWebsocketTransport(
    connection,
    new FastAPIWebsocketParams({
      audioInEnabled: true,
      audioOutEnabled: true,
      serializer: new ProtobufFrameSerializer(),
    }),
  );

  // Flux detects the start and end of each turn, so there is no VAD analyzer
  // and no Smart Turn model to load: the user aggregator follows Flux.
  const stt = new WorkersAIFluxSTTService({ ai: env.AI });

  const llm = new WorkersAILLMService({
    accountId: env.CLOUDFLARE_ACCOUNT_ID,
    apiToken: env.CLOUDFLARE_API_TOKEN,
    settings: new WorkersAILLMService.Settings({
      model: "@cf/meta/llama-3.3-70b-instruct-fp8-fast",
      systemInstruction:
        "You are a helpful assistant in a voice conversation. Answer briefly, in plain sentences.",
    }),
  });

  const tts = new WorkersAIAuraTTSService({
    ai: env.AI,
    sampleRate: OUTPUT_SAMPLE_RATE,
    settings: new WorkersAIAuraTTSService.Settings({ voice: "luna" }),
  });

  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, {
    cancelOnIdleTimeout: false,
    params: new PipelineParams({
      audioInSampleRate: INPUT_SAMPLE_RATE,
      audioOutSampleRate: OUTPUT_SAMPLE_RATE,
      enableMetrics: true,
      enableUsageMetrics: true,
    }),
  });

  const runner = new WorkerRunner({ handleSigint: false, handleSigterm: false });
  await runner.addWorkers(worker);

  transport.eventHandler("on_client_connected", async () => {
    context.addMessage({ role: "developer", content: "Please introduce yourself to the user." });
    await worker.queueFrames([new LLMRunFrame()]);
  });
  transport.eventHandler("on_client_disconnected", async () => {
    await runner.cancel();
  });

  await runner.run();
}

import { DurableObject } from "cloudflare:workers";
import { websocketFetchHandler } from "@pipedog/transport-websocket/fetch";

export class VoiceSession extends DurableObject<Env> {
  readonly #handler = websocketFetchHandler((connection) => runVoiceAgent(connection, this.env));

  override fetch(request: Request): Promise<Response> {
    return this.#handler(request);
  }
}

export default {
  async fetch(request: Request, env: Env & { VOICE_SESSION: DurableObjectNamespace<VoiceSession> }) {
    const id = new URL(request.url).pathname.split("/")[3];
    return env.VOICE_SESSION.getByName(id).fetch(request);
  },
};