refactor: move task attribute to function-based variants only [SEP-1686]

src/fastmcp/prompts/prompt.py

@@ -66,12 +66,6 @@ class Prompt(FastMCPComponent):
     arguments: list[PromptArgument] | None = Field(
         default=None, description="Arguments that can be passed to the prompt"
     )
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this prompt supports background task execution (SEP-1686)"
-        ),
-    ] = False
 
     def enable(self) -> None:
         super().enable()
@@ -164,6 +158,12 @@ 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
 
     @classmethod
     def from_function(

src/fastmcp/resources/resource.py

@@ -47,12 +47,6 @@ class Resource(FastMCPComponent):
         Annotations | None,
         Field(description="Optional annotations about the resource's behavior"),
     ] = None
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this resource supports background task execution (SEP-1686)"
-        ),
-    ] = False
 
     def enable(self) -> None:
         super().enable()
@@ -176,6 +170,12 @@ class FunctionResource(Resource):
     """
 
     fn: Callable[..., Any]
+    task: Annotated[
+        bool,
+        Field(
+            description="Whether this resource supports background task execution (SEP-1686)"
+        ),
+    ] = False
 
     @classmethod
     def from_function(

src/fastmcp/resources/template.py

@@ -103,12 +103,6 @@ class ResourceTemplate(FastMCPComponent):
     annotations: Annotations | None = Field(
         default=None, description="Optional annotations about the resource's behavior"
     )
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this resource template supports background task execution (SEP-1686)"
-        ),
-    ] = False
 
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}(uri_template={self.uri_template!r}, name={self.name!r}, description={self.description!r}, tags={self.tags})"
@@ -178,22 +172,14 @@ async def read(self, arguments: dict[str, Any]) -> str | bytes:
         )
 
     async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
-        """Create a resource from the template with the given parameters."""
-
-        async def resource_read_fn() -> str | bytes:
-            # Call function and check if result is a coroutine
-            result = await self.read(arguments=params)
-            return result
+        """Create a resource from the template with the given parameters.
 
