MCP Long-Running Tasks: Background Jobs in Python

Some tools are slow. A quarterly report, a full export or a site crawl can take minutes, and an MCP tool call that takes minutes holds up the agent, the user waiting on it, and sooner or later a client timeout. The fix is the same one web backends use: start the work in the background, hand back a job ID straight away, and let the caller check on it. In this guide you'll build an MCP server in Python whose slow report runs as a background job with progress, retries and cancellation, then put a real agent in front of it. You'll also see what progress notifications, cancellation and fire-and-forget tasks do on a normal tool call, and where the limits are. Every snippet ran against Promptise Foundry 1.3.2, with gpt-5-mini for the agents, and every output is what it printed.

Don't do the slow work inside the tool call. Register it as a job on a queue, so the agent submits it, gets a job ID back in milliseconds, and checks its status or fetches its result whenever it likes. With Promptise Foundry, that's MCPQueue and the @queue.job() decorator:

import asyncio

from promptise.mcp.server import MCPQueue, MCPServer

server = MCPServer("reports")
queue = MCPQueue(server, max_workers=2)


@queue.job(name="generate_report", timeout=120)
async def generate_report(region: str, quarter: str) -> dict:
    """Build the quarterly sales report for one region."""
    await asyncio.sleep(8)  # the slow part: queries, joins, rendering
    return {"region": region, "quarter": quarter, "url": f"https://reports.example.com/{region}-{quarter}.pdf"}


if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=8334)

Passing the server to MCPQueue adds five tools to it and starts the workers when the server starts. Submitting the eight-second report returns at once:

['queue_submit', 'queue_status', 'queue_result', 'queue_cancel', 'queue_list']
0.01s {"job_id": "0692ea0bc7ed9ce7", "status": "pending", "job_type": "generate_report"}

The rest of this guide builds this out with progress, retries, cancellation and an agent that uses it.

The tool call only puts the job on a queue. A pool of workers inside the server process picks jobs up and runs them, and every later call reads the job's record:

These are the five tools the queue adds. Pass tool_prefix="jobs" to MCPQueue and they become jobs_submit, jobs_status and so on.

A job is pending until a worker takes it, then running, and ends as completed, failed, timeout or cancelled.

• Python 3.10 or newer.

• Promptise Foundry 1.3.2 or newer from PyPI. The server and the queue need nothing else.

• An API key for a model provider for the agent steps. This guide uses OpenAI's gpt-5-mini; other providers work by changing the model string.

pip install "promptise>=1.3.2"
export OPENAI_API_KEY="sk-..."

The server makes quarterly sales reports, which take a while, and exports invoices to an accounting system that sometimes fails. The complete file, report_server.py, is in Step 4.

A job is an ordinary async function. Two extra parameters are filled in by the queue, based on their type annotation: a ProgressReporter that writes progress into the job's record, and a CancellationToken that tells the job when someone has cancelled it.

@queue.job(name="generate_report", timeout=120)
async def generate_report(
    region: str,
    quarter: str,
    progress: ProgressReporter,
    cancel: CancellationToken,
) -> dict:
    """Build the quarterly sales report for one region."""
    data = SALES[region.lower()]
    for done, step in enumerate(STEPS):
        cancel.check()  # stop here if someone called queue_cancel
        await progress.report(done, total=len(STEPS), message=step)
        await asyncio.sleep(2)  # the slow part: queries, joins, rendering
    return {
        "region": region,
        "quarter": quarter,
        "net_revenue": data["revenue"] - data["refunds"],
        "orders": data["orders"],
        "url": f"https://reports.example.com/{region.lower()}-{quarter.lower()}.pdf",
    }

• progress.report(done, total=4, message=...) stores done / total as a fraction, so status shows 0.25, not 25.

• cancel.check() raises when the job has been cancelled. Call it between steps, at points where stopping is safe.

