Workflow examples
The overview has eight complete workflows. Each one was created and run end to end:| Example | Shows |
|---|---|
| One-photo effect | A single step with a fixed prompt and a fallback model |
| Restyle a photo, then animate it | Chaining steps with $.step.primary |
| An LLM writes the prompt | An eachlabs-llm-router step feeding image and video steps |
| Two photos, one clip, with music | parallel steps, resizing and adding music |
| A reliable vision call | A fallback to another LLM provider and named outputs |
| Call an API, then use its data | An eachlabs-http-step call and its JSON fields |
| Let AI choose the path | A choice step routed by a vision model |
| Several scenes in parallel | Multi-step branches joined with merge-videos |
Run many inputs at once
Bulk trigger starts up to 10 runs of the same workflow version in one request. With a one-step image workflow (the one on Create Workflow):curl -X POST https://api.eachlabs.ai/v1/workflows/bulk-trigger/WF_ID/v1 \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_API_KEY" \
-d '{
"inputs": [
{"prompt": "A red bicycle against a white wall"},
{"prompt": "A blue bicycle against a white wall"}
],
"webhook_url": "https://your-app.com/webhooks/batch-complete"
}'
bulk_id and one execution_id per input. List the runs of a batch with:
curl "https://api.eachlabs.ai/v1/workflows/WF_ID/executions?bulk_id=BULK_ID" \
-H "Authorization: Bearer YOUR_API_KEY"
End-to-end Python
Create the quickstart workflow, run it and wait for the result:import time
import requests
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.eachlabs.ai/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
# 1. Create the workflow
workflow = requests.post(f"{BASE_URL}/workflows", headers=HEADERS, json={
"name": "Text to Image Generator",
"definition": {
"input_schema": {
"properties": {"prompt": {"type": "string", "required": True}}
},
"steps": [
{
"id": "write_prompt",
"type": "model",
"model": "eachlabs-llm-router",
"params": {
"model": "google/gemini-3.6-flash",
"messages": [
{"role": "system", "content": "Rewrite the idea as one vivid, detailed image prompt. Reply with the prompt only."},
{"role": "user", "content": "$.inputs.prompt"},
],
},
},
{
"id": "generate_image",
"type": "model",
"model": "nano-banana-2-text-to-image",
"params": {
"prompt": "$.write_prompt.primary.choices[0].message.content",
"aspect_ratio": "16:9",
},
},
],
},
}).json()
workflow_id = workflow["workflow_id"]
version_id = workflow["versions"][0]["version_id"]
print(f"Created workflow: {workflow_id}")
# 2. Run it
execution = requests.post(
f"{BASE_URL}/workflows/trigger/{workflow_id}/{version_id}",
headers=HEADERS,
json={"inputs": {"prompt": "A zen garden at sunrise"}},
).json()
execution_id = execution["execution_id"]
print(f"Execution started: {execution_id}")
# 3. Wait for the result
while True:
result = requests.get(f"{BASE_URL}/workflows/executions/{execution_id}", headers=HEADERS).json()
if result["status"] in ("completed", "failed", "cancelled"):
break
time.sleep(3)
if result["status"] == "completed":
print(f"Prompt: {result['step_outputs']['write_prompt']['primary']['choices'][0]['message']['content']}")
print(f"Image: {result['output']}")
else:
print(f"Error: {result.get('error')}: {result.get('error_cause')}")