Building long-running APIs on Ouro
How to build APIs that take hours or days to complete using webhooks for async result delivery
Last updated May 5, 2026
7 minute readOn this page
How to build APIs that take hours or days to complete using webhooks for async result delivery
Last updated May 5, 2026
7 minute readOn this page
Some API operations—like ML model training, large data processing, or complex simulations—can take hours or even days to complete. Standard HTTP requests aren't designed for this; they'll timeout long before your job finishes.
This guide shows you how to build long-running APIs on Ouro using an async webhook pattern that:
Instead of waiting for a response, your API should:
202 Accepted immediatelyOuro provides special headers on every request that enable this pattern:
| Header | Description |
|---|---|
ouro-webhook-url | URL to POST results when the job completes |
ouro-webhook-token | Secret token to authenticate webhook requests |
|
| Unique ID for this action (for logging progress) |
ouro-route-id | The service route being called |
ouro-route-org-id | Organization context for asset creation |
ouro-route-team-id | Team context for asset creation |
Here's a complete example using Modal for serverless compute. This pattern works with any hosting platform that supports background jobs.
Create a function that does the heavy work and sends results to the webhook when done:
Your FastAPI endpoint should validate the webhook URL, spawn the background job, and return immediately:
The key parts:
.spawn() — This starts the job without waiting (Modal-specific; other platforms have similar patterns)Keep users informed with real-time progress updates using Ouro's action logging system:
Progress messages appear in real-time in the Ouro UI, so users know their job is running and how far along it is.
When your job completes, POST to the webhook URL with this structure:
{
"status": "completed", # or "failed"
"ouro_action_id": "uuid-from-header",
"ouro_route_id": "uuid-from-header",
"response": {
# Your actual result data
# This can include files, datasets, or any structured output
}
}If your API creates assets (files, datasets, posts), include them under the same response keys declared by the route's x-ouro-output-assets.
For example, if the route declares an analysis_results file output:
import base64
response = {
"status": "completed",
"ouro_action_id": ouro_action_id,
"ouro_route_id": ouro_route_id,
"response": {
"analysis_results": {
"name": "Analysis Results",
"description": "Generated analysis output",
"filename": "results.json",
"type": "application/json",
"extension": "json",
"base64": base64.b64encode(json.dumps(results).encode()).decode(),
"org_id": ouro_org_id,
"team_id": ouro_team_id,
}
}
}For the full set of sync and async asset output shapes, see the route input and output assets guide.
Always send a webhook notification on failure so users aren't left waiting:
Set appropriate timeouts — Modal functions can run up to 24 hours. Set a timeout that matches your expected processing time plus buffer.
Log progress frequently — For jobs that take hours, send progress updates every few minutes so users know things are working.
Handle interruptions gracefully — Save checkpoints for long jobs so you can resume if something fails.
Validate early — Check inputs before spawning the background job. It's better to return a 400 error immediately than fail after hours of processing.
Include timing info — Let users know how long things took:
import time
start_time = time.time()
# ... processing ...
duration = time.time() - start_time
response = {
"result": result,
"processing_time_seconds": duration,
}The Chronos forecasting service uses this pattern for generating market reports that take 30-60 minutes to complete. Check out the implementation for a production example of long-running APIs on Ouro.
import os
import requests
from typing import Optional
import modal
app = modal.App("my-long-running-api")
# Your compute-intensive image
compute_image = modal.Image.debian_slim(python_version="3.11").pip_install(
"fastapi",
"requests",
# ... your ML/processing dependencies
)
def send_webhook_notification(
webhook_url: Optional[str],
webhook_token: Optional[str],
action_id: Optional[str],
route_id: Optional[str],
status: str, # "completed" or "failed"
response: dict,
):
"""Send results back to Ouro via webhook."""
if not webhook_url:
return
payload = {
"status": status,
"ouro_action_id": action_id,
"ouro_route_id": route_id,
"response": response,
}
headers = {"Content-Type": "application/json"}
if webhook_token:
headers["ouro-webhook-token"] = webhook_token
try:
requests.post(webhook_url, json=payload, headers=headers, timeout=30)
except Exception as e:
print(f"Webhook notification failed: {e}")
@app.function(
image=compute_image,
cpu=4,
timeout=3600, # 1 hour timeout
secrets=[modal.Secret.from_name("my-secrets")],
)
def process_long_job(
input_data: dict,
ouro_action_id: Optional[str] = None,
ouro_route_id: Optional[str] = None,
ouro_webhook_url: Optional[str] = None,
ouro_webhook_token: Optional[str] = None,
ouro_org_id: Optional[str] = None,
ouro_team_id: Optional[str] = None,
) -> dict:
"""
Your long-running computation.
This runs in the background after the API returns 202.
"""
try:
# === Your actual processing logic ===
result = do_expensive_computation(input_data)
# Send success notification
send_webhook_notification(
webhook_url=ouro_webhook_url,
webhook_token=ouro_webhook_token,
action_id=ouro_action_id,
route_id=ouro_route_id,
status="completed",
response=result,
)
return result
except Exception as e:
import traceback
error_result = {
"status": "error",
"error": str(e),
"traceback": traceback.format_exc(),
}
# Send failure notification
send_webhook_notification(
webhook_url=ouro_webhook_url,
webhook_token=ouro_webhook_token,
action_id=ouro_action_id,
route_id=ouro_route_id,
status="failed",
response=error_result,
)
return error_resultfrom fastapi import FastAPI, HTTPException, Header, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional
class JobRequest(BaseModel):
# Your input parameters
data: dict
config: Optional[dict] = None
@app.cls(image=api_image, secrets=[modal.Secret.from_name("my-secrets")])
@modal.concurrent(max_inputs=100)
class MyAPI:
@modal.asgi_app()
def fastapi_app(self):
api = FastAPI(title="Long-Running API")
@api.post("/process")
async def process_endpoint(
request: JobRequest,
ouro_action_id: Optional[str] = Header(None, alias="ouro-action-id"),
ouro_route_id: Optional[str] = Header(None, alias="ouro-route-id"),
ouro_webhook_url: Optional[str] = Header(None, alias="ouro-webhook-url"),
ouro_webhook_token: Optional[str] = Header(None, alias="ouro-webhook-token"),
ouro_org_id: Optional[str] = Header(None, alias="ouro-route-org-id"),
ouro_team_id: Optional[str] = Header(None, alias="ouro-route-team-id"),
):
"""
Start a long-running job.
Returns immediately with 202 Accepted.
Results delivered via webhook when complete.
"""
# Require webhook URL for async jobs
if not ouro_webhook_url:
raise HTTPException(
status_code=400,
detail="Missing ouro-webhook-url header. This endpoint requires async completion."
)
# Spawn the job (don't wait for result)
process_long_job.spawn(
input_data=request.data,
ouro_action_id=ouro_action_id,
ouro_route_id=ouro_route_id,
ouro_webhook_url=ouro_webhook_url,
ouro_webhook_token=ouro_webhook_token,
ouro_org_id=ouro_org_id,
ouro_team_id=ouro_team_id,
)
return JSONResponse(
status_code=status.HTTP_202_ACCEPTED,
content={
"status": "accepted",
"message": "Job started. You will be notified when it completes.",
},
)
return apifrom ouro import Ouro
def setup_logging(ouro_action_id: Optional[str], ouro_route_id: Optional[str]):
"""Set up Ouro logging for progress updates."""
if not ouro_action_id:
return None
ouro = Ouro(api_key=os.environ["OURO_API_KEY"])
def log_progress(message: str):
"""Send a progress message to Ouro."""
try:
ouro.client.post(
f"/actions/{ouro_action_id}/log",
json={
"message": message,
"asset_id": ouro_route_id,
"level": "info",
},
)
except Exception as e:
print(f"Failed to log to Ouro: {e}")
return log_progress
# Use it in your processing function:
@app.function(...)
def process_long_job(...):
log = setup_logging(ouro_action_id, ouro_route_id)
log("Starting data preprocessing...")
# Step 1
preprocessed = preprocess(input_data)
log("Preprocessing complete. Starting model training...")
# Step 2
model = train_model(preprocessed)
log("Training complete. Generating results...")
# Step 3
result = generate_output(model)
log("Job completed successfully!")
return resultdef handle_error(
error: Exception,
task_name: str,
ouro_webhook_url: Optional[str],
ouro_webhook_token: Optional[str],
ouro_action_id: Optional[str],
ouro_route_id: Optional[str],
) -> dict:
"""Handle errors and notify via webhook."""
import traceback
error_result = {
"status": "error",
"error": str(error),
"traceback": traceback.format_exc(),
}
print(f"Error in {task_name}: {error}")
if ouro_webhook_url:
send_webhook_notification(
webhook_url=ouro_webhook_url,
webhook_token=ouro_webhook_token,
action_id=ouro_action_id,
route_id=ouro_route_id,
status="failed",
response=error_result,
)
return error_result