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

Remove enable_tasks setting, enable task protocol by default

Browse source 
The task protocol (SEP-1686) is now always enabled - server always
registers task handlers and advertises task capabilities. Users still
opt into background execution at the server level (tasks=True) or
component level (task=True on tools, prompts, resources).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Chris Guidry 2025-12-09 09:34:42 -05:00
commit 350f723592
14 changed files with 191 additions and 315 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/servers/tasks.mdx
View file

@ -16,9 +16,6 @@ FastMCP implements the MCP background task protocol ([SEP-1686](https://modelcon
**What is Docket?** FastMCP's task system is powered by [Docket](https://github.com/chrisguidry/docket), originally built by [Prefect](https://prefect.io) to power [Prefect Cloud](https://www.prefect.io/prefect/cloud)'s managed task scheduling and execution service. Docket is the beating heart of Prefect's enterprise task infrastructure, processing millions of tasks daily across their multi-tenant SaaS platform. It's now open-sourced for the community.
</Tip>
<Note>
Background tasks are disabled by default in v2.14.0. Enable them with `FASTMCP_ENABLE_TASKS=true` or by passing `tasks=True` to the FastMCP constructor. This default will change in a future release.
</Note>
## What Are MCP Background Tasks?
@ -116,11 +113,8 @@ Conversely, when a component has `mode="required"` but the client doesn't reques
### Configuration
Background tasks require explicit opt-in via environment variable:
| Environment Variable | Default | Description |
|---------------------|---------|-------------|
| `FASTMCP_ENABLE_TASKS` | `false` | Enable the MCP task protocol |
| `FASTMCP_DOCKET_URL` | `memory://` | Backend URL (`memory://` or `redis://host:port/db`) |
## Backends

3
examples/tasks/.envrc
View file

@ -2,9 +2,6 @@
# This file is loaded by direnv (https://direnv.net/) when you cd into this directory
# Run `direnv allow` to enable automatic environment loading
# Enable MCP SEP-1686 task protocol support
export FASTMCP_ENABLE_TASKS=true
# Configure Docket backend URL
# Use Redis backend (requires docker-compose up)
export FASTMCP_DOCKET_URL=redis://localhost:24242/0

1
examples/tasks/README.md
View file

@ -51,7 +51,6 @@ fastmcp tasks worker server.py
| Variable | Default | Description |
|----------|---------|-------------|
| `FASTMCP_ENABLE_TASKS` | `false` | Enable MCP task protocol (SEP-1686) |
| `FASTMCP_DOCKET_URL` | `memory://` | Docket backend URL |
## Learn More

11
src/fastmcp/client/client.py
View file

@ -458,14 +458,9 @@ class Client(Generic[ClientTransportT]):
try:
with anyio.fail_after(timeout):
if fastmcp.settings.enable_tasks:
self._session_state.initialize_result = (
await _task_capable_initialize(self.session)
)
else:
self._session_state.initialize_result = (
await self.session.initialize()
)
self._session_state.initialize_result = await _task_capable_initialize(
self.session
)
return self._session_state.initialize_result
except TimeoutError as e:

10
src/fastmcp/client/transports.py
View file

@ -856,16 +856,14 @@ class FastMCPTransport(ClientTransport):
_enter_server_lifespan(server=self.server),
):
# Build experimental capabilities
import fastmcp
experimental_capabilities = {}
if fastmcp.settings.enable_tasks:
# Declare SEP-1686 task support
experimental_capabilities["tasks"] = {
# Declare SEP-1686 task support
experimental_capabilities = {
"tasks": {
"tools": True,
"prompts": True,
"resources": True,
}
}
tg.start_soon(
lambda: self.server._mcp_server.run(

11
src/fastmcp/server/http.py
View file

@ -19,7 +19,6 @@ from starlette.responses import Response
from starlette.routing import BaseRoute, Mount, Route
from starlette.types import Lifespan, Receive, Scope, Send
import fastmcp
from fastmcp.server.auth import AuthProvider
from fastmcp.server.auth.middleware import RequireAuthMiddleware
from fastmcp.utilities.logging import get_logger
@ -161,11 +160,10 @@ def create_sse_app(
async def handle_sse(scope: Scope, receive: Receive, send: Send) -> Response:
async with sse.connect_sse(scope, receive, send) as streams:
# Build experimental capabilities
experimental_capabilities = {}
if fastmcp.settings.enable_tasks:
# Declare SEP-1686 task support per final spec (lines 49-63)
# Nested structure: {list: {}, cancel: {}, requests: {tools: {call: {}}}}
experimental_capabilities["tasks"] = {
# Declare SEP-1686 task support per final spec (lines 49-63)
# Nested structure: {list: {}, cancel: {}, requests: {tools: {call: {}}}}
experimental_capabilities = {
"tasks": {
"list": {},
"cancel": {},
"requests": {
@ -174,6 +172,7 @@ def create_sse_app(
"resources": {"read": {}},
},
}
}
await server._mcp_server.run(
streams[0],

214
src/fastmcp/server/server.py
View file

@ -210,9 +210,7 @@ class FastMCP(Generic[LifespanResultT]):
sampling_handler_behavior: Literal["always", "fallback"] | None = None,
):
# Resolve server default for background task support
self._support_tasks_by_default: bool = (
tasks if tasks is not None else fastmcp.settings.enable_tasks
)
self._support_tasks_by_default: bool = tasks if tasks is not None else False
# Docket instance (set during lifespan for cross-task access)
self._docket = None
@ -689,43 +687,42 @@ class FastMCP(Generic[LifespanResultT]):
pass
# Check for task metadata and route appropriately
if fastmcp.settings.enable_tasks:
async with fastmcp.server.context.Context(fastmcp=self):
# Get resource including from mounted servers
resource = await self._get_resource_with_task_config(str(uri))
if resource and hasattr(resource, "task_config"):
task_mode = resource.task_config.mode # type: ignore[union-attr]
async with fastmcp.server.context.Context(fastmcp=self):
# Get resource including from mounted servers
resource = await self._get_resource_with_task_config(str(uri))
if resource and hasattr(resource, "task_config"):
task_mode = resource.task_config.mode # type: ignore[union-attr]
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Resource '{uri}' requires task-augmented execution",
)
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Resource '{uri}' requires task-augmented execution",
)
)
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
# For FunctionResource/FunctionResourceTemplate, use Docket
if isinstance(
resource,
FunctionResource | FunctionResourceTemplate,
):
task_meta_dict = task_meta.model_dump(exclude_none=True)
return await handle_resource_as_task(
self, str(uri), resource, task_meta_dict
)
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
# For FunctionResource/FunctionResourceTemplate, use Docket
if isinstance(
resource,
FunctionResource | FunctionResourceTemplate,
):
task_meta_dict = task_meta.model_dump(exclude_none=True)
return await handle_resource_as_task(
self, str(uri), resource, task_meta_dict
)
# Forbidden mode: task requested but mode="forbidden"
# Raise error since resources don't have isError field
if task_meta and task_mode == "forbidden":
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Resource '{uri}' does not support task-augmented execution",
)
# Forbidden mode: task requested but mode="forbidden"
# Raise error since resources don't have isError field
if task_meta and task_mode == "forbidden":
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Resource '{uri}' does not support task-augmented execution",
)
)
# Synchronous execution
result = await self._read_resource_mcp(uri)
@ -817,39 +814,38 @@ class FastMCP(Generic[LifespanResultT]):
pass
# Check for task metadata and route appropriately
if fastmcp.settings.enable_tasks:
async with fastmcp.server.context.Context(fastmcp=self):
prompts = await self.get_prompts()
prompt = prompts.get(name)
if prompt and hasattr(prompt, "task_config") and prompt.task_config:
task_mode = prompt.task_config.mode # type: ignore[union-attr]
async with fastmcp.server.context.Context(fastmcp=self):
prompts = await self.get_prompts()
prompt = prompts.get(name)
if prompt and hasattr(prompt, "task_config") and prompt.task_config:
task_mode = prompt.task_config.mode # type: ignore[union-attr]
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Prompt '{name}' requires task-augmented execution",
)
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Prompt '{name}' requires task-augmented execution",
)
)
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
task_meta_dict = task_meta.model_dump(exclude_none=True)
result = await handle_prompt_as_task(
self, name, arguments, task_meta_dict
)
return mcp.types.ServerResult(result)
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
task_meta_dict = task_meta.model_dump(exclude_none=True)
result = await handle_prompt_as_task(
self, name, arguments, task_meta_dict
)
return mcp.types.ServerResult(result)
# Forbidden mode: task requested but mode="forbidden"
# Raise error since prompts don't have isError field
if task_meta and task_mode == "forbidden":
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Prompt '{name}' does not support task-augmented execution",
)
# Forbidden mode: task requested but mode="forbidden"
# Raise error since prompts don't have isError field
if task_meta and task_mode == "forbidden":
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Prompt '{name}' does not support task-augmented execution",
)
)
# Synchronous execution
result = await self._get_prompt_mcp(name, arguments)
@ -859,9 +855,6 @@ class FastMCP(Generic[LifespanResultT]):
def _setup_task_protocol_handlers(self) -> None:
"""Register SEP-1686 task protocol handlers with SDK."""
if not fastmcp.settings.enable_tasks:
return
from mcp.types import (
CancelTaskRequest,
GetTaskPayloadRequest,
@ -1547,49 +1540,48 @@ class FastMCP(Generic[LifespanResultT]):
# No request context available - proceed without task metadata
pass
if fastmcp.settings.enable_tasks:
# Get tool from local manager, mounted servers, or proxy
tool = await self._get_tool_with_task_config(key)
if tool and hasattr(tool, "task_config"):
task_mode = tool.task_config.mode # type: ignore[union-attr]
# Get tool from local manager, mounted servers, or proxy
tool = await self._get_tool_with_task_config(key)
if tool and hasattr(tool, "task_config"):
task_mode = tool.task_config.mode # type: ignore[union-attr]
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Tool '{key}' requires task-augmented execution",
)
# Enforce mode="required" - must have task metadata
if task_mode == "required" and not task_meta:
raise McpError(
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Tool '{key}' requires task-augmented execution",
)
)
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
# For FunctionTool, use Docket for background execution
if isinstance(tool, FunctionTool):
task_meta_dict = task_meta.model_dump(exclude_none=True)
return await handle_tool_as_task(
self, key, arguments, task_meta_dict
)
# For ProxyTool/mounted tools, proceed with normal execution
# They will forward task metadata to their backend
# Forbidden mode: task requested but mode="forbidden"
# Return error result with returned_immediately=True
if task_meta and task_mode == "forbidden":
return mcp.types.CallToolResult(
content=[
mcp.types.TextContent(
type="text",
text=f"Tool '{key}' does not support task-augmented execution",
)
],
isError=True,
_meta={
"modelcontextprotocol.io/task": {
"returned_immediately": True
}
},
# Route to background if task metadata present and mode allows
if task_meta and task_mode != "forbidden":
# For FunctionTool, use Docket for background execution
if isinstance(tool, FunctionTool):
task_meta_dict = task_meta.model_dump(exclude_none=True)
return await handle_tool_as_task(
self, key, arguments, task_meta_dict
)
# For ProxyTool/mounted tools, proceed with normal execution
# They will forward task metadata to their backend
# Forbidden mode: task requested but mode="forbidden"
# Return error result with returned_immediately=True
if task_meta and task_mode == "forbidden":
return mcp.types.CallToolResult(
content=[
mcp.types.TextContent(
type="text",
text=f"Tool '{key}' does not support task-augmented execution",
)
],
isError=True,
_meta={
"modelcontextprotocol.io/task": {
"returned_immediately": True
}
},
)
# Synchronous execution (normal path)
result = await self._call_tool_middleware(key, arguments)
@ -2484,11 +2476,10 @@ class FastMCP(Generic[LifespanResultT]):
)
# Build experimental capabilities
experimental_capabilities = {}
if fastmcp.settings.enable_tasks:
# Declare SEP-1686 task support per final spec (lines 49-63)
# Nested structure: {list: {}, cancel: {}, requests: {tools: {call: {}}}}
experimental_capabilities["tasks"] = {
# Declare SEP-1686 task support per final spec (lines 49-63)
# Nested structure: {list: {}, cancel: {}, requests: {tools: {call: {}}}}
experimental_capabilities = {
"tasks": {
"list": {},
"cancel": {},
"requests": {
@ -2497,6 +2488,7 @@ class FastMCP(Generic[LifespanResultT]):
"resources": {"read": {}},
},
}
}
await self._mcp_server.run(
read_stream,

18
src/fastmcp/settings.py
View file

@ -209,24 +209,6 @@ class Settings(BaseSettings):
experimental: ExperimentalSettings = ExperimentalSettings()
# Tasks settings
enable_tasks: Annotated[
bool,
Field(
description=inspect.cleandoc(
"""
Enable MCP SEP-1686 task protocol support for background execution.
Server-side: Advertises task capabilities and handles task/* protocol
methods. Tools, prompts, and resources marked with task=True will
execute in the background via Docket.
Client-side: Advertises task capability to servers.
"""
),
),
] = False
docket: DocketSettings = DocketSettings()
enable_rich_tracebacks: Annotated[

11
tests/client/tasks/conftest.py
View file

@ -1,12 +1,3 @@
"""Shared fixtures for client task tests."""
import pytest
from fastmcp.utilities.tests import temporary_settings
@pytest.fixture(autouse=True)
async def enable_tasks():
"""Enable task protocol support for all client task tests."""
with temporary_settings(enable_tasks=True):
yield
# Task protocol is now always enabled - no fixture needed

11
tests/client/test_client.py
View file

@ -1,9 +1,7 @@
import asyncio
import sys
from typing import Any, cast
from unittest.mock import AsyncMock
import mcp
import pytest
from mcp import McpError
from mcp.client.auth import OAuthClientProvider
@ -415,12 +413,13 @@ async def test_client_connection(fastmcp_server):
assert not client.is_connected()
async def test_initialize_called_once(fastmcp_server, monkeypatch):
mock_initialize = AsyncMock()
monkeypatch.setattr(mcp.ClientSession, "initialize", mock_initialize)
async def test_initialize_called_once(fastmcp_server):
"""Test that initialization is called once and sets initialize_result."""
client = Client(transport=FastMCPTransport(fastmcp_server))
async with client:
assert mock_initialize.call_count == 1
# Verify that initialization succeeded by checking initialize_result
assert client.initialize_result is not None
assert client.initialize_result.serverInfo is not None
async def test_initialize_result_connected(fastmcp_server):

16
tests/server/tasks/conftest.py
View file

@ -1,17 +1,3 @@
"""Shared fixtures for task tests."""
import pytest
from fastmcp.utilities.tests import temporary_settings
@pytest.fixture(autouse=True)
async def enable_tasks():
"""Enable task protocol support for all task tests."""
with temporary_settings(enable_tasks=True):
# Verify enabled
import fastmcp
assert fastmcp.settings.enable_tasks, "Tasks should be enabled after fixture"
yield
# Task protocol is now always enabled - no fixture needed

86
tests/server/tasks/test_server_tasks_parameter.py
View file

@ -8,7 +8,6 @@ settings properly override the server default.
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.utilities.tests import temporary_settings
async def test_server_tasks_true_defaults_all_components():
@ -84,38 +83,21 @@ async def test_server_tasks_false_defaults_all_components():
await client.read_resource("test://resource", task=True)
async def test_server_tasks_none_uses_settings():
"""Server with tasks=None (or omitted) uses global settings."""
# Test with enable_tasks=True in settings
with temporary_settings(enable_tasks=True):
mcp = FastMCP("test") # tasks=None, should use settings
async def test_server_tasks_none_defaults_to_false():
"""Server with tasks=None (or omitted) defaults to False."""
mcp = FastMCP("test") # tasks=None, defaults to False
@mcp.tool()
async def my_tool() -> str:
return "tool result"
@mcp.tool()
async def my_tool() -> str:
return "tool result"
async with Client(mcp) as client:
# Tool should support background execution (from settings)
tool_task = await client.call_tool("my_tool", task=True)
assert not tool_task.returned_immediately
# Test with enable_tasks=False in settings
with temporary_settings(enable_tasks=False):
mcp2 = FastMCP("test2") # tasks=None, should use settings
@mcp2.tool()
async def my_tool2() -> str:
return "tool result"
async with Client(mcp2) as client:
# When enable_tasks=False, server doesn't advertise task capabilities.
# Client's task=True is ignored because server doesn't support tasks.
# Tool executes synchronously and succeeds.
tool_task = await client.call_tool("my_tool2", task=True)
assert tool_task.returned_immediately
result = await tool_task.result()
# Tool should execute successfully (synchronously)
assert "tool result" in str(result)
async with Client(mcp) as client:
# Tool should NOT support background execution (mode="forbidden" from default)
tool_task = await client.call_tool("my_tool", task=True)
assert tool_task.returned_immediately
result = await tool_task.result()
assert result.is_error
assert "does not support task-augmented execution" in str(result)
async def test_component_explicit_false_overrides_server_true():
@ -261,34 +243,32 @@ async def test_mixed_explicit_and_inherited():
async def test_server_tasks_parameter_sets_component_defaults():
"""Server tasks parameter sets component defaults but global settings gate protocol."""
# Server tasks=True sets component defaults, but enable_tasks must be True
with temporary_settings(enable_tasks=True):
mcp = FastMCP("test", tasks=True)
"""Server tasks parameter sets component defaults."""
# Server tasks=True sets component defaults
mcp = FastMCP("test", tasks=True)
@mcp.tool()
async def tool_inherits_true() -> str:
return "tool result"
@mcp.tool()
async def tool_inherits_true() -> str:
return "tool result"
async with Client(mcp) as client:
# Tool inherits tasks=True from server
tool_task = await client.call_tool("tool_inherits_true", task=True)
assert not tool_task.returned_immediately
async with Client(mcp) as client:
# Tool inherits tasks=True from server
tool_task = await client.call_tool("tool_inherits_true", task=True)
assert not tool_task.returned_immediately
# Server tasks=False sets component defaults
with temporary_settings(enable_tasks=True):
mcp2 = FastMCP("test2", tasks=False)
mcp2 = FastMCP("test2", tasks=False)
@mcp2.tool()
async def tool_inherits_false() -> str:
return "tool result"
@mcp2.tool()
async def tool_inherits_false() -> str:
return "tool result"
async with Client(mcp2) as client:
# Tool inherits tasks=False (mode="forbidden") - returns error
tool_task = await client.call_tool("tool_inherits_false", task=True)
assert tool_task.returned_immediately
result = await tool_task.result()
assert result.is_error
async with Client(mcp2) as client:
# Tool inherits tasks=False (mode="forbidden") - returns error
tool_task = await client.call_tool("tool_inherits_false", task=True)
assert tool_task.returned_immediately
result = await tool_task.result()
assert result.is_error
async def test_resource_template_inherits_server_tasks_default():

96
tests/server/tasks/test_task_capabilities.py
View file

@ -1,81 +1,47 @@
"""
Tests for SEP-1686 task capabilities declaration.
Verifies that the server correctly advertises task support based on settings.
Verifies that the server correctly advertises task support.
Task protocol is now always enabled.
"""
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.utilities.tests import temporary_settings
async def test_capabilities_include_tasks_when_enabled():
"""Server capabilities include tasks when enable_tasks=True."""
with temporary_settings(enable_tasks=True):
mcp = FastMCP("capability-test")
async def test_capabilities_include_tasks():
"""Server capabilities always include tasks."""
mcp = FastMCP("capability-test")
@mcp.tool()
async def test_tool() -> str:
return "test"
@mcp.tool()
async def test_tool() -> str:
return "test"
async with Client(mcp) as client:
# Get server initialization result which includes capabilities
init_result = client.initialize_result
async with Client(mcp) as client:
# Get server initialization result which includes capabilities
init_result = client.initialize_result
# Verify tasks capability is present
assert init_result.capabilities.experimental is not None
assert "tasks" in init_result.capabilities.experimental
tasks_cap = init_result.capabilities.experimental["tasks"]
assert tasks_cap == {
"tools": True,
"prompts": True,
"resources": True,
}
# Verify tasks capability is present
assert init_result.capabilities.experimental is not None
assert "tasks" in init_result.capabilities.experimental
tasks_cap = init_result.capabilities.experimental["tasks"]
assert tasks_cap == {
"tools": True,
"prompts": True,
"resources": True,
}
async def test_capabilities_exclude_tasks_when_disabled():
"""Server capabilities do NOT include tasks when enable_tasks=False."""
with temporary_settings(enable_tasks=False):
mcp = FastMCP("capability-test")
async def test_client_uses_task_capable_session():
"""Client uses task-capable initialization."""
mcp = FastMCP("client-cap-test")
@mcp.tool()
def test_tool() -> str:
return "test"
@mcp.tool()
async def test_tool() -> str:
return "test"
async with Client(mcp) as client:
# Get server initialization result
init_result = client.initialize_result
# Verify tasks capability is NOT present
if init_result.capabilities.experimental:
assert "tasks" not in init_result.capabilities.experimental
async def test_client_advertises_task_capability_when_enabled():
"""Client advertises experimental.tasks capability when enable_tasks=True."""
with temporary_settings(enable_tasks=True):
mcp = FastMCP("client-cap-test")
@mcp.tool()
async def test_tool() -> str:
return "test"
async with Client(mcp) as client:
# Client should have connected successfully with task capabilities
assert client.initialize_result is not None
async def test_client_does_not_advertise_tasks_when_disabled():
"""Client does NOT use custom session when enable_tasks=False."""
with temporary_settings(enable_tasks=False):
mcp = FastMCP("no-tasks-client-test")
@mcp.tool()
def test_tool() -> str:
return "test"
async with Client(mcp) as client:
# Session should be standard ClientSession, not our custom one
# The session should be a standard ClientSession
assert type(client.session).__name__ == "ClientSession"
async with Client(mcp) as client:
# Client should have connected successfully with task capabilities
assert client.initialize_result is not None
# Session should be a ClientSession (task-capable init uses standard session)
assert type(client.session).__name__ == "ClientSession"

2
tests/server/tasks/test_task_protocol.py
View file

@ -26,8 +26,6 @@ async def task_enabled_server():
"""A tool that always fails."""
raise ValueError("This tool always fails")
assert mcp._support_tasks_by_default
return mcp