• timeout=120 is the ceiling for one attempt. The default is 300 seconds, set by default_timeout on MCPQueue.

• max_workers=2 on the queue means at most two jobs run at the same time. The rest wait as pending.

Before an agent gets involved, check the queue with a plain MCP client. Start the server with python report_server.py, then run this:

import asyncio
import json
import time

from promptise.mcp.client import MCPClient


def show(result) -> dict:
    return json.loads(result.content[0].text)


async def main():
    async with MCPClient(url="http://127.0.0.1:8330/mcp") as client:
        print("Tools:", [t.name for t in await client.list_tools()])

        start = time.perf_counter()
        job = show(await client.call_tool(
            "queue_submit",
            {"job_type": "generate_report", "args": {"region": "EMEA", "quarter": "Q3"}},
        ))
        print(f"{time.perf_counter() - start:.2f}s submit ->", job)

        while True:
            await asyncio.sleep(1.5)
            status = show(await client.call_tool("queue_status", {"job_id": job["job_id"]}))
            print(f"{time.perf_counter() - start:.2f}s status ->", status)
            if status["status"] not in ("pending", "running"):
                break

        result = show(await client.call_tool("queue_result", {"job_id": job["job_id"]}))
        print(f"{time.perf_counter() - start:.2f}s result ->", json.dumps(result, indent=2))


asyncio.run(main())

Tools: ['queue_submit', 'queue_status', 'queue_result', 'queue_cancel', 'queue_list', 'wait_for_job']
0.01s submit -> {'job_id': '9bd14eed428d1947', 'status': 'pending', 'job_type': 'generate_report'}
1.59s status -> {'job_id': '9bd14eed428d1947', 'job_type': 'generate_report', 'status': 'running', 'priority': 'normal', 'progress': 0.0, 'attempts': 1, 'progress_message': 'Loading orders'}
3.10s status -> {'job_id': '9bd14eed428d1947', 'job_type': 'generate_report', 'status': 'running', 'priority': 'normal', 'progress': 0.25, 'attempts': 1, 'progress_message': 'Matching refunds'}
4.91s status -> {'job_id': '9bd14eed428d1947', 'job_type': 'generate_report', 'status': 'running', 'priority': 'normal', 'progress': 0.5, 'attempts': 1, 'progress_message': 'Computing totals'}
6.41s status -> {'job_id': '9bd14eed428d1947', 'job_type': 'generate_report', 'status': 'running', 'priority': 'normal', 'progress': 0.5, 'attempts': 1, 'progress_message': 'Computing totals'}
8.92s status -> {'job_id': '9bd14eed428d1947', 'job_type': 'generate_report', 'status': 'completed', 'priority': 'normal', 'progress': 1.0, 'attempts': 1, 'progress_message': 'Rendering the PDF'}
8.94s result -> {
  "job_id": "9bd14eed428d1947",
  "job_type": "generate_report",
  "status": "completed",
  "priority": "normal",
  "progress": 1.0,
  "attempts": 1,
  "progress_message": "Rendering the PDF",
  "result": {
    "region": "EMEA",
    "quarter": "Q3",
    "net_revenue": 402450.0,
    "orders": 1840,
    "url": "https://reports.example.com/emea-q3.pdf"
  }
}

The submit took a hundredth of a second, the status moved from step to step, and the result arrived about nine seconds later. wait_for_job in the tool list is a small tool you'll add in Step 4.

The model sees the five queue tools. queue_submit's description lists every registered job type with its docstring and the JSON Schema of its arguments, and job_type only accepts those names. What the tools can't say is how you want the agent to handle slow work, so put that in its instructions:

import asyncio

from promptise import build_agent
from promptise.config import HTTPServerSpec

INSTRUCTIONS = """You are a reporting assistant. Reports are slow, so they run as background jobs.

Job types on the reports server:
- generate_report: args {"region": "EMEA" or "Americas", "quarter": "Q1" to "Q4"}

To produce a report, submit it with queue_submit and check it with queue_status.
When the status is completed, fetch it with queue_result. Don't wait for a
slow job: tell the user the job ID and the progress message, and check again
when they ask."""


