tai42-backend-celery

A Celery execution backend for the TAI ecosystem. It implements the tai42_contract.backend.Backend surface — launching the worker runtime (worker / beat / flower). Fleet config propagation is not a backend concern: a backend-runtime process receives worker-bus ops (reload_config, reload_mcp, …) through the skeleton's own worker bus, exactly like a serving HTTP worker. This backend adds one thing on top of that: after a bus op applies, it re-forks this worker's prefork pool so the forked children re-inherit the updated tool registry (see Live reload and the prefork pool). It also ships the sync_task / schedule_task / async_task tool extensions (execute any tool through the queue) and the backend_* tool surface, including the scheduling marker tools the host's schedules API and backup round trip depend on.

The TAI ecosystem

TAI is an open-source runtime for MCP tools, agents, and workflows. An execution Backend is the runtime's execution strategy — one pluggable object that runs the background worker fleet (the processes that pull and execute enqueued work). Fleet config propagation is not a backend concern; live-reload operations reach each worker through the skeleton's own worker bus. This package is one such backend (Celery); siblings can back the same contract with other queues. The documentation site covers the platform-level story:

• Backend & fleet-control concept: https://tai42.ai/concepts/backends
• Build an execution backend (author guide): https://tai42.ai/guides/authors/backend
• Ecosystem catalog: https://tai42.ai/reference/catalog

