Code Examples

code examples

Copy-paste examples for every Atlas API workflow. Jobs use the async pattern: submit → poll → download, or use webhooks when callback signing is configured.

Or jump straight to a working app

Offline Example App

Next.js · Text/Audio → Video · 3 generation modes

→

Realtime Example App

Next.js · Live Avatar + LLM + TTS

→

Language

Requirespip install requests

Base URLhttps://api.atlasv1.com

Every generation endpoint is async. You submit a job and get back a job_id. Then either poll GET /v1/jobs/{id} until complete, or send X-Callback-URL to get notified via webhook when Atlas signing is configured. Download the result from GET /v1/jobs/{id}/result.

01beginner

Quickstart

The simplest possible call, submit an audio file and a face image, poll for completion, then download the MP4.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Submit job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", open("speech.mp3", "rb"), "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")

# Step 2, Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()

    if status["status"] == "completed":
        break
    elif status["status"] == "failed":
        raise Exception(f"Job failed: {status['error']}")

    print(f"Status: {status['status']}, queue: {status.get('queue_position', '?')}...")
    time.sleep(2)

# Step 3, Download result
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

02beginner

Full Pipeline

Two-step flow: generate speech with ElevenLabs (or any TTS), then feed it into avatar generation. TTS returns audio directly; only the video step is an async job.

Step 1: Text → speech audio via ElevenLabs (or any TTS provider)
Step 2: Audio + Image → MP4 video via /v1/generate

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Generate speech audio with ElevenLabs
tts_res = requests.post(
    "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
    headers={
        "xi-api-key": "YOUR_ELEVENLABS_KEY",
        "Content-Type": "application/json",
    },
    json={
        "text": "Hey! Welcome to Atlas. We turn any face into a talking avatar.",
        "model_id": "eleven_multilingual_v2",
    },
)
audio_bytes = tts_res.content

# Step 2, Generate avatar video
video_job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Video job submitted: {video_job['job_id']}")

# Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{video_job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] == "completed":
        break
    elif status["status"] == "failed":
        raise Exception(f"Job failed: {status['error']}")
    time.sleep(2)