async def main():
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={"reports": HTTPServerSpec(url="http://127.0.0.1:8330/mcp")},
        instructions=INSTRUCTIONS,
        trace_tools=True,
    )
    try:
        messages = [{"role": "user", "content": "I need the Q3 sales report for EMEA."}]
        result = await agent.ainvoke({"messages": messages})
        print(">>>", result["messages"][-1].content)

        await asyncio.sleep(10)  # the user does something else for a while

        messages = result["messages"] + [{"role": "user", "content": "Is it ready?"}]
        result = await agent.ainvoke({"messages": messages})
        print(">>>", result["messages"][-1].content)
    finally:
        await agent.shutdown()


asyncio.run(main())

→ Invoking tool: queue_submit with {'job_type': 'generate_report', 'args': {'region': 'EMEA', 'quarter': 'Q3'}}
✔ Tool result from queue_submit: {"job_id": "2c249ca28071c021", "status": "pending", "job_type": "generate_report"}
>>> Done — I submitted the Q3 EMEA sales report job.

- Job ID: 2c249ca28071c021
- Current status: pending

What I can do next:
- Check progress/status for this job (ask “check job 2c249ca28071c021”).
- Fetch the report once it’s completed (ask “get result 2c249ca28071c021” or “fetch job 2c249ca28071c021”).
- Cancel the job (ask “cancel job 2c249ca28071c021”).

Would you like me to check the status now or wait until you ask?
→ Invoking tool: queue_status with {'job_id': '2c249ca28071c021'}
✔ Tool result from queue_status: {"job_id": "2c249ca28071c021", "job_type": "generate_report", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 1, "progress_message": "Rendering the PDF"}
>>> The job is completed — it says "Rendering the PDF." Would you like me to fetch the report now?

This is the point of a queue. The first turn ended as soon as the job was queued, so the user got an answer instead of a spinner. The job kept running between turns, and the second turn found it completed. In this run the model then asked before fetching the result, although its instructions say to fetch it; in two more runs it called queue_result straight away and answered with the link. The job ID lives in the conversation history, which is why passing result["messages"] into the next turn matters.

Sometimes the user does want to wait. With only the queue tools, an agent waits by calling queue_status again and again, and each check costs a model round trip. With instructions to keep checking until the job finished, and no other tool to wait with, this is how a run went:

→ Invoking tool: queue_submit with {'job_type': 'generate_report', 'args': {'region': 'EMEA', 'quarter': 'Q3'}, 'priority': 'normal'}
✔ Tool result from queue_submit: {"job_id": "a611458e4cf72f8c", "status": "pending", "job_type": "generate_report"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.0, "attempts": 1, "progress_message": "Loading orders"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.0, "attempts": 1, "progress_message": "Loading orders"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.25, "attempts": 1, "progress_message": "Matching refunds"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.25, "attempts": 1, "progress_message": "Matching refunds"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.5, "attempts": 1, "progress_message": "Computing totals"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "running", "priority": "normal", "progress": 0.75, "attempts": 1, "progress_message": "Rendering the PDF"}
→ Invoking tool: queue_status with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_status: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 1, "progress_message": "Rendering the PDF"}
→ Invoking tool: queue_result with {'job_id': 'a611458e4cf72f8c'}
✔ Tool result from queue_result: {"job_id": "a611458e4cf72f8c", "job_type": "generate_report", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 1, "progress_message": "Rendering the PDF", "result": {"region": "EMEA", "quarter": "Q3", "net_revenue": 402450.0, "orders": 1840, "url": "https://reports.example.com/emea-q3.pdf"}}
>>> The Q3 EMEA sales report is ready.