Its only tai-* dependencies are tai42-contract (the Backend ABC, the CallbackSchema field shape, Manifest, ExtensionKind, and the tai42_app handle) and tai42-kit (settings base + cache, the pooled Redis client, schedule normalization, signature helpers, and the jq compiler). Beyond those it depends on its broker stack: celery, kombu, celery-redbeat, redis, celery-pydantic, flower, and makefun — plus fastmcp (the platform's tool substrate) for FastMCP-context handling when composing dispatch-branch signatures.

Install

Requires Python 3.13+. Install from PyPI into the environment that runs the server:

uv add tai42-backend-celery

Or from source — clone this repo and add it as an editable dependency; the tai42-* dependencies resolve in-tree from the workspace.

git clone https://github.com/tai42ai/tai42   # next to your app checkout
cd /path/to/your/app
uv add --editable ../tai42/plugins/backend-celery

Discovery

The host discovers this backend by importing its package — importing tai42_backend_celery registers everything through the global tai42_app handle as a side-effect (there is no entry-point): the CeleryBackend (tai42_app.backends.register_backend), the Celery application with its pool-child fork-safety hooks, the backend_* tool surface, and the sync_task / schedule_task / async_task BACKEND-kind tool extensions. Name the package in your manifest's backend_module field:

backend_module: tai42_backend_celery

The worker runtime is started through the host's backend launcher with one of:

tai backend worker [celery worker options...]
tai backend beat
tai backend flower

Fleet ops and the worker bus

Fleet config propagation lives in the skeleton's worker bus, internal to the app: a backend-runtime process joins that bus through the app's single long-lived subscription and receives each op (reload_config, reload_mcp, …) exactly like a serving HTTP worker — this backend carries no control-plane of its own. Its one obligation on top of the bus is the prefork-pool turnover: after every applied op that declares an effect (the tool registry or the compiled-template cache changed) the backend re-forks this worker's pool (see Live reload and the prefork pool).

The worker runs on a background thread so this process's event loop stays alive to keep reading the bus subscription; launch installs SIGTERM/SIGINT handlers that request a warm (then cold) Celery shutdown, since Celery skips its own signal handlers off the main thread.

Task execution

The sync_task / async_task extension branches dispatch the wrapped tool to the queue via the celery.tool_execution task. A dispatched tool runs once; retries are declared per tool with a ToolRetryPolicy and applied by the platform, never by Celery (Celery still retries publishing the task message on connection loss, the retry option below). Every task — dispatched or fired by beat — is bounded by the dispatch job budget CELERY_TASK_TIMEOUT (Celery's task_time_limit, enforced by the prefork pool). Task options exposed on every branch: queue, countdown, priority, retry, routing_key, expires, eta, and callback_kwargs — a callback schema chained via a Celery link that runs celery.callback_task on the result (jq condition gate, jq expression transform, optional follow-up tool).

schedule_task persists a RedBeat entry from the contract's typed cadence (tai42_contract.backend.scheduling: an interval with kind: "interval", every seconds and relative, which this backend honours, or a five-field crontab with kind: "crontab"). The backend maps the cadence to RedBeat's own stored shape, so RedBeat's entries are unchanged. Beat must run with the RedBeat scheduler, which launch("beat") configures automatically.

Worker fork safety: each freshly forked pool child evicts the monitoring vendor client through the contract's fork-safe writer.shutdown() (logged, never a silent telemetry disable) and, at child exit, flushes buffered spans and closes its task loops' pooled clients. On macOS a warning names the platform's fork/DNS hazard; prefer the solo or threads pool there.

Tools

Task/worker tools: backend_ping_worker, backend_list_active_workers, backend_task_status, backend_task_result (timeout-aware; re-raises task exceptions), backend_cancel_task, backend_active_tasks, backend_reserved_tasks, backend_scheduled_tasks, backend_registered_tasks, backend_worker_stats, backend_worker_queues. The fleet-view tools return Celery's inspect shape — a mapping of worker name to that worker's entries — so the keying is per worker. backend_list_failed_tasks raises NotImplementedError — Celery keeps no queryable failed-task index.

Schedule tools (RedBeat): backend_list_schedules (the contract's ScheduleListRow: name / tool / cadence / enabled / next_run_at_ts / next_run_at_iso, never the stored arguments), backend_get_schedule, backend_schedule_exists, backend_enable_schedule, backend_disable_schedule, backend_run_schedule_now, backend_update_schedule, backend_delete_schedule, plus the backup round trip backend_export_schedules / backend_import_schedules (the contract's ScheduleRecord rows, the fired tool split out of the stored kwargs; a ScheduleImportReport with skip/overwrite import mode keyed by name; per-row errors surfaced as {"index", "name", "error"}, never swallowed).

Configuration

Settings are read from the CELERY_ environment group (see CelerySettings):

Live reload and the prefork pool

After the worker bus applies an op in this process that declares an effect (the post-apply hook receives the op's effects), the backend re-forks this worker's prefork pool so its children re-inherit the parent's updated tool registry, and it CONFIRMS the restart rather than assuming it: it polls stats and active until every idle pre-restart child PID is gone and the pool is back to full size. Celery's pool_restart arms a per-child restart sentinel, and a child checks it before taking each task: an idle child recycles on its ~1s poll wakeup, while a child busy with a task finishes that task, exits after it and takes no new one, so it is not waited for (the confirmation logs its PID at INFO). That includes a task that published the very op being applied, which is itself waiting on this worker's reply. A pool whose idle pre-restart children do not leave, or that does not refill, fails the op loudly, naming the worker — turning this worker's bus reply into failed rather than reporting an applied op behind stale children. A query op such as list_failed_mcps declares no effect, so it never re-forks the pool.

The turnover runs inside the op apply, before this worker's terminal reply is sent, and the publisher waits only its bus apply window (TAI_BUS_APPLY_TIMEOUT, default 90s) for that reply. The host hands the fleet-op handler what is left of that window, measured from this worker's receipt of the op, and the confirmation budget sits a small margin under it: the confirm or raise reaches the publisher before its report cut, so a stalled turnover is recorded as a truthful failed rather than a timed_out guess. Raising TAI_BUS_APPLY_TIMEOUT raises the turnover budget under it.

Development

uv venv --python 3.13
uv pip install --no-sources --group dev --editable .
uv run --no-sync pytest --cov --cov-report=term-missing
uv run --no-sync ruff check .
uv run --no-sync ruff format --check .
uv run --no-sync pyright

License

Apache-2.0. See LICENSE and NOTICE.