Make extensions construction-only and improve the tasks example

Drop the public `MCPServer.add_extension`; extensions are fixed at construction
via `extensions=[...]` (the apply logic moves to a private `_apply_extension`,
with the `tools/call` interceptor composed once afterwards). This matches the
declarative design and removes the mid-connection mutation footgun. Rework the
tasks story around a `render_report` tool whose multi-step work motivates running
it as a task, with named `_start_task` / `_get_task` / `_task_result` helpers so
the client reads as a clear lifecycle.

Author: Kludex

docs/migration.md

@@ -414,15 +414,14 @@ reverse-DNS identifier and advertise it under `ServerCapabilities.extensions`
 (the 2026-07-28 capability map). An extension subclasses `mcp.server.mcpserver.Extension`
 and overrides only the contribution methods it needs: `tools()`/`resources()`/`methods()`
 (additive) and `intercept_tool_call()` (wraps `tools/call`). Pass instances at
-construction, or register later with `add_extension`:
+construction:
 
 ```python
 from mcp.server.mcpserver import MCPServer
 from mcp.server.apps import Apps
 from mcp.server.tasks import Tasks
 
 mcp = MCPServer("demo", extensions=[Apps(), Tasks()])
-# or: mcp.add_extension(Apps())
 ```
 
 Two reference extensions ship in their own modules:

examples/stories/apps/README.md

@@ -17,9 +17,9 @@ uv run python -m stories.apps.client --http
 
 ## What to look at
 
-- `server.py` `Apps()` + `mcp.add_extension(apps)` — the extension advertises
-  `io.modelcontextprotocol/ui` under `ServerCapabilities.extensions` and
-  contributes the UI-bound tool and its `ui://` resource. `MCPServer` itself
+- `server.py` `MCPServer("apps-example", extensions=[apps])` — the extension
+  advertises `io.modelcontextprotocol/ui` under `ServerCapabilities.extensions`
+  and contributes the UI-bound tool and its `ui://` resource. `MCPServer` itself
   never learns about "ui"; it applies a closed set of contributions.
 - `server.py` `@apps.tool(resource_uri=...)` — stamps `_meta.ui.resourceUri` on
   the tool; `add_html_resource` registers the matching `ui://` resource at

examples/stories/apps/server.py

@@ -26,7 +26,6 @@
 
 
 def build_server() -> MCPServer:
-    mcp = MCPServer("apps-example")
     apps = Apps()
 
     @apps.tool(resource_uri=RESOURCE_URI, title="Get Time", description="Return the current time.")
@@ -37,8 +36,7 @@ def get_time(ctx: Context) -> str:
         return now
 
     apps.add_html_resource(RESOURCE_URI, CLOCK_HTML, title="Clock")
-    mcp.add_extension(apps)
-    return mcp
+    return MCPServer("apps-example", extensions=[apps])
 
 
 if __name__ == "__main__":

examples/stories/manifest.toml

@@ -49,7 +49,7 @@ status = "deprecated"
 lowlevel = false
 
 [story.apps]
-# Extension API is MCPServer-tier (Apps decorators + add_extension); no lowlevel variant.
+# Extension API is MCPServer-tier (Apps decorators + extensions=[...]); no lowlevel variant.
 # The extensions capability map (SEP-2133) rides server/discover, a modern-only path, so
 # `main` pins "auto" (legacy initialize cannot carry it) and the leg is http-asgi.
 lowlevel   = false

examples/stories/tasks/README.md

@@ -18,15 +18,18 @@ uv run python -m stories.tasks.client --http
 
 ## What to look at
 
-- `server.py` `MCPServer(extensions=[Tasks()])` — opt in at construction. The
-  extension advertises `io.modelcontextprotocol/tasks` and serves `tasks/get`,
-  `tasks/result`, `tasks/cancel`, and `tasks/list`.
+- `server.py` `MCPServer("tasks-example", extensions=[Tasks()])` — opt in at
+  construction. The extension advertises `io.modelcontextprotocol/tasks` and
+  serves `tasks/get`, `tasks/result`, `tasks/cancel`, and `tasks/list`. The
+  `render_report` tool is the kind of slower, multi-step work a caller would
+  rather run as a task than block on.
 - `mcp.server.tasks.Tasks.intercept_tool_call` — the interceptive seam: a plain
   call passes through; a call with a `task` field is recorded and returned with
   the task id in `_meta["io.modelcontextprotocol/related-task"]`.