# Download the video
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{video_job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

03beginner

Text → Video (ElevenLabs)

Two-step convenience flow: call ElevenLabs TTS for audio, then submit it with a face image to Atlas for avatar video generation. No built-in TTS needed.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Generate speech audio with ElevenLabs
tts_res = requests.post(
    "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
    headers={
        "xi-api-key": "YOUR_ELEVENLABS_KEY",
        "Content-Type": "application/json",
    },
    json={
        "text": "Hello! This video was generated from just text and a photo.",
        "model_id": "eleven_multilingual_v2",
    },
)
audio_bytes = tts_res.content

# Step 2, Submit avatar video job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")

# Step 3, Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

if status["status"] == "failed":
    raise Exception(f"Job failed: {status['error']}")

# Step 4, Download MP4
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

External TTS Providers

05intermediate

ElevenLabs + Atlas

Use ElevenLabs for voice generation, then submit the audio to Atlas for avatar video. Same async poll pattern.

Also requirespip install elevenlabs

from elevenlabs import ElevenLabs
import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Generate speech with ElevenLabs
client = ElevenLabs(api_key="YOUR_ELEVENLABS_KEY")
audio = client.text_to_speech.convert(
    text="This voice is from ElevenLabs, but the avatar is Atlas.",
    voice_id="JBFqnCBsd6RMkjVDRZzb",
    output_format="mp3_44100_128",
)
audio_bytes = b"".join(audio)

# Submit avatar job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

# Poll
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

# Download
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

06intermediate

OpenAI TTS + Atlas

Use OpenAI's text-to-speech for voice generation, then submit to Atlas for the avatar video.

Also requirespip install openai

from openai import OpenAI
import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Generate speech with OpenAI TTS
speech = OpenAI().audio.speech.create(
    model="tts-1-hd",
    voice="nova",
    input="This voice is from OpenAI, but the avatar is Atlas.",
)

# Submit avatar job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", speech.content, "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

# Poll
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

# Download
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

Webhooks

09intermediate

Webhook Receiver

Skip polling where webhook signing is configured: send X-Callback-URL when submitting a job and Atlas can POST the result to your server when it's done. Polling remains supported for every job.

Submit with webhook

import requests

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Submit a video job with a webhook callback
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers={**headers, "X-Callback-URL": "https://yourapp.com/webhook/atlas"},
    files={
        "audio": ("speech.mp3", open("speech.mp3", "rb"), "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")
print("No polling needed, webhook will fire when complete")

Receive the webhook, Flask

from flask import Flask, request, jsonify
import hmac, hashlib, requests as http

app = Flask(__name__)
WEBHOOK_SECRET = "YOUR_API_KEY"

@app.route("/webhook/atlas", methods=["POST"])
def handle_atlas_webhook():
    sig = request.headers.get("X-Atlas-Signature", "")
    ts = request.headers.get("X-Atlas-Timestamp", "")
    body = request.get_data()

    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{ts}.".encode() + body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(sig, expected):
        return jsonify({"error": "invalid signature"}), 403

    payload = request.get_json()
    job_id = payload["job_id"]
    event = payload["event"]

    if event == "job.completed":
        result = http.get(
            payload["result_url"],
            headers={"Authorization": f"Bearer {WEBHOOK_SECRET}"},
        ).json()
        video = http.get(result["url"])
        with open(f"{job_id}.mp4", "wb") as f:
            f.write(video.content)
        print(f"Saved {job_id}.mp4")

    elif event == "job.failed":
        print(f"Job {job_id} failed: {payload['error_code']}")

    return jsonify({"received": True})

if __name__ == "__main__":
    app.run(port=8080)

10advanced

Webhook + FastAPI

A production-ready async webhook receiver with signature verification, background downloading, and proper error handling.

Requirespip install fastapi uvicorn httpx

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import hmac, hashlib, httpx, asyncio

app = FastAPI()
WEBHOOK_SECRET = "YOUR_API_KEY"

def verify_signature(body: bytes, sig: str, ts: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{ts}.".encode() + body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(sig, expected)

async def download_result(result_url: str, job_id: str):
    async with httpx.AsyncClient(timeout=60) as client:
        r = await client.get(
            result_url,
            headers={"Authorization": f"Bearer {WEBHOOK_SECRET}"},
        )
        data = r.json()
        file_r = await client.get(data["url"])
        ext = "mp4" if "video" in file_r.headers.get("content-type", "") else "mp3"
        with open(f"outputs/{job_id}.{ext}", "wb") as f:
            f.write(file_r.content)
        print(f"Downloaded {job_id}.{ext} ({len(file_r.content)} bytes)")

@app.post("/webhook/atlas")
async def atlas_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-Atlas-Signature", "")
    ts = request.headers.get("X-Atlas-Timestamp", "")

    if not verify_signature(body, sig, ts):
        raise HTTPException(status_code=403, detail="Invalid signature")

    payload = await request.json()
    job_id = payload["job_id"]
    event = payload["event"]

    if event == "job.completed":
        asyncio.create_task(download_result(payload["result_url"], job_id))
        return {"status": "downloading", "job_id": job_id}

    elif event == "job.failed":
        print(f"Job {job_id} failed: {payload.get('error_code')}")
        return {"status": "noted", "job_id": job_id}

    return {"status": "ignored"}

# Run: uvicorn webhook_server:app --port 8080

Realtime Avatar

11beginner

React Hook, `@northmodellabs/atlas-react`

Connect a live avatar in React with a single hook call. The useAtlasSession() hook handles all LiveKit wiring, room lifecycle, video/audio tracks, microphone, transcriptions, and cleanup.

Installnpm install @northmodellabs/atlas-react livekit-client

Frontend, React Component

app/avatar.tsx

import { useAtlasSession } from "@northmodellabs/atlas-react";
import { useRef, useState } from "react";

export default function AvatarPage() {
  const [faceFile, setFaceFile] = useState<File | null>(null);

  const session = useAtlasSession({
    createSession: async (face) => {
      const form = new FormData();
      if (face) form.append("face", face);
      const res = await fetch("/api/session", { method: "POST", body: form });
      return res.json();
    },
    deleteSession: async (id) => {
      await fetch(`/api/session/${id}`, { method: "DELETE" });
    },
  });

  return (
    <div>
      <div ref={session.videoRef} style={{ width: 512, height: 512 }} />

      {session.status === "idle" && (
        <div>
          <input
            type="file"
            accept="image/*"
            onChange={(e) => setFaceFile(e.target.files?.[0] ?? null)}
          />
          <button onClick={() => session.connect(faceFile)}>
            Start Session
          </button>
        </div>
      )}

      {session.status === "connecting" && <p>Connecting...</p>}

      {session.status === "connected" && (
        <div>
          <button onClick={() => session.setMicEnabled(session.muted)}>
            {session.muted ? "Unmute" : "Mute"}
          </button>
          <input
            type="range" min={0} max={100}
            value={session.volume}
            onChange={(e) => session.setVolume(Number(e.target.value))}
          />
          <span>Realtime session active</span>
          <button onClick={session.disconnect}>End Session</button>
        </div>
      )}

      {session.error && <p style={{ color: "red" }}>{session.error}</p>}

      <div>
        {session.messages.filter(m => m.final).map((msg) => (
          <p key={msg.id}><b>{msg.role}:</b> {msg.text}</p>
        ))}
      </div>
    </div>
  );
}

Backend, API Route (Next.js)

app/api/session/route.ts

const API_KEY = process.env.ATLAS_API_KEY!;
const API_URL = process.env.ATLAS_API_URL || "https://api.atlasv1.com";

export async function POST(req: Request) {
  const form = await req.formData();
  const res = await fetch(`${API_URL}/v1/realtime/session`, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}` },
    body: form,
  });
  return Response.json(await res.json(), { status: res.status });
}

app/api/session/[id]/route.ts

const API_KEY = process.env.ATLAS_API_KEY!;
const API_URL = process.env.ATLAS_API_URL || "https://api.atlasv1.com";

export async function DELETE(
  _req: Request,
  { params }: { params: { id: string } }
) {
  const res = await fetch(
    `${API_URL}/v1/realtime/session/${params.id}`,
    {
      method: "DELETE",
      headers: { Authorization: `Bearer ${API_KEY}` },
    }
  );
  return Response.json(await res.json(), { status: res.status });
}

Full docs: See the complete API reference for useAtlasSession() on npm, or in the API docs.

11.5intermediate

React Hook, Passthrough Mode

Bring your own LLM + TTS and use publishAudio() to send audio to the avatar for lip-sync. Atlas provides the GPU compute and WebRTC video.

app/avatar-passthrough.tsx

import { useAtlasSession } from "@northmodellabs/atlas-react";
import { useState } from "react";

export default function PassthroughPage() {
  const [faceFile, setFaceFile] = useState<File | null>(null);
  const [input, setInput] = useState("");

  const session = useAtlasSession({
    createSession: async (face) => {
      const form = new FormData();
      if (face) form.append("face", face);
      form.append("mode", "passthrough"); // Use passthrough mode
      const res = await fetch("/api/session", { method: "POST", body: form });
      return res.json();
    },
    deleteSession: async (id) => {
      await fetch(`/api/session/${id}`, { method: "DELETE" });
    },
  });

  async function handleSend() {
    if (!input.trim()) return;
    const text = input;
    setInput("");

    // Call your own LLM + TTS backend
    const res = await fetch("/api/chat", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ text }),
    });
    const { audio } = await res.json();

    // Publish audio to room, avatar lip-syncs, mic auto-mutes
    if (audio) {
      await session.publishAudio(audio);
    }
  }

  return (
    <div>
      <div ref={session.videoRef} style={{ width: 512, height: 512 }} />

      {session.status === "idle" && (
        <div>
          <input type="file" accept="image/*"
            onChange={(e) => setFaceFile(e.target.files?.[0] ?? null)} />
          <button onClick={() => session.connect(faceFile)}>Start</button>
        </div>
      )}

      {session.status === "connected" && (
        <div>
          <input value={input} onChange={(e) => setInput(e.target.value)}
            onKeyDown={(e) => e.key === "Enter" && handleSend()} />
          <button onClick={handleSend}>Send</button>
          <button onClick={session.disconnect}>End</button>
        </div>
      )}
    </div>
  );
}

publishAudio() accepts a base64 string, Blob, or ArrayBuffer. It automatically mutes your mic during playback, publishes the audio as a LiveKit track for avatar lip-sync, plays it locally, and cleans up when done. See API docs for full details.

12intermediate

Manual LiveKit Integration

If you prefer full control, connect to a realtime session using the LiveKit client SDK directly. Your backend creates the session and passes the token to the client.

import requests, json

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Create a realtime session
session = requests.post(
    "https://api.atlasv1.com/v1/realtime/session",
    headers={**headers, "Content-Type": "application/json"},
    json={"face_url": "https://example.com/face.jpg"},
).json()

print(f"Session ID: {session['session_id']}")
print(f"LiveKit URL: {session['livekit_url']}")
print(f"Token: {session['token'][:20]}...")
print(f"Room: {session['room']}")

# Connect with LiveKit Python SDK or pass token to a frontend client
# pip install livekit
from livekit import rtc

room = rtc.Room()

@room.on("track_subscribed")
def on_track(track, publication, participant):
    if track.kind == rtc.TrackKind.KIND_VIDEO:
        print(f"Video track received from {participant.identity}")
    elif track.kind == rtc.TrackKind.KIND_AUDIO:
        print(f"Audio track received from {participant.identity}")

await room.connect(session["livekit_url"], session["token"])
print("Connected!")

# When done
await room.disconnect()
requests.delete(
    f"https://api.atlasv1.com/v1/realtime/session/{session['session_id']}",
    headers=headers,
)

Simpler option: Use @northmodellabs/atlas-react to replace all this boilerplate with a single useAtlasSession() hook call.

Advanced Patterns

07advanced

Batch Processing

Generate multiple avatar videos from a list of scripts. Call ElevenLabs for each script (returns audio directly), then submit all video jobs to Atlas and poll in parallel.

Concurrency tip: Since jobs run server-side, you can submit all of them upfront, then poll in parallel. No need to wait for one to finish before submitting the next.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

scripts = [
    "Welcome to our product demo.",
    "Here's how the dashboard works.",
    "Thanks for watching! Try it free.",
]


def wait_for_job(job_id):
    while True:
        s = requests.get(
            f"https://api.atlasv1.com/v1/jobs/{job_id}",
            headers=headers,
        ).json()
        if s["status"] == "completed":
            return s
        elif s["status"] == "failed":
            raise Exception(f"Job {job_id} failed: {s['error']}")
        time.sleep(2)


# Step 1, Generate all speech audio with ElevenLabs
audio_clips = []
for text in scripts:
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": "YOUR_ELEVENLABS_KEY",
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
    )
    audio_clips.append(tts_res.content)
    print(f"TTS complete for: {text[:30]}...")
print(f"All {len(scripts)} TTS clips generated")

# Step 2, Submit all video jobs at once
video_jobs = []
for i, audio_bytes in enumerate(audio_clips):
    job = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers=headers,
        files={
            "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
            "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
        },
    ).json()
    video_jobs.append(job["job_id"])
    print(f"Video job {i + 1} submitted: {job['job_id']}")

# Step 3, Wait and download all videos
for i, vid_id in enumerate(video_jobs):
    wait_for_job(vid_id)
    result = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{vid_id}/result",
        headers=headers,
    ).json()
    video = requests.get(result["url"])
    with open(f"clip_{i + 1}.mp4", "wb") as f:
        f.write(video.content)
    print(f"Clip {i + 1}/{len(scripts)} saved")

08advanced

Production-Ready

A reusable function with proper error handling, rate limit retries, progress logging, and streaming file downloads. Supports both polling and webhook modes.

import requests
import time


def poll_job(job_id, headers, timeout=600, interval=2):
    """Poll a job until it completes, fails, or times out."""
    start = time.time()
    while time.time() - start < timeout:
        status = requests.get(
            f"https://api.atlasv1.com/v1/jobs/{job_id}",
            headers=headers,
        ).json()

        if status["status"] == "completed":
            return status
        elif status["status"] == "failed":
            raise Exception(
                f"Job {job_id} failed [{status.get('error_code')}]: {status.get('error')}"
            )

        time.sleep(interval)

    raise TimeoutError(f"Job {job_id} did not complete within {timeout}s")


def download_result(job_id, output_path, headers):
    """Download a completed job's output to disk."""
    result = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job_id}/result",
        headers=headers,
    ).json()
    file = requests.get(result["url"], stream=True)
    file.raise_for_status()

    with open(output_path, "wb") as f:
        for chunk in file.iter_content(chunk_size=8192):
            f.write(chunk)


def generate_avatar(text, image_path, output_path, api_key, elevenlabs_key):
    """Text + image → avatar video with ElevenLabs TTS and retry logic."""
    headers = {"Authorization": f"Bearer {api_key}"}

    # Step 1, Generate speech audio with ElevenLabs
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": elevenlabs_key,
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
        timeout=30,
    )
    tts_res.raise_for_status()
    audio_bytes = tts_res.content
    print(f"TTS complete, {len(audio_bytes)} bytes")

    # Step 2, Submit video job to Atlas
    vid_res = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers=headers,
        files={
            "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
            "image": (image_path, open(image_path, "rb"), "image/jpeg"),
        },
        timeout=30,
    )

    if vid_res.status_code == 429:
        wait = vid_res.json().get("retry_after_seconds", 10)
        print(f"Rate limited, waiting {wait}s")
        time.sleep(wait)
        return generate_avatar(text, image_path, output_path, api_key, elevenlabs_key)

    vid_res.raise_for_status()
    vid_job = vid_res.json()
    print(f"Video job: {vid_job['job_id']}")

    vid_status = poll_job(vid_job["job_id"], headers)
    print("Video complete")

    download_result(vid_job["job_id"], output_path, headers)

    return {
        "status": vid_status["status"],
    }


result = generate_avatar(
    text="Hello from Atlas!",
    image_path="face.jpg",
    output_path="output.mp4",
    api_key="YOUR_API_KEY",
    elevenlabs_key="YOUR_ELEVENLABS_KEY",
)
print(result)

Webhook alternative: For server-to-server workflows, send X-Callback-URL with the job request. Atlas can POST the result to your endpoint when webhook signing is configured; polling remains the fallback. See examples 9–10 for receiver code.

import requests

def generate_avatar_webhook(text, image_path, callback_url, api_key, elevenlabs_key):
    headers = {"Authorization": f"Bearer {api_key}"}

    # Generate speech audio with ElevenLabs
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": elevenlabs_key,
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
    )
    tts_res.raise_for_status()

    # Submit video job with webhook
    vid = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers={**headers, "X-Callback-URL": callback_url},
        files={
            "audio": ("speech.mp3", tts_res.content, "audio/mpeg"),
            "image": (image_path, open(image_path, "rb"), "image/jpeg"),
        },
        timeout=30,
    )
    vid.raise_for_status()
    print(f"Video job: {vid.json()['job_id']}, webhook will fire on completion")
    return vid.json()


generate_avatar_webhook(
    text="Hello from Atlas!",
    image_path="face.jpg",
    callback_url="https://yourapp.com/webhook/atlas",
    api_key="YOUR_API_KEY",
    elevenlabs_key="YOUR_ELEVENLABS_KEY",
)

Code Examples

code examples

Copy-paste examples for every Atlas API workflow. Jobs use the async pattern: submit → poll → download, or use webhooks when callback signing is configured.

Or jump straight to a working app

Offline Example App

Next.js · Text/Audio → Video · 3 generation modes

→

Realtime Example App

Next.js · Live Avatar + LLM + TTS

→

Language

Requirespip install requests

Base URLhttps://api.atlasv1.com

01beginner

Quickstart

The simplest possible call, submit an audio file and a face image, poll for completion, then download the MP4.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Submit job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", open("speech.mp3", "rb"), "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")

# Step 2, Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()

    if status["status"] == "completed":
        break
    elif status["status"] == "failed":
        raise Exception(f"Job failed: {status['error']}")

    print(f"Status: {status['status']}, queue: {status.get('queue_position', '?')}...")
    time.sleep(2)

# Step 3, Download result
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

02beginner

Full Pipeline

Two-step flow: generate speech with ElevenLabs (or any TTS), then feed it into avatar generation. TTS returns audio directly; only the video step is an async job.

Step 1: Text → speech audio via ElevenLabs (or any TTS provider)
Step 2: Audio + Image → MP4 video via /v1/generate

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Generate speech audio with ElevenLabs
tts_res = requests.post(
    "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
    headers={
        "xi-api-key": "YOUR_ELEVENLABS_KEY",
        "Content-Type": "application/json",
    },
    json={
        "text": "Hey! Welcome to Atlas. We turn any face into a talking avatar.",
        "model_id": "eleven_multilingual_v2",
    },
)
audio_bytes = tts_res.content

# Step 2, Generate avatar video
video_job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Video job submitted: {video_job['job_id']}")

# Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{video_job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] == "completed":
        break
    elif status["status"] == "failed":
        raise Exception(f"Job failed: {status['error']}")
    time.sleep(2)

# Download the video
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{video_job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

03beginner

Text → Video (ElevenLabs)

Two-step convenience flow: call ElevenLabs TTS for audio, then submit it with a face image to Atlas for avatar video generation. No built-in TTS needed.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Step 1, Generate speech audio with ElevenLabs
tts_res = requests.post(
    "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
    headers={
        "xi-api-key": "YOUR_ELEVENLABS_KEY",
        "Content-Type": "application/json",
    },
    json={
        "text": "Hello! This video was generated from just text and a photo.",
        "model_id": "eleven_multilingual_v2",
    },
)
audio_bytes = tts_res.content

# Step 2, Submit avatar video job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")

# Step 3, Poll until complete
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

if status["status"] == "failed":
    raise Exception(f"Job failed: {status['error']}")

# Step 4, Download MP4
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

print("Done, saved output.mp4")

External TTS Providers

05intermediate

ElevenLabs + Atlas

Use ElevenLabs for voice generation, then submit the audio to Atlas for avatar video. Same async poll pattern.

Also requirespip install elevenlabs

from elevenlabs import ElevenLabs
import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Generate speech with ElevenLabs
client = ElevenLabs(api_key="YOUR_ELEVENLABS_KEY")
audio = client.text_to_speech.convert(
    text="This voice is from ElevenLabs, but the avatar is Atlas.",
    voice_id="JBFqnCBsd6RMkjVDRZzb",
    output_format="mp3_44100_128",
)
audio_bytes = b"".join(audio)

# Submit avatar job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", audio_bytes, "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

# Poll
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

# Download
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

06intermediate

OpenAI TTS + Atlas

Use OpenAI's text-to-speech for voice generation, then submit to Atlas for the avatar video.

Also requirespip install openai

from openai import OpenAI
import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Generate speech with OpenAI TTS
speech = OpenAI().audio.speech.create(
    model="tts-1-hd",
    voice="nova",
    input="This voice is from OpenAI, but the avatar is Atlas.",
)

# Submit avatar job
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers=headers,
    files={
        "audio": ("speech.mp3", speech.content, "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

# Poll
while True:
    status = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job['job_id']}",
        headers=headers,
    ).json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

# Download
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result",
    headers=headers,
).json()
video = requests.get(result["url"])

with open("output.mp4", "wb") as f:
    f.write(video.content)

Webhooks

09intermediate

Webhook Receiver

Skip polling where webhook signing is configured: send X-Callback-URL when submitting a job and Atlas can POST the result to your server when it's done. Polling remains supported for every job.

Submit with webhook

import requests

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Submit a video job with a webhook callback
job = requests.post(
    "https://api.atlasv1.com/v1/generate",
    headers={**headers, "X-Callback-URL": "https://yourapp.com/webhook/atlas"},
    files={
        "audio": ("speech.mp3", open("speech.mp3", "rb"), "audio/mp3"),
        "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
    },
).json()

print(f"Job submitted: {job['job_id']}")
print("No polling needed, webhook will fire when complete")

Receive the webhook, Flask

from flask import Flask, request, jsonify
import hmac, hashlib, requests as http

app = Flask(__name__)
WEBHOOK_SECRET = "YOUR_API_KEY"

@app.route("/webhook/atlas", methods=["POST"])
def handle_atlas_webhook():
    sig = request.headers.get("X-Atlas-Signature", "")
    ts = request.headers.get("X-Atlas-Timestamp", "")
    body = request.get_data()

    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{ts}.".encode() + body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(sig, expected):
        return jsonify({"error": "invalid signature"}), 403

    payload = request.get_json()
    job_id = payload["job_id"]
    event = payload["event"]

    if event == "job.completed":
        result = http.get(
            payload["result_url"],
            headers={"Authorization": f"Bearer {WEBHOOK_SECRET}"},
        ).json()
        video = http.get(result["url"])
        with open(f"{job_id}.mp4", "wb") as f:
            f.write(video.content)
        print(f"Saved {job_id}.mp4")

    elif event == "job.failed":
        print(f"Job {job_id} failed: {payload['error_code']}")

    return jsonify({"received": True})

if __name__ == "__main__":
    app.run(port=8080)

10advanced

Webhook + FastAPI

A production-ready async webhook receiver with signature verification, background downloading, and proper error handling.

Requirespip install fastapi uvicorn httpx

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import hmac, hashlib, httpx, asyncio

app = FastAPI()
WEBHOOK_SECRET = "YOUR_API_KEY"

def verify_signature(body: bytes, sig: str, ts: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{ts}.".encode() + body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(sig, expected)

async def download_result(result_url: str, job_id: str):
    async with httpx.AsyncClient(timeout=60) as client:
        r = await client.get(
            result_url,
            headers={"Authorization": f"Bearer {WEBHOOK_SECRET}"},
        )
        data = r.json()
        file_r = await client.get(data["url"])
        ext = "mp4" if "video" in file_r.headers.get("content-type", "") else "mp3"
        with open(f"outputs/{job_id}.{ext}", "wb") as f:
            f.write(file_r.content)
        print(f"Downloaded {job_id}.{ext} ({len(file_r.content)} bytes)")

@app.post("/webhook/atlas")
async def atlas_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-Atlas-Signature", "")
    ts = request.headers.get("X-Atlas-Timestamp", "")

    if not verify_signature(body, sig, ts):
        raise HTTPException(status_code=403, detail="Invalid signature")

    payload = await request.json()
    job_id = payload["job_id"]
    event = payload["event"]

    if event == "job.completed":
        asyncio.create_task(download_result(payload["result_url"], job_id))
        return {"status": "downloading", "job_id": job_id}

    elif event == "job.failed":
        print(f"Job {job_id} failed: {payload.get('error_code')}")
        return {"status": "noted", "job_id": job_id}

    return {"status": "ignored"}

# Run: uvicorn webhook_server:app --port 8080

Realtime Avatar

11beginner

React Hook, `@northmodellabs/atlas-react`

Connect a live avatar in React with a single hook call. The useAtlasSession() hook handles all LiveKit wiring, room lifecycle, video/audio tracks, microphone, transcriptions, and cleanup.

Installnpm install @northmodellabs/atlas-react livekit-client

Frontend, React Component

app/avatar.tsx

import { useAtlasSession } from "@northmodellabs/atlas-react";
import { useRef, useState } from "react";

export default function AvatarPage() {
  const [faceFile, setFaceFile] = useState<File | null>(null);

  const session = useAtlasSession({
    createSession: async (face) => {
      const form = new FormData();
      if (face) form.append("face", face);
      const res = await fetch("/api/session", { method: "POST", body: form });
      return res.json();
    },
    deleteSession: async (id) => {
      await fetch(`/api/session/${id}`, { method: "DELETE" });
    },
  });

  return (
    <div>
      <div ref={session.videoRef} style={{ width: 512, height: 512 }} />

      {session.status === "idle" && (
        <div>
          <input
            type="file"
            accept="image/*"
            onChange={(e) => setFaceFile(e.target.files?.[0] ?? null)}
          />
          <button onClick={() => session.connect(faceFile)}>
            Start Session
          </button>
        </div>
      )}

      {session.status === "connecting" && <p>Connecting...</p>}

      {session.status === "connected" && (
        <div>
          <button onClick={() => session.setMicEnabled(session.muted)}>
            {session.muted ? "Unmute" : "Mute"}
          </button>
          <input
            type="range" min={0} max={100}
            value={session.volume}
            onChange={(e) => session.setVolume(Number(e.target.value))}
          />
          <span>Realtime session active</span>
          <button onClick={session.disconnect}>End Session</button>
        </div>
      )}

      {session.error && <p style={{ color: "red" }}>{session.error}</p>}

      <div>
        {session.messages.filter(m => m.final).map((msg) => (
          <p key={msg.id}><b>{msg.role}:</b> {msg.text}</p>
        ))}
      </div>
    </div>
  );
}

Backend, API Route (Next.js)

app/api/session/route.ts

const API_KEY = process.env.ATLAS_API_KEY!;
const API_URL = process.env.ATLAS_API_URL || "https://api.atlasv1.com";

export async function POST(req: Request) {
  const form = await req.formData();
  const res = await fetch(`${API_URL}/v1/realtime/session`, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}` },
    body: form,
  });
  return Response.json(await res.json(), { status: res.status });
}

