API DOCS

Authorization: Bearer <your_api_key>

{
  "name": "Atlas API",
  "version": "8.0",
  "endpoints": {
    "POST /v1/generate": "Audio + image → lip-sync avatar video",
    "GET /v1/jobs/{id}": "Get job status and details",
    "GET /v1/jobs/{id}/result": "Get presigned download URL for completed job output",
    "GET /v1/jobs": "List your recent jobs",
    "POST /v1/realtime/session": "Create realtime avatar session (conversation or passthrough mode)",
    "PATCH /v1/realtime/session/{id}": "Hot-swap face image mid-session",
    "GET /v1/realtime/session/{id}": "Get session status",
    "DELETE /v1/realtime/session/{id}": "End session and release GPU",
    "GET /v1/health": "Health check",
    "GET /v1/status": "System status",
    "GET /v1/me": "Your API key info and usage"
  },
  "authentication": "Authorization: Bearer <api_key>",
  "flow": {
    "offline": "POST → 202 {job_id} → poll GET /v1/jobs/{id} → GET /v1/jobs/{id}/result",
    "realtime": "POST /v1/realtime/session → connect to LiveKit with token → DELETE to end"
  }
}

{
  "job_id": "a1b2c3d4e5f6",
  "status": "pending",
  "message": "Job accepted. Poll GET /v1/jobs/a1b2c3d4e5f6 for status."
}

# Step 1 — Submit job
curl -X POST "https://api.atlasv1.com/v1/generate" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -F "audio=@speech.mp3" \
  -F "image=@face.jpg"
# Returns: {"job_id": "a1b2c3d4e5f6", "status": "pending", ...}

# Step 2 — Poll
curl "https://api.atlasv1.com/v1/jobs/a1b2c3d4e5f6" \
  -H "Authorization: Bearer YOUR_API_KEY"
# Returns: {"status": "completed", ...} when done

# Step 3 — Get presigned URL
curl "https://api.atlasv1.com/v1/jobs/a1b2c3d4e5f6/result" \
  -H "Authorization: Bearer YOUR_API_KEY"
# Returns: {"url": "https://...", "content_type": "video/mp4", "expires_in": 86400}

# Download from presigned URL (no auth needed)
curl -o output.mp4 "PRESIGNED_URL_FROM_ABOVE"

import requests, time

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

# Step 1 — Submit
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()

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

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

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

    time.sleep(2)

# Step 3 — Get presigned URL + download
result = requests.get(
    f"https://api.atlasv1.com/v1/jobs/{job_id}/result",
    headers=headers,
).json()

# Download from presigned URL (no auth needed)
video = requests.get(result["url"])
with open("output.mp4", "wb") as f:
    f.write(video.content)

print(f"Saved — {status['timing']['latency_ms'] / 1000:.1f}s")

{
  "jobs": [
    {
      "job_id": "a1b2c3d4e5f6",
      "type": "video",
      "status": "completed",
      "created_at": "2026-03-25T16:47:07Z",
      "completed_at": "2026-03-25T16:47:52Z",
      "latency_ms": 43000,
      "output_duration": null,
      "input_chars": null,
      "error_code": null
    },
  ],
  "count": 42,
  "limit": 20,
  "offset": 0
}

{
  "job_id": "a1b2c3d4e5f6",
  "type": "video",
  "status": "processing",
  "queue_position": 0,
  "input": {
    "text_preview": null,
    "language": null,
    "chars": null,
    "audio_size": 105000,
    "image_size": 52000
  },
  "output": {
    "duration": null,
    "size_bytes": null,
    "sample_rate": null
  },
  "error": null,
  "error_code": null,
  "timing": {
    "created_at": "2026-03-25T16:47:07Z",
    "started_at": "2026-03-25T16:47:09Z",
    "completed_at": null,
    "latency_ms": null
  }
}

{
  "job_id": "a1b2c3d4e5f6",
  "type": "video",
  "status": "completed",
  "queue_position": 0,
  "input": {
    "text_preview": null,
    "language": null,
    "chars": null,
    "audio_size": 105000,
    "image_size": 52000
  },
  "output": {
    "duration": null,
    "size_bytes": 4634000,
    "sample_rate": null,
    "has_result": true
  },
  "error": null,
  "error_code": null,
  "timing": {
    "created_at": "2026-03-25T16:47:07Z",
    "started_at": "2026-03-25T16:47:09Z",
    "completed_at": "2026-03-25T16:47:52Z",
    "latency_ms": 43000
  },
  "url": "https://storage.example.com/jobs/.../output.mp4?X-Amz-Algorithm=...",
  "expires_in": 86400,
  "result_url": "/v1/jobs/a1b2c3d4e5f6/result"
}

