From ef29b731ea9c402e2c3844b209f64a4a8474a34d Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 22 Jul 2026 08:03:32 -0400
Subject: [PATCH] Add server-side claim production for tasks; emit resultType
 discriminator
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Widen the tools/call result serialization (via a refcounted, modern-gated wrap
installed by TasksExtension) so a CreateTaskResult reaches the client instead of
being stripped by the CallToolResult|InputRequiredResult surface — the SDK ships
claim consumption but no production. Emit the resultType discriminator the
protocol requires (task on CreateTaskResult, complete on the tasks/* results);
the draft schema forbids it (additionalProperties:false), a contradiction
reported upstream. Closes compliance gaps G1/G4/G5.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 fastmcp_tasks/README.md                       |  78 +++++++++++-
 fastmcp_tasks/fastmcp_tasks/extension.py      |   7 ++
 fastmcp_tasks/fastmcp_tasks/models.py         |  42 ++++++-
 .../fastmcp_tasks/wire_production.py          | 111 ++++++++++++++++++
 tests/tasks/server/test_wire_models.py        |  47 +++++++-
 tests/tasks/server/test_wire_production.py    |  95 +++++++++++++++
 6 files changed, 369 insertions(+), 11 deletions(-)
 create mode 100644 fastmcp_tasks/fastmcp_tasks/wire_production.py
 create mode 100644 tests/tasks/server/test_wire_production.py

diff --git a/fastmcp_tasks/README.md b/fastmcp_tasks/README.md
index 53c9f0902..16cc8cc11 100644
--- a/fastmcp_tasks/README.md
+++ b/fastmcp_tasks/README.md
@@ -1,9 +1,83 @@
 # fastmcp-tasks
 
-`fastmcp-tasks` provides background task execution for FastMCP servers via the `io.modelcontextprotocol/tasks` extension ([SEP-2663](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2663)).
+A complete implementation of background tasks for the Model Context Protocol — the `io.modelcontextprotocol/tasks` extension defined in [SEP-2663](https://github.com/modelcontextprotocol/ext-tasks).
 
-It bundles the task-queue dependencies (powered by [docket](https://github.com/chrisguidry/docket)) that FastMCP's `tasks` extra requires. Install it alongside [FastMCP](https://gofastmcp.com) to run long-lived tools as background tasks instead of blocking a request for their full duration.
+The MCP tasks extension is a Final SEP, but as of this writing it ships in the ecosystem as a schema and a prose specification — no language SDK provides a working runtime for it. `fastmcp-tasks` is, to our knowledge, the first: a full server-side implementation of the protocol, built on the durable execution engine ([docket](https://github.com/chrisguidry/docket)) that FastMCP has run in production since v3. If you want to actually *run* MCP background tasks today, this is the implementation.
+
+## What background tasks are
+
+Most tool calls are synchronous: the client sends `tools/call` and holds the request open until the tool returns. That breaks down for work that takes minutes or hours — a long analysis, a batch job, a slow external API. The tasks extension lets a server answer such a call *immediately* with a durable task handle, then run the work in the background while the client polls for completion on its own schedule.
+
+The model is poll-based and stateless by construction, which is what makes it survive disconnects, server restarts, and load balancers:
+
+1. A client that supports tasks issues a normal `tools/call` with a per-request opt-in.
+2. The server decides whether to run it as a task. If it does, it returns a `CreateTaskResult` carrying a server-generated task id — right away, before the work starts.
+3. The client polls `tasks/get` until the task reaches a terminal state, then reads the result inlined in the response.
+4. `tasks/cancel` requests cancellation; `tasks/update` answers any input the task asks for mid-run.
+
+The server owns the task's durable state, so the client can poll across independent requests — from any process, after a crash, through any replica — with no session affinity required.
+
+## Usage
+
+Install it as the `tasks` extra on FastMCP:
 
 ```bash
 uv pip install "fastmcp[tasks]"
 ```
+
+Register the extension on your server and mark the tools that may run as tasks. The extension is where the backend is configured — point it at Redis for a distributed deployment, or leave it on the in-memory default for a single process:
+
+```python
+from fastmcp import FastMCP
+from fastmcp_tasks import TasksExtension
+
+mcp = FastMCP("Analytics")
+mcp.add_extension(TasksExtension(url="redis://localhost:6379/0"))
+
+
+@mcp.tool(task=True)
+async def analyze(dataset: str) -> str:
+    # Long-running work. The client gets a task handle immediately and
+    # polls for the result; this runs in a background worker.
+    ...
+```
+
+`task=True` is a declaration of intent — this tool *may* run as a task — while the server, per the spec, decides per call whether to actually task it. Use `TaskConfig` for finer control:
+
+```python
+from fastmcp_tasks import TaskConfig
+
+
+@mcp.tool(task=TaskConfig(mode="required"))
+async def must_run_async(n: int) -> int:
+    # Always runs as a task; a client that has not opted in is told so.
+    ...
+```
+
+Registering `TasksExtension` is required to serve `task=True` tools — the tool declares intent, the extension provides the engine. A `task=True` tool on a server with no tasks extension registered fails loudly at startup rather than silently running inline.
+
+### Running out-of-process workers
+
+For distributed deployments backed by Redis, run dedicated worker processes alongside your server:
+
+```bash
+python -m fastmcp_tasks.worker_cli worker server.py
+```
+
+Workers and servers that share a backend URL and queue name share a task queue, so you can scale execution independently of your request-serving frontends.
+
+## Configuration
+
+The backend is configured on the extension. Every option also has a `FASTMCP_DOCKET_*` environment variable, so an env-configured deployment can construct `TasksExtension()` with no arguments:
+
+| Option | Env var | Default | Description |
+| --- | --- | --- | --- |
+| `url` | `FASTMCP_DOCKET_URL` | `memory://` | Backend URL. `memory://` for single-process; `redis://host:port/db` for distributed workers. |
+| `name` | `FASTMCP_DOCKET_NAME` | `fastmcp` | Queue name. Servers and workers sharing a name and URL share a queue. |
+| `concurrency` | `FASTMCP_DOCKET_CONCURRENCY` | `10` | Maximum concurrent tasks per worker. |
+
+See the [FastMCP task documentation](https://gofastmcp.com/servers/tasks) for the full reference.
+
+## Status
+
+The tasks extension is an experimental MCP extension, and `fastmcp-tasks` tracks its draft schema. The protocol's shape is settled — SEP-2663 is Final — but field-level details may still move; this package versions independently so it can follow the schema without waiting on a FastMCP release.
diff --git a/fastmcp_tasks/fastmcp_tasks/extension.py b/fastmcp_tasks/fastmcp_tasks/extension.py
index 646baca3c..5b3e2be1f 100644
--- a/fastmcp_tasks/fastmcp_tasks/extension.py
+++ b/fastmcp_tasks/fastmcp_tasks/extension.py
@@ -236,6 +236,7 @@ def _install_worker_hooks() -> None:
         set_background_context_factory,
         set_worker_server_resolver,
     )
+    from fastmcp_tasks import wire_production
     from fastmcp_tasks.context import make_task_context, resolve_worker_server
     from fastmcp_tasks.input_store import elicit_in_task
 
@@ -244,6 +245,10 @@ def _install_worker_hooks() -> None:
     set_background_context_factory(make_task_context)
     set_worker_server_resolver(resolve_worker_server)
     set_task_elicitation_handler(elicit_in_task)
+    # Enable server-side production of the claimed CreateTaskResult on tools/call
+    # (the SDK ships only claim consumption). Refcounted independently but
+    # installed/released in lockstep with the worker hooks.
+    wire_production.install()
 
 
 def _release_worker_hooks() -> None:
@@ -252,6 +257,7 @@ def _release_worker_hooks() -> None:
         set_background_context_factory,
         set_worker_server_resolver,
     )
+    from fastmcp_tasks import wire_production
 
     global _active_worker_hook_holds
     _active_worker_hook_holds -= 1
@@ -260,3 +266,4 @@ def _release_worker_hooks() -> None:
         set_task_elicitation_handler(None)
         set_worker_server_resolver(None)
         set_background_context_factory(None)
+    wire_production.uninstall()
diff --git a/fastmcp_tasks/fastmcp_tasks/models.py b/fastmcp_tasks/fastmcp_tasks/models.py
index 1ce1cc07a..761b7ebf4 100644
--- a/fastmcp_tasks/fastmcp_tasks/models.py
+++ b/fastmcp_tasks/fastmcp_tasks/models.py
@@ -56,10 +56,10 @@ class _TaskFields(BaseModel):
     false` on the task arm forbids it (see module docstring).
     """
 
-    # Serialization aliases only: these result models are constructed by field
-    # name (the engine builds them) and dumped to camelCase by the runner
-    # (`model_dump(by_alias=True)`). Wire *validation* of results is the client's
-    # concern.
+    # Serialization aliases: the engine constructs these by field name and the
+    # runner dumps them to camelCase (`model_dump(by_alias=True)`). The
+    # claim-production wrap returns that dump unchanged, so no input alias is
+    # needed.
     model_config = ConfigDict(populate_by_name=True)
 
     task_id: str = Field(serialization_alias="taskId")
@@ -80,8 +80,22 @@ class CreateTaskResult(_TaskFields):
 
     A flat merge of `Result` and `Task` (SEP-2663): the finished task stub the
     client polls with `tasks/get`. Status is typically `working`.
+
+    `resultType` is the wire discriminator that distinguishes this from a
+    `CallToolResult` on the shared `tools/call` method: the modern result union
+    carries a required `resultType`, and the SDK's client-side `ResultClaim`
+    for tasks requires this model to pin it to `Literal["task"]`. The vendored
+    draft schema omits `resultType` from the task arm (its
+    `additionalProperties: false` forbids it) — a schema-vs-protocol
+    contradiction reported upstream. Protocol interop requires the field, so we
+    emit it; only this shape needs it (the `tasks/*` methods each have a single
+    result type and bypass the discriminated union).
     """
 
+    result_type: Literal["task"] = Field(
+        default="task", serialization_alias="resultType"
+    )
+
 
 class GetTaskResult(_TaskFields):
     """Result of `tasks/get`: the detailed task (`Result & DetailedTask`).
@@ -90,8 +104,16 @@ class GetTaskResult(_TaskFields):
     `input_requests` (input_required) alongside the flat task fields, matching
     the schema's 5-status union. The three payload fields default to `None` and
     are dropped from the wire dump for the statuses that do not use them.
+
+    `resultType` is `"complete"` (SEP-2663 L338): `tasks/get` itself completes
+    normally, whatever the task's own status. As with `CreateTaskResult`, the
+    draft schema's `additionalProperties: false` omits this field — a
+    contradiction reported upstream; protocol interop requires emitting it.
     """
 
+    result_type: Literal["complete"] = Field(
+        default="complete", serialization_alias="resultType"
+    )
     result: dict[str, Any] | None = None
     error: dict[str, Any] | None = None
     input_requests: dict[str, Any] | None = Field(
@@ -100,11 +122,19 @@ class GetTaskResult(_TaskFields):
 
 
 class UpdateTaskResult(Result):
-    """Empty acknowledgement for `tasks/update` (SEP-2663 `Result`)."""
+    """Acknowledgement for `tasks/update` (SEP-2663 `Result`, `resultType: "complete"`)."""
+
+    result_type: Literal["complete"] = Field(
+        default="complete", serialization_alias="resultType"
+    )
 
 
 class CancelTaskResult(Result):
-    """Empty acknowledgement for `tasks/cancel` (SEP-2663 `Result`)."""
+    """Acknowledgement for `tasks/cancel` (SEP-2663 `Result`, `resultType: "complete"`)."""
+
+    result_type: Literal["complete"] = Field(
+        default="complete", serialization_alias="resultType"
+    )
 
 
 class GetTaskParams(RequestParams):
diff --git a/fastmcp_tasks/fastmcp_tasks/wire_production.py b/fastmcp_tasks/fastmcp_tasks/wire_production.py
new file mode 100644
index 000000000..09ac42949
--- /dev/null
+++ b/fastmcp_tasks/fastmcp_tasks/wire_production.py
@@ -0,0 +1,111 @@
+"""Server-side production of the tasks extension's claimed `tools/call` result.
+
+The MCP SDK ships the *consumption* half of SEP-2133 claimed results — a client
+`ResultClaim` resolves an extension's result shape on a core method — but not
+the *production* half: nothing lets a server emit one. On the modern protocol
+the runner revalidates every `tools/call` result against
+`SERVER_RESULTS[("tools/call", "2026-07-28")]`, which admits only
+`CallToolResult | InputRequiredResult`. A returned `CreateTaskResult` is coerced
+through those `extra="ignore"` models and stripped to nothing — the `taskId`
+never reaches the client, so the tasks extension cannot create a task over the
+wire even though its `tasks/*` methods (being custom methods) serialize freely.
+
+This module supplies the missing production half. It wraps
+`mcp_types.methods.serialize_server_result` — which the runner looks up on the
+module at call time — so that a modern `tools/call` result tagged
+`resultType: "task"` is validated against `CreateTaskResult` and dumped as-is,
+routed by the discriminator rather than the ambiguous result union (an untagged
+task dict would otherwise be swallowed by the all-optional `InputRequiredResult`
+arm). Every other result delegates to the original serializer unchanged.
+
+The wrap is process-global but inert for anything that is not a tasks server: a
+server that never emits `resultType: "task"` never takes the task branch. It is
+installed and reference-counted by `TasksExtension.lifespan()` so it is present
+exactly while at least one tasks extension is running, and removed after the
+last one stops. It is gated to modern protocol versions because claimed result
+shapes exist only there.
+
+Removal trigger: when the SDK grows a first-class server-side claim-production
+API (mirroring the client `ResultClaim`), this wrap is deleted and
+`TasksExtension` declares its produced claim through that API instead. See the
+upstream report in the migration notes.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+import mcp_types.methods as _methods
+from mcp_types.version import MODERN_PROTOCOL_VERSIONS
+
+_TASK_RESULT_TYPE = "task"
+_TASK_AUGMENTED_METHOD = "tools/call"
+
+# Sentinel distinguishing "caller passed no surface" (the runner's path, which we
+# may divert) from an explicit surface another caller supplied (never diverted).
+_STOCK: Any = object()
+
+# The original module function, captured once. `None` until the first install.
+_original_serialize: Any = None
+_active_holds: int = 0
+
+
+def _serialize_with_task_production(
+    method: str,
+    version: str,
+    data: Mapping[str, Any],
+    *,
+    surface: Any = _STOCK,
+) -> dict[str, Any]:
+    """Serialize a server result, letting a tagged task result through.
+
+    A modern `tools/call` result carrying `resultType: "task"` is returned as
+    the producer already dumped it, rather than being validated against — and
+    stripped by — the stock `CallToolResult | InputRequiredResult` surface. This
+    is the same bypass the runner already applies to custom-method results
+    (which skip surface validation entirely); the producer built this dict from
+    a validated `CreateTaskResult`, so its shape is already correct. Every other
+    result — and any call that supplies an explicit `surface` — delegates to the
+    SDK's original serializer unchanged.
+    """
+    if (
+        surface is _STOCK
+        and method == _TASK_AUGMENTED_METHOD
+        and version in MODERN_PROTOCOL_VERSIONS
+        and isinstance(data, Mapping)
+        and data.get("resultType") == _TASK_RESULT_TYPE
+    ):
+        return dict(data)
+    if surface is _STOCK:
+        return _original_serialize(method, version, data)
+    return _original_serialize(method, version, data, surface=surface)
+
+
+def install() -> None:
+    """Install the task claim-production wrap (reference-counted, idempotent).
+
+    Safe to call from every `TasksExtension.lifespan()`: the first call captures
+    and replaces the SDK serializer, later calls only bump the reference count.
+    """
+    global _original_serialize, _active_holds
+    _active_holds += 1
+    if _original_serialize is not None:
+        return
+    _original_serialize = _methods.serialize_server_result
+    # Runtime attribute swap: the wrapper is call-compatible (it forwards
+    # `surface` when supplied and only diverts the runner's no-surface task
+    # path), but ty cannot verify a monkeypatch's signature match.
+    _methods.serialize_server_result = _serialize_with_task_production  # ty: ignore[invalid-assignment]
+
+
+def uninstall() -> None:
+    """Release one hold; restore the SDK serializer when the last one exits."""
+    global _original_serialize, _active_holds
+    _active_holds -= 1
+    if _active_holds > 0:
+        return
+    _active_holds = 0
+    if _original_serialize is not None:
+        _methods.serialize_server_result = _original_serialize
+        _original_serialize = None
diff --git a/tests/tasks/server/test_wire_models.py b/tests/tasks/server/test_wire_models.py
index 76419e0fe..111aedf49 100644
--- a/tests/tasks/server/test_wire_models.py
+++ b/tests/tasks/server/test_wire_models.py
@@ -9,6 +9,15 @@ The vendored schema composes results as `allOf[Result, Task]` where the Task arm
 carries `additionalProperties: false`; a stray `_meta` therefore fails
 validation. The models omit `_meta` and the runner's `exclude_none` dump keeps it
 out, which is exactly what these assertions check.
+
+**Known schema-vs-protocol contradiction:** the modern `tools/call` result union
+carries a required `resultType` discriminator, and the SDK's client-side
+`ResultClaim` requires `CreateTaskResult` to pin `resultType: "task"` — so we
+emit it. The draft schema's Task arm, however, forbids `resultType` (its
+`additionalProperties: false` does not list it). We validate the task *fields*
+against the schema with the discriminator stripped, and assert separately that
+the discriminator is present on the wire. This contradiction is reported
+upstream (the schema forbids a field the base protocol requires).
 """
 
 from __future__ import annotations
@@ -44,6 +53,19 @@ def _dump(model: Any) -> dict[str, Any]:
     return model.model_dump(by_alias=True, mode="json", exclude_none=True)
 
 
+def _dump_task_fields(model: Any) -> dict[str, Any]:
+    """Dump without the `resultType` discriminator the draft schema omits.
+
+    `resultType` is required by the protocol's result union but forbidden by the
+    schema's Task arm; strip it so the remaining task fields can be validated
+    against the schema. `test_create_task_result_emits_result_type_discriminator`
+    covers the discriminator itself.
+    """
+    dumped = _dump(model)
+    dumped.pop("resultType", None)
+    return dumped
+
+
 def test_create_task_result_matches_schema():
     result = CreateTaskResult(
         task_id="t1",
@@ -53,7 +75,24 @@ def test_create_task_result_matches_schema():
         ttl_ms=900000,
         poll_interval_ms=5000,
     )
-    _validate("CreateTaskResult", _dump(result))
+    _validate("CreateTaskResult", _dump_task_fields(result))
+
+
+def test_create_task_result_emits_result_type_discriminator():
+    """The protocol requires `resultType: "task"` to distinguish a tasked result.
+
+    The modern `tools/call` union discriminates on `resultType`, and the SDK's
+    `ResultClaim` for tasks pins the model to `Literal["task"]`; without it a
+    client cannot tell a task result from a `CallToolResult`.
+    """
+    result = CreateTaskResult(
+        task_id="t1",
+        status="working",
+        created_at=_ISO,
+        last_updated_at=_ISO,
+        ttl_ms=900000,
+    )
+    assert _dump(result)["resultType"] == "task"
 
 
 @pytest.mark.parametrize(
@@ -83,7 +122,7 @@ def test_get_task_result_matches_schema(status: TaskStatus, payload: dict[str, A
         poll_interval_ms=5000,
         **payload,
     )
-    _validate("GetTaskResult", _dump(result))
+    _validate("GetTaskResult", _dump_task_fields(result))
 
 
 def test_get_task_result_completed_omits_error_and_inputs():
@@ -111,8 +150,10 @@ def test_null_ttl_is_permitted_by_schema():
         ttl_ms=None,
     )
     dumped = result.model_dump(by_alias=True, mode="json", exclude_none=False)
-    # Drop the other None optionals the runner would also drop, keeping ttlMs=null.
+    # Drop the other None optionals the runner would also drop, keeping ttlMs=null,
+    # and the resultType the draft schema omits (see module docstring).
     dumped = {k: v for k, v in dumped.items() if v is not None or k == "ttlMs"}
+    dumped.pop("resultType", None)
     _validate("CreateTaskResult", dumped)
 
 
diff --git a/tests/tasks/server/test_wire_production.py b/tests/tasks/server/test_wire_production.py
new file mode 100644
index 000000000..ec7ef7418
--- /dev/null
+++ b/tests/tasks/server/test_wire_production.py
@@ -0,0 +1,95 @@
+"""The server-side claim-production wrap for the tasks extension.
+
+`wire_production` widens the SDK's `tools/call` result serialization so a
+`CreateTaskResult` (`resultType: "task"`) survives to the wire instead of being
+stripped by the `CallToolResult | InputRequiredResult` surface. These tests
+exercise the wrap at the exact boundary the server runner calls
+(`mcp_types.methods.serialize_server_result`), which is where the SDK otherwise
+drops the task fields.
+"""
+
+from __future__ import annotations
+
+import mcp_types.methods as methods
+import pytest
+from fastmcp_tasks import wire_production
+
+_MODERN = "2026-07-28"
+
+_TASK_DICT = {
+    "resultType": "task",
+    "taskId": "abc123",
+    "status": "working",
+    "createdAt": "2026-07-21T12:00:00+00:00",
+    "lastUpdatedAt": "2026-07-21T12:00:00+00:00",
+    "ttlMs": 900000,
+}
+
+
+@pytest.fixture
+def installed():
+    """Install the wrap for one test, guaranteeing removal."""
+    wire_production.install()
+    try:
+        yield
+    finally:
+        wire_production.uninstall()
+
+
+def test_without_wrap_task_fields_are_stripped():
+    """Baseline: the stock serializer drops the task fields (the gap we close)."""
+    out = methods.serialize_server_result("tools/call", _MODERN, dict(_TASK_DICT))
+    assert "taskId" not in out
+
+
+def test_wrap_preserves_task_result(installed):
+    out = methods.serialize_server_result("tools/call", _MODERN, dict(_TASK_DICT))
+    assert out["taskId"] == "abc123"
+    assert out["resultType"] == "task"
+    assert out["status"] == "working"
+
+
+def test_wrap_leaves_ordinary_tool_result_untouched(installed):
+    """A normal (non-task) tools/call result serializes exactly as before."""
+    complete = {"content": [{"type": "text", "text": "hi"}], "resultType": "complete"}
+    out = methods.serialize_server_result("tools/call", _MODERN, complete)
+    assert out["content"] == [{"type": "text", "text": "hi"}]
+    assert "taskId" not in out
+
+
+def test_wrap_delegates_non_diverted_calls(installed):
+    """Only a task-tagged tools/call is diverted; everything else delegates.
+
+    A `tools/list` call is never routed to task production, so its payload is
+    validated by the stock `ListToolsResult` surface exactly as without the
+    wrap — proven here by the stock validator rejecting an off-surface dict
+    rather than the wrap silently converting or swallowing it.
+    """
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        methods.serialize_server_result("tools/list", _MODERN, {"tools": []})
+
+
+def test_uninstall_restores_stock_serializer():
+    wire_production.install()
+    wrapped = methods.serialize_server_result
+    wire_production.uninstall()
+    assert methods.serialize_server_result is not wrapped
+    # And the task fields are stripped again once restored.
+    out = methods.serialize_server_result("tools/call", _MODERN, dict(_TASK_DICT))
+    assert "taskId" not in out
+
+
+def test_refcount_survives_nested_holds():
+    """Two holds (sibling extensions): the wrap stays until the last release."""
+    wire_production.install()
+    wire_production.install()
+    wire_production.uninstall()
+    # One hold remains; the wrap is still active.
+    out = methods.serialize_server_result("tools/call", _MODERN, dict(_TASK_DICT))
+    assert out["taskId"] == "abc123"
+    wire_production.uninstall()
+    # Last hold released; stock behavior restored.
+    out = methods.serialize_server_result("tools/call", _MODERN, dict(_TASK_DICT))
+    assert "taskId" not in out