app/api/session/[id]/route.ts

const API_KEY = process.env.ATLAS_API_KEY!;
const API_URL = process.env.ATLAS_API_URL || "https://api.atlasv1.com";

export async function DELETE(
  _req: Request,
  { params }: { params: { id: string } }
) {
  const res = await fetch(
    `${API_URL}/v1/realtime/session/${params.id}`,
    {
      method: "DELETE",
      headers: { Authorization: `Bearer ${API_KEY}` },
    }
  );
  return Response.json(await res.json(), { status: res.status });
}

Full docs: See the complete API reference for useAtlasSession() on npm, or in the API docs.

11.5intermediate

React Hook, Passthrough Mode

Bring your own LLM + TTS and use publishAudio() to send audio to the avatar for lip-sync. Atlas provides the GPU compute and WebRTC video.

app/avatar-passthrough.tsx

import { useAtlasSession } from "@northmodellabs/atlas-react";
import { useState } from "react";

export default function PassthroughPage() {
  const [faceFile, setFaceFile] = useState<File | null>(null);
  const [input, setInput] = useState("");

  const session = useAtlasSession({
    createSession: async (face) => {
      const form = new FormData();
      if (face) form.append("face", face);
      form.append("mode", "passthrough"); // Use passthrough mode
      const res = await fetch("/api/session", { method: "POST", body: form });
      return res.json();
    },
    deleteSession: async (id) => {
      await fetch(`/api/session/${id}`, { method: "DELETE" });
    },
  });

  async function handleSend() {
    if (!input.trim()) return;
    const text = input;
    setInput("");

    // Call your own LLM + TTS backend
    const res = await fetch("/api/chat", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ text }),
    });
    const { audio } = await res.json();

    // Publish audio to room, avatar lip-syncs, mic auto-mutes
    if (audio) {
      await session.publishAudio(audio);
    }
  }

  return (
    <div>
      <div ref={session.videoRef} style={{ width: 512, height: 512 }} />

      {session.status === "idle" && (
        <div>
          <input type="file" accept="image/*"
            onChange={(e) => setFaceFile(e.target.files?.[0] ?? null)} />
          <button onClick={() => session.connect(faceFile)}>Start</button>
        </div>
      )}

      {session.status === "connected" && (
        <div>
          <input value={input} onChange={(e) => setInput(e.target.value)}
            onKeyDown={(e) => e.key === "Enter" && handleSend()} />
          <button onClick={handleSend}>Send</button>
          <button onClick={session.disconnect}>End</button>
        </div>
      )}
    </div>
  );
}

