ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add server-side claim production for tasks; emit resultType discriminator

Browse source 
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>
This commit is contained in:
Jeremiah Lowin 2026-07-22 08:03:32 -04:00
commit ef29b731ea
No known key found for this signature in database
6 changed files with 369 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

78
fastmcp_tasks/README.md
View file

@ -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.

7
fastmcp_tasks/fastmcp_tasks/extension.py
View file

@ -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()

42
fastmcp_tasks/fastmcp_tasks/models.py
View file

@ -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):

111
fastmcp_tasks/fastmcp_tasks/wire_production.py Normal file
View file

@ -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

47
tests/tasks/server/test_wire_models.py
View file

@ -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)

95
tests/tasks/server/test_wire_production.py Normal file
View file

@ -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