Calling external services

Most production workflows reach outside their process: a REST API, a payment gateway, a third-party data feed. Those calls can be slow, intermittently unavailable, or silently wrong. Wrapping them in a Flux task gives you durability. The result is recorded the first time it succeeds, so a worker restart or workflow replay never re-issues the call. timeout, retry_max_attempts, and fallback cover the remaining failure modes.

The baseline: a simple HTTP task

The github_stars example in the Flux repository is the canonical starting point:

import httpx

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@task
async def get_stars(repo: str) -> int:
    url = f"https://api.github.com/repos/{repo}"
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        return response.json()["stargazers_count"]


@workflow
async def repo_summary(ctx: ExecutionContext[str]):
    return await get_stars(ctx.input)


if __name__ == "__main__":
    ctx = repo_summary.run("edurdias/flux")
    print(ctx.output)       # {'stars': 2, ...}
    print(ctx.has_succeeded)  # True

This works, but it has no protection against a slow API or a transient failure. If the request hangs for thirty seconds, the workflow hangs too. If the API returns a 503, the workflow fails permanently.

Adding a timeout

Pass timeout=N to cap each attempt at N seconds. Flux wraps the call in asyncio.wait_for and raises ExecutionTimeoutError when the limit is exceeded:

@task.with_options(
    timeout=10,   # seconds per attempt; 0 (the default) means no limit
)
async def get_repo_info(repo: str) -> dict:
    url = f"https://api.github.com/repos/{repo}"
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        data = response.json()
        return {
            "name": data["full_name"],
            "stars": data["stargazers_count"],
            "open_issues": data["open_issues_count"],
        }

Adding retries

Transient errors — a momentary 429, a brief DNS hiccup — are worth retrying. Pass retry_max_attempts to tell Flux how many additional attempts to make after the first failure:

@task.with_options(
    retry_max_attempts=3,   # 3 retries = 4 total attempts
    retry_delay=2,          # wait 2 s before the first retry
    retry_backoff=2,        # exponential backoff multiplier: 2 s -> 4 s -> 8 s
    timeout=10,
)
async def get_repo_info(repo: str) -> dict:
    url = f"https://api.github.com/repos/{repo}"
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        return response.json()

retry_backoff produces genuine exponential backoff: each retry waits the previous delay multiplied by retry_backoff, capped at 600 s. With retry_delay=2 and retry_backoff=2 the waits are 2 s → 4 s → 8 s. This lets a transient outage clear without hammering the service. See Reliability: errors and retries for the full retry mechanics.

For retry basics and the complete options surface, see Defining tasks: task with retries. This page focuses on the external-service pattern specifically: what to do when retries are not enough.

Adding a fallback for unavailability

Sometimes an external service is down for minutes, not milliseconds. Retrying three times with a two-second delay does not help if the outage lasts an hour. Use fallback to return a safe substitute value instead of failing the workflow:

def repo_info_unavailable(repo: str) -> dict:
    """Return a sentinel value when the API is unreachable after all retries."""
    return {
        "name": repo,
        "stars": None,
        "open_issues": None,
        "available": False,
    }


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    timeout=10,
    fallback=repo_info_unavailable,
)
async def get_repo_info(repo: str) -> dict:
    url = f"https://api.github.com/repos/{repo}"
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        data = response.json()
        return {
            "name": data["full_name"],
            "stars": data["stargazers_count"],
            "open_issues": data["open_issues_count"],
            "available": True,
        }

The fallback receives the same arguments as the primary function. Its return value is recorded as the task’s output and the workflow continues normally, with ctx.has_succeeded set to True. The fallback runs at most once and is not retried.

Complete example: HTTP task with timeout, retries, and fallback

The full pattern: caps each attempt, retries on transient failures, and falls back when the service is unavailable:

import httpx

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


def github_unavailable(owner: str, repo: str) -> dict:
    return {
        "full_name": f"{owner}/{repo}",
        "stars": None,
        "open_issues": None,
        "available": False,
    }


@task.with_options(
    name="github-repo-{owner}-{repo}",
    retry_max_attempts=3,
    retry_delay=2,
    timeout=10,
    fallback=github_unavailable,
)
async def fetch_github_repo(owner: str, repo: str) -> dict:
    url = f"https://api.github.com/repos/{owner}/{repo}"
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        data = response.json()
        return {
            "full_name": data["full_name"],
            "stars": data["stargazers_count"],
            "open_issues": data["open_issues_count"],
            "available": True,
        }


@workflow
async def audit_repos(ctx: ExecutionContext[list[list[str]]]):
    results = []
    for owner, repo in ctx.input:
        info = await fetch_github_repo(owner, repo)
        results.append(info)
    return results


if __name__ == "__main__":
    ctx = audit_repos.run([["edurdias", "flux"], ["python", "cpython"]])
    for r in ctx.output:
        status = "live" if r["available"] else "unavailable"
        print(f"{r['full_name']}: {r['stars']} stars ({status})")

Idempotency

Flux records a task’s output after it succeeds and skips the call on replay. That guarantee covers the Flux side. If the external service is not idempotent (calling it twice has a different effect than calling it once), you carry that responsibility in the task.

Read-only calls such as GET requests and read queries are safe by nature. Write calls (POST, PUT, charges, email sends) need a client-side idempotency key derived from a stable identifier like the workflow execution ID. Avoid side effects in the fallback as well: the fallback is also recorded, and a fallback that mutates external state can leave inconsistencies.

See Reliability: idempotency for a full treatment.

Timeout enforcement

Flux enforces timeout via asyncio.wait_for. When the limit is exceeded, Flux raises ExecutionTimeoutError and feeds it into the normal retry/fallback/rollback chain. The counter resets per attempt: timeout=10, retry_max_attempts=3 can spend up to 40 seconds on calls alone before a fallback or failure.

See Reliability: timeouts for workflow-level timeouts and server-side enforcement.