12intermediate

Manual LiveKit Integration

If you prefer full control, connect to a realtime session using the LiveKit client SDK directly. Your backend creates the session and passes the token to the client.

import requests, json

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Create a realtime session
session = requests.post(
    "https://api.atlasv1.com/v1/realtime/session",
    headers={**headers, "Content-Type": "application/json"},
    json={"face_url": "https://example.com/face.jpg"},
).json()

print(f"Session ID: {session['session_id']}")
print(f"LiveKit URL: {session['livekit_url']}")
print(f"Token: {session['token'][:20]}...")
print(f"Room: {session['room']}")

# Connect with LiveKit Python SDK or pass token to a frontend client
# pip install livekit
from livekit import rtc

room = rtc.Room()

@room.on("track_subscribed")
def on_track(track, publication, participant):
    if track.kind == rtc.TrackKind.KIND_VIDEO:
        print(f"Video track received from {participant.identity}")
    elif track.kind == rtc.TrackKind.KIND_AUDIO:
        print(f"Audio track received from {participant.identity}")

await room.connect(session["livekit_url"], session["token"])
print("Connected!")

# When done
await room.disconnect()
requests.delete(
    f"https://api.atlasv1.com/v1/realtime/session/{session['session_id']}",
    headers=headers,
)

Simpler option: Use @northmodellabs/atlas-react to replace all this boilerplate with a single useAtlasSession() hook call.

