Add TaskConfig for SEP-1686 execution modes

Expose the full MCP task execution modes (forbidden/optional/required)
via TaskConfig instead of just boolean task=True/False.

docs/servers/tasks.mdx

@@ -61,6 +61,41 @@ When a client requests background execution, the call returns immediately with a
 Background tasks require async functions. Attempting to use `task=True` with a sync function raises a `ValueError` at registration time.
 </Warning>
 
+## Execution Modes
+
+For fine-grained control over task execution behavior, use `TaskConfig` instead of the boolean shorthand. The MCP task protocol defines three execution modes:
+
+| Mode | Client calls without task | Client calls with task |
+|------|--------------------------|------------------------|
+| `"forbidden"` | Executes synchronously | Error: task not supported |
+| `"optional"` | Executes synchronously | Executes as background task |
+| `"required"` | Error: task required | Executes as background task |
+
+```python
+from fastmcp import FastMCP, TaskConfig
+
+mcp = FastMCP("MyServer")
+
+# Supports both sync and background execution (default when task=True)
+@mcp.tool(task=TaskConfig(mode="optional"))
+async def flexible_task() -> str:
+    return "Works either way"
+
+# Requires background execution - errors if client doesn't request task
+@mcp.tool(task=TaskConfig(mode="required"))
+async def must_be_background() -> str:
+    return "Only runs as a background task"
+
+# No task support (default when task=False or omitted)
+@mcp.tool(task=TaskConfig(mode="forbidden"))
+async def sync_only() -> str:
+    return "Never runs as background task"
+```
+
+The boolean shortcuts map to these modes:
+- `task=True` → `TaskConfig(mode="optional")`
+- `task=False` → `TaskConfig(mode="forbidden")`
+
 ### Server-Wide Default
 
 To enable background task support for all components by default, pass `tasks=True` to the constructor. Individual decorators can still override this with `task=False`.
@@ -75,7 +110,9 @@ If your server defines any synchronous tools, resources, or prompts, you will ne
 
 ### Graceful Degradation
 
-When a client requests background execution (`task=True` in the request) but the component doesn't support it (`task=False` on the decorator), FastMCP executes synchronously and returns the result inline. This follows the SEP-1686 specification for graceful degradation—clients can always request background execution without worrying about server capabilities.
+When a client requests background execution but the component has `mode="forbidden"`, FastMCP executes synchronously and returns the result inline. This follows the SEP-1686 specification for graceful degradation—clients can always request background execution without worrying about server capabilities.
+
+Conversely, when a component has `mode="required"` but the client doesn't request background execution, FastMCP returns an error indicating that task execution is required.
 
 ### Configuration
 

src/fastmcp/__init__.py

@@ -14,6 +14,7 @@ if settings.log_enabled:
 
 from fastmcp.server.server import FastMCP
 from fastmcp.server.context import Context
+from fastmcp.server.tasks.config import TaskConfig
 import fastmcp.server
 
 from fastmcp.client import Client
@@ -31,6 +32,7 @@ __all__ = [
     "Client",
     "Context",
     "FastMCP",
+    "TaskConfig",
     "client",
     "settings",
 ]

src/fastmcp/prompts/prompt.py

@@ -15,6 +15,7 @@ from pydantic import Field, TypeAdapter
 
 from fastmcp.exceptions import PromptError
 from fastmcp.server.dependencies import get_context, without_injected_parameters
+from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.json_schema import compress_schema
 from fastmcp.utilities.logging import get_logger
@@ -120,7 +121,7 @@ class Prompt(FastMCPComponent):
         tags: set[str] | None = None,
         enabled: bool | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionPrompt:
         """Create a Prompt from a function.
 
@@ -158,12 +159,10 @@ class FunctionPrompt(Prompt):
     """A prompt that is a function."""
 
     fn: Callable[..., PromptResult | Awaitable[PromptResult]]
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this prompt supports background task execution (SEP-1686)"
-        ),
-    ] = False
+    task_config: Annotated[
+        TaskConfig,
+        Field(description="Background task execution configuration (SEP-1686)."),
+    ] = Field(default_factory=lambda: TaskConfig(mode="forbidden"))
 
     @classmethod
     def from_function(
@@ -176,7 +175,7 @@ class FunctionPrompt(Prompt):
         tags: set[str] | None = None,
         enabled: bool | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionPrompt:
         """Create a Prompt from a function.
 
@@ -201,6 +200,15 @@ class FunctionPrompt(Prompt):
 
         description = description or inspect.getdoc(fn)
 
+        # Normalize task to TaskConfig and validate
+        if task is None:
+            task_config = TaskConfig(mode="forbidden")
+        elif isinstance(task, bool):
+            task_config = TaskConfig.from_bool(task)
+        else:
+            task_config = task
+        task_config.validate_function(fn, func_name)
+
         # if the fn is a callable class, we need to get the __call__ method from here out
         if not inspect.isroutine(fn):
             fn = fn.__call__
@@ -208,13 +216,6 @@ class FunctionPrompt(Prompt):
         if isinstance(fn, staticmethod):
             fn = fn.__func__  # type: ignore[assignment]
 
-        # Validate that task=True requires async functions (after unwrapping)
-        if task and not inspect.iscoroutinefunction(fn):
-            raise ValueError(
-                f"Prompt '{func_name}' uses a sync function but has task=True. "
-                "Background tasks require async functions. Set task=False to disable."
-            )
-
         # Wrap fn to handle dependency resolution internally
         wrapped_fn = without_injected_parameters(fn)
         type_adapter = get_cached_typeadapter(wrapped_fn)
@@ -271,7 +272,7 @@ class FunctionPrompt(Prompt):
             enabled=enabled if enabled is not None else True,
             fn=wrapped_fn,
             meta=meta,
-            task=task if task is not None else False,
+            task_config=task_config,
         )
 
     def _convert_string_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:

src/fastmcp/resources/resource.py

