Async AI Tasks: asyncio.create_task, gather, Timeout, and Cancellation for LLM Workloads in 2026

TL;DR — Async AI tasks: create_task, gather, timeout, cancellation, BackgroundTasks. FastAPI Async: "Starlette and FastAPI based on AnyIO, compatible with asyncio and Trio. Python main language for ML/DL — FastAPI very good match." FastAPI BackgroundTasks: "BackgroundTasks object, async def or normal def. Tied to process lifespan — lost on crash." OneUptime: "Simple BackgroundTasks for lightweight work. Celery or custom queue for heavy CPU-bound." Dev.to: "BackgroundTasks tied to process lifespan. For tasks that must survive restarts, use external queue." Learn more with async backend, FastAPI tutorial, job queue, and streaming SSE.

FastAPI Docs introduces the built-in option: "FastAPI creates the BackgroundTasks object and passes it as parameter. Function can be async def or normal def — FastAPI handles correctly. If no need for task to wait, use BackgroundTasks."

Dev.to adds the caveat: "Like raw asyncio tasks, BackgroundTasks are tied to the lifespan of the FastAPI process. If the server crashes or restarts, any uncompleted BackgroundTasks are lost. For tasks that must survive restarts, use external queue."

Async AI Task Architecture

Async Task API Reference

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class TaskPattern(Enum):
    CONCURRENT = "concurrent"
    TIMEOUT = "timeout"
    CANCEL = "cancel"
    TRACKING = "tracking"