-- `client.py` — sends a task-augmented `tools/call` via `client.session` (the
-  `task` field and `tasks/*` methods are outside the spec verbs `Client`
-  exposes), then drives the lifecycle through `tasks/get` and `tasks/result`.
+- `client.py` `main` — start the call as a task, read its `tasks/get` status,
+  then fetch the payload with `tasks/result`. The `task` field and `tasks/*`
+  methods are outside the spec verbs `Client` exposes, so the thin
+  `_start_task` / `_get_task` / `_task_result` helpers wrap `client.session`.
 
 ## Caveats
 

examples/stories/tasks/client.py

@@ -1,45 +1,55 @@
-"""Request task-augmented execution, then drive the task lifecycle via `tasks/*`."""
+"""Start a tool call as a task, then poll the task to completion and fetch its result.
+
+`Client` exposes only spec verbs, so the `task` augmentation and the `tasks/*`
+methods drop to `client.session`. The thin `_start_task` / `_get_task` /
+`_task_result` helpers keep that `cast` noise out of the story below; `main`
+itself reads as: kick off the work, see it as a task, collect the report.
+"""
 
 from typing import cast
 
 import mcp_types as types
 
-from mcp.client import Client
+from mcp.client import Client, ClientSession
 from mcp.server.tasks import EXTENSION_ID, RELATED_TASK_META_KEY
 from stories._harness import Target, run_client
 
 
+async def _start_task(session: ClientSession, name: str, arguments: dict[str, object]) -> types.CallToolResult:
+    """Call a tool with task augmentation; the result carries the task id in `_meta`."""
+    request = types.CallToolRequest(
+        params=types.CallToolRequestParams(name=name, arguments=arguments, task=types.TaskMetadata(ttl=60))
+    )
+    return await session.send_request(cast("types.ClientRequest", request), types.CallToolResult)
+
+
+async def _get_task(session: ClientSession, task_id: str) -> types.GetTaskResult:
+    request = types.GetTaskRequest(params=types.GetTaskRequestParams(task_id=task_id))
+    return await session.send_request(cast("types.ClientRequest", request), types.GetTaskResult)
+
+
+async def _task_result(session: ClientSession, task_id: str) -> types.CallToolResult:
+    request = types.GetTaskPayloadRequest(params=types.GetTaskPayloadRequestParams(task_id=task_id))
+    return await session.send_request(cast("types.ClientRequest", request), types.CallToolResult)
+
+
 async def main(target: Target, *, mode: str = "auto") -> None:
     async with Client(target, mode=mode) as client:
         # The extensions capability map rides `server/discover` (modern only); a legacy
         # connection (today's stdio) omits it, so assert it only when present.
         if client.server_capabilities.extensions is not None:
             assert client.server_capabilities.extensions == {EXTENSION_ID: {"list": {}, "cancel": {}}}
 
