Avatar only (your own audio)

import { Room, TrackKind } from "@livekit/rtc-node";
 
const API = "https://api.avatargo.ai/v1";
async function call(method, path, body) {
  const res = await fetch(API + path, {
    method,
    headers: { authorization: `Bearer ${process.env.AVATARGO_API_KEY}`, ...(body && { "content-type": "application/json" }) },
    body: body && JSON.stringify(body),
  });
  const json = res.status === 204 ? null : await res.json();
  if (!res.ok) throw new Error(`${json.error.code}: ${json.error.message}`);
  return json;
}
 
// 1. A session for a face, with your server sending the audio. Give sessionToken to the page.
export function startSession(avatarId) {
  return call("POST", "/auth/session-token", { avatarId, sessionOptions: { audioPassthrough: true } });
}
 
// 3. Join the room as the sender, and wait until the avatar is ready.
export async function joinAsSender(sessionId) {
  const ingress = await call("POST", `/sessions/${sessionId}/audio-ingress`);
  const room = new Room();
  await room.connect(ingress.livekitUrl, ingress.token, { autoSubscribe: false });
  // The avatar calls this when it has finished speaking an utterance (or was interrupted).
  let finished = () => {};
  room.localParticipant.registerRpcMethod("lk.playback_finished", async () => (finished(), ""));
  // Joining is not readiness: audio sent before the avatar publishes its video is dropped.
  const ready = () => [...(room.remoteParticipants.get(ingress.avatarIdentity)?.trackPublications.values() ?? [])]
    .some((t) => t.kind === TrackKind.KIND_VIDEO);
  while (!ready()) await new Promise((r) => setTimeout(r, 250));
 
  // One stream per utterance: pcm_s16le, 24 kHz, mono. Closing it ends the utterance.
  // Resolves once the avatar has spoken it: sending is faster than speaking.
  const { topic, sampleRate, channels } = ingress.audio;
  async function say(pcm) {
    const spoken = new Promise((resolve) => (finished = resolve));
    const writer = await room.localParticipant.streamBytes({
      name: `utterance-${Date.now()}`, topic, destinationIdentities: [ingress.avatarIdentity],
      attributes: { sample_rate: String(sampleRate), num_channels: String(channels) },
    });
    for (let i = 0; i < pcm.length; i += 15_000) await writer.write(pcm.subarray(i, i + 15_000));
    await writer.close();
    await spoken;
  }
  return { room, say };
}
 
// Stop what the avatar is saying now; end the session.
export const interrupt = (sessionId) => call("POST", `/sessions/${sessionId}/interrupt`);
export async function end(sessionId, room) {
  await call("POST", `/sessions/${sessionId}/stop`);
  await room.disconnect();
}

import asyncio, json, os, time, urllib.request
from livekit import rtc
 
API = "https://api.avatargo.ai/v1"
 
def call(method, path, body=None):
    headers = {"Authorization": f"Bearer {os.environ['AVATARGO_API_KEY']}"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(API + path, method=method, headers=headers,
                                 data=json.dumps(body).encode() if body is not None else None)
    with urllib.request.urlopen(req) as res:  # an error raises HTTPError; its body is {"error": {"code", "message"}}
        raw = res.read()
        return json.loads(raw) if raw else None
 
# 1. A session for a face, with your server sending the audio. Give sessionToken to the page.
async def start_session(avatar_id):
    return await asyncio.to_thread(call, "POST", "/auth/session-token",
                                   {"avatarId": avatar_id, "sessionOptions": {"audioPassthrough": True}})
 
# 3. Join the room as the sender, and wait until the avatar is ready.
async def join_as_sender(session_id):
    ingress = await asyncio.to_thread(call, "POST", f"/sessions/{session_id}/audio-ingress")
    room = rtc.Room()
    await room.connect(ingress["livekitUrl"], ingress["token"], rtc.RoomOptions(auto_subscribe=False))
    # The avatar calls this when it has finished speaking an utterance (or was interrupted).
    finished = [asyncio.get_running_loop().create_future()]
    async def on_finished(_data):
        if not finished[0].done():
            finished[0].set_result(None)
        return ""
    room.local_participant.register_rpc_method("lk.playback_finished", on_finished)
    # Joining is not readiness: audio sent before the avatar publishes its video is dropped.
    def ready():
        p = room.remote_participants.get(ingress["avatarIdentity"])
        return p is not None and any(t.kind == rtc.TrackKind.KIND_VIDEO for t in p.track_publications.values())
    while not ready():
        await asyncio.sleep(0.25)
 
    # One stream per utterance: pcm_s16le, 24 kHz, mono. Closing it ends the utterance.
    # Returns once the avatar has spoken it: sending is faster than speaking.
    audio = ingress["audio"]
    async def say(pcm: bytes):
        finished[0] = asyncio.get_running_loop().create_future()
        writer = await room.local_participant.stream_bytes(
            f"utterance-{time.time()}", topic=audio["topic"], destination_identities=[ingress["avatarIdentity"]],
            attributes={"sample_rate": str(audio["sampleRate"]), "num_channels": str(audio["channels"])})
        for i in range(0, len(pcm), 15_000):
            await writer.write(pcm[i:i + 15_000])
        await writer.aclose()
        await finished[0]
    return room, say
 
# Stop what the avatar is saying now; end the session.
async def interrupt(session_id):
    await asyncio.to_thread(call, "POST", f"/sessions/{session_id}/interrupt")
 
async def end(session_id, room):
    await asyncio.to_thread(call, "POST", f"/sessions/{session_id}/stop")
    await room.disconnect()