- Report URL: https://reports.example.com/emea-q3.pdf
- Job ID: a611458e4cf72f8c (status: completed — Rendering the PDF)
- Key metrics:
  - Net revenue: 402,450.00
  - Orders: 1,840
…

Seven status checks in 15 seconds, each one a model round trip, before the status said completed and the agent fetched the result. Every check reached the server: its log shows a POST /mcp request for each of the nine tool calls, plus three for the connection handshake. Two more runs checked eight times each. It works, but the model sets the pace, not the job, and every check costs a model call.

A better way is one tool that waits on the server side and returns when the job finishes or the time is up. It uses the queue's own status() method, so it reads the same record as queue_status. Here it is in the complete server:

import asyncio
import time
from typing import Annotated

from pydantic import Field

from promptise.mcp.server import CancellationToken, MCPQueue, MCPServer, ProgressReporter

server = MCPServer("reports")
queue = MCPQueue(server, max_workers=2)

# A stand-in for your warehouse and your accounting system.
SALES = {
    "emea": {"orders": 1840, "revenue": 412_300.0, "refunds": 9_850.0},
    "americas": {"orders": 2310, "revenue": 598_120.0, "refunds": 14_200.0},
}
STEPS = ["Loading orders", "Matching refunds", "Computing totals", "Rendering the PDF"]
export_attempts: dict[str, int] = {}


@queue.job(name="generate_report", timeout=120)
async def generate_report(
    region: str,
    quarter: str,
    progress: ProgressReporter,
    cancel: CancellationToken,
) -> dict:
    """Build the quarterly sales report for one region."""
    data = SALES[region.lower()]
    for done, step in enumerate(STEPS):
        cancel.check()  # stop here if someone called queue_cancel
        await progress.report(done, total=len(STEPS), message=step)
        await asyncio.sleep(2)  # the slow part: queries, joins, rendering
    return {
        "region": region,
        "quarter": quarter,
        "net_revenue": data["revenue"] - data["refunds"],
        "orders": data["orders"],
        "url": f"https://reports.example.com/{region.lower()}-{quarter.lower()}.pdf",
    }


@queue.job(name="export_invoices", timeout=60, max_retries=2, backoff_base=2.0)
async def export_invoices(month: str) -> dict:
    """Send one month of invoices to the accounting system."""
    export_attempts[month] = export_attempts.get(month, 0) + 1
    await asyncio.sleep(1)
    if export_attempts[month] == 1:
        raise ConnectionError("accounting API answered 503 Service Unavailable")
    return {"month": month, "invoices": 214, "file": f"invoices-{month}.csv"}


@server.tool()
async def wait_for_job(
    job_id: str,
    seconds: Annotated[int, Field(ge=1, le=25, description="How long to wait at most.")] = 20,
) -> dict:
    """Wait for a background job to finish, then return its status.

    Use this instead of calling queue_status again and again. It returns as soon as
    the job is completed, failed, cancelled or timed out, or when the time is up.
    """
    deadline = time.monotonic() + seconds
    while True:
        status = await queue.status(job_id)
        if status["status"] not in ("pending", "running") or time.monotonic() >= deadline:
            return status
        await asyncio.sleep(0.5)


if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=8330)

The 25-second cap keeps each call well inside client timeouts; if the job needs longer, the agent simply calls it again. The agent's instructions now list both job types and say: "When the user wants to wait for the result, call wait_for_job instead of polling queue_status". Asked for two things at once, it ran them side by side:

→ Invoking tool: queue_submit with {'job_type': 'export_invoices', 'args': {'month': '2026-09'}}
→ Invoking tool: queue_submit with {'job_type': 'generate_report', 'args': {'region': 'Americas', 'quarter': 'Q3'}}
✔ Tool result from queue_submit: {"job_id": "64e97691b024d484", "status": "pending", "job_type": "export_invoices"}
✔ Tool result from queue_submit: {"job_id": "9876bc0f5750118b", "status": "pending", "job_type": "generate_report"}
→ Invoking tool: wait_for_job with {'job_id': '64e97691b024d484', 'seconds': 25}
→ Invoking tool: wait_for_job with {'job_id': '9876bc0f5750118b', 'seconds': 25}
✔ Tool result from wait_for_job: {"job_id": "64e97691b024d484", "job_type": "export_invoices", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 2}
✔ Tool result from wait_for_job: {"job_id": "9876bc0f5750118b", "job_type": "generate_report", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 1, "progress_message": "Rendering the PDF"}
→ Invoking tool: queue_result with {'job_id': '64e97691b024d484'}
✔ Tool result from queue_result: {"job_id": "64e97691b024d484", "job_type": "export_invoices", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 2, "result": {"month": "2026-09", "invoices": 214, "file": "invoices-2026-09.csv"}}
→ Invoking tool: queue_result with {'job_id': '9876bc0f5750118b'}
✔ Tool result from queue_result: {"job_id": "9876bc0f5750118b", "job_type": "generate_report", "status": "completed", "priority": "normal", "progress": 1.0, "attempts": 1, "progress_message": "Rendering the PDF", "result": {"region": "Americas", "quarter": "Q3", "net_revenue": 583920.0, "orders": 2310, "url": "https://reports.example.com/americas-q3.pdf"}}
>>> All done — both jobs finished successfully.

1) Export invoices (September 2026)
- Job ID: 64e97691b024d484
- Status: completed
- Month: 2026-09
- Invoices exported: 214
- Output file: invoices-2026-09.csv

2) Q3 Americas report
- Job ID: 9876bc0f5750118b
- Status: completed
- Region: Americas
- Quarter: Q3
- Net revenue: $583,920.00
- Orders: 2,310
- Report (PDF): https://reports.example.com/americas-q3.pdf
…

In two of three runs the model asked to wait 25 seconds, the most le=25 allows, and in the third it kept the default of 20: since Promptise 1.3.1 the limit is part of the schema the model sees. A client that asks for more anyway gets Input validation error: 60 is greater than the maximum of 25, so each call stays well inside client timeouts. One wait per job was enough, and the export's retry happened without the agent having to do anything: the export came back with "attempts": 2. The tool's description does the persuading, too: with the polling instructions from above and wait_for_job on the server, the agent went straight from queue_submit to wait_for_job and never called queue_status.

The model still makes its own choices. In three runs of this agent it fetched both results every time, but nothing in the tools makes it do so, and on an earlier Promptise version one run waited for both jobs but fetched only the export's result and never mentioned the report. If a result must not be dropped, check the job list yourself after the turn rather than relying on the model to.

export_invoices fails on its first attempt with a ConnectionError, the kind of transient error worth retrying. max_retries=2 allows two more attempts, and backoff_base=2.0 sets the wait between them: backoff_base * 2 ** (attempt - 1), so 2 seconds, then 4. A small client that prints each change of status shows the whole cycle:

 0.1s {'job_id': 'e2acc662b9ff1675', 'job_type': 'export_invoices', 'status': 'running', 'priority': 'normal', 'progress': 0.0, 'attempts': 1}
 1.1s {'job_id': 'e2acc662b9ff1675', 'job_type': 'export_invoices', 'status': 'pending', 'priority': 'normal', 'progress': 0.0, 'attempts': 1, 'error': 'Attempt 1 failed: accounting API answered 503 Service Unavailable. Retrying in 2.0s.'}
 3.1s {'job_id': 'e2acc662b9ff1675', 'job_type': 'export_invoices', 'status': 'running', 'priority': 'normal', 'progress': 0.0, 'attempts': 2, 'error': 'Attempt 1 failed: accounting API answered 503 Service Unavailable. Retrying in 2.0s.'}
 4.1s {'job_id': 'e2acc662b9ff1675', 'job_type': 'export_invoices', 'status': 'completed', 'priority': 'normal', 'progress': 1.0, 'attempts': 2}
