tai42-backend-arq

An arq execution backend for the TAI ecosystem. It implements the tai42_contract.backend.Backend surface — the worker runtime (launch) — and layers the platform's background-execution features over arq: queued and awaited tool runs, recurring schedules (interval or crontab) with export/import backup, and result-chaining callbacks. Fleet propagation of config changes is not this backend's concern: a backend-runtime process receives fleet ops through the app's own internal worker bus, exactly like a serving HTTP worker.

The TAI ecosystem

TAI is an open-source runtime for MCP tools, agents, and workflows. An execution Backend is "how work runs beyond the request" — a pluggable strategy the runtime uses to queue tool executions on a worker fleet. Fanning control operations (manifest updates, config/tool/MCP reloads) out across every worker is the skeleton's worker bus, not the backend — a backend-runtime process just joins that bus like any other worker. This package is one such backend (arq over Redis); any package can back the same contract, so this repo is this provider's own full doc home, and the documentation site covers the platform-level story:

• Backend concept: https://tai42.ai/concepts/backends
• Build a 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, CallbackSchema, Manifest, and the tai42_app handle) and tai42-kit[jq] (settings base, schedule normalization, signature helpers, jq). Beyond those it depends on the broker stack — arq, croniter, orjson, makefun, click — plus fastmcp (the platform's tool substrate) and pydantic / pydantic-settings / pydantic-core.

Install

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

uv add tai42-backend-arq

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

Discovery

The host discovers this backend by importing its package — importing tai42_backend_arq registers everything through the global tai42_app handle as a side-effect (there is no entry-point): the ArqBackend (@tai42_app.backends.register_backend), the backend_* tool surface, the sync_task / schedule_task / async_task BACKEND-kind tool extensions. The enqueue side (enqueues, job status and results, schedule hashes) runs over the platform's pooled Redis client, so its connections are pooled per configuration generation and closed when that generation retires. Name the package in your manifest's backend_module field:

backend_module: tai42_backend_arq

Start a worker through the host's backend CLI; everything after worker is this backend's own option surface (see tai42_backend_arq.worker.main):

tai backend worker --max-jobs 10

Configuration

Settings are read from the ARQ_ environment group (see ArqSettings):

ARQ_MANIFEST_KEY, ARQ_TASK_TIMEOUT, and ARQ_TOOL_NAME_ARG mirror the host's backend settings defaults — both sides must agree on these values without sharing code, so override them only in lockstep with the host.

Worker CLI options (after the worker subcommand): --redis-url (defaults to ARQ_REDIS_URL), --burst, --keep-result (default 3600; 0 disables result retention), --queue-name, --max-jobs (default 10), --job-timeout (defaults to ARQ_TASK_TIMEOUT), --poll-delay (default 0.5), --max-tries (default 5), --health-check-interval (default 60). The worker runs with arq's allow_abort_jobs enabled — task cancellation and schedule replace/delete rely on abort processing. A dispatched tool runs once: the backend never retries a tool body (a tool that wants retries declares a ToolRetryPolicy). arq itself re-runs a job cut by a cold stop (CancelledError), bounded by --max-tries — arq's own delivery semantics.

Fleet ops

Config-change propagation across the worker fleet (manifest updates, config/tool/MCP reloads) is the app's own internal worker bus, not a backend concern — a backend-runtime process receives those ops through the app's bus subscription exactly like a serving HTTP worker. See the skeleton's worker-bus concept: https://tai42.ai/concepts/worker-bus

Schedules

Recurring runs live in per-schedule Redis hashes (arq:schedule:{name}) driven by a self-rescheduling task_scheduler job; a startup watchdog restarts schedules whose pending job was lost. The schedule tools (backend_list_schedules — the contract's ScheduleListRow: name / tool / cadence / enabled / next_run_at_ts / next_run_at_iso / error, never the stored arguments —, backend_get_schedule, backend_delete_schedule, backend_enable_schedule, backend_disable_schedule, backend_run_schedule_now, backend_schedule_exists, backend_update_schedule) operate on those hashes, and backend_export_schedules / backend_import_schedules round-trip the contract's ScheduleRecord rows for backup (per-row import errors surfaced as {"index", "name", "error"}, never swallowed). A schedule's cadence is the contract's typed ScheduleCadence (tai42_contract.backend.scheduling); the hash stores its JSON dump in the schedule field. arq cannot honour a relative interval, so relative: true is refused on create, update and import.

Upgrading: reset the schedule store

The schedule field's stored format is the typed cadence. A hash written in the earlier format makes the schedule tools (list, export, get, update) raise ScheduleStoreFormatError naming the schedule; arq keeps firing such a hash from its cron_or_interval field. Reset the store and re-create the schedules:

redis-cli --scan --pattern 'arq:schedule:*' | xargs redis-cli del

The pattern is ArqSettings.arq_schedule_pattern (arq:schedule:*).

Task execution

The sync_task / async_task extension branches dispatch the wrapped tool to the queue via the tool_execution job; a sync_task wait re-raises a failed job's stored failure as TaskFailedError (aborted jobs included). Task options exposed on every branch: countdown (defer by seconds), eta (ISO datetime; defer until then), expires (seconds the queued job stays runnable), and callback_kwargs — a callback schema chained via a callback_job that runs over the primary job's result (jq condition gate, jq expression transform, optional follow-up tool). schedule_task registers a recurring schedule (see below).

Tools

Task/worker tools use arq's public Job status/result API: backend_task_status, backend_task_result (timeout-aware; a task's stored failure re-raises as TaskFailedError carrying the original exception's type, repr, and traceback text — result payloads describe an unserializable value in a tagged JSON shape the deserializer revives on read), backend_cancel_task (a stored abort replays as arq's confirmed-abort verdict; stored failures are reported with their detail), backend_active_tasks (a flat job-id-keyed map — arq has no per-worker attribution), backend_reserved_tasks (a flat list of due job ids), backend_scheduled_tasks (job id → due time in epoch milliseconds), backend_list_failed_tasks (failed/aborted results within the keep-result window, as {"task_id", "error"} rows carrying the stored failure detail). Capabilities arq has no reliable data model for raise NotImplementedError loudly: backend_registered_tasks, backend_worker_stats, backend_worker_queues, backend_ping_worker, backend_list_active_workers.

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.