Advanced Patterns

07advanced

Batch Processing

Generate multiple avatar videos from a list of scripts. Call ElevenLabs for each script (returns audio directly), then submit all video jobs to Atlas and poll in parallel.

Concurrency tip: Since jobs run server-side, you can submit all of them upfront, then poll in parallel. No need to wait for one to finish before submitting the next.

import requests, time

API_KEY = "YOUR_API_KEY"
headers = {"Authorization": f"Bearer {API_KEY}"}

scripts = [
    "Welcome to our product demo.",
    "Here's how the dashboard works.",
    "Thanks for watching! Try it free.",
]


def wait_for_job(job_id):
    while True:
        s = requests.get(
            f"https://api.atlasv1.com/v1/jobs/{job_id}",
            headers=headers,
        ).json()
        if s["status"] == "completed":
            return s
        elif s["status"] == "failed":
            raise Exception(f"Job {job_id} failed: {s['error']}")
        time.sleep(2)


# Step 1, Generate all speech audio with ElevenLabs
audio_clips = []
for text in scripts:
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": "YOUR_ELEVENLABS_KEY",
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
    )
    audio_clips.append(tts_res.content)
    print(f"TTS complete for: {text[:30]}...")
print(f"All {len(scripts)} TTS clips generated")

# Step 2, Submit all video jobs at once
video_jobs = []
for i, audio_bytes in enumerate(audio_clips):
    job = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers=headers,
        files={
            "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
            "image": ("face.jpg", open("face.jpg", "rb"), "image/jpeg"),
        },
    ).json()
    video_jobs.append(job["job_id"])
    print(f"Video job {i + 1} submitted: {job['job_id']}")