-        return Resource.from_function(
-            fn=resource_read_fn,
-            uri=uri,
-            name=self.name,
-            description=self.description,
-            mime_type=self.mime_type,
-            tags=self.tags,
-            enabled=self.enabled,
-            task=self.task,
+        The base implementation does not support background tasks.
+        Use FunctionResourceTemplate for task support.
+        """
+        raise NotImplementedError(
+            "Subclasses must implement create_resource(). "
+            "Use FunctionResourceTemplate for task support."
         )
 
     def to_mcp_template(
@@ -245,6 +231,31 @@ 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
+
+    async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
+        """Create a resource from the template with the given parameters."""
+
+        async def resource_read_fn() -> str | bytes:
+            # Call function and check if result is a coroutine
+            result = await self.read(arguments=params)
+            return result
+
+        return Resource.from_function(
+            fn=resource_read_fn,
+            uri=uri,
+            name=self.name,
+            description=self.description,
+            mime_type=self.mime_type,
+            tags=self.tags,
+            enabled=self.enabled,
+            task=self.task,
+        )
 
     async def read(self, arguments: dict[str, Any]) -> str | bytes:
         """Read the resource content."""

src/fastmcp/server/server.py

@@ -60,9 +60,9 @@
 from fastmcp.prompts import Prompt
 from fastmcp.prompts.prompt import FunctionPrompt
 from fastmcp.prompts.prompt_manager import PromptManager
-from fastmcp.resources.resource import Resource
+from fastmcp.resources.resource import FunctionResource, Resource
 from fastmcp.resources.resource_manager import ResourceManager
-from fastmcp.resources.template import ResourceTemplate
+from fastmcp.resources.template import FunctionResourceTemplate, ResourceTemplate
 from fastmcp.server.auth import AuthProvider
 from fastmcp.server.http import (
     StarletteWithLifespan,
@@ -400,48 +400,21 @@ async def _docket_lifespan(self) -> AsyncIterator[None]:
                 self._docket = docket
 
                 # Register local task-enabled tools/prompts/resources with Docket
+                # Only function-based variants support background tasks
                 for tool in self._tool_manager._tools.values():
-                    if not hasattr(tool, "fn"):
-                        continue
-                    supports_task = (
-                        tool.task
-                        if tool.task is not None
-                        else self._support_tasks_by_default
-                    )
-                    if supports_task:
+                    if isinstance(tool, FunctionTool) and tool.task:
                         docket.register(tool.fn)
 
                 for prompt in self._prompt_manager._prompts.values():
-                    if not hasattr(prompt, "fn"):
-                        continue
-                    supports_task = (
-                        prompt.task
-                        if prompt.task is not None
-                        else self._support_tasks_by_default
-                    )
-                    if supports_task:
+                    if isinstance(prompt, FunctionPrompt) and prompt.task:
                         docket.register(prompt.fn)
 
                 for resource in self._resource_manager._resources.values():
-                    if not hasattr(resource, "fn"):
-                        continue
-                    supports_task = (
-                        resource.task
-                        if resource.task is not None
-                        else self._support_tasks_by_default
-                    )
-                    if supports_task:
+                    if isinstance(resource, FunctionResource) and resource.task:
                         docket.register(resource.fn)
 
                 for template in self._resource_manager._templates.values():
-                    if not hasattr(template, "fn"):
-                        continue
-                    supports_task = (
-                        template.task
-                        if template.task is not None
-                        else self._support_tasks_by_default
-                    )
-                    if supports_task:
+                    if isinstance(template, FunctionResourceTemplate) and template.task:
                         docket.register(template.fn)
 
                 # Set Docket in ContextVar so CurrentDocket can access it
@@ -602,7 +575,11 @@ async def handler(req: mcp.types.ReadResourceRequest) -> mcp.types.ServerResult:
                 async with fastmcp.server.context.Context(fastmcp=self):
                     try:
                         resource = await self._resource_manager.get_resource(uri)
-                        if resource and resource.task:
+                        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(
@@ -671,7 +648,7 @@ async def handler(req: mcp.types.GetPromptRequest) -> mcp.types.ServerResult:
                 async with fastmcp.server.context.Context(fastmcp=self):
                     prompts = await self.get_prompts()
                     prompt = prompts.get(name)
-                    if prompt and prompt.task:
+                    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(
@@ -1349,7 +1326,7 @@ async def _call_tool_mcp(
                 if task_meta and fastmcp.settings.enable_tasks:
                     # Task metadata present - check if tool supports background execution
                     tool = self._tool_manager._tools.get(key)
-                    if tool and tool.task:
+                    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)

src/fastmcp/tools/tool.py

@@ -136,12 +136,6 @@ class Tool(FastMCPComponent):
         ToolResultSerializerType | None,
         Field(description="Optional custom serializer for tool results"),
     ] = None
-    task: Annotated[
-        bool,
-        Field(
-            description="Whether this tool supports background task execution (SEP-1686)"
-        ),
-    ] = False
 
     @model_validator(mode="after")
     def _validate_tool_name(self) -> Tool:
@@ -179,24 +173,15 @@ def to_mcp_tool(
         elif self.annotations and self.annotations.title:
             title = self.annotations.title
 
-        # Auto-populate task execution mode based on tool.task flag if not explicitly set
-        # Per SEP-1686: tools declare task support via execution.task
-        # task values: "never" (no task support), "optional" (supports both), "always" (requires task)
-        annotations = self.annotations
-        execution = None
-        if self.task:
-            # Tool supports background execution - use "optional" to allow both immediate and task execution
-            execution = ToolExecution(task="optional")
-
         return MCPTool(
             name=overrides.get("name", self.name),
             title=overrides.get("title", title),
             description=overrides.get("description", self.description),
             inputSchema=overrides.get("inputSchema", self.parameters),
             outputSchema=overrides.get("outputSchema", self.output_schema),
             icons=overrides.get("icons", self.icons),
-            annotations=overrides.get("annotations", annotations),
-            execution=overrides.get("execution", execution),
+            annotations=overrides.get("annotations", self.annotations),
+            execution=overrides.get("execution"),
             _meta=overrides.get(
                 "_meta", self.get_meta(include_fastmcp_meta=include_fastmcp_meta)
             ),
@@ -284,6 +269,35 @@ def from_tool(
 
 class FunctionTool(Tool):
     fn: Callable[..., Any]
+    task: Annotated[
+        bool,
+        Field(
+            description="Whether this tool supports background task execution (SEP-1686)"
+        ),
+    ] = False
+
+    def to_mcp_tool(
+        self,
+        *,
+        include_fastmcp_meta: bool | None = None,
+        **overrides: Any,
+    ) -> MCPTool:
+        """Convert the FastMCP tool to an MCP tool.
+
+        Extends the base implementation to add task execution mode if enabled.
+        """
+        # Get base MCP tool from parent
+        mcp_tool = super().to_mcp_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")
+
+        return mcp_tool
 
     @classmethod
     def from_function(

tests/server/middleware/test_logging.py

@@ -332,7 +332,7 @@ async def test_on_message_with_resource_template_in_payload(
 
         assert get_log_lines(caplog) == snapshot(
             [
-                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "{\\"name\\":\\"tmpl\\",\\"title\\":null,\\"description\\":null,\\"icons\\":null,\\"tags\\":[],\\"meta\\":null,\\"enabled\\":true,\\"uri_template\\":\\"tmpl://{id}\\",\\"mime_type\\":\\"text/plain\\",\\"parameters\\":{\\"id\\":{\\"type\\":\\"string\\"}},\\"annotations\\":null,\\"task\\":false}", "payload_type": "ResourceTemplate"}',
+                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "{\\"name\\":\\"tmpl\\",\\"title\\":null,\\"description\\":null,\\"icons\\":null,\\"tags\\":[],\\"meta\\":null,\\"enabled\\":true,\\"uri_template\\":\\"tmpl://{id}\\",\\"mime_type\\":\\"text/plain\\",\\"parameters\\":{\\"id\\":{\\"type\\":\\"string\\"}},\\"annotations\\":null}", "payload_type": "ResourceTemplate"}',
                 '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
             ]
         )

tests/server/tasks/test_sync_function_task_disabled.py

@@ -8,6 +8,9 @@
 import pytest
 
 from fastmcp import FastMCP
+from fastmcp.prompts.prompt import FunctionPrompt
+from fastmcp.resources.resource import FunctionResource
+from fastmcp.tools.tool import FunctionTool
 
 
 async def test_sync_tool_with_explicit_task_true_raises():
@@ -91,8 +94,9 @@ async def async_tool(x: int) -> int:
         """An async tool."""
         return x * 2
 
-    # Tool should have task=True
+    # Tool should have task=True and be a FunctionTool
     tool = await mcp.get_tool("async_tool")
+    assert isinstance(tool, FunctionTool)
     assert tool.task is True
 
 
@@ -105,8 +109,9 @@ async def async_prompt() -> str:
         """An async prompt."""
         return "Hello"
 
-    # Prompt should have task=True
+    # Prompt should have task=True and be a FunctionPrompt
     prompt = await mcp.get_prompt("async_prompt")
+    assert isinstance(prompt, FunctionPrompt)
     assert prompt.task is True
 
 
@@ -119,8 +124,9 @@ async def async_resource() -> str:
         """An async resource."""
         return "data"
 
-    # Resource should have task=True
+    # Resource should have task=True and be a FunctionResource
     resource = await mcp._resource_manager.get_resource("test://async")
+    assert isinstance(resource, FunctionResource)
     assert resource.task is True
 
 
@@ -134,6 +140,7 @@ def sync_tool(x: int) -> int:
         return x * 2
 
     tool = await mcp.get_tool("sync_tool")
+    assert isinstance(tool, FunctionTool)
     assert tool.task is False
 
 
@@ -147,6 +154,7 @@ def sync_prompt() -> str:
         return "Hello"
 
     prompt = await mcp.get_prompt("sync_prompt")
+    assert isinstance(prompt, FunctionPrompt)
     assert prompt.task is False
 
 
@@ -160,6 +168,7 @@ def sync_resource() -> str:
         return "data"
 
     resource = await mcp._resource_manager.get_resource("test://sync")
+    assert isinstance(resource, FunctionResource)
     assert resource.task is False