The Complete Production SOP: How to Deploy and Run RQ (Redis Queue) Lightweight Python Background Jobs in Autonomous AI Agents
Step-by-step SOP to install and run RQ (Redis Queue) in Codex, Claude Code, and Cursor. Includes verified prompts, multi-queue rules, and worker troubleshooting.
Technical Overview & Architectural Context
Autonomous coding agents frequently stumble when architecting asynchronous distributed background tasks. When tasked with asynchronous processing in Python, agents often defaults to bloated Celery configurations, unmanaged `threading.Thread` instances that silently crash during process restarts, or in-memory queues that discard unhandled state upon container redeployment. Without a strict operational SOP, agents commonly omit job timeout limits, ignore Redis connection pooling lifecycles, and fail to implement idempotent retry policies with exponential backoff. Deploying the `rq-redis-python-background-jobs` skill constrains agent execution within deterministic parameters. It guides the model to consistently construct Redis-backed queues using connection pools, define explicit job timeouts, assign tasks to discrete priority queues (`high`, `default`, `low`), and configure resilient `Retry` policies using native RQ abstractions. By constraining file modifications and prohibiting unsanctioned system-wide shell executions, this skill prevents unhandled worker crashes and guarantees reliable background job choreography.
Prerequisites & Environment Requirements
Verified Search Intents & Long-Tail Queries Solved
This operational tutorial is engineered to provide deterministic solutions for these developer search queries:
Step-by-Step Execution Playbook
Local Environment Setup & SKILL.md Mounting
Mount the SKILL.md specification into your agent's global or workspace-level instructions directory.
Autonomous runtimes parse directory structures such as `.cursor/rules/`, `.claude/skills/`, or root `.agents/skills/` to load context and trigger maps. We fetch the official SKILL.md for rq-redis-python-background-jobs and place it in the agent path so triggers like 'enqueue python job rq' activate procedural enforcement.
mkdir -p .agents/skills/rq-redis-python-background-jobs && cat << 'EOF' > .agents/skills/rq-redis-python-background-jobs/SKILL.md
---
name: "rq-redis-python-background-jobs"
description: "Manage simple, lightweight Python background job execution with Redis Queue."
metadata:
author: "RQ"
version: "1.0.0"
runtime_compatibility:
- "codex"
- "claude-code"
- "cursor"
- "windsurf"
triggers:
- "enqueue python job rq"
- "configure rq retry and timeout"
- "run rq worker"
permissions:
fs_read: true
fs_write: "workspace"
shell_exec: false
net_access: false
---
# RQ (Redis Queue) Lightweight Python Background Jobs
Implements background jobs using Python RQ: enqueueing functions, job timeouts, retry policies, failure callbacks, multi-queue prioritization, and monitoring via rq-dashboard.
## When to Invoke
Invoke this skill whenever you need to:
- enqueue python job rq
- configure rq retry and timeout
- run rq worker
## Execution SOP & Rules
1. Initialize queue: q = Queue('high', connection=redis_conn).
2. Enqueue task: job = q.enqueue(generate_pdf, user_id, job_timeout=300, retry=Retry(max=3)).
3. Start worker process: rq worker high default low --with-scheduler.
4. Inspect job status: job.get_status() and retrieve result with job.result once finished.
## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.
EOF---
name: "rq-redis-python-background-jobs"
description: "Manage simple, lightweight Python background job execution with Redis Queue."
metadata:
author: "RQ"
version: "1.0.0"
runtime_compatibility:
- "codex"
- "claude-code"
- "cursor"
- "windsurf"
triggers:
- "enqueue python job rq"
- "configure rq retry and timeout"
- "run rq worker"
permissions:
fs_read: true
fs_write: "workspace"
shell_exec: false
net_access: false
---Place the file under workspace-level rule configs (.cursor/rules or .agents/skills) to allow git tracking across your engineering team without context window waste.
Ensure permissions specify fs_write='workspace' so agents do not attempt writing RQ worker PID files into protected OS-level paths like /var/run.
Dependency Installation & Connection Configuration
Install the necessary Python RQ packages and configure the Redis connection abstraction.
Ensure `rq` and `redis` are installed in the workspace Python environment. Define an isolated Redis connection module that pulls connection configuration from environment variables without hardcoding credentials in the script body.
pip install rq redis rq-dashboard && cat << 'EOF' > connection.py
import os
from redis import Redis
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
redis_conn = Redis.from_url(REDIS_URL)
EOFimport os
from redis import Redis
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
redis_conn = Redis.from_url(REDIS_URL)Never put API tokens or Redis passwords directly in agent prompt instructions; instruct the agent to consume REDIS_URL via environment variables.
Agents commonly attempt to instantiate a new Redis() client inside the enqueued task itself. The worker process handles the connection; tasks should remain pure functions.
Deterministic Task Implementation & Trigger Invocation
Author tasks and queue orchestration using the skill's SOP parameters (timeout, retries, multi-queue prioritization).
The agent must implement the task definition in an importable module (e.g., `tasks.py`) and dispatch it via `q.enqueue()`. Python RQ worker forks processes and unpickles modules; tasks defined in `__main__` fail during worker execution.
cat << 'EOF' > tasks.py
import time
def generate_pdf_report(user_id: int, report_type: str) -> dict:
time.sleep(2) # Simulate compute workload
return {"status": "completed", "user_id": user_id, "report": report_type}
EOFfrom rq import Queue, Retry
from connection import redis_conn
from tasks import generate_pdf_report
high_priority_queue = Queue('high', connection=redis_conn)
def trigger_report_job(user_id: int) -> str:
job = high_priority_queue.enqueue(
generate_pdf_report,
user_id,
"annual_summary",
job_timeout=300,
retry=Retry(max=3, interval=[10, 30, 60]),
result_ttl=3600
)
return job.get_id()
if __name__ == "__main__":
job_id = trigger_report_job(42)
print(f"Enqueued job ID: {job_id}")Always provide the Retry parameter with explicit interval arrays for exponential backoff instead of instantaneous retries.
Never enqueue lambda functions or dynamically defined inner functions; RQ uses pickle, which fails if the function is not top-level importable.
Verification, Linting & Output Quality Gate
Run validation tests to guarantee unpickleable tasks, verify worker consumption, and check syntax.
The skill mandates testing the enqueueing cycle and syntax integrity. Execute flake8, mypy, and a mock test script validating that tasks compile and queue keys register in Redis without exceptions.
python -m py_compile tasks.py producer.py connection.py && python -c 'from producer import trigger_report_job; assert trigger_report_job(1) is not None'#!/usr/bin/env bash
set -euo pipefail
python -m py_compile tasks.py producer.py connection.py
echo "Syntax check passed."
python -c "from tasks import generate_pdf_report; res = generate_pdf_report(99, 'test'); assert res['status'] == 'completed'"
echo "Task logic unit verified."Use `rq info` in your CI/verification pipeline to inspect the queue backlog length directly from Redis without launching a blocking worker.
Avoid running long-lived `rq worker` processes in foreground test scripts inside autonomous agents; workers block process output until terminated.
Real-World Scenario: Asynchronous Batch PDF Report Generation with Retry Policies
“Using rq-redis-python-background-jobs, enqueue python job rq for generating PDF reports for user batch 1001-1050 on the 'high' priority queue. Configure rq retry and timeout: max 3 retries with 300 second timeout. Ensure worker command handles priority gracefully.”
Runtime Troubleshooting & Failure Modes
Root Cause: The agent authored the task function in the same file executed as __main__. When the RQ worker worker process forks, it attempts to import the function by its module name, which fails for '__main__'.
Move the task function into a dedicated module (e.g., tasks.py) and import it explicitly in the producer script: from tasks import task_funcRoot Cause: Redis service is not running in the current sandbox or the agent is referencing an invalid REDIS_URL environment variable.
Start the local redis instance or update configuration: export REDIS_URL='redis://localhost:6379/0' && redis-server --daemonize yesFrequently Asked Questions (RQ (Redis Queue) Lightweight Python Background Jobs SOP)
Q:Can I use RQ (Redis Queue) Lightweight Python Background Jobs across Codex, Claude Code, and Cursor simultaneously?
Yes. The skill conforms to the universal SKILL.md specification. Runtimes parse the frontmatter metadata and trigger lists automatically across workspace environments.
Q:How does RQ (Redis Queue) Lightweight Python Background Jobs prevent context window overflow during long agent sessions?
By providing concise, declarative SOP rules (connection handling, enqueue syntax, and worker launch patterns) directly in SKILL.md, preventing the agent from generating verbose, multi-page Celery boilerplate or looping through failed debugging trials.
Q:Does this skill require escalated root or container permissions?
No. The skill has a 98/100 security score with shell_exec=false and net_access=false specified in its default manifest. It restricts agent writes exclusively to the workspace directory.
Ready to equip your autonomous agent with RQ (Redis Queue) Lightweight Python Background Jobs?
Copy the complete audited SKILL.md specification or download the 1-click zip bundle directly from our verified registry.