{
  "job_id": "a1b2c3d4e5f6",
  "type": "video",
  "status": "failed",
  "queue_position": 0,
  "input": {
    "text_preview": null,
    "language": null,
    "chars": null,
    "audio_size": 105000,
    "image_size": 52000
  },
  "output": {
    "duration": null,
    "size_bytes": null,
    "sample_rate": null
  },
  "error": "Processing failed. Please try again.",
  "error_code": "generation_failed",
  "timing": {
    "created_at": "2026-03-25T16:47:07Z",
    "started_at": "2026-03-25T16:47:09Z",
    "completed_at": "2026-03-25T16:47:40Z",
    "latency_ms": 31000
  }
}

{
  "url": "https://storage.example.com/jobs/.../output.mp4?X-Amz-Algorithm=...",
  "content_type": "video/mp4",
  "expires_in": 86400
}

{
  "authenticated": true,
  "key_prefix": "sk_live_A7xQ",
  "name": "My API Key",
  "tier": "starter",
  "requests_used": 14,
  "rate_limit": {
    "requests_per_minute": 30,
    "remaining": 28,
    "resets_in": "42s"
  },
  "billing": "pay_as_you_go"
}

{
  "status": "operational",
  "services": {
    "avatar_generation": "available",
    "voice_synthesis": "available"
  }
}

{
  "status": "all systems go",
  "services": {
    "avatar_generation": "online"
  }
}

{
  "session_id": "ses_a1b2c3d4e5f6...",
  "livekit_url": "wss://your-livekit-instance.livekit.cloud",
  "token": "<livekit_jwt_token>",
  "room": "atlas-rt-ses_a1b2c3d4e5f6...",
  "mode": "conversation",
  "max_duration_seconds": 3600,
  "pricing": "$10/hour, prorated per second"
}

{
  "session_id": "ses_x9y8z7w6v5u4...",
  "livekit_url": "wss://your-livekit-instance.livekit.cloud",
  "token": "<livekit_jwt_token>",
  "room": "atlas-rt-ses_x9y8z7w6v5u4...",
  "mode": "passthrough",
  "max_duration_seconds": 3600,
  "pricing": "$6/hour, prorated per second"
}

import requests

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

# Create a rendering 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",
        "mode": "passthrough",
    },
).json()

print(f"Session: {session['session_id']}")
print(f"Mode: {session['mode']}")  # "passthrough" (rendering)
print(f"LiveKit URL: {session['livekit_url']}")
print(f"Token: {session['token'][:20]}...")

# Connect to LiveKit with the token, publish your TTS audio track,
# and subscribe to the avatar video track.
# Use livekit-client (JS) or livekit SDK (Python) to connect.

{
  "session_id": "ses_a1b2c3d4e5f6...",
  "status": "active",
  "room": "atlas-rt-ses_a1b2c3d4e5f6...",
  "started_at": "2026-04-01T01:30:00Z",
  "ended_at": null,
  "duration_seconds": 142.5
}

{
  "session_id": "ses_a1b2c3d4e5f6...",
  "status": "ended",
  "room": "atlas-rt-ses_a1b2c3d4e5f6...",
  "started_at": "2026-04-01T01:30:00Z",
  "ended_at": "2026-04-01T01:35:22Z",
  "duration_seconds": 322.0
}

{
  "session_id": "ses_a1b2c3d4e5f6...",
  "status": "active",
  "message": "Face updated successfully."
}

{
  "session_id": "ses_a1b2c3d4e5f6...",
  "status": "ended",
  "duration_seconds": 322.0,
  "estimated_cost": "$0.89"
}

{
  "error": "already_ended",
  "message": "Session already ended."
}

import { Room, RoomEvent } from "livekit-client";

async function startRealtimeAvatar(faceUrl?: string) {
  // Step 1 — Create session via your backend (which calls Atlas API)
  const res = await fetch("/api/realtime-session", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ face_url: faceUrl }),
  });
  const { session_id, livekit_url, token, room: roomName } = await res.json();

  // Step 2 — Connect to LiveKit room
  const room = new Room();

  room.on(RoomEvent.TrackSubscribed, (track) => {
    if (track.kind === "video") {
      const element = track.attach();
      document.getElementById("avatar-container")?.appendChild(element);
    }
  });

  await room.connect(livekit_url, token);
  console.log("Connected to room:", roomName);

  return { room, sessionId: session_id };
}

// Usage
const { room, sessionId } = await startRealtimeAvatar(
  "https://example.com/face.jpg"
);

// Send text to make the avatar speak
room.localParticipant.publishData(
  new TextEncoder().encode(JSON.stringify({
    type: "speak",
    text: "Hello! How can I help you today?",
  })),
  { reliable: true }
);