# Step 3, Wait and download all videos
for i, vid_id in enumerate(video_jobs):
    wait_for_job(vid_id)
    result = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{vid_id}/result",
        headers=headers,
    ).json()
    video = requests.get(result["url"])
    with open(f"clip_{i + 1}.mp4", "wb") as f:
        f.write(video.content)
    print(f"Clip {i + 1}/{len(scripts)} saved")

08advanced

Production-Ready

A reusable function with proper error handling, rate limit retries, progress logging, and streaming file downloads. Supports both polling and webhook modes.

import requests
import time


def poll_job(job_id, headers, timeout=600, interval=2):
    """Poll a job until it completes, fails, or times out."""
    start = time.time()
    while time.time() - start < timeout:
        status = requests.get(
            f"https://api.atlasv1.com/v1/jobs/{job_id}",
            headers=headers,
        ).json()

        if status["status"] == "completed":
            return status
        elif status["status"] == "failed":
            raise Exception(
                f"Job {job_id} failed [{status.get('error_code')}]: {status.get('error')}"
            )

        time.sleep(interval)

    raise TimeoutError(f"Job {job_id} did not complete within {timeout}s")


def download_result(job_id, output_path, headers):
    """Download a completed job's output to disk."""
    result = requests.get(
        f"https://api.atlasv1.com/v1/jobs/{job_id}/result",
        headers=headers,
    ).json()
    file = requests.get(result["url"], stream=True)
    file.raise_for_status()

    with open(output_path, "wb") as f:
        for chunk in file.iter_content(chunk_size=8192):
            f.write(chunk)


