Synchronous vs Asynchronous
Egg API supports both synchronous and asynchronous generation:
| Model | Type | Response |
|---|
nanobanana | Synchronous | Results returned immediately |
nanobanana-pro | Synchronous | Results returned immediately |
veo3.1 | Asynchronous | Task ID returned, poll for results |
Async Workflow
For async models (like veo3.1), you need to:
- Submit - Create a generation task
- Poll - Check the task status periodically
- 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
| Status | Description |
|---|
pending | Task is queued, waiting to start |
processing | Task is being processed |
completed | Task finished successfully |
failed | Task 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
Recommended Polling Interval
- 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