CODE EXAMPLES

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)

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)

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")

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)

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

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>Latency: {session.latency}ms</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>
  );
}

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 });
}

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 });
}

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>
  );
}

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,
)

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")

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(f"Video complete — {vid_status['timing']['latency_ms'] / 1000:.1f}s")

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

    return {
        "video_time_s": vid_status["timing"]["latency_ms"] / 1000,
    }


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",
)