# What Is the Purpose of the Task Queue in Calliope?

> Discover the purpose of the task queue in Calliope. It allows asynchronous background work by decoupling heavy processing from the HTTP request cycle, improving API responsiveness.

- Repository: [chrisimmel/calliope](https://github.com/chrisimmel/calliope)
- Tags: internals
- Published: 2026-02-27

---

**The task queue in Calliope enables asynchronous, long-running background work by decoupling computationally expensive operations from the HTTP request cycle, allowing the API to return immediately while heavy processing continues in the background.**

The task queue is a foundational component of the [chrisimmel/calliope](https://github.com/chrisimmel/calliope) open-source storytelling engine. It provides a robust mechanism for handling time-consuming tasks—such as generating story frames, running image analysis, or converting audio to text—without blocking the main application thread or risking request timeouts.

## Decoupling Heavy Processing from HTTP Requests

Long-running operations in Calliope can take seconds to minutes to complete. By pushing these jobs onto the **task queue**, the API returns an immediate response to clients while the actual work proceeds asynchronously. This architecture prevents request timeouts and maintains system responsiveness during intensive workloads like frame generation or media analysis.

## Unified Abstraction for Multiple Runtimes

The task queue provides a consistent interface regardless of the deployment environment. In development, Calliope uses an in-memory implementation called `LocalTaskQueue`, while production environments can switch to **Google Cloud Tasks (`GCPTaskQueue`)** without modifying calling code.

The factory function in [`calliope/tasks/factory.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/factory.py) handles this runtime selection automatically:

```python

# In calliope/tasks/factory.py

_task_queue_instance = None

def configure_task_queue():
    global _TASK_QUEUE_INSTANCE
    if _TASK_QUEUE_INSTANCE is None:
        _TASK_QUEUE_INSTANCE = get_task_queue()
        if isinstance(_TASK_QUEUE_INSTANCE, LocalTaskQueue):
            from .handlers import register_handlers
            register_handlers(_TASK_QUEUE_INSTANCE)
    return _TASK_QUEUE_INSTANCE

```

This pattern ensures that task enqueueing and status checking remain identical across local development and cloud deployments.

## Tracking Task Execution State

Every task in the queue is represented by a `Task` model containing metadata such as `task_id`, `task_type`, `status`, and timestamps. These status updates are persisted to Firebase, enabling real-time monitoring through both the user interface and the dedicated API endpoint.

The status endpoint in [`calliope/routes/v2/tasks.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/routes/v2/tasks.py) exposes this state:

```python

# calliope/routes/v2/tasks.py

@router.get("/status/{task_id}")
async def get_task_status(task_id: str):
    task_queue = configure_task_queue()
    status = await task_queue.get_task_status(task_id)
    if not status:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return status

```

Clients can poll `GET /v2/tasks/status/{task_id}` to track progress or detect failures.

## Extensibility Through Task Handlers

Concrete work is performed by async functions registered in [`calliope/tasks/handlers.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/handlers.py). The queue dispatches tasks to these handlers based on the `task_type` string provided during enqueueing.

For example, the `"add_frame"` task type maps to the `add_frame_task` coroutine. New task types can be added by implementing additional handlers and registering them via the `register_handlers` function, which is automatically invoked at startup when using `LocalTaskQueue`.

## Enqueuing and Executing Background Jobs

To enqueue a task from application code, use the factory to obtain a queue instance and call `enqueue`:

```python
from calliope.tasks.factory import configure_task_queue

# Get a ready‑to‑use queue (local or GCP depending on environment)

task_queue = configure_task_queue()

payload = {
    "story_id": "c12345",
    "client_id": "client_xyz",
    "snippets": [
        {"snippet_type": "text", "content": "A sunrise over mountains"}
    ],
    # optional: source_ip_address, extra_fields, etc.

}

# Enqueue the job – returns the generated task_id

task_id = await task_queue.enqueue("add_frame", payload)
print(f"Enqueued add_frame task, id={task_id}")

```

The `enqueue` method is defined in the abstract interface [`calliope/tasks/queue.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/queue.py) and implemented in [`calliope/tasks/local_queue.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/local_queue.py) for development environments.

## Summary

- **Asynchronous Processing**: The task queue offloads expensive operations from the HTTP request cycle, preventing timeouts and keeping the API responsive.
- **Environment Agnostic**: A factory pattern in [`calliope/tasks/factory.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/factory.py) switches between `LocalTaskQueue` and `GCPTaskQueue` without changing application code.
- **State Persistence**: Task metadata and status updates are stored in Firebase, accessible via the `/v2/tasks/status/{task_id}` endpoint.
- **Modular Handlers**: Work is executed by registered handlers in [`calliope/tasks/handlers.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/handlers.py), making it straightforward to extend with new background job types.

## Frequently Asked Questions

### How does Calliope choose between local and GCP task queues?

The `configure_task_queue()` function in [`calliope/tasks/factory.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/factory.py) inspects the environment configuration to determine which implementation to instantiate. For local development, it returns `LocalTaskQueue`, while production deployments can return `GCPTaskQueue` based on environment variables or settings, ensuring seamless transitions between environments.

### What types of operations should use the task queue in Calliope?

Any operation that takes significant time or resources should use the queue, including frame generation for stories, image analysis, audio-to-text conversion, and other AI or media processing tasks. This ensures the main API thread remains available to handle new requests while expensive work runs in the background.

### How can I check the status of a background task in Calliope?

Query the `GET /v2/tasks/status/{task_id}` endpoint, which retrieves the current state from Firebase via the queue's `get_task_status` method implemented in [`calliope/tasks/local_queue.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/local_queue.py) (or the GCP variant in production). The response includes the task's current status, timestamps, and any error details if the job failed.

### How do I add custom task handlers to Calliope?

Create a new async function in [`calliope/tasks/handlers.py`](https://github.com/chrisimmel/calliope/blob/main/calliope/tasks/handlers.py) and register it within the `register_handlers()` function to map a task type string to your handler. For `LocalTaskQueue`, this registration happens automatically at startup when `configure_task_queue()` is first called, binding the task type to your handler function.