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
A config that registers a task backend also needs the worker bus and a shared
hooks registry: tai serve, tai backend, a config reload and a config write all
refuse it unless TAI_BUS_REDIS_URL and HOOKS_REDIS_URL are set (each falls back
to TAI_DEFAULT_REDIS_URL). The backend process matches the events it raises
against its own hooks registry, so an in-memory one would run none of the hooks
registered through the server.
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):
| Env var | Default | Purpose |
|---|---|---|
ARQ_REDIS_URL |
redis://localhost:6379/0 |
Redis connection for the queue and schedules |
ARQ_REDIS_MAX_CONNECTIONS |
— | Optional cap on the pool's connections |
ARQ_RETRY_ATTEMPTS |
0 |
Times a failed Redis connect or command is retried, each retry logged as a warning; 0 makes one attempt |
ARQ_RETRY_ON_TIMEOUT |
false |
With ARQ_RETRY_ATTEMPTS at 0, retry a timed-out command once at once |
ARQ_SOCKET_TIMEOUT |
— | Seconds a Redis command may wait on its socket; redis-py's default when unset |
ARQ_SOCKET_CONNECT_TIMEOUT |
1 |
Seconds a Redis connect may take |
ARQ_CALLBACK_TIMEOUT |
5 |
Seconds a callback job waits for its predecessor to complete |
ARQ_MANIFEST_KEY |
MANIFEST_KEY |
Name of the env var the host exports the manifest under for worker processes |
ARQ_TASK_TIMEOUT |
300 |
The dispatch job budget in seconds: the longest a job may run on the worker (the --job-timeout default), and how long a synchronous branch tool waits for its queued job's result |
ARQ_TOOL_NAME_ARG |
backend_tool_name |
kwargs key carrying the target tool name into a queued execution |
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.
ARQ_REDIS_URL and ARQ_REDIS_MAX_CONNECTIONS fall back to the shared
TAI_DEFAULT_REDIS_URL / TAI_DEFAULT_REDIS_MAX_CONNECTIONS when unset.
Every Redis connection of this backend — the enqueue side and the worker — is
built from these settings. The worker's connect at start follows
ARQ_RETRY_ATTEMPTS, so with the default 0 a worker started before Redis is
reachable fails at once. A running worker rides a Redis outage only within that
same retry: arq's poll loop has no reconnect of its own, so an outage longer
than the retry ends the worker.
Worker CLI options (after the worker subcommand): --redis-url (defaults to
ARQ_REDIS_URL; the other connection settings still apply), --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.
1 maintainer
Metadata
1 maintainer
Release files for tai42-backend-arq 7.0.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| tai42_backend_arq-7.0.1.tar.gz | 83.0 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| tai42_backend_arq-7.0.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 126.6 kB
Release files / tai42_backend_arq-7.0.1.tar.gz
| Download URL | tai42_backend_arq-7.0.1.tar.gz |
|---|---|
| Size | 83.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
19c6d2f22baf49a49b2c31465e73d0cb7a3afadaf5f2b14772167d8338f0a19a
|
|
BLAKE2b-256 checksum How to use checksums |
e5c9a6ee20bfa81f67fe321a48d671ea96ce2559c3fe60b851091e0b941f1ce9
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / tai42_backend_arq-7.0.1-py3-none-any.whl
| Download URL | tai42_backend_arq-7.0.1-py3-none-any.whl |
|---|---|
| Size | 43.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
6a678335b74ae6b7c06323a39849a01629320ce60aee61b3191d263a836dc428
|
|
BLAKE2b-256 checksum How to use checksums |
f00fd008ee93417c23bf0a1beac4f737c9dd443ead367695a409b3f9c2205cd7
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Security for tai42-backend-arq 7.0.1
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