@dataclass
class AsyncAITasksGuide:
    """Async AI tasks implementation guide."""

    def get_concurrent_tasks(self) -> str:
        """Concurrent LLM tasks with gather."""
        return (
            "# === CONCURRENT TASKS ===\n"
            "import asyncio\n"
            "import httpx\n"
            "import json\n"
            "\n"
            "OLLAMA_URL = (\n"
            "    'http://localhost:11434')\n"
            "\n"
            "async def call_llm(\n"
            "    prompt: str,\n"
            "    model: str = 'llama3.2'):\n"
            "    '''Call LLM asynchronously.''' \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=300.0\n"
            "    ) as client:\n"
            "        resp = await client.post(\n"
            "            f'{OLLAMA_URL}/api/chat',\n"
            "            json={\n"
            "                'model': model,\n"
            "                'messages': [{\n"
            "                  'role': 'user',\n"
            "                  'content': prompt}]})\n"
            "        return resp.json()['message']\n"
            "          ['content']\n"
            "\n"
            "# Pattern 1: gather\n"
            "async def multi_prompt(prompts):\n"
            "    '''Run multiple LLM calls\n"
            "    concurrently.''' \n"
            "    tasks = [\n"
            "        asyncio.create_task(\n"
            "            call_llm(p))\n"
            "        for p in prompts]\n"
            "    results = await (\n"
            "      asyncio.gather(\n"
            "        *tasks,\n"
            "        return_exceptions=True))\n"
            "    # return_exceptions collects\n"
            "    #   errors instead of raising\n"
            "    return [\n"
            "        r if not isinstance(\n"
            "          r, Exception) else None\n"
            "        for r in results]\n"
            "\n"
            "# Pattern 2: TaskGroup (3.11+)\n"
            "async def task_group(prompts):\n"
            "    '''Structured concurrency.''' \n"
            "    results = {}\n"
            "    async with asyncio.TaskGroup(\n"
            "      ) as tg:\n"
            "        for i, p in enumerate(\n"
            "          prompts):\n"
            "            task = tg.create_task(\n"
            "                call_llm(p))\n"
            "            results[i] = task\n"
            "    # All tasks complete here\n"
            "    return {i: t.result()\n"
            "      for i, t in results.items()}\n"
            "\n"
            "# Pattern 3: wait FIRST_COMPLETED\n"
            "async def first_response(prompts):\n"
            "    '''Return first LLM to\n"
            "    respond.''' \n"
            "    tasks = [\n"
            "        asyncio.create_task(\n"
            "            call_llm(p))\n"
            "        for p in prompts]\n"
            "    done, pending = await (\n"
            "      asyncio.wait(\n"
            "        tasks,\n"
            "        return_when=\n"
            "          asyncio\n"
            "          .FIRST_COMPLETED))\n"
            "    # Cancel remaining\n"
            "    for t in pending:\n"
            "        t.cancel()\n"
            "    # Return first result\n"
            "    for t in done:\n"
            "        return t.result()"
        )

    def get_timeout(self) -> str:
        """Timeout handling for AI tasks."""
        return (
            "# === TIMEOUT ===\n"
            "\n"
            "# Pattern 1: wait_for\n"
            "async def llm_with_timeout(\n"
            "    prompt, timeout=300):\n"
            "    '''LLM call with timeout.''' \n"
            "    try:\n"
            "        result = await (\n"
            "          asyncio.wait_for(\n"
            "            call_llm(prompt),\n"
            "            timeout=timeout))\n"
            "        return result\n"
            "    except asyncio.TimeoutError:\n"
            "        # Task cancelled by\n"
            "        #   wait_for on timeout\n"
            "        return {'error': 'timeout'}\n"
            "\n"
            "# Pattern 2: timeout context\n"
            "#   (Python 3.11+)\n"
            "async def llm_scoped_timeout(\n"
            "    prompt, timeout=300):\n"
            "    '''Scoped timeout with\n"
            "    context manager.''' \n"
            "    try:\n"
            "        async with asyncio.timeout(\n"
            "            timeout):\n"
            "            result = await (\n"
            "              call_llm(prompt))\n"
            "            return result\n"
            "    except TimeoutError:\n"
            "        return {'error': 'timeout'}\n"
            "\n"
            "# Pattern 3: per-task timeout\n"
            "async def multi_with_timeout(\n"
            "    prompts, timeout=300):\n"
            "    '''Multiple tasks each\n"
            "    with timeout.''' \n"
            "    tasks = [\n"
            "        asyncio.wait_for(\n"
            "            call_llm(p),\n"
            "            timeout=timeout)\n"
            "        for p in prompts]\n"
            "    results = await (\n"
            "      asyncio.gather(\n"
            "        *tasks,\n"
            "        return_exceptions=True))\n"
            "    return results"
        )

    def get_cancellation(self) -> str:
        """Task cancellation patterns."""
        return (
            "# === CANCELLATION ===\n"
            "\n"
            "class TaskRegistry:\n"
            "    '''Registry for tracking\n"
            "    and cancelling tasks.''' \n"
            "    \n"
            "    def __init__(self):\n"
            "        self.tasks = {}  # id -> Task\n"
            "    \n"
            "    def create(self, task_id,\n"
            "        coro):\n"
            "        task = asyncio.create_task(\n"
            "            coro)\n"
            "        self.tasks[task_id] = task\n"
            "        return task\n"
            "    \n"
            "    async def cancel(self,\n"
            "        task_id):\n"
            "        if task_id in self.tasks:\n"
            "            task = self.tasks[\n"
            "              task_id]\n"
            "            task.cancel()\n"
            "            try:\n"
            "                await task\n"
            "            except (\n"
            "              asyncio\n"
            "              .CancelledError):\n"
            "                pass\n"
            "            del self.tasks[\n"
            "              task_id]\n"
            "    \n"
            "    def remove(self, task_id):\n"
            "        self.tasks.pop(\n"
            "          task_id, None)\n"
            "\n"
            "registry = TaskRegistry()\n"
            "\n"
            "# Cancel on client disconnect\n"
            "from fastapi import Request\n"
            "\n"
            "@app.post('/v1/generate')\n"
            "async def generate(\n"
            "    request: Request,\n"
            "    prompt: str):\n"
            "    task_id = str(uuid.uuid4())\n"
            "    \n"
            "    async def run_and_track():\n"
            "        try:\n"
            "            result = await (\n"
            "              call_llm(prompt))\n"
            "            # Store result in Redis\n"
            "            await redis.set(\n"
            "                f'task:{task_id}',\n"
            "                json.dumps({\n"
            "                  'status':'complete',\n"
            "                  'result': result}))\n"
            "        except asyncio\n"
            "          .CancelledError:\n"
            "            # Cleanup on cancel\n"
            "            await redis.set(\n"
            "                f'task:{task_id}',\n"
            "                json.dumps({\n"
            "                  'status':'cancelled'}))\n"
            "            raise  # Re-raise\n"
            "        except Exception as e:\n"
            "            await redis.set(\n"
            "                f'task:{task_id}',\n"
            "                json.dumps({\n"
            "                  'status':'failed',\n"
            "                  'error': str(e)}))\n"
            "        finally:\n"
            "            registry.remove(task_id)\n"
            "    \n"
            "    registry.create(\n"
            "        task_id, run_and_track())\n"
            "    return {'task_id': task_id}\n"
            "\n"
            "# Shield critical cleanup\n"
            "async def safe_cleanup():\n"
            "    '''Shield from cancellation.''' \n"
            "    await asyncio.shield(\n"
            "        close_connections())"
        )

    def get_background_tasks(self) -> str:
        """FastAPI BackgroundTasks for AI."""
        return (
            "# === FASTAPI BACKGROUNDTASKS ===\n"
            "from fastapi import BackgroundTasks\n"
            "\n"
            "async def log_usage(\n"
            "    user_id, tokens, model):\n"
            "    '''Log token usage.''' \n"
            "    # Non-blocking log\n"
            "    await redis.incrby(\n"
            "        f'usage:{user_id}',\n"
            "        tokens)\n"
            "\n"
            "async def update_cache(\n"
            "    key, value):\n"
            "    '''Update cache.''' \n"
            "    await redis.set(\n"
            "        key, value, ex=3600)\n"
            "\n"
            "@app.post('/v1/chat')\n"
            "async def chat(\n"
            "    prompt: str,\n"
            "    bg: BackgroundTasks):\n"
            "    '''Chat with background\n"
            "    logging.''' \n"
            "    result = await call_llm(\n"
            "        prompt)\n"
            "    \n"
            "    # Add background tasks\n"
            "    bg.add_task(\n"
            "        log_usage,\n"
            "        user_id='user1',\n"
            "        tokens=100,\n"
            "        model='llama3.2')\n"
            "    bg.add_task(\n"
            "        update_cache,\n"
            "        f'cache:{hash(prompt)}',\n"
            "        result)\n"
            "    \n"
            "    # Response sent immediately\n"
            "    #   tasks run after\n"
            "    return {'result': result}\n"
            "\n"
            "# WARNING: BackgroundTasks\n"
            "#   lost on server crash.\n"
            "#   For persistent tasks,\n"
            "#   use Arq or Celery."
        )

    def get_task_tracking(self) -> str:
        """Long-running task tracking with Redis."""
        return (
            "# === TASK TRACKING ===\n"
            "import uuid\n"
            "import json\n"
            "from fastapi import FastAPI\n"
            "\n"
            "app = FastAPI()\n"
            "\n"
            "@app.post('/v1/async-generate')\n"
            "async def async_generate(\n"
            "    prompt: str):\n"
            "    '''Submit long-running task.''' \n"
            "    task_id = str(uuid.uuid4())\n"
            "    \n"
            "    # Store initial status\n"
            "    await redis.set(\n"
            "        f'task:{task_id}',\n"
            "        json.dumps({\n"
            "            'status': 'pending',\n"
            "            'result': None}),\n"
            "        ex=3600)  # 1h TTL\n"
            "    \n"
            "    # Create background task\n"
            "    async def run():\n"
            "        await redis.set(\n"
            "            f'task:{task_id}',\n"
            "            json.dumps({\n"
            "                'status':'running'}))\n"
            "        try:\n"
            "            result = await (\n"
            "              call_llm(prompt))\n"
            "            await redis.set(\n"
            "                f'task:{task_id}',\n"
            "                json.dumps({\n"
            "                  'status':'complete',\n"
            "                  'result': result}),\n"
            "                ex=3600)\n"
            "        except Exception as e:\n"
            "            await redis.set(\n"
            "                f'task:{task_id}',\n"
            "                json.dumps({\n"
            "                  'status':'failed',\n"
            "                  'error': str(e)}),\n"
            "                ex=3600)\n"
            "    \n"
            "    asyncio.create_task(run())\n"
            "    return {'task_id': task_id}\n"
            "\n"
            "@app.get('/v1/tasks/{task_id}')\n"
            "async def get_task_status(\n"
            "    task_id: str):\n"
            "    '''Poll task status.''' \n"
            "    data = await redis.get(\n"
            "        f'task:{task_id}')\n"
            "    if not data:\n"
            "        raise HTTPException(\n"
            "          404, 'Not found')\n"
            "    return json.loads(data)\n"
            "\n"
            "@app.post(\n"
            "  '/v1/tasks/{task_id}/cancel')\n"
            "async def cancel_task(\n"
            "    task_id: str):\n"
            "    '''Cancel running task.''' \n"
            "    await registry.cancel(task_id)\n"
            "    return {'status': 'cancelled'}"
        )

    def get_production_tips(self) -> dict:
        """Production tips."""
        return {
            "create_task": "asyncio.create_task schedules coroutine on event loop. Returns Task object. Fire-and-forget if not awaited.",
            "gather": "asyncio.gather runs tasks concurrently. return_exceptions=True to collect errors. Not sequential.",
            "taskgroup": "asyncio.TaskGroup (3.11+) for structured concurrency. Automatic cancellation on error. Better error propagation.",
            "wait_for": "asyncio.wait_for sets timeout. Raises TimeoutError. Cancels task on timeout. Set 300s for LLM, 60s for embeddings.",
            "timeout_ctx": "asyncio.timeout (3.11+) context manager. Scoped timeout without cancelling task directly.",
            "cancel": "task.cancel() raises CancelledError. Cleanup in finally. Close connections, stop LLM generation.",
            "shield": "asyncio.shield protects from cancellation. Use for critical cleanup that must complete.",
            "background": "BackgroundTasks for lightweight post-response work. Lost on crash. Use Arq/Celery for persistence.",
            "tracking": "Store task status in Redis. task_id, status (pending/running/complete/failed), result. TTL for cleanup.",
            "registry": "TaskRegistry dict for active tasks. Cancel by task_id. Remove on completion. Clean up on disconnect.",
        }

