{"id":"0c0942a9-99cc-4f49-abe2-ba943e5b860f","slug":"pgtaskqueue","kind":"mcp","name":"pgtaskqueue","summary":"A task queue whose only infrastructure is PostgreSQL: SKIP LOCKED claims, asyncio workers, schedules, webhooks, and an optional FastAPI HTTP/MCP API with JWT auth and a web UI.","owner":"mamedshahmaliyev","repoUrl":"https://github.com/mamedshahmaliyev/pgtaskqueue","license":"MIT","licenseTier":"permissive","trust":"community","stars":0,"match":null,"fewStars":true,"format":"mcp","createdAt":"2026-10-18T16:10:00.000Z","addedAt":"2026-10-21T09:54:59.718Z","logoUrl":"https://avatars.githubusercontent.com/u/20942559?v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["databases","productivity","design-media"],"tags":["asyncio","background-jobs","fastapi","job-queue","postgresql","python","task-queue"],"runtime":"python","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-26T11:38:23.108Z","sample":false,"url":"https://conduits.dev/tools/pgtaskqueue","language":"Python","transports":["stdio"],"install":{"pypi":"pgtaskqueue"},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# pgtaskqueue\n\nA task queue for Python that lives in the PostgreSQL you already run.\n\nTasks are rows. Workers claim them with `FOR UPDATE SKIP LOCKED` and run them with\n`asyncio`. There is no broker, no result backend and no scheduler process: one database,\nyour workers, and optionally an HTTP API.\n\nTo try it without installing anything, use the [public demo](#try-the-demo).\n\n```python\nfrom pgtaskqueue import TaskContext, task\n\n\n@task(\"report.build\", max_attempts=5, timeout=300)\nasync def build_report(ctx: TaskContext, payload: dict) -> dict:\n    ctx.log(f\"building for {payload['tenant']}\")\n    # A step's result is saved on the task: a retry resumes after it, not before it.\n    path = await ctx.step(\"extract\", lambda: extract_to_file(payload[\"tenant\"]))\n    count = await ctx.step(\"load\", lambda: load(path))\n    return {\"rows\": count}\n```\n\n```python\nawait queue.enqueue(\"report.build\", {\"tenant\": \"acme\"}, metadata={\"tenant\": \"acme\"})\n```\n\n```bash\ntq worker --executors myapp.tasks --slots 50\n```\n\n```sql\nSELECT metadata->>'tenant' AS tenant, count(*), avg(spent_seconds)\nFROM tq_tasks WHERE name = 'report.build' AND completed_at > now() - interval '7 days'\nGROUP BY 1;\n```\n\n## Try the demo\n\nA running copy, with the web UI, the HTTP API, the MCP server and two workers:\n\n| | |\n|---|---|\n| Web UI | <https://pgtaskqueue.mmdsh.org> |\n| API reference | <https://pgtaskqueue.mmdsh.org/docs> |\n| MCP server | `https://pgtaskqueue.mmdsh.org/mcp` |\n\nSign in with one of these accounts. The password for all of them is **`demo-password`**.\n\n| Account | Role | What it can do |\n|---|---|---|\n| `demo-manager` | manager | What an operator can, and manage accounts no wider than its own |\n| `demo-operator` | operator | See and manage every task, pause queues, manage schedules |\n| `demo-submitter` | submitter | Enqueue tasks, and see and manage only its own |\n| `demo-viewer` | viewer | See everything, change nothing |\n\nFrom a terminal:\n\n```bash\nURL=https://pgtaskqueue.mmdsh.org\nTOKEN=$(curl -s $URL/v1/auth/token -d grant_type=password \\\n  -d username=demo-submitter -d password=demo-password \\\n  | python3 -c 'import json, sys; print(json.load(sys.stdin)[\"access_token\"])')\n\ncurl -s $URL/v1/tasks -H \"Authorization: Bearer $TOKEN\" \\\n  -H 'Content-Type: application/json' \\\n  -d '{\"name\": \"demo.flaky\", \"payload\": {\"fail_times\": 1}}'\n```\n\nThings to know before you try it:\n\n- **It is public and shared.** Everyone uses the same accounts and sees the same tasks. Put\n  nothing private in a payload.\n- **It is reset every hour.** Your tasks, schedules and accounts are deleted, and the\n  demo data is seeded again.\n- **It is limited.** Each account may have only a few tasks waiting and running, and enqueue\n  only a few a minute. A **429** means wait a minute.\n- **It is not a benchmark.** It runs on a small shared server. Please do not load-test it;\n  run your own copy with the [quick start](#quick-start) instead.\n\n[docs/demo.md](docs/demo.md) has the limits, the task names you can run, what to try in each\nrole, webhooks and MCP.\n\n## Why keep the queue in PostgreSQL\n\nThe point is **manageability, control and analytics**, not speed.\n\n- **You can see everything.** Each task is a row that stays: payload, status, attempts,\n  timings, output, error with traceback, the worker that ran it, who enqueued it. Each log\n  line is a row. Questions about your background work are SQL queries, joined to your own\n  tables.\n- **You can change things while they run.** Reprioritise, reschedule or edit a waiting\n  task. Cancel a running one and have it stop. Pause a queue. Requeue last night's failures\n  once the cause is fixed.\n- **Tasks commit with your data.** Enqueue in the same transaction as the business write,\n  and either both happen or neither does. No outbox, no lost message.\n- **Nothing new to operate.** No broker to size, secure, monitor and back up. Your\n  PostgreSQL backups already cover the queue.\n\nWhat it costs is raw throughput: about 4,000 tasks a second end to end on a laptop, where a\nRedis-backed queue does far more. If your tasks do real work, you will not notice. If you\nneed to move tens of thousands of tiny messages a second, use a broker.\n[docs/comparison.md](docs/comparison.md) is an honest account of when to choose something\nelse, including a list of what this project does not have.\n\n## What is in it\n\n**The queue**\n\n- Atomic claims with `SKIP LOCKED` over a partial index: claim time stays at 2 ms from ten\n  thousand rows to a million\n- Priorities, delays and scheduled times, deduplication keys, batch enqueue\n- Periodic tasks: a cron expression with a timezone, or a fixed interval, declared next\n  to the task or created through the API. No scheduler process: workers fire them\n- Retries with exponential backoff and jitter; `Retry`, `Snooze` and `NonRetryable` to\n  decide per failure; per-task timeouts\n- Heartbeats, and recovery of tasks whose worker died\n- Cancellation that reaches a running task\n- At-least-once delivery, with fencing so a worker that was presumed dead cannot overwrite\n  the task's next attempt\n\n**Workers**\n\n- One `asyncio` event loop per process, structured with `TaskGroup`; thousands of\n  concurrent I/O-bound tasks per process\n- Slots and pools: `--tasks 'image.*:2,ALL:20'` gives each kind of work its own limit\n- Plain `def` executors run in a thread, which is stopped when the task is canceled\n- Each kind of work can have its own image and its own `requirements.txt`; the API never\n  imports task code\n- Graceful shutdown that hands unfinished tasks back\n- Workers that exit when idle (or after N tasks) and are restarted fresh by\n  `restart: unless-stopped`\n- A small HTTP protocol for workers that cannot reach the database, in any language\n\n**Per task**\n\n- Logs stored with the task, including what libraries log while it runs\n- Progress, and named steps whose results survive a retry\n- Child tasks\n- Signed webhook callbacks when it finishes, delivered with retries\n\n**API and access**\n\n- FastAPI REST API with OpenAPI docs, long-polling for results\n- OAuth2 token endpoint, JWT access and refresh tokens, HS256 or key pairs, or your own\n  identity provider\n- Five roles, plus per-account ACLs on queue names and task names\n- An MCP server at `/mcp`, so an AI assistant can use the queue with the same permissions\n- A web UI at `/ui`: a dashboard, the task list with filters, enqueue, cancel, kill and\n  restart, one task with its live log, and statistics. No build step\n\n**Operations**\n\n- `tq` command line: migrate, run, enqueue, inspect, cancel, requeue, pause, schedule,\n  stats, and `tq seed` for demonstration data\n- Docker Compose project: PostgreSQL, migrations, API, two example workers\n- Works behind PgBouncer in transaction mode\n- Alembic migrations that coexist with your application's own\n- Benchmarks you can re-run\n\n## Quick start\n\nYou need Docker. This starts PostgreSQL, the API and two workers with example tasks.\n\n```bash\ngit clone https://github.com/mamedshahmaliyev/pgtaskqueue.git\ncd pgtaskqueue\ncp .env.example .env\n```\n\nEdit `.env` and set the two required values:\n\n```bash\nTQ_JWT_SECRET=<output of: openssl rand -hex 32>\nTQ_BOOTSTRAP_ADMIN_PASSWORD=<a password for the first admin account>\n```\n\nStart it:\n\n```bash\ndocker compose up -d --build\ndocker compose ps\n```\n\nRun a task and wait for its result:\n\n```bash\ndocker compose exec worker tq enqueue demo.echo '{\"message\": \"hello\"}' --wait\n```\n\n```json\n{\n  \"status\": \"succeeded\",\n  \"output\": {\n    \"attempt\": 1,\n    \"message\": \"hello\"\n  },\n  \"error\": null\n}\n```\n\nRun one that fails halfway, is retried, and resumes at the step that failed:\n\n```bash\ndocker compose exec worker tq enqueue report.build '{\"rows\": 500, \"fail_once_at\": \"transform\"}' --wait\ndocker compose exec worker tq tasks --name report.build --limit 1     # copy the id\ndocker compose exec worker tq show <id> --logs\n```\n\n```\n08:39:28.761393 #1 info    [extract] step 'extract': started\n08:39:28.966322 #1 info    [extract] extracted 500 rows\n08:39:28.970236 #1 info    [extract] step 'extract': done\n08:39:28.972667 #1 info    [transform] step 'transform': started\n08:39:28.973673 #1 warning [transform] attempt 1/3 failed: RuntimeError: transform failed on purpose (attempt 1) — retrying in 1.0s\n08:39:30.076241 #2 info    [transform] step 'extract': reusing the result of an earlier attempt\n08:39:30.078463 #2 info    [transform] step 'transform': started\n08:39:30.081190 #2 info    [transform] step 'transform': done\n08:39:30.082206 #2 info    [load] step 'load': started\n08:39:30.198403 #2 info    [load] step 'load': done\n```\n\nA task that only the second worker can run, because only its image has Pillow:\n\n```bash\ndocker compose exec worker tq enqueue image.render '{\"width\": 1600, \"height\": 900}' --wait\n```\n\nSee what happened:\n\n```bash\ndocker compose exec worker tq stats\ndocker compose exec worker tq workers\ndocker compose exec db psql -U tq task_queue -c \\\n  \"SELECT name, status, count(*), round(avg(spent_seconds)::numeric, 3) AS avg_seconds\n   FROM tq_tasks GROUP BY 1, 2 ORDER BY 1, 2\"\n```\n\n### Over HTTP\n\nThe API is on <http://localhost:8010>, with interactive documentation at\n<http://localhost:8010/docs> and the [web UI](docs/ui.md) at <http://localhost:8010/ui>.\nSign in there with the admin account from your `.env`.\n\n```bash\nTOKEN=$(curl -s http://localhost:8010/v1/auth/token \\\n  -d grant_type=password -d username=admin -d password='<your admin password>' \\\n  | python3 -c 'import sys, json; print(json.load(sys.stdin)[\"access_token\"])')\n\ncurl -s http://localhost:8010/v1/tasks \\\n  -H \"Authorization: Bearer $TOKEN\" -H 'Content-Type: application/json' \\\n  -d '{\"name\": \"demo.echo\", \"payload\": {\"message\": \"over http\"}}'\n\n# Wait up to 20 seconds for it to finish, then return it:\ncurl -s \"http://localhost:8010/v1/tasks/<id>?wait=20\" -H \"Authorization: Bearer $TOKEN\"\n```\n\nTo stop: `docker compose down` (add `-v` to delete the data as well).\n\n## Using it in your own project\n\nIt is not on PyPI yet. Install from the repository:\n\n```bash\npip install \"pgtaskqueue @ git+https://github.com/mamedshahmaliyev/pgtaskqueue.git\"          # queue + worker\npip install \"pgtaskqueue[api] @ git+https://github.com/mamedshahmaliyev/pgtaskqueue.git\"     # + HTTP API, auth, MCP\n```\n\nPoint it at a database and create its tables. They are all prefixed `tq_`, so your\napplication's own database is a fine place for them:\n\n```bash\nexport TQ_DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/myapp\ntq migrate\n```\n\nWrite a task:\n\n```python\n# myapp/tasks.py\nimport httpx\nfrom pydantic import BaseModel\n\nfrom pgtaskqueue import Retry, TaskContext, task\n\n\nclass Welcome(BaseModel):\n    user_id: int\n    email: str\n\n\n@task(\"mail.welcome\", max_attempts=5, timeout=30)\nasync def send_welcome(ctx: TaskContext, payload: Welcome) -> dict:\n    async with httpx.AsyncClient() as http:\n        response = await http.post(\"https://mail.example.com/send\", json=payload.model_dump())\n    if response.status_code == 429:\n        raise Retry(\"rate limited\", delay=60)\n    response.raise_for_status()\n    ctx.log(f\"sent to {payload.email}\")\n    return {\"message_id\": response.json()[\"id\"]}\n```\n\nRun a worker for it:\n\n```bash\ntq worker --executors myapp.tasks --slots 50\n```\n\nEnqueue from your application:\n\n```python\nfrom pgtaskqueue import Database, TaskQueue\n\ndb = Database()  # reads TQ_DATABASE_URL\nqueue = TaskQueue(db)\n\ntask = await queue.enqueue(\"mail.welcome\", {\"user_id\": 7, \"email\": \"ada@example.com\"})\n\n# or inside your own transaction, so the task exists if and only if the user does:\nasync with db.transaction() as tx:\n    ...  # insert the user with tx\n    await queue.enqueue(\"mail.welcome\", {\"user_id\": 7, \"email\": \"ada@example.com\"}, session=tx)\n```\n\n[examples/](examples/) has runnable versions of all of this: executors, a second worker\nwith its own dependencies, an API client, a webhook receiver, and a worker that talks only\nHTTP.\n\n## How it performs\n\nMeasured on one laptop (Apple M2 Max), with PostgreSQL 18 in Docker on the same machine and\ntasks that do nothing, so the numbers are the queue's own overhead:\n\n| | |\n|---|---|\n| Enqueue | 2,600/s one at a time from concurrent callers; 23,600/s in batches |\n| End to end | 1,200/s with one worker process; 4,100/s with four |\n| Pickup latency, idle worker | 8 ms median |\n| Concurrent tasks in one process | 10,000, using 129 MB |\n| Claim time at 10,000 rows and at 1,000,000 rows | 2 ms and 2 ms |\n\nThese are not server numbers, and they are not a promise for your workload.\n[docs/benchmarks.md](docs/benchmarks.md) has the method, the full tables, what was not\nmeasured, and how to run the same benchmarks on your hardware in ten minutes.\n\n## Documentation\n\n| | |\n|---|---|\n| [Architecture](docs/architecture.md) | How it works and why: the claim query, fencing, heartbeats, guarantees, the schema |\n| [Writing tasks](docs/executors.md) | Executors, payloads, retries, steps, logs, blocking work, testing |\n| [Periodic tasks](docs/schedules.md) | Cron expressions and intervals, when a schedule fires, missed runs, timezones |\n| [Running workers](docs/workers.md) | Slots and pools, one worker per kind of work, stopping, idle exit |\n| [HTTP API](docs/api.md) | Endpoints, webhooks, the remote-worker protocol |\n| [Auth](docs/auth.md) | Tokens, roles, ACLs, signing keys, external identity providers |\n| [MCP server](docs/mcp.md) | Connecting an AI assistant |\n| [Web UI](docs/ui.md) | The pages, what each role sees, demo data, how the browser session works |\n| [Public demo](docs/demo.md) | The demo's accounts, limits and resets, and how to host one |\n| [Infrastructure](docs/infrastructure.md) | Compose, images, PostgreSQL settings, connections, deploys, backups |\n| [Configuration](docs/configuration.md) | Every `TQ_` setting and every `tq` command |\n| [Analytics](docs/analytics.md) | SQL for throughput, durations, failures, tenants, dashboards |\n| [Benchmarks](docs/benchmarks.md) | Numbers, method and limits |\n| [Comparison](docs/comparison.md) | Against Celery and Redis, other PostgreSQL queues, workflow engines |\n| [AGENTS.md](AGENTS.md) | For contributors and coding agents: layout, invariants, how to change things |\n\n## Status\n\nVersion 0.1. New, and not yet proven in production.\n\n- Developed and tested on PostgreSQL 18 and Python 3.13. The test suite also passes on\n  Python 3.12 and 3.14. No other PostgreSQL version has been tested.\n- Works behind PgBouncer in transaction mode (tested with 1.26). Other poolers, and\n  Kubernetes, have not been tested.\n- The API and the schema may still change before 1.0. Schema changes come as migrations.\n\nIt does not limit how often a kind of task runs, and it has no workflow graphs.\nSee [what this project does not have](docs/comparison.md#what-this-project-does-not-have).\n\n## Development\n\n```bash\nuv sync --all-extras\ndocker compose up -d db\nuv run pytest -q\nuv run ruff check . && uv run ruff format --check .\n```\n\nThe tests run against a real PostgreSQL, in a database of their own (`task_queue_test`).\n[CONTRIBUTING.md](CONTRIBUTING.md) and [AGENTS.md](AGENTS.md) have the rest. To report a\nsecurity problem, see [SECURITY.md](SECURITY.md).\n\n## License\n\n[MIT](LICENSE)\n"}