result: {'month': '2026-09', 'invoices': 214, 'file': 'invoices-2026-09.csv'}

What to notice:

• Between attempts the job is `pending` again, with the failure in error. Once a retry succeeds, error is cleared, and attempts shows how many tries it took.

• Wrong arguments never become a job. queue_submit checks args against the job function's signature, so a job submitted without its quarter argument was refused at once, with nothing queued and nothing retried. The error names the field and shows the schema to use:

• Timeouts aren't retried. A job that runs past its timeout ends as timeout with Job timed out after 1s in error, on the first attempt.

• The worker is free during the backoff. A failed attempt schedules its retry and hands the worker back at once. With max_workers=1, a second job finished 0.9 seconds after the first one failed, while the first was still waiting out its four-second backoff.

{
  "error": {
    "code": "INVALID_JOB_ARGUMENTS",
    "message": "Invalid arguments for job type 'needs_args': quarter: Field required",
    "retryable": false,
    "suggestion": "Pass `args` matching this schema: {\"additionalProperties\":false,\"properties\":{\"quarter\":{\"type\":\"string\"},\"region\":{\"type\":\"string\"}},\"required\":[\"region\",\"quarter\"],\"type\":\"object\"}",
    "details": {
      "field_errors": {
        "quarter": "Field required"
      }
    }
  }
}

queue_cancel marks the job as cancelled, sets its CancellationToken and cancels the job's task right away. Whatever the job returns afterwards is discarded:

import asyncio
import json

from promptise.mcp.client import MCPClient


def show(result) -> dict:
    return json.loads(result.content[0].text)


async def main():
    async with MCPClient(url="http://127.0.0.1:8330/mcp") as client:
        job = show(await client.call_tool(
            "queue_submit", {"job_type": "generate_report", "args": {"region": "Americas", "quarter": "Q3"}}
        ))
        await asyncio.sleep(3)
        print("before:", show(await client.call_tool("queue_status", {"job_id": job["job_id"]})))
        print("cancel:", show(await client.call_tool("queue_cancel", {"job_id": job["job_id"]})))
        await asyncio.sleep(10)  # long after the job would have finished
        print("later: ", show(await client.call_tool("queue_result", {"job_id": job["job_id"]})))


asyncio.run(main())

before: {'job_id': '66fb8063521fdc32', 'job_type': 'generate_report', 'status': 'running', 'priority': 'normal', 'progress': 0.25, 'attempts': 1, 'progress_message': 'Matching refunds'}
cancel: {'job_id': '66fb8063521fdc32', 'status': 'cancelled'}
later:  {'job_id': '66fb8063521fdc32', 'job_type': 'generate_report', 'status': 'cancelled', 'priority': 'normal', 'progress': 0.25, 'attempts': 1, 'progress_message': 'Matching refunds'}

The job was in its second step and stopped there, so it never reached "Computing totals", and ten seconds later it's still cancelled with no result.

Not everything needs a queue. A tool that takes ten or twenty seconds can stay a normal tool and still tell the client how it's doing. ProgressReporter and CancellationToken work there too, injected with Depends:

@server.tool()
async def crawl_site(pages: int, progress: ProgressReporter = Depends(ProgressReporter)) -> dict:
    """Crawl a number of pages and report progress as it goes."""
    for page in range(1, pages + 1):
        await asyncio.sleep(0.5)
        await progress.report(page, total=pages, message=f"Crawled page {page} of {pages}")
    return {"pages": pages}

Here, report() sends MCP progress notifications to the client while the call is still running. A client only receives them if it asked for them by sending a progress token with the call. The official MCP Python SDK does that when you pass a progress_callback:

    async def on_progress(progress, total, message):
        print(f"  progress {progress}/{total}: {message}")

    async with streamablehttp_client(URL) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            print("official client, with progress_callback:")
            r = await session.call_tool("crawl_site", {"pages": 3}, progress_callback=on_progress)
            print("  result:", r.content[0].text)

