AsyncMQ

Async task queues, workers, retries, scheduling, and operations visibility for Python.

Documentation: https://asyncmq.dymmond.com

Source Code: https://github.com/dymmond/asyncmq

Supported Version: the latest released version is the supported version.

AsyncMQ is a background job runtime for Python services built on asyncio and anyio. It gives applications a queue API, worker runtime, retry and dead-letter behavior, repeatable scheduling, flow primitives, multiple backends, a CLI, and a packaged Lilya/Jinja operations dashboard.

Why AsyncMQ

• Task registration for Python services with @task, .enqueue(), .delay(), and .send().
• Queue and worker APIs for retries, backoff, delayed jobs, cancellation, pause/resume, cleanup, and DLQ operations.
• Backend options for Redis, PostgreSQL, MongoDB, RabbitMQ, and local memory-backed development.
• A production operations console that is packaged with AsyncMQ and works without Node.js or a frontend build pipeline.
• Clear runtime ownership: workers own execution, backends own durable queue state, and the dashboard consumes that state.

AsyncMQ is not a hosted queue service and does not promise exactly-once execution. Production task handlers should be idempotent and safe to retry.

Install

pip install asyncmq

Optional backend extras:

pip install "asyncmq[postgres]"
pip install "asyncmq[mongo]"
pip install "asyncmq[aio-pika]"
pip install "asyncmq[all]"

Quickstart

Start with the in-memory backend for local development:

# myapp/settings.py
from asyncmq.backends.memory import InMemoryBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = InMemoryBackend()
    worker_concurrency = 1

export ASYNCMQ_SETTINGS_MODULE=myapp.settings.AppSettings

Define a task:

# myapp/tasks.py
from asyncmq.tasks import task


@task(queue="emails", retries=2, ttl=300)
async def send_welcome(email: str) -> str:
    return f"sent welcome email to {email}"

Enqueue work:

# producer.py
import anyio

from asyncmq.queues import Queue
from myapp.tasks import send_welcome


async def main() -> None:
    queue = Queue("emails")
    job_id = await send_welcome.enqueue("alice@example.com", backend=queue.backend)
    print("enqueued", job_id)


anyio.run(main)

Run a worker:

asyncmq worker start emails --concurrency 1

Inspect from the CLI:

asyncmq queue list
asyncmq queue info emails
asyncmq job list --queue emails --state waiting
asyncmq job list --queue emails --state failed

Production Backend Example

Use a shared backend configuration for producers, workers, and the dashboard.

# myapp/settings.py
from asyncmq.backends.redis import RedisBackend
from asyncmq.conf.global_settings import Settings
from asyncmq.core.utils.dashboard import DashboardConfig


class AppSettings(Settings):
    secret_key = "replace-with-a-secret-from-your-secret-manager"
    backend = RedisBackend("redis://redis:6379/0")
    worker_concurrency = 8
    scan_interval = 1.0

    @property
    def dashboard_config(self) -> DashboardConfig:
        return DashboardConfig(
            secret_key=self.secret_key,
            dashboard_url_prefix="/asyncmq",
            path="/asyncmq",
            https_only=True,
        )

Workers and the dashboard can run in different services as long as they use the same ASYNCMQ_SETTINGS_MODULE and backend credentials.

Operations Dashboard

AsyncMQ includes a native dashboard built with Lilya, Jinja templates rendered by the server, and packaged static assets.

# myapp/dashboard.py
from lilya.apps import Lilya

from asyncmq.contrib.dashboard.admin import AsyncMQAdmin

app = Lilya()
admin = AsyncMQAdmin(
    enable_login=True,
    backend=auth_backend,  # Provide an AuthBackend implementation.
    url_prefix="/asyncmq",
)
admin.include_in(app)

The dashboard supports queue inspection, worker health, job lists, failed job tracebacks, DLQ actions, repeatables, metrics, runtime events, audit history, and deployments behind reverse proxies at /, /asyncmq/, and nested prefixes such as /operations/asyncmq/.

Read the Dashboard guide for authentication, separate dashboard/worker services, proxy setup, and Nginx examples.

Runtime Shape

flowchart LR
    Producer["Producer service"] --> Task["@task enqueue"]
    Task --> Queue["Queue API"]
    Queue --> Backend["Backend state"]
    Backend --> Worker["Worker runtime"]
    Worker --> Handler["Task handler"]
    Worker --> Backend
    Backend --> Dashboard["Operations dashboard"]
    Backend --> CLI["asyncmq CLI"]

Documentation Map

• Installation
• Quickstart
• Core Concepts
• Queues
• Workers
• Jobs
• Schedulers
• Flows
• CLI Reference
• Dashboard
• Production Operations
• Troubleshooting