Async AI Tasks Checklist

• [ ] asyncio.create_task: schedule coroutine on event loop, returns Task object, fire-and-forget
• [ ] asyncio.gather: wait for multiple tasks concurrently — not sequential
• [ ] asyncio.gather: return_exceptions=True to collect errors instead of raising
• [ ] asyncio.TaskGroup (Python 3.11+): structured concurrency, automatic cancellation on error
• [ ] asyncio.wait: more control — FIRST_COMPLETED, ALL_COMPLETED, return_when
• [ ] asyncio.wait_for: timeout for task, raises TimeoutError, cancels task on timeout
• [ ] asyncio.timeout (Python 3.11+): context manager for scoped timeout
• [ ] Set timeout based on task type: 300s for LLM, 60s for embeddings
• [ ] Handle TimeoutError gracefully — return error or retry
• [ ] Don't use signal.alarm — not async-safe
• [ ] httpx timeout for HTTP-level control in addition to asyncio timeout
• [ ] Combine timeout with semaphore for bounded concurrency
• [ ] task.cancel(): cancel running task, raises CancelledError
• [ ] try/except CancelledError for cleanup in task
• [ ] finally block to close connections and stop LLM generation
• [ ] await task after cancel to suppress CancelledError if needed
• [ ] asyncio.gather(..., return_exceptions=True) to handle cancellation in group
• [ ] Cancel on client disconnect in FastAPI
• [ ] Cancel on timeout
• [ ] Cancel on user interrupt (WebSocket message)
• [ ] asyncio.shield(): protect critical operations from cancellation — cleanup must complete
• [ ] FastAPI BackgroundTasks: lightweight post-response work, async def or normal def
• [ ] BackgroundTasks: response sent immediately, task runs after
• [ ] BackgroundTasks: for email, logging, cache update — NOT heavy CPU-bound
• [ ] BackgroundTasks: tied to process lifespan — lost on server crash or restart
• [ ] For tasks that must survive restarts: use external queue (Arq, Celery)
• [ ] Task tracking: store task_id, status (pending/running/complete/failed), result in Redis
• [ ] Task tracking: GET /tasks/{task_id} for status polling
• [ ] Task tracking: GET /tasks/{task_id}/result for result when complete
• [ ] Task tracking: TTL on Redis keys for cleanup (1h)
• [ ] Task tracking: WebSocket push for real-time status updates
• [ ] Task tracking: webhook callback when task complete
• [ ] Task tracking: progress tracking — update percentage in Redis
• [ ] TaskRegistry: dict of active tasks for cancellation by task_id
• [ ] TaskRegistry: remove on completion, clean up on disconnect
• [ ] Fan-out pattern: gather multiple LLM calls for parallel model queries
• [ ] Pipeline pattern: chained async tasks (extract → embed → store)
• [ ] Fire-and-forget: create_task without await for logging, notifications
• [ ] Long-running: task tracking + polling + Redis status storage
• [ ] Starlette and FastAPI based on AnyIO — compatible with asyncio and Trio
• [ ] Python is main language for ML/DL — FastAPI very good match for Data Science
• [ ] Read async backend for async patterns
• [ ] Read FastAPI tutorial for API setup
• [ ] Read job queue for persistent tasks
• [ ] Read streaming SSE for streaming
• [ ] Test: concurrent tasks complete faster than sequential
• [ ] Test: timeout cancels task and returns error
• [ ] Test: cancellation stops LLM generation and cleans up
• [ ] Test: BackgroundTasks run after response sent
• [ ] Test: task status transitions: pending → running → complete
• [ ] Test: task cancellation via API endpoint
• [ ] Test: gather with return_exceptions handles errors gracefully
• [ ] Test: shield protects critical cleanup from cancellation
• [ ] Document task patterns, timeout strategy, cancellation flow, tracking schema, registry design