-        # `Client` exposes only spec verbs, so task-augmented calls and the
-        # `tasks/*` methods drop to `client.session` (see custom_methods/). The
-        # casts satisfy the closed `ClientRequest` union; at runtime the body
-        # only calls `.model_dump()`.
-        session = client.session
-        call = types.CallToolRequest(
-            params=types.CallToolRequestParams(
-                name="echo", arguments={"text": "async"}, task=types.TaskMetadata(ttl=60)
-            )
-        )
-        result = await session.send_request(cast("types.ClientRequest", call), types.CallToolResult)
-        assert result.meta is not None, result
-        task_id = result.meta[RELATED_TASK_META_KEY]["taskId"]
-        assert isinstance(result.content[0], types.TextContent)
-        assert result.content[0].text == "async", result
-
-        get = types.GetTaskRequest(params=types.GetTaskRequestParams(task_id=task_id))
-        status = await session.send_request(cast("types.ClientRequest", get), types.GetTaskResult)
-        assert status.status == "completed", status
-
-        payload_req = types.GetTaskPayloadRequest(params=types.GetTaskPayloadRequestParams(task_id=task_id))
-        payload = await session.send_request(cast("types.ClientRequest", payload_req), types.CallToolResult)
-        assert isinstance(payload.content[0], types.TextContent)
-        assert payload.content[0].text == "async", payload
+        started = await _start_task(client.session, "render_report", {"title": "Q3", "sections": 2})
+        task_id = started.meta[RELATED_TASK_META_KEY]["taskId"] if started.meta else None
+        assert task_id is not None, started
+
+        task = await _get_task(client.session, task_id)
+        assert task.status == "completed", task
+
+        report = await _task_result(client.session, task_id)
+        assert isinstance(report.content[0], types.TextContent)
+        assert report.content[0].text.startswith("# Q3"), report
 
 
 if __name__ == "__main__":

examples/stories/tasks/server.py

@@ -1,9 +1,11 @@
 """Tasks: task-augmented tool execution via the interceptive half of the extension API.
 
-`Tasks` is an opt-in `Extension`. It intercepts `tools/call`: a plain call passes
-through, but a call carrying a `task` field is recorded under a task id and
-returned with that id in `_meta`. It also serves the `tasks/*` methods so a
-client can poll status and fetch the payload.
+`Tasks` is an opt-in `Extension`. It intercepts `tools/call`: a plain call runs
+inline and returns its `CallToolResult`, but a call carrying a `task` field is
+recorded under a task id and returned with that id in
+`_meta["io.modelcontextprotocol/related-task"]`, so the client can poll
+`tasks/get` / `tasks/result` instead of blocking. `render_report` is the kind of
+slower, multi-step tool a caller would rather run as a task.
 """
 
 from mcp.server.mcpserver import MCPServer
@@ -14,9 +16,10 @@
 def build_server() -> MCPServer:
     mcp = MCPServer("tasks-example", extensions=[Tasks()])
 
-    @mcp.tool(description="Echo the input back as plain text.", structured_output=False)
-    def echo(text: str) -> str:
-        return text
+    @mcp.tool(description="Render a multi-section report for the given title.", structured_output=False)
+    def render_report(title: str, sections: int) -> str:
+        body = "\n".join(f"## Section {n}\n(generated)" for n in range(1, sections + 1))
+        return f"# {title}\n\n{body}"
 
     return mcp
 

src/mcp/server/mcpserver/server.py

@@ -55,7 +55,7 @@
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
-from mcp.server.context import ServerMiddleware, ServerRequestContext
+from mcp.server.context import ServerRequestContext
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, Server
 from mcp.server.lowlevel.server import lifespan as default_lifespan
