Handling Long-Running Tools: Async Jobs & Progress Notifications

Some enterprise operations, large data exports, batch document processing, complex report generation, genuinely take minutes, not seconds. Forcing these into a single synchronous tools/call either times out the connection or blocks the agent (and the underlying transport) far longer than is healthy. MCP’s answer is the submit-poll pattern, optionally paired with progress notifications, and getting it right in production means the job’s state needs to survive more than just a successful run, it needs to survive a server restart mid-job.

The Submit-Poll Pattern, Backed by Durable Storage

import uuid, asyncio

@mcp.tool(name="submit_data_export")
async def submit_data_export(input: ExportInput) -> dict:
    """
    Submits a data export job and returns immediately with a job ID.
    Poll check_export_status with this ID; do not expect the export
    to be complete when this call returns.
    """
    job_id = str(uuid.uuid4())
    # Persisted to Redis/Postgres from the start, NOT an in-memory dict,
    # so the job's existence and progress survive a pod restart mid-job.
    await job_store.create(job_id, status="queued", progress=0.0, result=None)
    asyncio.create_task(_run_export(job_id, input))
    return {"job_id": job_id, "status": "queued"}

@mcp.tool(name="check_export_status")
async def check_export_status(job_id: str) -> dict:
    """Poll this tool with a job_id from submit_data_export to check progress or retrieve the result."""
    job = await job_store.get(job_id)
    if job is None:
        return {"isError": True, "message": f"No job found with id '{job_id}'"}
    return job

async def _run_export(job_id: str, input: ExportInput):
    await job_store.update(job_id, status="running")
    total_records = await export_service.count(input)
    processed = 0
    async for batch in export_service.stream_batches(input):
        await export_service.write_batch(batch)
        processed += len(batch)
        # Persisted after every batch, not just at the end, so a mid-job
        # kill loses at most one batch's worth of progress, not everything.
        await job_store.update(job_id, progress=processed / total_records)
    download_url = await export_service.finalize(job_id)
    await job_store.update(job_id, status="complete", result={"download_url": download_url})

sequenceDiagram
    participant Agent
    participant MCP as MCP Server
    participant Store as Job Store (Redis/Postgres)
    participant Worker as Background Worker

    Agent->>MCP: tools/call submit_data_export
    MCP->>Store: create job (persisted, not in-memory)
    MCP->>Worker: enqueue job (async, non-blocking)
    MCP-->>Agent: {job_id: "abc123", status: "queued"}
    Note over Agent: Agent can do other work,\ndoes not hold connection open

    loop Poll every N seconds
        Agent->>MCP: tools/call check_export_status(job_id)
        MCP->>Store: read current progress
        Store-->>MCP: {status: "running", progress: 0.4}
        MCP-->>Agent: {status: "running", progress: 0.4}
    end

    Worker->>Store: job complete, result persisted
    Agent->>MCP: tools/call check_export_status(job_id)
    MCP->>Store: read final state
    Store-->>MCP: {status: "complete", result: {...}}
    MCP-->>Agent: {status: "complete", result: {download_url: "..."}}

    style Agent fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style MCP fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Store fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Worker fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Surviving a Rolling Deploy Mid-Job

Connecting this directly back to Lesson 8’s graceful shutdown discussion: a rolling deploy sends SIGTERM to the pod running the background worker task, waits out the grace period, then force-kills it. If the job’s progress lived only in that process’s memory (a plain dict, as an early, simpler version of this pattern might use), the job simply vanishes the moment the process is killed, with no record anywhere that it ever existed past whatever was last written. Persisting job state after every meaningful unit of progress, as _run_export does above after each batch, rather than only at job completion, bounds the loss to “whatever work happened since the last persisted checkpoint” instead of “the entire job,” and, critically, means any replica that receives the next poll request (not necessarily the one that started the job) can read the correct current state from the shared store.

# On graceful shutdown (Lesson 8's SIGTERM handler), give in-flight background
# jobs a chance to reach their next checkpoint rather than being killed
# at an arbitrary point; jobs that do not finish in time simply resume
# from their last persisted checkpoint whenever a worker next picks them up.
async def graceful_worker_shutdown(active_job_ids: set[str], grace_seconds: float):
    deadline = time.monotonic() + grace_seconds
    while active_job_ids and time.monotonic() < deadline:
        await asyncio.sleep(0.5)  # Let in-flight batches reach their next persisted checkpoint
    # Any job still in "running" state at this point resumes correctly on
    # its next poll, since progress was persisted externally throughout.

Progress Notifications as an Alternative to Polling

For clients and transports that support server-initiated messages, the server can push progress updates proactively instead of requiring the caller to poll.

@mcp.tool(name="submit_data_export")
async def submit_data_export(input: ExportInput, context: mcp.Context) -> dict:
    """Same job, but pushes progress notifications instead of requiring polling."""
    job_id = str(uuid.uuid4())
    asyncio.create_task(_run_export_with_progress(job_id, input, context))
    return {"job_id": job_id, "status": "queued"}

async def _run_export_with_progress(job_id: str, input: ExportInput, context: mcp.Context):
    total = await export_service.count(input)
    processed = 0
    async for batch in export_service.stream_batches(input):
        await export_service.write_batch(batch)
        processed += len(batch)
        await context.report_progress(progress=processed / total, message=f"Exported {processed}/{total} records")
    await context.report_progress(progress=1.0, message="Export complete")

Use polling when the caller’s framework does not support server-initiated messages or when the job may outlive a single connection (the caller can reconnect later and re-poll the same job_id, which is exactly why durable job storage matters regardless of which mechanism is used); use progress notifications when a smoother, connection-lived UX is wanted and the transport supports it. Many production servers implement both, defaulting to whichever the connecting client’s capabilities negotiation (Lesson 2) indicates it supports, but both, notably, depend on the same underlying durability discipline: job state that lives in external storage, not in the memory of whichever process happened to start the job.

This closes Module 9. The final module covers scaling these servers under real load, CI/CD with schema contracts, and a capstone that assembles every piece from this course into one coherent enterprise MCP ecosystem.