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:

sequenceDiagram
    autonumber
    participant Client
    participant Runtime
    participant Engine as Task Engine
    participant Tool

    Client->>Runtime: tools/call<br/>_meta.task requested
    Runtime->>Runtime: verify task capability and tool execution policy
    Runtime->>Engine: create task
    Engine-->>Client: task accepted / pending response
    Engine->>Tool: run tool with task_context and cancel_token
    Tool->>Engine: task_context.send_progress(...)
    Engine-->>Client: notifications/progress
    Engine-->>Client: notifications/tasks/status

    alt user input required
        Tool->>Engine: task_context.require_input(...)
        Engine-->>Client: notifications/tasks/status waiting
        Client->>Runtime: elicitation response
        Runtime->>Engine: resume task
    end

    alt client cancels
        Client->>Runtime: tasks/cancel
        Runtime->>Engine: cancel task
        Engine->>Tool: cancel_token trips
        Engine-->>Client: notifications/tasks/status cancelled
    else tool completes
        Tool-->>Engine: content and structuredContent
        Engine-->>Client: notifications/tasks/status complete
        Client->>Runtime: tasks/result
        Runtime->>Engine: load stored result
        Engine-->>Client: final MCP result shape
    end

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.