What Is the Purpose of the Task Queue in Calliope?

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 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 handles this runtime selection automatically:

# 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 exposes this state:

# 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. 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:

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 and implemented in 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 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, 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 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 (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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}