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

Merge pull request #2578 from jlowin/remove-enable-tasks

Browse source 
Remove enable_tasks setting, enable task protocol by default
This commit is contained in:
Chris Guidry 2025-12-09 10:16:52 -05:00 • committed by GitHub
commit 9ea57f9ef8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 217 additions and 361 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:

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

@ -36,6 +36,7 @@ from fastmcp.client.auth.oauth import OAuth
from fastmcp.mcp_config import MCPConfig, infer_transport_type_from_url
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.server import FastMCP
from fastmcp.server.tasks.capabilities import get_task_capabilities
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
@ -856,16 +857,7 @@ 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"] = {
"tools": True,
"prompts": True,
"resources": True,
}
experimental_capabilities = get_task_capabilities()
tg.start_soon(
lambda: self.server._mcp_server.run(

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

@ -19,9 +19,9 @@ 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.server.tasks.capabilities import get_task_capabilities
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
@ -161,19 +161,7 @@ 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"] = {
"list": {},
"cancel": {},
"requests": {
"tools": {"call": {}},
"prompts": {"get": {}},
"resources": {"read": {}},
},
}
experimental_capabilities = get_task_capabilities()
await server._mcp_server.run(
streams[0],

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

@ -75,6 +75,7 @@ from fastmcp.server.http import (
)
from fastmcp.server.low_level import LowLevelServer
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.tasks.capabilities import get_task_capabilities
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.server.tasks.handlers import (
handle_prompt_as_task,
@ -211,9 +212,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
@ -690,43 +689,46 @@ 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 self._should_enable_component(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)
@ -818,39 +820,43 @@ 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 self._should_enable_component(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)
@ -860,9 +866,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,
@ -1548,49 +1551,52 @@ 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 self._should_enable_component(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)
@ -2485,19 +2491,7 @@ 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"] = {
"list": {},
"cancel": {},
"requests": {
"tools": {"call": {}},
"prompts": {"get": {}},
"resources": {"read": {}},
},
}
experimental_capabilities = get_task_capabilities()
await self._mcp_server.run(
read_stream,

28
src/fastmcp/server/tasks/__init__.py
View file

@ -3,43 +3,19 @@
This module implements protocol-level background task execution for MCP servers.
"""
from fastmcp.server.tasks.capabilities import get_task_capabilities
from fastmcp.server.tasks.config import TaskConfig, TaskMode
from fastmcp.server.tasks.converters import (
convert_prompt_result,
convert_resource_result,
convert_tool_result,
)
from fastmcp.server.tasks.handlers import (
handle_prompt_as_task,
handle_resource_as_task,
handle_tool_as_task,
)
from fastmcp.server.tasks.keys import (
build_task_key,
get_client_task_id_from_key,
parse_task_key,
)
from fastmcp.server.tasks.protocol import (
tasks_cancel_handler,
tasks_get_handler,
tasks_list_handler,
tasks_result_handler,
)
__all__ = [
"TaskConfig",
"TaskMode",
"build_task_key",
"convert_prompt_result",
"convert_resource_result",
"convert_tool_result",
"get_client_task_id_from_key",
"handle_prompt_as_task",
"handle_resource_as_task",
"handle_tool_as_task",
"get_task_capabilities",
"parse_task_key",
"tasks_cancel_handler",
"tasks_get_handler",
"tasks_list_handler",
"tasks_result_handler",
]

22
src/fastmcp/server/tasks/capabilities.py Normal file
View file

@ -0,0 +1,22 @@
"""SEP-1686 task capabilities declaration."""
from typing import Any
def get_task_capabilities() -> dict[str, Any]:
"""Return the SEP-1686 task capabilities structure.
This is the standard capabilities map advertised to clients,
declaring support for list, cancel, and request operations.
"""
return {
"tasks": {
"list": {},
"cancel": {},
"requests": {
"tools": {"call": {}},
"prompts": {"get": {}},
"resources": {"read": {}},
},
}
}

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

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

@ -1,81 +1,44 @@
"""
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
from fastmcp.server.tasks import get_task_capabilities
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 == get_task_capabilities()["tasks"]
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