FAQ

How do you run concurrent AI tasks with asyncio in Python?

Use asyncio.create_task to schedule coroutines concurrently and asyncio.gather to wait for multiple tasks. FastAPI Async: "Starlette and FastAPI based on AnyIO, compatible with asyncio and Trio. Python is main language for Data Science, Machine Learning, Deep Learning — FastAPI very good match." Concurrent tasks: (1) task1 = asyncio.create_task(call_llm(prompt1)). (2) task2 = asyncio.create_task(call_llm(prompt2)). (3) results = await asyncio.gather(task1, task2). (4) gather runs tasks concurrently — not sequential. (5) return_exceptions=True to collect errors instead of raising. (6) asyncio.wait() for more control (FIRST_COMPLETED, ALL_COMPLETED). (7) asyncio.TaskGroup (Python 3.11+) for structured concurrency. (8) Each task runs on same event loop — no threads needed for I/O.

How do you handle timeouts for long-running AI tasks?

Use asyncio.wait_for to set a timeout and asyncio.timeout (Python 3.11+) as a context manager. Timeout: (1) result = await asyncio.wait_for(call_llm(prompt), timeout=300). (2) Raises asyncio.TimeoutError if exceeded. (3) asyncio.timeout(300) as async context manager (Python 3.11+). (4) Cancel task on timeout to stop LLM generation. (5) Set timeout based on task type: 300s for LLM, 60s for embeddings. (6) Handle TimeoutError gracefully — return error or retry. (7) Don't use signal.alarm — not async-safe. (8) httpx timeout for HTTP-level control. (9) Combine with semaphore for bounded concurrency. (10) Log timeout events for monitoring.