official client, with progress_callback:
  progress 1.0/3.0: Crawled page 1 of 3
  progress 2.0/3.0: Crawled page 2 of 3
  progress 3.0/3.0: Crawled page 3 of 3
  result: {"pages": 3}
Promptise MCPClient, without progress_callback:
  result: {"pages": 3}
Promptise MCPClient, with progress_callback:
  progress 1.0/3.0: Crawled page 1 of 3
  progress 2.0/3.0: Crawled page 2 of 3
  progress 3.0/3.0: Crawled page 3 of 3
  result: {"pages": 3}

Promptise's own MCP client asks for progress when you pass progress_callback= to call_tool, and sends no progress token without one, so the server then sends nothing. An agent gets the same with build_agent(on_tool_progress=...), and trace_tools=True prints each notification:

import asyncio

from promptise import build_agent
from promptise.config import HTTPServerSpec


async def on_tool_progress(tool_name, progress, total, message):
    print(f"  [{tool_name}] {progress:.0f}/{total:.0f} {message}")


async def main():
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={"direct": HTTPServerSpec(url="http://127.0.0.1:8332/mcp")},
        on_tool_progress=on_tool_progress,
        trace_tools=True,
    )
    try:
        result = await agent.ainvoke({"messages": [{"role": "user", "content": "Crawl 3 pages."}]})
        print(">>>", result["messages"][-1].content)
    finally:
        await agent.shutdown()


asyncio.run(main())

→ Invoking tool: crawl_site with {'pages': 3}
… crawl_site progress 1/3: Crawled page 1 of 3
  [crawl_site] 1/3 Crawled page 1 of 3
… crawl_site progress 2/3: Crawled page 2 of 3
  [crawl_site] 2/3 Crawled page 2 of 3
… crawl_site progress 3/3: Crawled page 3 of 3
  [crawl_site] 3/3 Crawled page 3 of 3
✔ Tool result from crawl_site: {"pages": 3}
>>> Started crawling 3 pages. Progress: 3/3 pages crawled. Completed.

The model never sees progress notifications; they're for a person watching. All it got was the tool's result, {"pages": 3}, and its answer is built from that. If the agent needs to know how far along something is, use a queue job, whose progress the agent can read with queue_status.

Cancelling a normal tool call works much like cancelling a job. When the client sends an MCP cancellation, Promptise sets the tool's CancellationToken and gives the handler cancel_grace_period (5 seconds by default, set on MCPServer) to stop on its own before its task is cancelled. A tool that waited on its token logged this when the official client cancelled it:

15:27:13 watch_feed: token set, token.is_cancelled=True

So cancel.wait() and cancel.check() work on a normal tool too. Still put cleanup that must run in a try/finally: a handler that ignores the token is cancelled once the grace period is over.

BackgroundTasks is for small jobs after a tool has done its real work, such as sending a receipt or writing an analytics event. You add callables during the tool call, and they run after the response has gone out:

async def send_receipt(order_id: str) -> None:
    await asyncio.sleep(3)  # a slow email API
    print(f"{time.strftime('%X')} receipt sent for {order_id}", flush=True)


@server.tool()
async def place_order(order_id: str, bg: BackgroundTasks = Depends(BackgroundTasks)) -> dict:
    """Place an order and email the receipt afterwards."""
    bg.add(send_receipt, order_id)
    print(f"{time.strftime('%X')} handler returned for {order_id}", flush=True)
    return {"order_id": order_id, "status": "placed"}

Errors in background tasks are logged and never reach the caller. Here's the timing over HTTP. The server printed:

15:27:11 handler returned for A-1001
15:27:14 receipt sent for A-1001

And the client calling place_order printed:

15:27:11 place_order answered after 0.01s: {"order_id": "A-1001", "status": "placed"}