def generate_avatar(text, image_path, output_path, api_key, elevenlabs_key):
    """Text + image → avatar video with ElevenLabs TTS and retry logic."""
    headers = {"Authorization": f"Bearer {api_key}"}

    # Step 1, Generate speech audio with ElevenLabs
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": elevenlabs_key,
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
        timeout=30,
    )
    tts_res.raise_for_status()
    audio_bytes = tts_res.content
    print(f"TTS complete, {len(audio_bytes)} bytes")

    # Step 2, Submit video job to Atlas
    vid_res = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers=headers,
        files={
            "audio": ("speech.mp3", audio_bytes, "audio/mpeg"),
            "image": (image_path, open(image_path, "rb"), "image/jpeg"),
        },
        timeout=30,
    )

    if vid_res.status_code == 429:
        wait = vid_res.json().get("retry_after_seconds", 10)
        print(f"Rate limited, waiting {wait}s")
        time.sleep(wait)
        return generate_avatar(text, image_path, output_path, api_key, elevenlabs_key)

    vid_res.raise_for_status()
    vid_job = vid_res.json()
    print(f"Video job: {vid_job['job_id']}")

    vid_status = poll_job(vid_job["job_id"], headers)
    print("Video complete")

    download_result(vid_job["job_id"], output_path, headers)

    return {
        "status": vid_status["status"],
    }