How do you cancel async AI tasks in Python?

Use task.cancel() to cancel a running task and handle asyncio.CancelledError. Cancellation: (1) task = asyncio.create_task(call_llm(prompt)). (2) task.cancel() to cancel. (3) Raises CancelledError in the task. (4) try/except CancelledError for cleanup. (5) finally block to close connections. (6) await task to suppress cancellation if needed. (7) asyncio.gather(..., return_exceptions=True) to handle cancellation in group. (8) Cancel on client disconnect in FastAPI. (9) Cancel on timeout. (10) Cancel on user interrupt (WebSocket message). (11) Clean up resources: close httpx client, stop LLM generation. (12) Shielded tasks: asyncio.shield() to prevent cancellation of critical operations.

How do you use FastAPI BackgroundTasks for AI workloads?

Use BackgroundTasks for lightweight non-blocking work after response is sent. FastAPI Docs: "FastAPI creates BackgroundTasks object and passes as parameter. Function can be async def or normal def — FastAPI handles correctly. If no need for task to wait, use BackgroundTasks. Tied to lifespan of FastAPI process — if server crashes, uncompleted tasks lost." OneUptime: "FastAPI includes simple BackgroundTasks class for lightweight background work. For heavy CPU-bound jobs, use Celery or custom task queue." Dev.to: "Like raw asyncio tasks, BackgroundTasks tied to lifespan of FastAPI process. If server crashes or restarts, uncompleted BackgroundTasks lost. For tasks that must survive restarts, use external queue." BackgroundTasks: (1) async def send_notification(email): ... (2) @app.post('/generate') async def generate(bg: BackgroundTasks): bg.add_task(send_notification, email). (3) Response sent immediately, task runs after. (4) Lightweight: email, logging, cache update. (5) NOT for heavy CPU-bound: use Celery/Arq. (6) Lost on server crash — use external queue for persistence.

How do you track async task status for long-running AI jobs?

Store task status in Redis or in-memory dict, return task ID, client polls for status. Task tracking: (1) task_id = str(uuid.uuid4()). (2) Store in Redis: {task_id: {status: pending, result: None}}. (3) asyncio.create_task(run_task(task_id, prompt)). (4) Update status: pending to running to complete/failed. (5) GET /tasks/{task_id} returns current status. (6) GET /tasks/{task_id}/result returns result when complete. (7) TTL on Redis keys for cleanup. (8) WebSocket push for real-time updates. (9) Webhook callback when complete. (10) Progress tracking: update percentage in Redis. (11) Task registry: dict of active tasks for cancellation. (12) Cleanup completed tasks after TTL expiry.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →