taskiq

Configure Taskiq background tasks with Redis broker setup and .kiq() enqueue patterns.

Updated Sep 13, 2026
One-click install
npx skills add https://github.com/abdulazeezoj/monovella-poc --skill taskiq-abdulazeezoj
Or copy as Structured Prompt for Agent
Please help me install this Agent Skill.
Skill: taskiq
Source: https://github.com/abdulazeezoj/monovella-poc/tree/main/.agents/skills/taskiq
Command: npx skills add https://github.com/abdulazeezoj/monovella-poc --skill taskiq-abdulazeezoj

SYSTEM DOCUMENTATION & REQUIREMENTS

💡 This Skill requires taskiq, taskiq-redis, and includes references (resource) components.

What problem does it solve? Adding background tasks to a FastAPI project with Taskiq involves non-obvious wiring: tasks silently fail to register if their module is never imported, and the API and worker processes share one broker object with subtle lifecycle rules. This Skill encodes this project's exact conventions so new tasks land correctly the first time. ## Core Features & Use Cases - Task creation guide: Step-by-step walkthrough for adding a new @broker.task function under src/api/tasks/, wiring its import at the bottom of worker.py, and enqueueing it from a route with .kiq(). - Broker and two-process model reference: Explains the Redis ListQueueBroker setup, the inline-call vs. .kiq() distinction, and why the API process only enqueues while a separate taskiq worker process executes. - Gotcha diagnosis: Documents the silent task-discovery failure (missing import in worker.py), the broker.startup() double-run guard, and the Redis socket_timeout idle-worker crash. - Use Case: You add a send_email task, wire an endpoint that returns a task ID, but the task never runs — this Skill points you to the missing import at the bottom of worker.py before you chase broker connectivity. ## Quick Start Use the taskiq skill to add a new background task that sends a welcome email and enqueue it from a signup endpoint.

Frequently Asked Questions about taskiq

High-intent search queries and answers about installing and using this skill.

FAQPage Schema
How do I add a background task with Taskiq in FastAPI?

Create an async function decorated with @broker.task under src/api/tasks/, import that module at the bottom of worker.py after the broker is constructed, then enqueue it from a route with await task.kiq(...). Run a separate worker process with taskiq worker api.worker:broker --app-dir src.

What is the difference between calling a Taskiq task directly and using .kiq()?

Calling await add(1, 2) runs the function inline in the current process like any other function. Calling await add.kiq(1, 2) serializes the call onto the broker queue and returns a TaskiqTask immediately, while a separate worker process executes it later.

Why does my Taskiq worker report zero tasks or task not found?

The worker only discovers tasks whose modules have been imported, since @broker.task registers on import. If worker.py does not import your new tasks module at the bottom of the file, the task never registers even though enqueueing appears to succeed.

Taskiq vs Celery for a FastAPI project — which should I use?

Taskiq is async-first and matches FastAPI's async model, making it the natural fit for async task functions. Celery is the sync counterpart and suits projects that prefer its ecosystem; this project uses Taskiq with a Redis ListQueueBroker.

Why does my idle Taskiq worker crash with a Redis timeout?

ListQueueBroker.listen() issues a blocking BRPOP with no server-side timeout, but redis-py defaults socket_timeout to 5 seconds, raising a TimeoutError on idle polls. This project passes socket_timeout=None to ListQueueBroker to prevent the crash.

How do I test Taskiq tasks without running a worker?

Call the task function directly with await add(1, 2) in unit and integration tests, since the test suite uses ASGITransport without lifespan and no worker consumes the queue. Only e2e tests start a real worker subprocess and call broker.startup().