result = generate_avatar(
    text="Hello from Atlas!",
    image_path="face.jpg",
    output_path="output.mp4",
    api_key="YOUR_API_KEY",
    elevenlabs_key="YOUR_ELEVENLABS_KEY",
)
print(result)

import requests

def generate_avatar_webhook(text, image_path, callback_url, api_key, elevenlabs_key):
    headers = {"Authorization": f"Bearer {api_key}"}

    # Generate speech audio with ElevenLabs
    tts_res = requests.post(
        "https://api.elevenlabs.io/v1/text-to-speech/JBFqnCBsd6RMkjVDRZzb",
        headers={
            "xi-api-key": elevenlabs_key,
            "Content-Type": "application/json",
        },
        json={"text": text, "model_id": "eleven_multilingual_v2"},
    )
    tts_res.raise_for_status()

    # Submit video job with webhook
    vid = requests.post(
        "https://api.atlasv1.com/v1/generate",
        headers={**headers, "X-Callback-URL": callback_url},
        files={
            "audio": ("speech.mp3", tts_res.content, "audio/mpeg"),
            "image": (image_path, open(image_path, "rb"), "image/jpeg"),
        },
        timeout=30,
    )
    vid.raise_for_status()
    print(f"Video job: {vid.json()['job_id']}, webhook will fire on completion")
    return vid.json()


generate_avatar_webhook(
    text="Hello from Atlas!",
    image_path="face.jpg",
    callback_url="https://yourapp.com/webhook/atlas",
    api_key="YOUR_API_KEY",
    elevenlabs_key="YOUR_ELEVENLABS_KEY",
)

code examples

Or jump straight to a working app

Quickstart

Full Pipeline

Text → Video (ElevenLabs)

ElevenLabs + Atlas

OpenAI TTS + Atlas

Webhook Receiver

Submit with webhook

Receive the webhook, Flask

Webhook + FastAPI

React Hook, @northmodellabs/atlas-react

Frontend, React Component

Backend, API Route (Next.js)

React Hook, Passthrough Mode

Manual LiveKit Integration

Batch Processing

Production-Ready

code examples

Or jump straight to a working app

Quickstart

Full Pipeline

Text → Video (ElevenLabs)

ElevenLabs + Atlas

OpenAI TTS + Atlas

Webhook Receiver

Submit with webhook

Receive the webhook, Flask

Webhook + FastAPI

React Hook, @northmodellabs/atlas-react

Frontend, React Component

Backend, API Route (Next.js)

React Hook, Passthrough Mode

Manual LiveKit Integration

Batch Processing

Production-Ready

React Hook, `@northmodellabs/atlas-react`

React Hook, `@northmodellabs/atlas-react`