Centralize task capabilities, add component filtering

Addresses code review feedback:
- Extract `get_task_capabilities()` to avoid duplicating the SEP-1686
  capability structure across transports
- Add `_should_enable_component()` check before task routing for tools,
  resources, and prompts to respect enable/tag filtering
- Simplify tasks/__init__.py to avoid circular import issues

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

Co-Authored-By: Claude <noreply@anthropic.com>

src/fastmcp/client/transports.py

@@ -36,6 +36,7 @@ from fastmcp.client.auth.oauth import OAuth
 from fastmcp.mcp_config import MCPConfig, infer_transport_type_from_url
 from fastmcp.server.dependencies import get_http_headers
 from fastmcp.server.server import FastMCP
+from fastmcp.server.tasks.capabilities import get_task_capabilities
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
 
@@ -856,14 +857,7 @@ class FastMCPTransport(ClientTransport):
                 _enter_server_lifespan(server=self.server),
             ):
                 # Build experimental capabilities
-                # Declare SEP-1686 task support
-                experimental_capabilities = {
-                    "tasks": {
-                        "tools": True,
-                        "prompts": True,
-                        "resources": True,
-                    }
-                }
+                experimental_capabilities = get_task_capabilities()
 
                 tg.start_soon(
                     lambda: self.server._mcp_server.run(

src/fastmcp/server/http.py

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

src/fastmcp/server/server.py

@@ -74,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.capabilities import get_task_capabilities
 from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.server.tasks.handlers import (
     handle_prompt_as_task,
@@ -690,7 +691,11 @@ class FastMCP(Generic[LifespanResultT]):
             async with fastmcp.server.context.Context(fastmcp=self):
                 # Get resource including from mounted servers
                 resource = await self._get_resource_with_task_config(str(uri))
-                if resource and hasattr(resource, "task_config"):
+                if (
+                    resource
+                    and self._should_enable_component(resource)
+                    and hasattr(resource, "task_config")
+                ):
                     task_mode = resource.task_config.mode  # type: ignore[union-attr]
 
                     # Enforce mode="required" - must have task metadata
@@ -817,7 +822,12 @@ class FastMCP(Generic[LifespanResultT]):
             async with fastmcp.server.context.Context(fastmcp=self):
                 prompts = await self.get_prompts()
                 prompt = prompts.get(name)
-                if prompt and hasattr(prompt, "task_config") and prompt.task_config:
+                if (
+                    prompt
+                    and self._should_enable_component(prompt)
+                    and hasattr(prompt, "task_config")
+                    and prompt.task_config
+                ):
                     task_mode = prompt.task_config.mode  # type: ignore[union-attr]
 
                     # Enforce mode="required" - must have task metadata
@@ -1542,7 +1552,11 @@ class FastMCP(Generic[LifespanResultT]):
 
                 # Get tool from local manager, mounted servers, or proxy
                 tool = await self._get_tool_with_task_config(key)
-                if tool and hasattr(tool, "task_config"):
+                if (
+                    tool
+                    and self._should_enable_component(tool)
+                    and hasattr(tool, "task_config")
+                ):
                     task_mode = tool.task_config.mode  # type: ignore[union-attr]
 
                     # Enforce mode="required" - must have task metadata
@@ -2476,19 +2490,7 @@ class FastMCP(Generic[LifespanResultT]):
                     )
 
                     # Build experimental capabilities
-                    # Declare SEP-1686 task support per final spec (lines 49-63)
-                    # Nested structure: {list: {}, cancel: {}, requests: {tools: {call: {}}}}
-                    experimental_capabilities = {
-                        "tasks": {
-                            "list": {},
-                            "cancel": {},
-                            "requests": {
-                                "tools": {"call": {}},
-                                "prompts": {"get": {}},
-                                "resources": {"read": {}},
-                            },
-                        }
-                    }
+                    experimental_capabilities = get_task_capabilities()
 
                     await self._mcp_server.run(
                         read_stream,

src/fastmcp/server/tasks/__init__.py

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

src/fastmcp/server/tasks/capabilities.py

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

tests/server/tasks/test_task_capabilities.py

@@ -7,6 +7,7 @@ Task protocol is now always enabled.
 
 from fastmcp import FastMCP
 from fastmcp.client import Client
+from fastmcp.server.tasks import get_task_capabilities
 
 
 async def test_capabilities_include_tasks():
@@ -25,11 +26,7 @@ async def test_capabilities_include_tasks():
         assert init_result.capabilities.experimental is not None
         assert "tasks" in init_result.capabilities.experimental
         tasks_cap = init_result.capabilities.experimental["tasks"]
-        assert tasks_cap == {
-            "tools": True,
-            "prompts": True,
-            "resources": True,
-        }
+        assert tasks_cap == get_task_capabilities()["tasks"]
 
 
 async def test_client_uses_task_capable_session():