Building long-running APIs on Ouro

How to build APIs that take hours or days to complete using webhooks for async result delivery

6 minute read

Last updated December 8, 2025

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:

Returns immediately so requests don't timeout
Processes work in the background
Notifies Ouro when the job completes (or fails)
Provides real-time progress updates to users

The webhook pattern

Instead of waiting for a response, your API should:

Accept the request and return 202 Accepted immediately
Process in the background using async workers
Send results via webhook when complete

Ouro 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
`ouro-action-id`	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

Example implementation

Here's a complete example using Modal for serverless compute. This pattern works with any hosting platform that supports background jobs.

1. Define your long-running function

Create a function that does the heavy work and sends results to the webhook when done:

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_result

2. Create the API endpoint

Your FastAPI endpoint should validate the webhook URL, spawn the background job, and return immediately:

from 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 api

The key parts:

Require the webhook URL — If it's missing, the caller won't get results
Use .spawn() — This starts the job without waiting (Modal-specific; other platforms have similar patterns)
Return 202 Accepted — Indicates the request was accepted for processing

Adding progress logging

Keep users informed with real-time progress updates using Ouro's action logging system:

from 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 result

Progress messages appear in real-time in the Ouro UI, so users know their job is running and how far along it is.

Webhook response format

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
    }
}

Returning assets

If your API creates assets (files, datasets, posts), include them in the response:

import base64
 
response = {
    "status": "completed",
    "ouro_action_id": ouro_action_id,
    "ouro_route_id": ouro_route_id,
    "response": {
        "file": {
            "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,
        }
    }
}

Error handling

Always send a webhook notification on failure so users aren't left waiting:

def 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

Best practices

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,
}

Real-world example

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.

Next steps

Deploying ML models with Modal — Learn how to deploy your model to Modal
How to monetize APIs — Set up pricing for your long-running API
Services documentation — Deep dive into Ouro's service layer

See all guides