Tasks

pymcp-kit supports task-aware tool execution, result polling, cancellation, and progress reporting. Tasks are enabled by default at the capability level, but each tool still decides whether task augmentation is allowed.

Task Execution Flow

Task-aware tools can return immediately with task metadata, continue in the background, emit status/progress notifications, and expose the final structured result through tasks/result:

Opt A Tool Into Tasks

Use execution.taskSupport on the tool definition:

• "optional": the tool can run directly or as a task
• "required": the caller must use task augmentation
• "forbidden": the tool only runs as a normal request

from pymcp.registry import tool_registry


@tool_registry.register(
    execution={
        "taskSupport": "optional",
        "runner": "async",
    }
)
async def build_index(repo: str) -> str:
    return f"Indexed {repo}"

Injected Task Helpers

When the tool signature asks for them, the runtime can inject:

• cancel_token
• task_context
• request_context

task_context exposes helpers like:

• await task_context.send_progress(...)
• await task_context.require_input(...)
• await task_context.set_working(...)

cancel_token lets the tool stop cooperatively.

Example: Progress, Cancellation, And Elicitation

import asyncio

from pymcp.registry import tool_registry
from pymcp.runtime.context import RequestContext
from pymcp.session import request_elicitation
from pymcp.tasks import CancellationToken


@tool_registry.register(
    execution={
        "taskSupport": "optional",
        "runner": "async",
    }
)
async def review_repository(
    repo: str,
    cancel_token: CancellationToken | None = None,
    task_context=None,
    request_context: RequestContext | None = None,
) -> dict[str, object]:
    stages = ["clone", "index", "summarize"]

    for index, stage in enumerate(stages, start=1):
        if cancel_token is not None:
            cancel_token.check_cancelled()
        await asyncio.sleep(0)
        if task_context is not None:
            await task_context.send_progress(
                index,
                total=len(stages),
                message=f"Finished {stage} for {repo}",
            )

    if (
        task_context is not None
        and request_context is not None
        and request_context.session_id is not None
    ):
        await task_context.require_input("Choose the publish target before continuing.")
        _, response = await request_elicitation(
            request_context.app,
            request_context.session_id,
            {"message": "Which environment should receive the report?"},
            task_id=task_context.task_id,
        )
        await task_context.set_working("Continuing after user input.")

    return {
        "content": [{"type": "text", "text": f"Review complete for {repo}"}],
        "structuredContent": {"repo": repo, "status": "complete"},
    }

Task RPC Surface

The built-in handlers expose:

• tasks/list
• tasks/get
• tasks/cancel
• tasks/result

Task-related flows also emit:

• notifications/tasks/status
• notifications/progress

If a tool returns a structured MCP result directly, tasks/result preserves that result shape instead of flattening it into plain text.

Ownership And Access

Task visibility follows the caller context:

• without auth, tasks are scoped to the session
• with auth, tasks are bound to the authenticated principal

That means the same authenticated principal can access its tasks across sessions, while a different principal cannot.

Notes

• If a client did not negotiate task support for tools/call, the runtime ignores task metadata and processes the tool call normally.
• Stdio and Streamable HTTP both support task side-channel traffic such as status and progress updates.