LiveKit integration

import asyncio
import base64
import json
import os

import aiohttp
from dotenv import load_dotenv
from elevenlabs import AsyncElevenLabs
from livekit import agents, rtc
from livekit.agents import JobContext, WorkerOptions, cli

load_dotenv()

elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"]

USER_INPUT_RATE = 16000
AGENT_OUTPUT_RATE = 24000

async def signed_url() -> str:
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
        agent_id=SPEECH_ENGINE_ID,
    )
    return response.signed_url

async def entrypoint(ctx: JobContext):
    el_ws_ready: asyncio.Future[aiohttp.ClientWebSocketResponse] = (
        asyncio.get_running_loop().create_future()
    )

    async def pump_user_audio(track: rtc.Track):
        el_ws = await el_ws_ready
        stream = rtc.AudioStream(
            track, sample_rate=USER_INPUT_RATE, num_channels=1,
        )
        async for event in stream:
            payload = base64.b64encode(bytes(event.frame.data)).decode()
            await el_ws.send_str(json.dumps({"user_audio_chunk": payload}))

    # Register the subscriber BEFORE ctx.connect() so we don't miss tracks
    # that get auto-subscribed during the connection handshake.
    @ctx.room.on("track_subscribed")
    def on_track_subscribed(track, publication, participant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return
        if participant.identity == ctx.room.local_participant.identity:
            return
        asyncio.create_task(pump_user_audio(track))

    await ctx.connect()

    # Publish a track for the agent's synthesized audio.
    source = rtc.AudioSource(sample_rate=AGENT_OUTPUT_RATE, num_channels=1)
    track = rtc.LocalAudioTrack.create_audio_track("elevenlabs-agent", source)
    await ctx.room.local_participant.publish_track(
        track,
        rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE),
    )

    # Open the Speech Engine conversation WebSocket.
    http = aiohttp.ClientSession()
    el_ws = await http.ws_connect(await signed_url())
    await el_ws.send_str(json.dumps({"type": "conversation_initiation_client_data"}))
    el_ws_ready.set_result(el_ws)

    async def el_to_room():
        async for msg in el_ws:
            if msg.type != aiohttp.WSMsgType.TEXT:
                continue
            event = json.loads(msg.data)
            etype = event.get("type")
            if etype == "audio":
                pcm = base64.b64decode(event["audio_event"]["audio_base_64"])
                samples_per_channel = len(pcm) // 2
                frame = rtc.AudioFrame(
                    pcm, AGENT_OUTPUT_RATE, 1, samples_per_channel,
                )
                await source.capture_frame(frame)
            elif etype == "interruption":
                source.clear_queue()
            elif etype == "ping":
                event_id = event.get("ping_event", {}).get("event_id")
                await el_ws.send_str(json.dumps({
                    "type": "pong", "event_id": event_id,
                }))

    pump_task = asyncio.create_task(el_to_room())

    async def cleanup():
        pump_task.cancel()
        await el_ws.close()
        await http.close()

    ctx.add_shutdown_callback(cleanup)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        agent_name="elevenlabs-bridge",
    ))

import {
  type JobContext,
  WorkerOptions,
  cli,
  defineAgent,
} from "@livekit/agents";
import {
  AudioFrame,
  AudioSource,
  AudioStream,
  LocalAudioTrack,
  RoomEvent,
  TrackKind,
  TrackPublishOptions,
  TrackSource,
} from "@livekit/rtc-node";
import { ElevenLabsClient } from "@elevenlabs/elevenlabs-js";
import WebSocket from "ws";
import { fileURLToPath } from "node:url";
import "dotenv/config";

const elevenlabs = new ElevenLabsClient({
  apiKey: process.env.ELEVENLABS_API_KEY,
});
const SPEECH_ENGINE_ID = process.env.SPEECH_ENGINE_ID!;

const USER_INPUT_RATE = 16000;
const AGENT_OUTPUT_RATE = 24000;

async function signedUrl(): Promise<string> {
  const response = await elevenlabs.conversationalAi.conversations.getSignedUrl({
    agentId: SPEECH_ENGINE_ID,
  });
  return response.signedUrl;
}

export default defineAgent({
  entry: async (ctx: JobContext) => {
    let resolveElReady: (ws: WebSocket) => void;
    const elReady = new Promise<WebSocket>((resolve) => {
      resolveElReady = resolve;
    });

    // Register the subscriber BEFORE ctx.connect() so we don't miss
    // tracks that get auto-subscribed during the connection handshake.
    ctx.room.on(RoomEvent.TrackSubscribed, (track, _pub, participant) => {
      if (track.kind !== TrackKind.KIND_AUDIO) return;
      if (participant.identity === ctx.room.localParticipant?.identity) return;

      (async () => {
        const ws = await elReady;
        const stream = new AudioStream(track, {
          sampleRate: USER_INPUT_RATE,
          numChannels: 1,
        });
        for await (const frame of stream) {
          const payload = Buffer.from(
            frame.data.buffer,
            frame.data.byteOffset,
            frame.data.byteLength,
          ).toString("base64");
          ws.send(JSON.stringify({ user_audio_chunk: payload }));
        }
      })();
    });

    await ctx.connect();

    const source = new AudioSource(AGENT_OUTPUT_RATE, 1);
    const track = LocalAudioTrack.createAudioTrack("elevenlabs-agent", source);
    const publishOptions = new TrackPublishOptions();
    publishOptions.source = TrackSource.SOURCE_MICROPHONE;
    await ctx.room.localParticipant!.publishTrack(track, publishOptions);

    const ws = new WebSocket(await signedUrl());
    await new Promise<void>((resolve, reject) => {
      ws.once("open", () => resolve());
      ws.once("error", reject);
    });
    ws.send(JSON.stringify({ type: "conversation_initiation_client_data" }));
    resolveElReady!(ws);

    // Serialize captureFrame calls — concurrent captures throw
    // InvalidState in the rtc-node native layer.
    let captureChain: Promise<unknown> = Promise.resolve();

    ws.on("message", (raw) => {
      const event = JSON.parse(raw.toString());
      if (event.type === "audio") {
        const pcm = Buffer.from(event.audio_event.audio_base_64, "base64");
        const samples = new Int16Array(
          pcm.buffer, pcm.byteOffset, pcm.byteLength / 2,
        );
        const frame = new AudioFrame(
          samples, AGENT_OUTPUT_RATE, 1, samples.length,
        );
        captureChain = captureChain
          .then(() => source.captureFrame(frame))
          .catch((err) => console.warn("captureFrame:", err.message));
      } else if (event.type === "interruption") {
        source.clearQueue();
      } else if (event.type === "ping") {
        ws.send(JSON.stringify({
          type: "pong", event_id: event.ping_event?.event_id,
        }));
      }
    });

    ctx.addShutdownCallback(async () => {
      ws.close();
    });
  },
});

cli.runApp(new WorkerOptions({
  agent: fileURLToPath(import.meta.url),
  agentName: "elevenlabs-bridge",
}));

python bridge.py dev

npx tsx bridge.mts dev