@@ -210,9 +210,9 @@ def __init__(
         configure_logging(self.settings.log_level)
 
         self._extensions: list[Extension] = []
-        self._extension_interceptor: ServerMiddleware[LifespanResultT] | None = None
         for extension in extensions or ():
-            self.add_extension(extension)
+            self._apply_extension(extension)
+        self._install_extension_interceptor()
 
     @property
     def name(self) -> str:
@@ -254,20 +254,13 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         """
         return self._lowlevel_server.session_manager
 
-    def add_extension(self, extension: Extension) -> None:
-        """Register an opt-in MCP extension (SEP-2133).
+    def _apply_extension(self, extension: Extension) -> None:
+        """Apply one opt-in extension's contributions through the public surface.
 
-        Applies the extension's contributions through the server's public surface:
-        its tools and resources are registered, its request methods are wired onto
-        the low-level server, and its `tools/call` interceptor joins a single
-        composed `tools/call` middleware. The extension's settings are advertised
-        under `ServerCapabilities.extensions[extension.identifier]`.
-
-        Args:
-            extension: The extension to install.
-
-        Raises:
-            ValueError: If an extension with the same identifier is already registered.
+        Registers its tools/resources/methods and advertises its settings under
+        `ServerCapabilities.extensions[extension.identifier]`. Extensions are fixed
+        at construction, so this is private; the `tools/call` interceptor is
+        composed once afterwards by `_install_extension_interceptor`.
         """
         if any(e.identifier == extension.identifier for e in self._extensions):
             raise ValueError(f"Extension {extension.identifier!r} is already registered")
@@ -281,16 +274,15 @@ def add_extension(self, extension: Extension) -> None:
             self._lowlevel_server.add_request_handler(method.method, method.params_type, method.handler)
 
         self._lowlevel_server.extensions[extension.identifier] = extension.settings()
-        self._refresh_extension_interceptor()
 
-    def _refresh_extension_interceptor(self) -> None:
-        """Rebuild the single composed `tools/call` interceptor from all extensions."""
-        if self._extension_interceptor is not None:
-            self._lowlevel_server.middleware.remove(self._extension_interceptor)
-            self._extension_interceptor = None
+    def _install_extension_interceptor(self) -> None:
+        """Compose every extension's `tools/call` interceptor into one middleware.
+
+        Installed only when at least one extension overrides `intercept_tool_call`,
+        so a server with purely additive extensions adds no middleware.
+        """
         if any(type(e).intercept_tool_call is not Extension.intercept_tool_call for e in self._extensions):
-            self._extension_interceptor = compose_tool_call_interceptor(self._extensions)
-            self._lowlevel_server.middleware.append(self._extension_interceptor)
+            self._lowlevel_server.middleware.append(compose_tool_call_interceptor(self._extensions))
 
     @overload
     def run(self, transport: Literal["stdio"] = ...) -> None: ...

tests/server/mcpserver/test_extension.py

@@ -177,13 +177,6 @@ def test_duplicate_extension_identifier_raises() -> None:
         MCPServer("test", extensions=[_SettingsExt(), _SettingsExt()])
 
 
-def test_add_extension_after_construction_rejects_duplicate_identifier() -> None:
-    """SDK-defined: `add_extension` enforces the same uniqueness as the constructor."""
-    server = MCPServer("test", extensions=[_SettingsExt()])
-    with pytest.raises(ValueError):
-        server.add_extension(_SettingsExt())
-
-
 async def test_extension_method_reachable_via_session_send_request() -> None:
     """SDK-defined: an `Extension` overriding `methods()` wires a new request verb
     onto the low-level server, reachable through `client.session.send_request`."""
@@ -221,25 +214,22 @@ async def test_short_circuiting_interceptor_replaces_tool_result() -> None:
 
 
 def test_plain_extension_installs_no_tool_call_interceptor() -> None:
-    """SDK-defined: an extension that does not override `intercept_tool_call` leaves
-    `_extension_interceptor` unset and adds no middleware - the composed
-    interceptor exists only when at least one extension overrides it."""
+    """SDK-defined: an extension that does not override `intercept_tool_call` adds no
+    middleware - the composed interceptor exists only when at least one extension
+    overrides it."""
     baseline = len(MCPServer("test")._lowlevel_server.middleware)
     server = MCPServer("test", extensions=[_AdditiveExt()])
 
-    assert server._extension_interceptor is None
     assert len(server._lowlevel_server.middleware) == baseline
 
 
 def test_overriding_extension_installs_one_tool_call_interceptor() -> None:
-    """SDK-defined: registering an extension that overrides `intercept_tool_call`
-    composes exactly one middleware and records it as `_extension_interceptor`."""
+    """SDK-defined: an extension that overrides `intercept_tool_call` composes exactly
+    one additional `tools/call` middleware."""
     baseline = len(MCPServer("test")._lowlevel_server.middleware)
     server = MCPServer("test", extensions=[_ReplacingExt()])
 
-    assert server._extension_interceptor is not None
     assert len(server._lowlevel_server.middleware) == baseline + 1
-    assert server._lowlevel_server.middleware[-1] is server._extension_interceptor
 
 
 async def test_default_interceptor_passes_through_alongside_an_overriding_one() -> None: