Run background tasks with a worker

10xGraph runs durable background work on TaskIQ. You write tasks with @broker.task, the API sends them, a 10xgraph worker process runs them, and a 10xgraph scheduler process fires scheduled ones. The worker loads your graph, checkpointer, store and cache store the same way the API does, so a task can inject and run the graph.

For fire-and-forget work inside a node, use BackgroundTaskManager instead.

Install

pip install "10xgraph-api[jobs]"           # Redis
pip install "10xgraph-api[jobs-rabbitmq]"  # RabbitMQ, results on Redis
pip install "10xgraph-api[jobs-kafka]"     # Kafka, results on Redis

The supported brokers are Redis Streams, RabbitMQ and Kafka. Results, progress and dynamic schedules always live in Redis. Without the extra, the worker refuses to start and prints the command.

Configure

Add a worker key to 10xgraph.json. Status records, cancel flags and schedule records live in the API’s cache store, so set cache_store.type to redis too.

{
  "agent": "app.graph:app",
  "cache_store": {"type": "redis", "url": "${REDIS_URL}"},
  "worker": {
    "broker": {"type": "redis_stream", "url": "${REDIS_URL}"},
    "tasks": ["app.tasks"],
    "result_ttl": 86400,
    "retries": 3
  }
}

Write a task

# app/tasks.py
from typing import Annotated

from injectq.integrations.taskiq import InjectTaskiq
from taskiq import TaskiqDepends
from taskiq.depends.progress_tracker import ProgressTracker, TaskState
from tenxgraph.core.graph import CompiledGraph
from tenxgraph_api.jobs import TaskContext, broker


@broker.task(task_name="make_report", http=True)  # http=True lets the API start it; the default is False
async def make_report(
    topic: str,                                                    # set by the caller
    config: dict,                                                  # filled by the server
    ctx: Annotated[TaskContext, TaskiqDepends()],
    tracker: Annotated[ProgressTracker, TaskiqDepends()],
    graph: Annotated[CompiledGraph, InjectTaskiq(CompiledGraph)],  # InjectQ value
) -> dict:
    await tracker.set_progress(TaskState.STARTED, {"step": 1})
    if await ctx.cancelled():
        return {"cancelled": True}
    ...

Tasks are TaskIQ tasks, so everything in TaskIQ works. What 10xGraph adds:

• ctx: Annotated[TaskContext, TaskiqDepends()] gives task_id, task_name, config and await ctx.cancelled().
• A config parameter gets the run config (user_id, user and task_id) only if you declare it. user is the caller’s Identity as a dict. Pass it to graph.ainvoke(..., config=config) to run as the caller. Without a label it is {"task_id": ...} only.
• Inject values with InjectTaskiq. Plain Inject[...] does not work in tasks.
• Progress is TaskIQ’s ProgressTracker.
• TaskCancelled is the error a task cancelled before it started returns to the caller.

An argument typed as a pydantic model is not guaranteed to arrive as one. TaskIQ parses each argument against its type hint, and when parsing fails it logs a warning and passes the raw value (a dict) to your task, which then fails later, for example with AttributeError. Tip: call Model.model_validate(arg) at the top of the task to fail fast with a clear error.

The task name is the task_name you set, here make_report. Without it TaskIQ uses <module>:<function> (app.tasks:make_report).

Run the worker and the scheduler

10xgraph dev        # the API: starts and schedules tasks, does not run them
10xgraph worker     # runs tasks
10xgraph scheduler  # sends scheduled tasks

Both read 10xgraph.json from the current folder (--config to change it), refuse to start without a worker key, and import the entries in worker.tasks. Every other flag goes to TaskIQ unchanged (--workers, --ack-type, --log-level). 10xgraph worker refuses --use-process-pool. 10xgraph scheduler adds --update-interval 10 unless you pass one, which is how soon it sees changed schedules.

Run exactly one scheduler. Two would send every scheduled task twice. Run as many workers as you need.

10xgraph build generates both processes for you. With --docker-compose it adds a worker service and, if you ask for it, a scheduler service built from the same image as the API. With --k8s it adds a <service>-worker and a <service>-scheduler Deployment (the scheduler has one replica and strategy: Recreate, so a rollout never runs two). Use --worker and --scheduler to choose, or answer the questions when build asks. With the memory broker no worker is generated, because tasks run inside the API. See Deploy with Docker and Kubernetes.

Start a task from Python

Use TaskIQ’s own call:

await make_report.kiq("sales")

The run config is then {"task_id": ...} only. To pass one, set the tenxgraph_config label (a JSON string):

import json

await make_report.kicker().with_labels(
    tenxgraph_config=json.dumps({"user_id": "u1"})
).kiq("sales")

Schedules

A fixed schedule goes on the task and is sent by 10xgraph scheduler:

@broker.task(schedule=[{"cron": "0 2 * * *"}])
async def nightly_cleanup() -> None: ...

A common use is maintenance work, such as pruning old tool execution ledger rows. See the prune job example.

For dynamic schedules use TaskIQ’s schedule_by_cron, schedule_by_time or schedule_by_interval with the source from tenxgraph_api.jobs.scheduler:

from tenxgraph_api.jobs.scheduler import sources

await make_report.kicker().schedule_by_cron(sources[-1], "0 2 * * *", "sales")

sources[-1] is the Redis source. It uses the result backend’s Redis, so it exists with every broker except memory. The same schedules are available over HTTP, below.

Start a task over HTTP

Only tasks marked http=True can be started over HTTP; any other task answers 404. The routes use the same auth as the graph routes.

# run now; answers 202 with {"data": {"task_id": "..."}}
curl -X POST http://localhost:8000/v1/tasks/make_report \
  -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" \
  -d '{"kwargs": {"topic": "sales"}}'

# read it
curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/v1/tasks/$TASK_ID

# schedule it: exactly one of cron, at (ISO time, once) or every (seconds)
curl -X POST http://localhost:8000/v1/tasks/make_report \
  -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" \
  -d '{"kwargs": {"topic": "sales"}, "schedule": {"cron": "0 2 * * *"}}'

# change the time
curl -X PATCH http://localhost:8000/v1/tasks/$SCHEDULE_ID \
  -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" \
  -d '{"schedule": {"every": 3600}}'

The status is queued, running, success, failure or cancelled. Rules:

• The body is {"args": [], "kwargs": {}, "schedule": ...}.
• Callers set only the task’s plain arguments. config, user, user_id, injected parameters, unknown names and extra positional arguments are refused with 422. The server fills the run config from the caller.
• Tasks and schedules belong to the user who started them. GET and cancel follow the caller’s reach (own, tenant or all); starting a task and PATCH stay on the caller’s own. Anything outside the caller’s reach gets 404, the same as for an unknown id.
• There is no list route. Keep the ids you get back.
• A schedule runs as the user who created it, with that user’s roles as they are when each run starts. It lives until you cancel it; a one-time at schedule expires after its run result does.
• Scheduling needs the Redis result backend; with the memory broker or result backend the answer is 501.
• In production the error field is the exception class name only.
• Grant tasks:trigger, tasks:read and tasks:cancel to the roles that need them, in auth.roles. There is no tasks:pause. See API authentication.

Auth in the worker

With auth on, the worker checks every run again before it executes, because the role table may have changed since the task was queued:

1. The stored identity is rebuilt from its claims with the current auth.roles the worker booted with. A __system__ user id is refused.
2. The worker calls authorize(identity, "tasks:trigger", task_name) on your BaseAuth class. If it refuses, the stored identity is invalid, or the check raises, the run does not execute and its status is failure (fail closed).
3. The rebuilt identity goes into config["user"] and TaskContext.config.

Token expiry and revocation are not checked, because a queued run has no live token. A user whose role loses tasks:trigger stops running at the next run, and a revoked user’s already queued tasks still run until their role changes. Restart workers after you change auth.roles.

System identities

A task that no user started (a cron schedule on the task, or a call from your own code) has no caller. Give it a name and the identity it runs as:

{
  "auth": {
    "method": "jwt",
    "roles": {"member": {"permissions": ["graph:*", "threads:*", "tasks:trigger"], "reach": "own"}},
    "system_identities": {"nightly": {"roles": ["member"], "tenant": {"org_id": "acme"}}}
  }
}

@broker.task(schedule=[{"cron": "0 2 * * *"}], identity="nightly")
async def nightly_cleanup(config: dict) -> None: ...

The run executes as Identity(kind="system", user_id="__system__:nightly") with the roles and tenant of that entry, resolved like a token holding them. Rules:

• The identity label never overrides a caller. A task started by a user runs as that user.
• An unknown name with auth on aborts the run. With auth off the task runs without an identity and logs one warning.
• The system identity still needs tasks:trigger through its roles, and its reach comes from its roles.
• HTTP callers cannot set labels.

Monitoring and errors

The worker logs every task and sends the same steps as events to the event publishers bound in your container (Kafka, Redis, RabbitMQ, console or custom). AI tracing backends (Phoenix, Logfire, LangSmith, OTEL) do not receive them, because a task is not a graph run. A failed task is not published at all (metadata.source is task, metadata.phase names the TaskIQ hook):

A failed task sends no event. on_error logs it at ERROR level with the traceback and counts it in tenxgraph.task.runs with status="failure". See Publishers, AI tracing and metrics.

Metrics and traces come through OpenTelemetry, set up like the API (one setup shared by the API and the worker): OTEL_ENABLED=true, OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_SERVICE_NAME, OTEL_LEVEL. Each task becomes a span, and two metrics are exported:

With the metrics block in 10xgraph.json enabled, the worker exports these metrics itself, by OTLP push or a Prometheus scrape port (metrics.prometheus.worker_port). See Metrics with Prometheus and Grafana. Without the block, the metrics go to the same endpoint as the traces (OTLP gRPC), or to the console when no endpoint is set outside production; point that endpoint at an OpenTelemetry collector and export from the collector to Prometheus. OTEL_METRIC_EXPORT_INTERVAL (milliseconds, default 60000) sets how often metrics are pushed.

Cancel and progress

POST /v1/tasks/{id}/cancel sets a flag. A queued task is then not run and ends as cancelled. A running task is not killed: it stops when it checks await ctx.cancelled() and returns, and then ends as success with whatever it returned. Check it between steps of long work.

Progress is what the task passes to tracker.set_progress(state, meta). GET /v1/tasks/{id} shows it under progress, for example {"state": "STARTED", "meta": {"step": 2}}.

Delivery guarantees

What happens when a worker dies in the middle of a task depends on the broker:

Write tasks so a rerun is safe. Retries after an error are opt-in: @broker.task(retry_on_error=True), up to retries.

Limits

• Brokers: Redis Streams, RabbitMQ and Kafka. Results and schedules: Redis. Nothing else for now.
• Kafka: a task lost when its worker dies (see above). The default topic taskiq_topic has one partition, so only one worker in the group reads it; create the topic with more partitions to spread work. A new consumer group starts at the latest offset, so tasks sent before the first worker ever joined are skipped. Building the Kafka broker connects to Kafka at once, so the API and worker need it reachable at startup.
• The cancel check and the argument guard do not reach process-pool workers, so the worker refuses that mode.
• One scheduler.
• No list route and no way to kill a running task.