@@ -20,6 +20,7 @@ from pydantic import (
 from typing_extensions import Self
 
 from fastmcp.server.dependencies import get_context, without_injected_parameters
+from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.types import (
     get_fn_name,
@@ -77,7 +78,7 @@ class Resource(FastMCPComponent):
         enabled: bool | None = None,
         annotations: Annotations | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionResource:
         return FunctionResource.from_function(
             fn=fn,
@@ -170,12 +171,10 @@ class FunctionResource(Resource):
     """
 
     fn: Callable[..., Any]
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this resource supports background task execution (SEP-1686)"
-        ),
-    ] = False
+    task_config: Annotated[
+        TaskConfig,
+        Field(description="Background task execution configuration (SEP-1686)."),
+    ] = Field(default_factory=lambda: TaskConfig(mode="forbidden"))
 
     @classmethod
     def from_function(
@@ -191,24 +190,22 @@ class FunctionResource(Resource):
         enabled: bool | None = None,
         annotations: Annotations | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionResource:
         """Create a FunctionResource from a function."""
         if isinstance(uri, str):
             uri = AnyUrl(uri)
 
-        # Validate that task=True requires async functions
-        # Handle callable classes and staticmethods before checking
-        fn_to_check = fn
-        if not inspect.isroutine(fn) and callable(fn):
-            fn_to_check = fn.__call__
-        if isinstance(fn_to_check, staticmethod):
-            fn_to_check = fn_to_check.__func__
-        if task and not inspect.iscoroutinefunction(fn_to_check):
-            raise ValueError(
-                f"Resource '{name or get_fn_name(fn)}' uses a sync function but has task=True. "
-                "Background tasks require async functions. Set task=False to disable."
-            )
+        func_name = name or get_fn_name(fn)
+
+        # Normalize task to TaskConfig and validate
+        if task is None:
+            task_config = TaskConfig(mode="forbidden")
+        elif isinstance(task, bool):
+            task_config = TaskConfig.from_bool(task)
+        else:
+            task_config = task
+        task_config.validate_function(fn, func_name)
 
         # Wrap fn to handle dependency resolution internally
         wrapped_fn = without_injected_parameters(fn)
@@ -225,7 +222,7 @@ class FunctionResource(Resource):
             enabled=enabled if enabled is not None else True,
             annotations=annotations,
             meta=meta,
-            task=task if task is not None else False,
+            task_config=task_config,
         )
 
     async def read(self) -> str | bytes:

src/fastmcp/resources/template.py

@@ -18,6 +18,7 @@ from pydantic import (
 
 from fastmcp.resources.resource import Resource
 from fastmcp.server.dependencies import get_context, without_injected_parameters
+from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.json_schema import compress_schema
 from fastmcp.utilities.types import get_cached_typeadapter
@@ -136,7 +137,7 @@ class ResourceTemplate(FastMCPComponent):
         enabled: bool | None = None,
         annotations: Annotations | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionResourceTemplate:
         return FunctionResourceTemplate.from_function(
             fn=fn,
@@ -231,12 +232,10 @@ class FunctionResourceTemplate(ResourceTemplate):
     """A template for dynamically creating resources."""
 
     fn: Callable[..., Any]
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this resource template supports background task execution (SEP-1686)"
-        ),
-    ] = False
+    task_config: Annotated[
+        TaskConfig,
+        Field(description="Background task execution configuration (SEP-1686)."),
+    ] = Field(default_factory=lambda: TaskConfig(mode="forbidden"))
 
     async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
         """Create a resource from the template with the given parameters."""
@@ -254,7 +253,7 @@ class FunctionResourceTemplate(ResourceTemplate):
             mime_type=self.mime_type,
             tags=self.tags,
             enabled=self.enabled,
-            task=self.task,
+            task=self.task_config,
         )
 
     async def read(self, arguments: dict[str, Any]) -> str | bytes:
@@ -302,7 +301,7 @@ class FunctionResourceTemplate(ResourceTemplate):
         enabled: bool | None = None,
         annotations: Annotations | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionResourceTemplate:
         """Create a template from a function."""
 
@@ -373,6 +372,15 @@ class FunctionResourceTemplate(ResourceTemplate):
 
         description = description or inspect.getdoc(fn)
 
+        # Normalize task to TaskConfig and validate
+        if task is None:
+            task_config = TaskConfig(mode="forbidden")
+        elif isinstance(task, bool):
+            task_config = TaskConfig.from_bool(task)
+        else:
+            task_config = task
+        task_config.validate_function(fn, func_name)
+
         # if the fn is a callable class, we need to get the __call__ method from here out
         if not inspect.isroutine(fn):
             fn = fn.__call__
@@ -380,13 +388,6 @@ class FunctionResourceTemplate(ResourceTemplate):
         if isinstance(fn, staticmethod):
             fn = fn.__func__
 
-        # Validate that task=True requires async functions (after unwrapping)
-        if task and not inspect.iscoroutinefunction(fn):
-            raise ValueError(
-                f"Resource template '{func_name}' uses a sync function but has task=True. "
-                "Background tasks require async functions. Set task=False to disable."
-            )
-
         wrapper_fn = without_injected_parameters(fn)
         type_adapter = get_cached_typeadapter(wrapper_fn)
         parameters = type_adapter.json_schema()
@@ -408,5 +409,5 @@ class FunctionResourceTemplate(ResourceTemplate):
             enabled=enabled if enabled is not None else True,
             annotations=annotations,
             meta=meta,
-            task=task if task is not None else False,
+            task_config=task_config,
         )

src/fastmcp/server/server.py

@@ -35,11 +35,14 @@ from docket import Docket, Worker
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
 from mcp.server.stdio import stdio_server
+from mcp.shared.exceptions import McpError
 from mcp.types import (
+    METHOD_NOT_FOUND,
     Annotations,
     AnyFunction,
     CallToolRequestParams,
     ContentBlock,
+    ErrorData,
     GetPromptResult,
     ToolAnnotations,
 )
@@ -71,6 +74,7 @@ from fastmcp.server.http import (
 )
 from fastmcp.server.low_level import LowLevelServer
 from fastmcp.server.middleware import Middleware, MiddlewareContext
+from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.server.tasks.handlers import (
     handle_prompt_as_task,
     handle_resource_as_task,
@@ -401,21 +405,34 @@ class FastMCP(Generic[LifespanResultT]):
 
                 # Register local task-enabled tools/prompts/resources with Docket
                 # Only function-based variants support background tasks
+                # Register components where task execution is not "forbidden"
                 for tool in self._tool_manager._tools.values():
-                    if isinstance(tool, FunctionTool) and tool.task:
+                    if (
+                        isinstance(tool, FunctionTool)
+                        and tool.task_config.mode != "forbidden"
+                    ):
                         docket.register(tool.fn)
 
                 for prompt in self._prompt_manager._prompts.values():
-                    if isinstance(prompt, FunctionPrompt) and prompt.task:
-                        # task=True requires async fn (validated at creation time)
+                    if (
+                        isinstance(prompt, FunctionPrompt)
+                        and prompt.task_config.mode != "forbidden"
+                    ):
+                        # task execution requires async fn (validated at creation time)
                         docket.register(cast(Callable[..., Awaitable[Any]], prompt.fn))
 
                 for resource in self._resource_manager._resources.values():
-                    if isinstance(resource, FunctionResource) and resource.task:
+                    if (
+                        isinstance(resource, FunctionResource)
+                        and resource.task_config.mode != "forbidden"
+                    ):
                         docket.register(resource.fn)
 
                 for template in self._resource_manager._templates.values():
-                    if isinstance(template, FunctionResourceTemplate) and template.task:
+                    if (
+                        isinstance(template, FunctionResourceTemplate)
+                        and template.task_config.mode != "forbidden"
+                    ):
                         docket.register(template.fn)
 
                 # Set Docket in ContextVar so CurrentDocket can access it
@@ -572,20 +589,37 @@ class FastMCP(Generic[LifespanResultT]):
                 pass
 
             # Check for task metadata and route appropriately
-            if task_meta and fastmcp.settings.enable_tasks:
+            if fastmcp.settings.enable_tasks:
                 async with fastmcp.server.context.Context(fastmcp=self):
                     try:
                         resource = await self._resource_manager.get_resource(uri)
-                        if (
-                            resource
-                            and isinstance(resource, FunctionResource)
-                            and resource.task
-                        ):
-                            # Convert TaskMetadata to dict for handler
-                            task_meta_dict = task_meta.model_dump(exclude_none=True)
-                            return await handle_resource_as_task(
-                                self, str(uri), resource, task_meta_dict
-                            )
+                        if resource and isinstance(resource, FunctionResource):
+                            task_mode = resource.task_config.mode
+
+                            # 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="forbidden" - must not have task metadata
+                            if task_mode == "forbidden" and task_meta:
+                                raise McpError(
+                                    ErrorData(
+                                        code=METHOD_NOT_FOUND,
+                                        message=f"Resource '{uri}' does not support 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)
+                                return await handle_resource_as_task(
+                                    self, str(uri), resource, task_meta_dict
+                                )
                     except NotFoundError:
                         pass
 
@@ -645,22 +679,38 @@ class FastMCP(Generic[LifespanResultT]):
                 pass
 
             # Check for task metadata and route appropriately
-            if task_meta and fastmcp.settings.enable_tasks:
+            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 isinstance(prompt, FunctionPrompt) and prompt.task:
-                        # Convert TaskMetadata to dict for handler
-                        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)
-                    else:
-                        logger.debug(
-                            f"[{self.name}] Prompt {name} does not support background execution, "
-                            "ignoring task metadata and executing synchronously"
-                        )
+                    if prompt and isinstance(prompt, FunctionPrompt):
+                        task_mode = prompt.task_config.mode
+
+                        # 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="forbidden" - must not have task metadata
+                        if task_mode == "forbidden" and task_meta:
+                            raise McpError(
+                                ErrorData(
+                                    code=METHOD_NOT_FOUND,
+                                    message=f"Prompt '{name}' does not support 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)
 
             # Synchronous execution
             result = await self._get_prompt_mcp(name, arguments)
@@ -1324,23 +1374,35 @@ class FastMCP(Generic[LifespanResultT]):
                     # No request context available - proceed without task metadata
                     pass
 
-                if task_meta and fastmcp.settings.enable_tasks:
-                    # Task metadata present - check if tool supports background execution
+                if fastmcp.settings.enable_tasks:
                     tool = self._tool_manager._tools.get(key)
-                    if tool and isinstance(tool, FunctionTool) and tool.task:
-                        # Route to background execution
-                        # Convert TaskMetadata to dict for handler
-                        task_meta_dict = task_meta.model_dump(exclude_none=True)
-                        return await handle_tool_as_task(
-                            self, key, arguments, task_meta_dict
-                        )
-                    else:
-                        # Graceful degradation per SEP-1686 spec
-                        logger.debug(
-                            f"[{self.name}] Tool {key} does not support background execution, "
-                            "ignoring task metadata and executing synchronously"
-                        )
-                        # Fall through to synchronous execution
+                    if tool and isinstance(tool, FunctionTool):
+                        task_mode = tool.task_config.mode
+
+                        # 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="forbidden" - must not have task metadata
+                        if task_mode == "forbidden" and task_meta:
+                            raise McpError(
+                                ErrorData(
+                                    code=METHOD_NOT_FOUND,
+                                    message=f"Tool '{key}' does not support 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)
+                            return await handle_tool_as_task(
+                                self, key, arguments, task_meta_dict
+                            )
 
                 # Synchronous execution (normal path)
                 result = await self._call_tool_middleware(key, arguments)
@@ -1652,7 +1714,7 @@ class FastMCP(Generic[LifespanResultT]):
         exclude_args: list[str] | None = None,
         meta: dict[str, Any] | None = None,
         enabled: bool | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionTool: ...
 
     @overload
@@ -1670,7 +1732,7 @@ class FastMCP(Generic[LifespanResultT]):
         exclude_args: list[str] | None = None,
         meta: dict[str, Any] | None = None,
         enabled: bool | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> Callable[[AnyFunction], FunctionTool]: ...
 
     def tool(
@@ -1687,7 +1749,7 @@ class FastMCP(Generic[LifespanResultT]):
         exclude_args: list[str] | None = None,
         meta: dict[str, Any] | None = None,
         enabled: bool | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> Callable[[AnyFunction], FunctionTool] | FunctionTool:
         """Decorator to register a tool.
 
@@ -1761,8 +1823,8 @@ class FastMCP(Generic[LifespanResultT]):
             fn = name_or_fn
             tool_name = name  # Use keyword name if provided, otherwise None
 
-            # Resolve task parameter to concrete boolean
-            supports_task: bool = (
+            # Resolve task parameter
+            supports_task: bool | TaskConfig = (
                 task if task is not None else self._support_tasks_by_default
             )
 
@@ -1875,7 +1937,7 @@ class FastMCP(Generic[LifespanResultT]):
         enabled: bool | None = None,
         annotations: Annotations | dict[str, Any] | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
         """Decorator to register a function as a resource.
 
@@ -1952,8 +2014,8 @@ class FastMCP(Generic[LifespanResultT]):
                     )
                 )
 
-            # Resolve task parameter to concrete boolean
-            supports_task: bool = (
+            # Resolve task parameter
+            supports_task: bool | TaskConfig = (
                 task if task is not None else self._support_tasks_by_default
             )
 
@@ -2041,7 +2103,7 @@ class FastMCP(Generic[LifespanResultT]):
         tags: set[str] | None = None,
         enabled: bool | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionPrompt: ...
 
     @overload
@@ -2056,7 +2118,7 @@ class FastMCP(Generic[LifespanResultT]):
         tags: set[str] | None = None,
         enabled: bool | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> Callable[[AnyFunction], FunctionPrompt]: ...
 
     def prompt(
@@ -2070,7 +2132,7 @@ class FastMCP(Generic[LifespanResultT]):
         tags: set[str] | None = None,
         enabled: bool | None = None,
         meta: dict[str, Any] | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> Callable[[AnyFunction], FunctionPrompt] | FunctionPrompt:
         """Decorator to register a prompt.
 
@@ -2161,8 +2223,8 @@ class FastMCP(Generic[LifespanResultT]):
             fn = name_or_fn
             prompt_name = name  # Use keyword name if provided, otherwise None
 
-            # Resolve task parameter to concrete boolean
-            supports_task: bool = (
+            # Resolve task parameter
+            supports_task: bool | TaskConfig = (
                 task if task is not None else self._support_tasks_by_default
             )
 

src/fastmcp/server/tasks/__init__.py

@@ -3,6 +3,7 @@
 This module implements protocol-level background task execution for MCP servers.
 """
 
+from fastmcp.server.tasks.config import TaskConfig, TaskMode
 from fastmcp.server.tasks.converters import (
     convert_prompt_result,
     convert_resource_result,
@@ -26,6 +27,8 @@ from fastmcp.server.tasks.protocol import (
 )
 
 __all__ = [
+    "TaskConfig",
+    "TaskMode",
     "build_task_key",
     "convert_prompt_result",
     "convert_resource_result",

src/fastmcp/server/tasks/config.py

@@ -0,0 +1,89 @@
+"""TaskConfig for MCP SEP-1686 background task execution modes.
+
+This module defines the configuration for how tools, resources, and prompts
+handle task-augmented execution as specified in SEP-1686.
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any, Literal
+
+# Task execution modes per SEP-1686 / MCP ToolExecution.taskSupport
+TaskMode = Literal["forbidden", "optional", "required"]
+
+
+@dataclass
+class TaskConfig:
+    """Configuration for MCP background task execution (SEP-1686).
+
+    Controls how a component handles task-augmented requests:
+
+    - "forbidden": Component does not support task execution. Clients must not
+      request task augmentation; server returns -32601 if they do.
+    - "optional": Component supports both synchronous and task execution.
+      Client may request task augmentation or call normally.
+    - "required": Component requires task execution. Clients must request task
+      augmentation; server returns -32601 if they don't.
+
+    Example:
+        ```python
+        from fastmcp import FastMCP
+        from fastmcp.server.tasks import TaskConfig
+
+        mcp = FastMCP("MyServer")
+
+        # Background execution required
+        @mcp.tool(task=TaskConfig(mode="required"))
+        async def long_running_task(): ...
+
+        # Supports both modes (default when task=True)
+        @mcp.tool(task=TaskConfig(mode="optional"))
+        async def flexible_task(): ...
+        ```
+    """
+
+    mode: TaskMode = "optional"
+
+    @classmethod
+    def from_bool(cls, value: bool) -> TaskConfig:
+        """Convert boolean task flag to TaskConfig.
+
+        Args:
+            value: True for "optional" mode, False for "forbidden" mode.
+
+        Returns:
+            TaskConfig with appropriate mode.
+        """
+        return cls(mode="optional" if value else "forbidden")
+
+    def validate_function(self, fn: Callable[..., Any], name: str) -> None:
+        """Validate that function is compatible with this task config.
+
+        Task execution requires async functions. Raises ValueError if mode
+        is "optional" or "required" but function is synchronous.
+
+        Args:
+            fn: The function to validate (handles callable classes and staticmethods).
+            name: Name for error messages.
+
+        Raises:
+            ValueError: If task execution is enabled but function is sync.
+        """
+        if self.mode == "forbidden":
+            return
+
+        # Unwrap callable classes and staticmethods
+        fn_to_check = fn
+        if not inspect.isroutine(fn) and callable(fn):
+            fn_to_check = fn.__call__
+        if isinstance(fn_to_check, staticmethod):
+            fn_to_check = fn_to_check.__func__
+
+        if not inspect.iscoroutinefunction(fn_to_check):
+            raise ValueError(
+                f"'{name}' uses a sync function but has task execution enabled. "
+                "Background tasks require async functions."
+            )

src/fastmcp/server/tasks/converters.py

@@ -12,8 +12,6 @@ from typing import TYPE_CHECKING, Any
 import mcp.types
 import pydantic_core
 
-from fastmcp.tools.tool import ToolResult, _convert_to_content
-
 if TYPE_CHECKING:
     from fastmcp.server.server import FastMCP
 
@@ -35,6 +33,10 @@ async def convert_tool_result(
     Returns:
         CallToolResult with properly formatted content and structured content
     """
+    # Import here to avoid circular import:
+    # tools/tool.py -> tasks/config.py -> tasks/__init__.py -> converters.py -> tools/tool.py
+    from fastmcp.tools.tool import ToolResult, _convert_to_content
+
     # Get the tool to access its configuration
     tool = await server.get_tool(tool_name)
 

src/fastmcp/tools/tool.py

@@ -30,6 +30,7 @@ from typing_extensions import TypeVar
 
 import fastmcp
 from fastmcp.server.dependencies import get_context, without_injected_parameters
+from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.json_schema import compress_schema
 from fastmcp.utilities.logging import get_logger
@@ -201,7 +202,7 @@ class Tool(FastMCPComponent):
         serializer: ToolResultSerializerType | None = None,
         meta: dict[str, Any] | None = None,
         enabled: bool | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionTool:
         """Create a Tool from a function."""
         return FunctionTool.from_function(
@@ -269,12 +270,10 @@ class Tool(FastMCPComponent):
 
 class FunctionTool(Tool):
     fn: Callable[..., Any]
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this tool supports background task execution (SEP-1686)"
-        ),
-    ] = False
+    task_config: Annotated[
+        TaskConfig,
+        Field(description="Background task execution configuration (SEP-1686)."),
+    ] = Field(default_factory=lambda: TaskConfig(mode="forbidden"))
 
     def to_mcp_tool(
         self,
@@ -291,11 +290,10 @@ class FunctionTool(Tool):
             include_fastmcp_meta=include_fastmcp_meta, **overrides
         )
 
-        # Add task execution mode if this tool supports background tasks
-        # Per SEP-1686: tools declare task support via execution.task
-        # task values: "never" (no task support), "optional" (supports both), "always" (requires task)
-        if self.task and "execution" not in overrides:
-            mcp_tool.execution = ToolExecution(task="optional")
+        # Add task execution mode per SEP-1686
+        # Only set execution if not overridden and mode is not "forbidden"
+        if self.task_config.mode != "forbidden" and "execution" not in overrides:
+            mcp_tool.execution = ToolExecution(taskSupport=self.task_config.mode)
 
         return mcp_tool
 
@@ -314,7 +312,7 @@ class FunctionTool(Tool):
         serializer: ToolResultSerializerType | None = None,
         meta: dict[str, Any] | None = None,
         enabled: bool | None = None,
-        task: bool | None = None,
+        task: bool | TaskConfig | None = None,
     ) -> FunctionTool:
         """Create a Tool from a function."""
         if exclude_args and fastmcp.settings.deprecation_warnings:
@@ -328,25 +326,21 @@ class FunctionTool(Tool):
                 stacklevel=2,
             )
 
-        # Validate that task=True requires async functions
-        # Handle callable classes and staticmethods before checking
-        fn_to_check = fn
-        if not inspect.isroutine(fn) and callable(fn):
-            fn_to_check = fn.__call__
-        if isinstance(fn_to_check, staticmethod):
-            fn_to_check = fn_to_check.__func__
-        if task and not inspect.iscoroutinefunction(fn_to_check):
-            fn_name = name or getattr(fn, "__name__", repr(fn))
-            raise ValueError(
-                f"Tool '{fn_name}' uses a sync function but has task=True. "
-                "Background tasks require async functions. Set task=False to disable."
-            )
-
         parsed_fn = ParsedFunction.from_function(fn, exclude_args=exclude_args)
+        func_name = name or parsed_fn.name
 
-        if name is None and parsed_fn.name == "<lambda>":
+        if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
+        # Normalize task to TaskConfig and validate
+        if task is None:
+            task_config = TaskConfig(mode="forbidden")
+        elif isinstance(task, bool):
+            task_config = TaskConfig.from_bool(task)
+        else:
+            task_config = task
+        task_config.validate_function(fn, func_name)
+
         if isinstance(output_schema, NotSetT):
             final_output_schema = parsed_fn.output_schema
         else:
@@ -375,7 +369,7 @@ class FunctionTool(Tool):
             serializer=serializer,
             meta=meta,
             enabled=enabled if enabled is not None else True,
-            task=task if task is not None else False,
+            task_config=task_config,
         )
 
     async def run(self, arguments: dict[str, Any]) -> ToolResult:

tests/client/tasks/test_task_result_caching.py

@@ -132,25 +132,84 @@ async def test_result_and_await_share_cache():
         assert id(result_via_method) == id(result_via_await)
 
 
-async def test_immediate_task_caches_result():
-    """Immediate tasks (graceful degradation) also cache results."""
-    call_count = 0
+async def test_forbidden_mode_tool_caches_error_result():
+    """Tools with task=False (mode=forbidden) cache error results."""
     mcp = FastMCP("test")
 
-    # Tool with task=False - will execute immediately
     @mcp.tool(task=False)
     async def non_task_tool() -> int:
+        return 1
+
+    async with Client(mcp) as client:
+        # Request as task, but mode="forbidden" will reject with error
+        task = await client.call_tool("non_task_tool", task=True)
+
+        # Should be immediate (error returned immediately)
+        assert task.returned_immediately
+
+        result1 = await task.result()
+        result2 = await task.result()
+        result3 = await task.result()
+
+        # All should return cached error
+        assert result1.is_error
+        assert "does not support task-augmented execution" in str(result1)
+
+        # Verify they're the same object (cached)
+        assert result1 is result2 is result3
+
+
+async def test_forbidden_mode_prompt_raises_error():
+    """Prompts with task=False (mode=forbidden) raise error."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+
+    mcp = FastMCP("test")
+
+    @mcp.prompt(task=False)
+    async def non_task_prompt() -> str:
+        return "Immediate"
+
+    async with Client(mcp) as client:
+        # Prompts with mode="forbidden" raise McpError when called with task=True
+        with pytest.raises(McpError):
+            await client.get_prompt("non_task_prompt", task=True)
+
+
+async def test_forbidden_mode_resource_raises_error():
+    """Resources with task=False (mode=forbidden) raise error."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+
+    mcp = FastMCP("test")
+
+    @mcp.resource("file://immediate.txt", task=False)
+    async def non_task_resource() -> str:
+        return "Immediate"
+
+    async with Client(mcp) as client:
+        # Resources with mode="forbidden" raise McpError when called with task=True
+        with pytest.raises(McpError):
+            await client.read_resource("file://immediate.txt", task=True)
+
+
+async def test_immediate_task_caches_result():
+    """Immediate tasks (optional mode called without background) cache results."""
+    call_count = 0
+    mcp = FastMCP("test", tasks=True)
+
+    # Tool with task=True (optional mode) - but without docket will execute immediately
+    @mcp.tool(task=True)
+    async def task_tool() -> int:
         nonlocal call_count
         call_count += 1
         return call_count
 
     async with Client(mcp) as client:
-        # Request as task, but server will execute immediately
-        task = await client.call_tool("non_task_tool", task=True)
-
-        # Should be immediate (graceful degradation)
-        assert task.returned_immediately
+        # Call with task=True
+        task = await client.call_tool("task_tool", task=True)
 
+        # Get result multiple times
         result1 = await task.result()
         result2 = await task.result()
         result3 = await task.result()
@@ -164,56 +223,6 @@ async def test_immediate_task_caches_result():
         assert result1 is result2 is result3
 
 
-async def test_immediate_prompt_task_caches_result():
-    """Immediate prompt tasks cache results."""
-    call_count = 0
-    mcp = FastMCP("test")
-
-    @mcp.prompt(task=False)
-    async def non_task_prompt() -> str:
-        nonlocal call_count
-        call_count += 1
-        return f"Immediate: {call_count}"
-
-    async with Client(mcp) as client:
-        task = await client.get_prompt("non_task_prompt", task=True)
-
-        # Should be immediate
-        assert task.returned_immediately
-
-        result1 = await task.result()
-        result2 = await task.result()
-
-        # Verify caching
-        assert result1 is result2
-        assert result1.messages[0].content.text == "Immediate: 1"
-
-
-async def test_immediate_resource_task_caches_result():
-    """Immediate resource tasks cache results."""
-    call_count = 0
-    mcp = FastMCP("test")
-
-    @mcp.resource("file://immediate.txt", task=False)
-    async def non_task_resource() -> str:
-        nonlocal call_count
-        call_count += 1
-        return f"Immediate: {call_count}"
-
-    async with Client(mcp) as client:
-        task = await client.read_resource("file://immediate.txt", task=True)
-
-        # Should be immediate
-        assert task.returned_immediately
-
-        result1 = await task.result()
-        result2 = await task.result()
-
-        # Verify caching
-        assert result1 is result2
-        assert result1[0].text == "Immediate: 1"
-
-
 async def test_cache_persists_across_mixed_access_patterns():
     """Cache works correctly when mixing result() and await."""
     mcp = FastMCP("test")

tests/server/tasks/test_server_tasks_parameter.py

@@ -49,7 +49,10 @@ async def test_server_tasks_true_defaults_all_components():
 
 
 async def test_server_tasks_false_defaults_all_components():
-    """Server with tasks=False makes all components default to NOT supporting tasks."""
+    """Server with tasks=False makes all components default to mode=forbidden."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+
     mcp = FastMCP("test", tasks=False)
 
     @mcp.tool()
@@ -65,17 +68,20 @@ async def test_server_tasks_false_defaults_all_components():
         return "resource result"
 
     async with Client(mcp) as client:
-        # Tool should execute immediately (graceful degradation)
+        # Tool with mode="forbidden" returns error when called with task=True
         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)
 
-        # Prompt should execute immediately (graceful degradation)
-        prompt_task = await client.get_prompt("my_prompt", task=True)
-        assert prompt_task.returned_immediately
+        # Prompt with mode="forbidden" raises McpError when called with task=True
+        with pytest.raises(McpError):
+            await client.get_prompt("my_prompt", task=True)
 
-        # Resource should execute immediately (graceful degradation)
-        resource_task = await client.read_resource("test://resource", task=True)
-        assert resource_task.returned_immediately
+        # Resource with mode="forbidden" raises McpError when called with task=True
+        with pytest.raises(McpError):
+            await client.read_resource("test://resource", task=True)
 
 
 async def test_server_tasks_none_uses_settings():
@@ -102,9 +108,14 @@ async def test_server_tasks_none_uses_settings():
             return "tool result"
 
         async with Client(mcp2) as client:
-            # Tool should execute immediately (from settings)
+            # 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 def test_component_explicit_false_overrides_server_true():
@@ -126,9 +137,12 @@ async def test_component_explicit_false_overrides_server_true():
         assert "no_task_tool" not in docket.tasks  # task=False means not registered
         assert "default_tool" in docket.tasks  # Inherits tasks=True
 
-        # Explicit False should execute immediately despite server default
+        # Explicit False (mode="forbidden") returns error when called with task=True
         no_task = await client.call_tool("no_task_tool", task=True)
         assert no_task.returned_immediately
+        result = await no_task.result()
+        assert result.is_error
+        assert "does not support task-augmented execution" in str(result)
 
         # Default should support background execution
         default_task = await client.call_tool("default_tool", task=True)
@@ -158,13 +172,18 @@ async def test_component_explicit_true_overrides_server_false():
         task = await client.call_tool("task_tool", task=True)
         assert not task.returned_immediately
 
-        # Default should execute immediately
+        # Default (mode="forbidden") returns error when called with task=True
         default = await client.call_tool("default_tool", task=True)
         assert default.returned_immediately
+        result = await default.result()
+        assert result.is_error
 
 
 async def test_mixed_explicit_and_inherited():
     """Mix of explicit True/False/None on different components."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+
     mcp = FastMCP("test", tasks=True)  # Server default is True
 
     @mcp.tool()
@@ -216,17 +235,19 @@ async def test_mixed_explicit_and_inherited():
         explicit_true = await client.call_tool("explicit_true_tool", task=True)
         assert not explicit_true.returned_immediately
 
+        # Explicit False (mode="forbidden") returns error
         explicit_false = await client.call_tool("explicit_false_tool", task=True)
         assert explicit_false.returned_immediately
+        result = await explicit_false.result()
+        assert result.is_error
 
         # Prompts
         inherited_prompt_task = await client.get_prompt("inherited_prompt", task=True)
         assert not inherited_prompt_task.returned_immediately
 
-        explicit_false_prompt_task = await client.get_prompt(
-            "explicit_false_prompt", task=True
-        )
-        assert explicit_false_prompt_task.returned_immediately
+        # Explicit False prompt (mode="forbidden") raises McpError
+        with pytest.raises(McpError):
+            await client.get_prompt("explicit_false_prompt", task=True)
 
         # Resources
         inherited_resource_task = await client.read_resource(
@@ -234,10 +255,9 @@ async def test_mixed_explicit_and_inherited():
         )
         assert not inherited_resource_task.returned_immediately
 
-        explicit_false_resource_task = await client.read_resource(
-            "test://explicit_false", task=True
-        )
-        assert explicit_false_resource_task.returned_immediately
+        # Explicit False resource (mode="forbidden") raises McpError
+        with pytest.raises(McpError):
+            await client.read_resource("test://explicit_false", task=True)
 
 
 async def test_server_tasks_parameter_sets_component_defaults():
@@ -264,9 +284,11 @@ async def test_server_tasks_parameter_sets_component_defaults():
             return "tool result"
 
         async with Client(mcp2) as client:
-            # Tool inherits tasks=False from server (graceful degradation)
+            # 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():
@@ -285,6 +307,9 @@ async def test_resource_template_inherits_server_tasks_default():
 
 async def test_multiple_components_same_name_different_tasks():
     """Different component types with same name can have different task settings."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+
     mcp = FastMCP("test", tasks=False)
 
     @mcp.tool(task=True)
@@ -300,6 +325,6 @@ async def test_multiple_components_same_name_different_tasks():
         tool_task = await client.call_tool("shared_name", task=True)
         assert not tool_task.returned_immediately
 
-        # Prompt inheriting False should execute immediately
-        prompt_task = await client.get_prompt("shared_name_prompt", task=True)
-        assert prompt_task.returned_immediately
+        # Prompt inheriting False (mode="forbidden") raises McpError
+        with pytest.raises(McpError):
+            await client.get_prompt("shared_name_prompt", task=True)

tests/server/tasks/test_sync_function_task_disabled.py

@@ -17,7 +17,9 @@ async def test_sync_tool_with_explicit_task_true_raises():
     """Sync tool with task=True raises ValueError."""
     mcp = FastMCP("test")
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.tool(task=True)
         def sync_tool(x: int) -> int:
@@ -29,7 +31,9 @@ async def test_sync_tool_with_inherited_task_true_raises():
     """Sync tool inheriting task=True from server raises ValueError."""
     mcp = FastMCP("test", tasks=True)
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.tool()  # Inherits task=True from server
         def sync_tool(x: int) -> int:
@@ -41,7 +45,9 @@ async def test_sync_prompt_with_explicit_task_true_raises():
     """Sync prompt with task=True raises ValueError."""
     mcp = FastMCP("test")
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.prompt(task=True)
         def sync_prompt() -> str:
@@ -53,7 +59,9 @@ async def test_sync_prompt_with_inherited_task_true_raises():
     """Sync prompt inheriting task=True from server raises ValueError."""
     mcp = FastMCP("test", tasks=True)
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.prompt()  # Inherits task=True from server
         def sync_prompt() -> str:
@@ -65,7 +73,9 @@ async def test_sync_resource_with_explicit_task_true_raises():
     """Sync resource with task=True raises ValueError."""
     mcp = FastMCP("test")
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.resource("test://sync", task=True)
         def sync_resource() -> str:
@@ -77,7 +87,9 @@ async def test_sync_resource_with_inherited_task_true_raises():
     """Sync resource inheriting task=True from server raises ValueError."""
     mcp = FastMCP("test", tasks=True)
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
 
         @mcp.resource("test://sync")  # Inherits task=True from server
         def sync_resource() -> str:
@@ -94,10 +106,10 @@ async def test_async_tool_with_task_true_remains_enabled():
         """An async tool."""
         return x * 2
 
-    # Tool should have task=True and be a FunctionTool
+    # Tool should have task mode="optional" and be a FunctionTool
     tool = await mcp.get_tool("async_tool")
     assert isinstance(tool, FunctionTool)
-    assert tool.task is True
+    assert tool.task_config.mode == "optional"
 
 
 async def test_async_prompt_with_task_true_remains_enabled():
@@ -109,10 +121,10 @@ async def test_async_prompt_with_task_true_remains_enabled():
         """An async prompt."""
         return "Hello"
 
-    # Prompt should have task=True and be a FunctionPrompt
+    # Prompt should have task mode="optional" and be a FunctionPrompt
     prompt = await mcp.get_prompt("async_prompt")
     assert isinstance(prompt, FunctionPrompt)
-    assert prompt.task is True
+    assert prompt.task_config.mode == "optional"
 
 
 async def test_async_resource_with_task_true_remains_enabled():
@@ -124,10 +136,10 @@ async def test_async_resource_with_task_true_remains_enabled():
         """An async resource."""
         return "data"
 
-    # Resource should have task=True and be a FunctionResource
+    # Resource should have task mode="optional" and be a FunctionResource
     resource = await mcp._resource_manager.get_resource("test://async")
     assert isinstance(resource, FunctionResource)
-    assert resource.task is True
+    assert resource.task_config.mode == "optional"
 
 
 async def test_sync_tool_with_task_false_works():
@@ -141,7 +153,7 @@ async def test_sync_tool_with_task_false_works():
 
     tool = await mcp.get_tool("sync_tool")
     assert isinstance(tool, FunctionTool)
-    assert tool.task is False
+    assert tool.task_config.mode == "forbidden"
 
 
 async def test_sync_prompt_with_task_false_works():
@@ -155,7 +167,7 @@ async def test_sync_prompt_with_task_false_works():
 
     prompt = await mcp.get_prompt("sync_prompt")
     assert isinstance(prompt, FunctionPrompt)
-    assert prompt.task is False
+    assert prompt.task_config.mode == "forbidden"
 
 
 async def test_sync_resource_with_task_false_works():
@@ -169,7 +181,7 @@ async def test_sync_resource_with_task_false_works():
 
     resource = await mcp._resource_manager.get_resource("test://sync")
     assert isinstance(resource, FunctionResource)
-    assert resource.task is False
+    assert resource.task_config.mode == "forbidden"
 
 
 # =============================================================================
@@ -187,7 +199,7 @@ async def test_async_callable_class_tool_with_task_true_works():
 
     # Callable classes use Tool.from_function() directly
     tool = Tool.from_function(AsyncCallableTool(), task=True)
-    assert tool.task is True
+    assert tool.task_config.mode == "optional"
 
 
 async def test_async_callable_class_prompt_with_task_true_works():
@@ -200,7 +212,7 @@ async def test_async_callable_class_prompt_with_task_true_works():
 
     # Callable classes use Prompt.from_function() directly
     prompt = Prompt.from_function(AsyncCallablePrompt(), task=True)
-    assert prompt.task is True
+    assert prompt.task_config.mode == "optional"
 
 
 async def test_sync_callable_class_tool_with_task_true_raises():
@@ -211,5 +223,7 @@ async def test_sync_callable_class_tool_with_task_true_raises():
         def __call__(self, x: int) -> int:
             return x * 2
 
-    with pytest.raises(ValueError, match="uses a sync function but has task=True"):
+    with pytest.raises(
+        ValueError, match="uses a sync function but has task execution enabled"
+    ):
         Tool.from_function(SyncCallableTool(), task=True)

tests/server/tasks/test_task_config_modes.py

@@ -0,0 +1,349 @@
+"""Tests for TaskConfig mode enforcement (SEP-1686).
+
+Tests that the server correctly enforces task execution modes:
+- "forbidden": No task support, error if client requests task
+- "optional": Supports both sync and task execution
+- "required": Requires task execution, error if client doesn't request task
+"""
+
+import pytest
+
+from fastmcp import FastMCP, TaskConfig
+from fastmcp.client import Client
+from fastmcp.exceptions import ToolError
+
+
+class TestTaskConfigNormalization:
+    """Test that boolean task values normalize correctly to TaskConfig."""
+
+    async def test_task_true_normalizes_to_optional(self):
+        """task=True should normalize to TaskConfig(mode='optional')."""
+        mcp = FastMCP("test", tasks=False)  # Disable default task support
+
+        @mcp.tool(task=True)
+        async def my_tool() -> str:
+            return "ok"
+
+        tool = await mcp._tool_manager.get_tool("my_tool")
+        assert tool is not None
+        assert tool.task_config.mode == "optional"  # type: ignore[attr-defined]
+
+    async def test_task_false_normalizes_to_forbidden(self):
+        """task=False should normalize to TaskConfig(mode='forbidden')."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=False)
+        async def my_tool() -> str:
+            return "ok"
+
+        tool = await mcp._tool_manager.get_tool("my_tool")
+        assert tool is not None
+        assert tool.task_config.mode == "forbidden"  # type: ignore[attr-defined]
+
+    async def test_task_config_passed_directly(self):
+        """TaskConfig should be preserved when passed directly."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="required"))
+        async def my_tool() -> str:
+            return "ok"
+
+        tool = await mcp._tool_manager.get_tool("my_tool")
+        assert tool is not None
+        assert tool.task_config.mode == "required"  # type: ignore[attr-defined]
+
+    async def test_default_task_inherits_server_default(self):
+        """Default task value should inherit from server default."""
+        # Server with tasks disabled
+        mcp_no_tasks = FastMCP("test", tasks=False)
+
+        @mcp_no_tasks.tool()
+        def my_tool_sync() -> str:
+            return "ok"
+
+        tool = await mcp_no_tasks._tool_manager.get_tool("my_tool_sync")
+        assert tool is not None
+        assert tool.task_config.mode == "forbidden"  # type: ignore[attr-defined]
+
+        # Server with tasks enabled
+        mcp_tasks = FastMCP("test", tasks=True)
+
+        @mcp_tasks.tool()
+        async def my_tool_async() -> str:
+            return "ok"
+
+        tool2 = await mcp_tasks._tool_manager.get_tool("my_tool_async")
+        assert tool2 is not None
+        assert tool2.task_config.mode == "optional"  # type: ignore[attr-defined]
+
+
+class TestToolModeEnforcement:
+    """Test mode enforcement for tools."""
+
+    @pytest.fixture
+    def server(self):
+        """Create server with tools in different modes."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="required"))
+        async def required_tool() -> str:
+            """Tool that requires task execution."""
+            return "required result"
+
+        @mcp.tool(task=TaskConfig(mode="forbidden"))
+        async def forbidden_tool() -> str:
+            """Tool that forbids task execution."""
+            return "forbidden result"
+
+        @mcp.tool(task=TaskConfig(mode="optional"))
+        async def optional_tool() -> str:
+            """Tool that supports both modes."""
+            return "optional result"
+
+        return mcp
+
+    async def test_required_mode_without_task_returns_error(self, server):
+        """Required mode returns error when called without task metadata."""
+        async with Client(server) as client:
+            # The server returns isError=True, which the client converts to ToolError
+            with pytest.raises(ToolError) as exc_info:
+                await client.call_tool("required_tool", {})
+
+            assert "requires task-augmented execution" in str(exc_info.value)
+
+    async def test_required_mode_with_task_succeeds(self, server):
+        """Required mode succeeds when called with task metadata."""
+        async with Client(server) as client:
+            task = await client.call_tool("required_tool", {}, task=True)
+            assert task is not None
+            result = await task.result()
+            assert result.data == "required result"
+
+    async def test_forbidden_mode_with_task_returns_error(self, server):
+        """Forbidden mode returns error when called with task metadata."""
+        async with Client(server) as client:
+            # Call with task=True should fail
+            task = await client.call_tool("forbidden_tool", {}, task=True)
+            assert task is not None
+            # The task should have returned immediately with an error
+            assert task.returned_immediately
+            result = await task.result()
+            # Check for error in the result
+            assert result.is_error
+
+    async def test_forbidden_mode_without_task_succeeds(self, server):
+        """Forbidden mode succeeds when called without task metadata."""
+        async with Client(server) as client:
+            result = await client.call_tool("forbidden_tool", {})
+            assert "forbidden result" in str(result)
+
+    async def test_optional_mode_without_task_succeeds(self, server):
+        """Optional mode succeeds when called without task metadata."""
+        async with Client(server) as client:
+            result = await client.call_tool("optional_tool", {})
+            assert "optional result" in str(result)
+
+    async def test_optional_mode_with_task_succeeds(self, server):
+        """Optional mode succeeds when called with task metadata."""
+        async with Client(server) as client:
+            task = await client.call_tool("optional_tool", {}, task=True)
+            assert task is not None
+            result = await task.result()
+            assert result.data == "optional result"
+
+
+class TestResourceModeEnforcement:
+    """Test mode enforcement for resources."""
+
+    @pytest.fixture
+    def server(self):
+        """Create server with resources in different modes."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.resource("resource://required", task=TaskConfig(mode="required"))
+        async def required_resource() -> str:
+            """Resource that requires task execution."""
+            return "required content"
+
+        @mcp.resource("resource://forbidden", task=TaskConfig(mode="forbidden"))
+        async def forbidden_resource() -> str:
+            """Resource that forbids task execution."""
+            return "forbidden content"
+
+        @mcp.resource("resource://optional", task=TaskConfig(mode="optional"))
+        async def optional_resource() -> str:
+            """Resource that supports both modes."""
+            return "optional content"
+
+        return mcp
+
+    async def test_required_resource_without_task_returns_error(self, server):
+        """Required mode returns error when read without task metadata."""
+        from mcp.shared.exceptions import McpError
+        from mcp.types import METHOD_NOT_FOUND
+
+        async with Client(server) as client:
+            with pytest.raises(McpError) as exc_info:
+                await client.read_resource("resource://required")
+
+            assert exc_info.value.error.code == METHOD_NOT_FOUND
+            assert "requires task-augmented execution" in exc_info.value.error.message
+
+    async def test_required_resource_with_task_succeeds(self, server):
+        """Required mode succeeds when read with task metadata."""
+        async with Client(server) as client:
+            task = await client.read_resource("resource://required", task=True)
+            assert task is not None
+            result = await task.result()
+            # Result is a list of resource contents
+            assert "required content" in str(result)
+
+    async def test_forbidden_resource_without_task_succeeds(self, server):
+        """Forbidden mode succeeds when read without task metadata."""
+        async with Client(server) as client:
+            result = await client.read_resource("resource://forbidden")
+            assert "forbidden content" in str(result)
+
+
+class TestPromptModeEnforcement:
+    """Test mode enforcement for prompts."""
+
+    @pytest.fixture
+    def server(self):
+        """Create server with prompts in different modes."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.prompt(task=TaskConfig(mode="required"))
+        async def required_prompt() -> str:
+            """Prompt that requires task execution."""
+            return "required message"
+
+        @mcp.prompt(task=TaskConfig(mode="forbidden"))
+        async def forbidden_prompt() -> str:
+            """Prompt that forbids task execution."""
+            return "forbidden message"
+
+        @mcp.prompt(task=TaskConfig(mode="optional"))
+        async def optional_prompt() -> str:
+            """Prompt that supports both modes."""
+            return "optional message"
+
+        return mcp
+
+    async def test_required_prompt_without_task_returns_error(self, server):
+        """Required mode returns error when called without task metadata."""
+        from mcp.shared.exceptions import McpError
+        from mcp.types import METHOD_NOT_FOUND
+
+        async with Client(server) as client:
+            with pytest.raises(McpError) as exc_info:
+                await client.get_prompt("required_prompt")
+
+            assert exc_info.value.error.code == METHOD_NOT_FOUND
+            assert "requires task-augmented execution" in exc_info.value.error.message
+
+    async def test_required_prompt_with_task_succeeds(self, server):
+        """Required mode succeeds when called with task metadata."""
+        async with Client(server) as client:
+            task = await client.get_prompt("required_prompt", task=True)
+            assert task is not None
+            result = await task.result()
+            # Result contains the prompt messages
+            assert "required message" in str(result)
+
+    async def test_forbidden_prompt_without_task_succeeds(self, server):
+        """Forbidden mode succeeds when called without task metadata."""
+        async with Client(server) as client:
+            result = await client.get_prompt("forbidden_prompt")
+            assert "forbidden message" in str(result.messages[0].content)  # type: ignore[attr-defined]
+
+
+class TestToolExecutionMetadata:
+    """Test that ToolExecution.taskSupport is set correctly in tool metadata."""
+
+    async def test_optional_tool_exposes_task_support(self):
+        """Tools with task enabled should expose taskSupport in metadata."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="optional"))
+        async def my_tool() -> str:
+            return "ok"
+
+        async with Client(mcp) as client:
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "my_tool")
+            assert tool.execution is not None
+            assert tool.execution.taskSupport == "optional"  # type: ignore[attr-defined]
+
+    async def test_required_tool_exposes_task_support(self):
+        """Tools with mode=required should expose taskSupport='required'."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="required"))
+        async def my_tool() -> str:
+            return "ok"
+
+        async with Client(mcp) as client:
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "my_tool")
+            assert tool.execution is not None
+            assert tool.execution.taskSupport == "required"  # type: ignore[attr-defined]
+
+    async def test_forbidden_tool_has_no_execution(self):
+        """Tools with mode=forbidden should not expose execution metadata."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="forbidden"))
+        async def my_tool() -> str:
+            return "ok"
+
+        async with Client(mcp) as client:
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "my_tool")
+            assert tool.execution is None
+
+
+class TestSyncFunctionValidation:
+    """Test that sync functions cannot have task execution enabled."""
+
+    def test_sync_function_with_task_true_raises(self):
+        """Sync functions should raise ValueError when task=True."""
+        mcp = FastMCP("test", tasks=False)
+
+        with pytest.raises(ValueError, match="sync function"):
+
+            @mcp.tool(task=True)
+            def sync_tool() -> str:
+                return "ok"
+
+    def test_sync_function_with_required_mode_raises(self):
+        """Sync functions should raise ValueError with mode='required'."""
+        mcp = FastMCP("test", tasks=False)
+
+        with pytest.raises(ValueError, match="sync function"):
+
+            @mcp.tool(task=TaskConfig(mode="required"))
+            def sync_tool() -> str:
+                return "ok"
+
+    def test_sync_function_with_optional_mode_raises(self):
+        """Sync functions should raise ValueError with mode='optional'."""
+        mcp = FastMCP("test", tasks=False)
+
+        with pytest.raises(ValueError, match="sync function"):
+
+            @mcp.tool(task=TaskConfig(mode="optional"))
+            def sync_tool() -> str:
+                return "ok"
+
+    async def test_sync_function_with_forbidden_mode_ok(self):
+        """Sync functions should work fine with mode='forbidden'."""
+        mcp = FastMCP("test", tasks=False)
+
+        @mcp.tool(task=TaskConfig(mode="forbidden"))
+        def sync_tool() -> str:
+            return "ok"
+
+        tool = await mcp._tool_manager.get_tool("sync_tool")
+        assert tool is not None
+        assert tool.task_config.mode == "forbidden"  # type: ignore[attr-defined]

tests/server/tasks/test_task_prompts.py

@@ -69,20 +69,24 @@ async def test_prompt_task_executes_in_background(prompt_server):
         assert "comprehensive" in result.messages[0].content.text.lower()
 
 
-async def test_graceful_degradation_prompt_without_task_flag(prompt_server):
-    """Prompts with task=False execute synchronously even with task metadata."""
+async def test_forbidden_mode_prompt_rejects_task_calls(prompt_server):
+    """Prompts with task=False (mode=forbidden) reject task-augmented calls."""
+    from mcp.shared.exceptions import McpError
+    from mcp.types import METHOD_NOT_FOUND
 
     @prompt_server.prompt(task=False)  # Explicitly disable task support
     async def sync_only_prompt(topic: str) -> str:
         return f"Sync prompt: {topic}"
 
     async with Client(prompt_server) as client:
-        # Try to call with task metadata - should execute synchronously
-        task = await client.get_prompt("sync_only_prompt", {"topic": "test"}, task=True)
+        # Calling with task=True when task=False should raise McpError
+        import pytest
 
-        # Should have executed immediately (graceful degradation)
-        assert task.returned_immediately
+        with pytest.raises(McpError) as exc_info:
+            await client.get_prompt("sync_only_prompt", {"topic": "test"}, task=True)
 
-        # Can get result without waiting
-        result = await task.result()
-        assert "Sync prompt: test" in result.messages[0].content.text
+        # New behavior: mode="forbidden" returns METHOD_NOT_FOUND error
+        assert exc_info.value.error.code == METHOD_NOT_FOUND
+        assert (
+            "does not support task-augmented execution" in exc_info.value.error.message
+        )

tests/server/tasks/test_task_resources.py

@@ -84,22 +84,25 @@ async def test_resource_template_with_task(resource_server):
         assert '"userId": "123"' in result[0].text
 
 
-async def test_graceful_degradation_resource_without_task_flag(resource_server):
-    """Resources with task=False execute synchronously even with task metadata."""
+async def test_forbidden_mode_resource_rejects_task_calls(resource_server):
+    """Resources with task=False (mode=forbidden) reject task-augmented calls."""
+    import pytest
+    from mcp.shared.exceptions import McpError
+    from mcp.types import METHOD_NOT_FOUND
 
     @resource_server.resource(
-        "file://sync.txt", task=False
+        "file://sync.txt/", task=False
     )  # Explicitly disable task support
     async def sync_only_resource() -> str:
         return "Sync content"
 
     async with Client(resource_server) as client:
-        # Try to call with task metadata - should execute synchronously
-        task = await client.read_resource("file://sync.txt", task=True)
+        # Calling with task=True when task=False should raise McpError
+        with pytest.raises(McpError) as exc_info:
+            await client.read_resource("file://sync.txt", task=True)
 
-        # Should have executed immediately (graceful degradation)
-        assert task.returned_immediately
-
-        # Can get result without waiting
-        result = await task.result()
-        assert "Sync content" in result[0].text
+        # New behavior: mode="forbidden" returns METHOD_NOT_FOUND error
+        assert exc_info.value.error.code == METHOD_NOT_FOUND
+        assert (
+            "does not support task-augmented execution" in exc_info.value.error.message
+        )

tests/server/tasks/test_task_tools.py

@@ -88,16 +88,15 @@ async def test_tool_task_executes_in_background(tool_server):
         assert result.data == "completed"
 
 
-async def test_graceful_degradation_tool_without_task_flag(tool_server):
-    """Tools with task=False execute synchronously even with task metadata."""
+async def test_forbidden_mode_tool_rejects_task_calls(tool_server):
+    """Tools with task=False (mode=forbidden) reject task-augmented calls."""
     async with Client(tool_server) as client:
-        # Try to call with task metadata - server should execute synchronously
+        # Calling with task=True when task=False should return error
         task = await client.call_tool("sync_only_tool", {"message": "test"}, task=True)
         assert task
         assert task.returned_immediately
 
         result = await task.result()
-        assert "Sync: test" in str(result)
-
-        status = await task.status()
-        assert status.status == "completed"
+        # New behavior: mode="forbidden" returns an error
+        assert result.is_error
+        assert "does not support task-augmented execution" in str(result)

tests/server/test_tool_annotations.py

@@ -222,7 +222,7 @@ async def test_tool_functionality_with_annotations():
 
 
 async def test_task_execution_auto_populated_for_task_enabled_tool():
-    """Test that execution.task is automatically set when tool has task=True."""
+    """Test that execution.taskSupport is automatically set when tool has task=True."""
     mcp = FastMCP("Test Server")
 
     @mcp.tool(task=True)
@@ -235,7 +235,7 @@ async def test_task_execution_auto_populated_for_task_enabled_tool():
         assert len(tools_result) == 1
         assert tools_result[0].name == "background_tool"
         assert tools_result[0].execution is not None
-        assert tools_result[0].execution.task == "optional"
+        assert tools_result[0].execution.taskSupport == "optional"  # type: ignore[attr-defined]
 
 
 async def test_task_execution_omitted_for_task_disabled_tool():

tests/tools/test_tool.py

@@ -52,7 +52,7 @@ class TestToolFromFunction:
                     "type": "object",
                     "x-fastmcp-wrap-result": True,
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
                 "fn": HasName("add"),
             }
         )
@@ -100,7 +100,7 @@ class TestToolFromFunction:
                     "type": "object",
                     "x-fastmcp-wrap-result": True,
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
                 "fn": HasName("fetch_data"),
             }
         )
@@ -135,7 +135,7 @@ class TestToolFromFunction:
                     "type": "object",
                     "x-fastmcp-wrap-result": True,
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
             }
         )
 
@@ -169,7 +169,7 @@ class TestToolFromFunction:
                     "type": "object",
                     "x-fastmcp-wrap-result": True,
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
             }
         )
 
@@ -211,7 +211,7 @@ class TestToolFromFunction:
                     "type": "object",
                 },
                 "output_schema": {"additionalProperties": True, "type": "object"},
-                "task": False,
+                "task": {"mode": "forbidden"},
                 "fn": HasName("create_user"),
             }
         )
@@ -274,7 +274,7 @@ class TestToolFromFunction:
                     "required": ["x"],
                     "type": "object",
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
             }
         )
 
@@ -307,7 +307,7 @@ class TestToolFromFunction:
                     "required": ["_a", "_b"],
                     "type": "object",
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
             }
         )
 
@@ -362,7 +362,7 @@ class TestToolFromFunction:
                     "type": "object",
                     "x-fastmcp-wrap-result": True,
                 },
-                "task": False,
+                "task": {"mode": "forbidden"},
             }
         )