Skip to main content

Synchronous vs Asynchronous

Egg API supports both synchronous and asynchronous generation:
ModelTypeResponse
nanobananaSynchronousResults returned immediately
nanobanana-proSynchronousResults returned immediately
veo3.1AsynchronousTask ID returned, poll for results

Async Workflow

For async models (like veo3.1), you need to:
  1. Submit - Create a generation task
  2. Poll - Check the task status periodically
  3. Retrieve - Get the results when complete

Step 1: Submit a Task

Submit your generation request:
curl -X POST https://api.eggapi.ai/v1/generate \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "veo3.1",
    "prompt": "A cat playing with a ball of yarn",
    "duration": 5,
    "aspect_ratio": "16:9"
  }'
Response (202 Accepted):
{
  "data": {
    "id": "gen_abc123def456",
    "status": "pending",
    "model": "veo3.1",
    "type": "video_generate",
    "output": null,
    "error": null,
    "cost": "$0.0000",
    "created_at": "2024-01-15T10:30:00Z",
    "started_at": null,
    "completed_at": null
  },
  "error": null
}

Step 2: Poll for Status

Check the task status periodically:
curl https://api.eggapi.ai/v1/tasks/gen_abc123def456 \
  -H "Authorization: Bearer YOUR_API_KEY"

Task Statuses

StatusDescription
pendingTask is queued, waiting to start
processingTask is being processed
completedTask finished successfully
failedTask failed

Processing Response

{
  "data": {
    "id": "gen_abc123def456",
    "status": "processing",
    "model": "veo3.1",
    "type": "video_generate",
    "output": null,
    "error": null,
    "cost": "$0.0000",
    "created_at": "2024-01-15T10:30:00Z",
    "started_at": "2024-01-15T10:30:05Z",
    "completed_at": null
  },
  "error": null
}

Step 3: Get Results

When status is completed:
{
  "data": {
    "id": "gen_abc123def456",
    "status": "completed",
    "model": "veo3.1",
    "type": "video_generate",
    "output": {
      "url": "https://cdn.eggapi.ai/outputs/abc123.mp4",
      "duration_seconds": 5
    },
    "error": null,
    "cost": "$0.4000",
    "created_at": "2024-01-15T10:30:00Z",
    "started_at": "2024-01-15T10:30:05Z",
    "completed_at": "2024-01-15T10:31:30Z"
  },
  "error": null
}

Polling Best Practices

  • Start polling after 3-5 seconds
  • Poll every 5 seconds for videos
  • Use exponential backoff if needed

Example Implementation

import requests
import time

def wait_for_task(task_id, api_key, max_wait=300):
    """Wait for an async task to complete."""
    url = f"https://api.eggapi.ai/v1/tasks/{task_id}"
    headers = {"Authorization": f"Bearer {api_key}"}

    start_time = time.time()
    poll_interval = 5

    while time.time() - start_time < max_wait:
        response = requests.get(url, headers=headers)
        task = response.json()["data"]

        if task["status"] == "completed":
            return task

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

        print(f"Status: {task['status']}...")
        time.sleep(poll_interval)

    raise Exception("Task timed out")

# Usage
task = wait_for_task("gen_abc123", "YOUR_API_KEY")
print(f"Video URL: {task['output']['url']}")

Handling Failures

When a task fails, the response includes an error message:
{
  "data": {
    "id": "gen_abc123def456",
    "status": "failed",
    "error": "Content moderation: prompt contains prohibited content",
    ...
  },
  "error": null
}
Common failure reasons:
  • Content moderation violations
  • Invalid prompt
  • Provider issues (our system automatically retries with backup providers)
Egg API automatically handles provider failures by retrying with backup providers. You typically won’t see provider-related failures unless all providers are unavailable.

Listing Tasks

View all your tasks with pagination:
curl "https://api.eggapi.ai/v1/tasks?page=1&per_page=20" \
  -H "Authorization: Bearer YOUR_API_KEY"
This is useful for:
  • Building task dashboards
  • Tracking generation history
  • Debugging issues