// When done — disconnect client and end session
room.disconnect();
await fetch(`/api/realtime-session/${sessionId}`, { method: "DELETE" });

// app/api/realtime-session/route.ts
export async function POST(req: Request) {
  const { face_url, mode } = await req.json();

  const res = await fetch("https://api.atlasv1.com/v1/realtime/session", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.ATLAS_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ face_url, mode }), // "conversation" (Interactive) | "passthrough" (Rendering)
  });

  return Response.json(await res.json(), { status: res.status });
}

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

function AvatarPage() {
  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(); // { sessionId, livekitUrl, token }
    },
    deleteSession: async (id) => {
      await fetch(`/api/session/${id}`, { method: "DELETE" });
    },
  });

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

      {session.status === "idle" && (
        <button onClick={() => session.connect(myFaceFile)}>Start</button>
      )}

      {session.status === "connected" && (
        <>
          <button onClick={() => session.setMicEnabled(session.muted)}>
            {session.muted ? "Unmute" : "Mute"}
          </button>
          <button onClick={session.disconnect}>End</button>
        </>
      )}

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

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

// After connecting, publish audio from your own TTS:
async function handleUserMessage(text: string) {
  // 1. Call your LLM
  const llmResponse = await fetch("/api/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text }),
  });
  const { audio } = await llmResponse.json(); // base64 audio from your TTS

  // 2. Publish to the room — avatar lip-syncs, mic auto-mutes
  const handle = await session.publishAudio(audio);

  // Optional: stop early
  // handle.stop();
}

// 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 contentType = req.headers.get("content-type") || "";

  // Supports both form uploads and JSON (for rendering mode)
  let body: BodyInit;
  let headers: Record<string, string> = { Authorization: `Bearer ${API_KEY}` };

  if (contentType.includes("multipart/form-data")) {
    body = await req.formData();
  } else {
    const json = await req.json();
    body = JSON.stringify(json); // { face_url, mode: "passthrough" } for rendering
    headers["Content-Type"] = "application/json";
  }

  const res = await fetch(`${API_URL}/v1/realtime/session`, {
    method: "POST",
    headers,
    body,
  });
  return Response.json(await res.json(), { status: res.status });
}

// app/api/session/[id]/route.ts
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 hmac, hashlib

def verify_atlas_webhook(body: bytes, signature: str, timestamp: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        f"{timestamp}.{body.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# In your Flask/FastAPI handler:
# signature = request.headers["X-Atlas-Signature"]
# timestamp = request.headers["X-Atlas-Timestamp"]
# verify_atlas_webhook(request.body, signature, timestamp, YOUR_SECRET)

{
  "error": "error_code",
  "message": "Human-readable description"
}

{
  "error": "unauthorized",
  "message": "Missing Authorization header. Use: Authorization: Bearer <api_key>"
}

{
  "error": "unauthorized",
  "message": "Invalid Authorization format. Use: Authorization: Bearer <api_key>"
}

{
  "error": "not_ready",
  "message": "Job is still processing. Poll GET /v1/jobs/{id} until status is completed."
}

{
  "error": "rate_limit_exceeded",
  "message": "Slow down! You've hit the limit of 30 requests per minute.",
  "retry_after_seconds": 12
}

{
  "error": "queue_unavailable",
  "message": "Job queue is temporarily unavailable. Please try again in a moment."
}

from elevenlabs import ElevenLabs
import requests, time

client = ElevenLabs(api_key="YOUR_ELEVENLABS_KEY")
audio = client.text_to_speech.convert(
    text="Hello, welcome to our demo.",
    voice_id="JBFqnCBsd6RMkjVDRZzb",
    output_format="mp3_44100_128",
)
audio_bytes = b"".join(audio)

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

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

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

r = requests.get(f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result", headers=headers).json()
video = requests.get(r["url"])
with open("output.mp4", "wb") as f:
    f.write(video.content)

from openai import OpenAI
import requests, time

speech = OpenAI().audio.speech.create(
    model="tts-1-hd", voice="nova",
    input="Hello, welcome to our demo.",
)

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

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

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

r = requests.get(f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result", headers=headers).json()
video = requests.get(r["url"])
with open("output.mp4", "wb") as f:
    f.write(video.content)

from deepgram import DeepgramClient
import requests, time

deepgram = DeepgramClient("YOUR_DEEPGRAM_KEY")
deepgram.speak.v("1").save(
    "speech.mp3",
    {"text": "Hello, welcome to our demo."},
    {"model": "aura-asteria-en"},
)

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

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

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

r = requests.get(f"https://api.atlasv1.com/v1/jobs/{job['job_id']}/result", headers=headers).json()
video = requests.get(r["url"])
with open("output.mp4", "wb") as f:
    f.write(video.content)