The caller got its answer at once, and the receipt went out three seconds later. On a graceful shutdown the server waits for background tasks that are still running. Under TestClient they run before call_tool returns, so a test can check their effects. They have no ID, status or retries, so make anything slow or likely to fail a queue job.

Here's how to choose:

These are true of Promptise Foundry 1.3.2. Each one was checked in a run or in the source.

• Jobs live in memory. The only backend that ships is InMemoryQueueBackend. When the server restarts, every job and result is gone: asked about a job submitted before the restart, the server answered JOB_NOT_FOUND. There's no persistence and no Redis backend. You can write your own against the QueueBackend protocol, seven async methods from enqueue to count, but nothing re-runs jobs that were running when a process died.

• One queue per process. With two replicas, a job submitted to one is JOB_NOT_FOUND on the other. Behind a load balancer, either keep a session on one replica or run one replica. A shared backend alone doesn't fix cancellation, because the cancellation signal is held in the memory of the process running the job.

• Job ownership needs authentication. On a server with require_auth=True, or with MCPQueue(auth=True), each caller sees and cancels only its own jobs: in a test with two API keys, the second caller's queue_list was empty, and the first caller's job answered it JOB_NOT_FOUND. A caller with the admin role sees every job of its own tenant. Without authentication, all callers share one anonymous owner, so don't expose the queue tools that way on a shared server.

• Results expire. Finished jobs are removed after result_ttl, one hour by default, checked every cleanup_interval of 60 seconds.

• Testing needs a manual start. TestClient doesn't run startup hooks, so queue workers don't start and submitted jobs stay pending. Call await queue.start() in your test and await queue.stop() after it. call_tool returns MCP content, so read the job with json.loads(resp[0].text).

How do I fix an MCP tool timeout?

Stop making the client wait for the whole job. Turn the slow tool into a queue job, so the call that starts it returns a job ID at once and nothing long-running is left to time out. Through Promptise's MCP client, plain tool calls that took 45 seconds and 330 seconds both came back fine, so when you do hit a timeout, check the client you use, any proxy in between, and TimeoutMiddleware on the server. Whatever the limit, a job ID sidesteps it.

How do MCP progress notifications work?

The client sends a progress token with a tool call, and the server sends notifications/progress messages tied to that token while the call runs, each with a number, an optional total and a message. In Promptise, ProgressReporter.report() sends them, and does nothing when the client didn't ask; Promptise's client asks when you pass progress_callback= (or on_tool_progress= to build_agent). They show in client UIs; the model doesn't see them. The MCP specification has the details.

What's the difference between BackgroundTasks and an MCP job queue?

BackgroundTasks runs small follow-up work after one tool call has answered, with no ID, status or retries. MCPQueue runs jobs on separate workers, gives each a job ID, and adds progress, retries, timeouts and cancellation. Use the queue for anything that takes noticeable time.

Can the MCP job queue use Redis?

Not out of the box in Promptise 1.3. MCPQueue takes any object that implements the QueueBackend protocol through its backend argument, so you can write a Redis or Postgres backend yourself. The Queue & Background Jobs page shows the methods to implement.

Should the agent wait for a background job?

Usually not. The value of a job is that the conversation can move on, as in Step 3. When the user does want to wait, give the agent a tool that waits on the server side, like wait_for_job in Step 4, instead of letting it poll.

• Queue & Background Jobs: every MCPQueue option, priorities, backends and health checks.

• Resilience Patterns: BackgroundTasks, ProgressReporter, CancellationToken and the circuit breaker.

• Production Features: caching, rate limits, timeouts and the order to add middleware in.

• Deployment: running an MCP server over HTTP in production.

• How to Connect MCP Servers to Your AI Agent in Python: the agent and client basics this guide builds on.

• OpenAPI to MCP: Turn Any REST API into an MCP Server: generate a server from an existing API, then add jobs for its slow endpoints.