Extract docstring code examples into type-checked companion files

Move inline code examples from docstrings in `src/mcp/` into standalone
companion files at `examples/snippets/docstrings/mcp/`, mirroring the
source tree structure. The `scripts/sync_snippets.py` script keeps the
docstring content in sync with the companion files via
`<!-- snippet-source #RegionName -->` markers.

This ensures all docstring examples are checked by pyright and ruff,
catching type errors and style drift that would otherwise go unnoticed
in raw docstring text. The pattern follows the TypeScript SDK's approach
of one companion file per source file, with each example in a named
function whose parameters supply typed context.

19 companion files cover 42 code examples across the public API surface:
`MCPServer`, `Client`, `ClientSession`, `Context`, `ResponseRouter`,
`ServerTaskContext`, `ExperimentalTaskHandlers`,
`ClientCredentialsOAuthProvider`, `PrivateKeyJWTOAuthProvider`,
`SignedJWTParameters`, and others.

All source markers use path-less form (`#RegionName`) — the companion
path is derived automatically from the target file location. Region
names follow `ClassName_methodName_variant` without abbreviation.

Pyright execution environment for the companion files suppresses only
"unused artifact" diagnostics inherent to example code
(`reportUnusedFunction`, `reportUnusedVariable`, `reportAbstractUsage`,
`reportUnusedClass`, `reportPrivateUsage`). All actual type-checking
rules remain enabled.

Author: jonathanhefner

examples/snippets/docstrings/mcp/client/auth/extensions/client_credentials.py

@@ -0,0 +1,95 @@
+"""Companion examples for src/mcp/client/auth/extensions/client_credentials.py docstrings."""
+
+from __future__ import annotations
+
+from mcp.client.auth import TokenStorage
+from mcp.client.auth.extensions.client_credentials import (
+    ClientCredentialsOAuthProvider,
+    PrivateKeyJWTOAuthProvider,
+    SignedJWTParameters,
+    static_assertion_provider,
+)
+
+
+async def fetch_token_from_identity_provider(*, audience: str) -> str: ...
+
+
+def ClientCredentialsOAuthProvider_init(my_token_storage: TokenStorage) -> None:
+    # region ClientCredentialsOAuthProvider_init
+    provider = ClientCredentialsOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        client_secret="my-client-secret",
+    )
+    # endregion ClientCredentialsOAuthProvider_init
+
+
+def static_assertion_provider_usage(my_token_storage: TokenStorage, my_prebuilt_jwt: str) -> None:
+    # region static_assertion_provider_usage
+    provider = PrivateKeyJWTOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+    )
+    # endregion static_assertion_provider_usage
+
+
+def SignedJWTParameters_usage(my_token_storage: TokenStorage, private_key_pem: str) -> None:
+    # region SignedJWTParameters_usage
+    jwt_params = SignedJWTParameters(
+        issuer="my-client-id",
+        subject="my-client-id",
+        signing_key=private_key_pem,
+    )
+    provider = PrivateKeyJWTOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        assertion_provider=jwt_params.create_assertion_provider(),
+    )
+    # endregion SignedJWTParameters_usage
+
+
+def PrivateKeyJWTOAuthProvider_workloadIdentity(my_token_storage: TokenStorage) -> None:
+    # region PrivateKeyJWTOAuthProvider_workloadIdentity
+    async def get_workload_identity_token(audience: str) -> str:
+        # Fetch JWT from your identity provider
+        # The JWT's audience must match the provided audience parameter
+        return await fetch_token_from_identity_provider(audience=audience)
+
+    provider = PrivateKeyJWTOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        assertion_provider=get_workload_identity_token,
+    )
+    # endregion PrivateKeyJWTOAuthProvider_workloadIdentity
+
+
+def PrivateKeyJWTOAuthProvider_staticJWT(my_token_storage: TokenStorage, my_prebuilt_jwt: str) -> None:
+    # region PrivateKeyJWTOAuthProvider_staticJWT
+    provider = PrivateKeyJWTOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+    )
+    # endregion PrivateKeyJWTOAuthProvider_staticJWT
+
+
+def PrivateKeyJWTOAuthProvider_sdkSigned(my_token_storage: TokenStorage, private_key_pem: str) -> None:
+    # region PrivateKeyJWTOAuthProvider_sdkSigned
+    jwt_params = SignedJWTParameters(
+        issuer="my-client-id",
+        subject="my-client-id",
+        signing_key=private_key_pem,
+    )
+    provider = PrivateKeyJWTOAuthProvider(
+        server_url="https://api.example.com",
+        storage=my_token_storage,
+        client_id="my-client-id",
+        assertion_provider=jwt_params.create_assertion_provider(),
+    )
+    # endregion PrivateKeyJWTOAuthProvider_sdkSigned

examples/snippets/docstrings/mcp/client/client.py

@@ -0,0 +1,24 @@
+"""Companion examples for src/mcp/client/client.py docstrings."""
+
+from __future__ import annotations
+
+import asyncio
+
+
+def Client_usage() -> None:
+    # region Client_usage
+    from mcp.client import Client
+    from mcp.server.mcpserver import MCPServer
+
+    server = MCPServer("test")
+
+    @server.tool()
+    def add(a: int, b: int) -> int:
+        return a + b
+
+    async def main():
+        async with Client(server) as client:
+            result = await client.call_tool("add", {"a": 1, "b": 2})
+
+    asyncio.run(main())
+    # endregion Client_usage

examples/snippets/docstrings/mcp/client/experimental/task_handlers.py

@@ -0,0 +1,36 @@
+"""Companion examples for src/mcp/client/experimental/task_handlers.py docstrings."""
+
+from __future__ import annotations
+
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+
+from mcp import types
+from mcp.client.experimental.task_handlers import ExperimentalTaskHandlers
+from mcp.client.session import ClientSession
+from mcp.shared._context import RequestContext
+from mcp.shared.session import SessionMessage
+
+
+async def my_get_task_handler(
+    context: RequestContext[ClientSession],
+    params: types.GetTaskRequestParams,
+) -> types.GetTaskResult | types.ErrorData: ...
+
+
+async def my_list_tasks_handler(
+    context: RequestContext[ClientSession],
+    params: types.PaginatedRequestParams | None,
+) -> types.ListTasksResult | types.ErrorData: ...
+
+
+def ExperimentalTaskHandlers_usage(
+    read_stream: MemoryObjectReceiveStream[SessionMessage | Exception],
+    write_stream: MemoryObjectSendStream[SessionMessage],
+) -> None:
+    # region ExperimentalTaskHandlers_usage
+    handlers = ExperimentalTaskHandlers(
+        get_task=my_get_task_handler,
+        list_tasks=my_list_tasks_handler,
+    )
+    session = ClientSession(read_stream, write_stream, experimental_task_handlers=handlers)
+    # endregion ExperimentalTaskHandlers_usage

examples/snippets/docstrings/mcp/client/experimental/tasks.py

@@ -0,0 +1,60 @@
+"""Companion examples for src/mcp/client/experimental/tasks.py docstrings."""
+
+from __future__ import annotations
+
+import anyio
+
+from mcp.client.session import ClientSession
+from mcp.types import CallToolResult
+
+
+async def module_overview(session: ClientSession) -> None:
+    # region module_overview
+    # Call a tool as a task
+    result = await session.experimental.call_tool_as_task("tool_name", {"arg": "value"})
+    task_id = result.task.task_id
+
+    # Get task status
+    status = await session.experimental.get_task(task_id)
+
+    # Get task result when complete
+    if status.status == "completed":
+        result = await session.experimental.get_task_result(task_id, CallToolResult)
+
+    # List all tasks
+    tasks = await session.experimental.list_tasks()
+
+    # Cancel a task
+    await session.experimental.cancel_task(task_id)
+    # endregion module_overview
+
+
+async def ExperimentalClientFeatures_call_tool_as_task_usage(session: ClientSession) -> None:
+    # region ExperimentalClientFeatures_call_tool_as_task_usage
+    # Create task
+    result = await session.experimental.call_tool_as_task("long_running_tool", {"input": "data"})
+    task_id = result.task.task_id
+
+    # Poll for completion
+    while True:
+        status = await session.experimental.get_task(task_id)
+        if status.status == "completed":
+            break
+        await anyio.sleep(0.5)
+
+    # Get result
+    final = await session.experimental.get_task_result(task_id, CallToolResult)
+    # endregion ExperimentalClientFeatures_call_tool_as_task_usage
+
+
+async def ExperimentalClientFeatures_poll_task_usage(session: ClientSession, task_id: str) -> None:
+    # region ExperimentalClientFeatures_poll_task_usage
+    async for status in session.experimental.poll_task(task_id):
+        print(f"Status: {status.status}")
+        if status.status == "input_required":
+            # Handle elicitation request via tasks/result
+            pass
+
+    # Task is now terminal, get the result
+    result = await session.experimental.get_task_result(task_id, CallToolResult)
+    # endregion ExperimentalClientFeatures_poll_task_usage

examples/snippets/docstrings/mcp/client/session.py

@@ -0,0 +1,13 @@
+"""Companion examples for src/mcp/client/session.py docstrings."""
+
+from __future__ import annotations
+
+from mcp.client.session import ClientSession
+from mcp.types import CallToolResult
+
+
+async def ClientSession_experimental_usage(session: ClientSession, task_id: str) -> None:
+    # region ClientSession_experimental_usage
+    status = await session.experimental.get_task(task_id)
+    result = await session.experimental.get_task_result(task_id, CallToolResult)
+    # endregion ClientSession_experimental_usage

examples/snippets/docstrings/mcp/client/session_group.py

@@ -0,0 +1,19 @@
+"""Companion examples for src/mcp/client/session_group.py docstrings."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.client.session_group import ClientSessionGroup
+
+
+async def ClientSessionGroup_usage(server_params: list[Any]) -> None:
+    # region ClientSessionGroup_usage
+    def name_fn(name: str, server_info: Any) -> str:
+        return f"{server_info.name}_{name}"
+
+    async with ClientSessionGroup(component_name_hook=name_fn) as group:
+        for server_param in server_params:
+            await group.connect_to_server(server_param)
+        ...
+    # endregion ClientSessionGroup_usage

examples/snippets/docstrings/mcp/server/experimental/request_context.py

@@ -0,0 +1,31 @@
+"""Companion examples for src/mcp/server/experimental/request_context.py docstrings."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.context import ServerRequestContext
+from mcp.server.experimental.task_context import ServerTaskContext
+from mcp.types import CallToolRequestParams, CallToolResult, CreateTaskResult, TextContent
+
+
+def Experimental_run_task_usage() -> None:
+    # region Experimental_run_task_usage
+    async def handle_tool(
+        ctx: ServerRequestContext[Any, Any],
+        params: CallToolRequestParams,
+    ) -> CreateTaskResult:
+        async def work(task: ServerTaskContext) -> CallToolResult:
+            result = await task.elicit(
+                message="Are you sure?",
+                requested_schema={"type": "object", "properties": {"confirm": {"type": "boolean"}}},
+            )
+            if result.action == "accept" and result.content:
+                confirmed = result.content.get("confirm", False)
+            else:
+                confirmed = False
+            return CallToolResult(content=[TextContent(text="Done" if confirmed else "Cancelled")])
+
+        return await ctx.experimental.run_task(work)
+
+    # endregion Experimental_run_task_usage

examples/snippets/docstrings/mcp/server/experimental/task_context.py

@@ -0,0 +1,24 @@
+"""Companion examples for src/mcp/server/experimental/task_context.py docstrings."""
+
+from __future__ import annotations
+
+from mcp.server.experimental.task_context import ServerTaskContext
+from mcp.types import CallToolResult, TextContent
+
+
+async def ServerTaskContext_usage(task: ServerTaskContext) -> None:
+    # region ServerTaskContext_usage
+    async def my_task_work(task: ServerTaskContext) -> CallToolResult:
+        await task.update_status("Starting...")
+
+        result = await task.elicit(
+            message="Continue?",
+            requested_schema={"type": "object", "properties": {"ok": {"type": "boolean"}}},
+        )
+
+        if result.action == "accept" and result.content and result.content.get("ok"):
+            return CallToolResult(content=[TextContent(text="Done!")])
+        else:
+            return CallToolResult(content=[TextContent(text="Cancelled")])
+
+    # endregion ServerTaskContext_usage

examples/snippets/docstrings/mcp/server/experimental/task_support.py

@@ -0,0 +1,33 @@
+"""Companion examples for src/mcp/server/experimental/task_support.py docstrings."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.lowlevel.server import Server
+from mcp.shared.experimental.tasks.message_queue import TaskMessageQueue
+from mcp.shared.experimental.tasks.store import TaskStore
+
+
+# Stubs for undefined references in examples
+class RedisTaskStore(TaskStore):  # type: ignore[abstract]
+    def __init__(self, redis_url: str) -> None: ...
+
+
+class RedisTaskMessageQueue(TaskMessageQueue):  # type: ignore[abstract]
+    def __init__(self, redis_url: str) -> None: ...
+
+
+def TaskSupport_simple(server: Server[Any]) -> None:
+    # region TaskSupport_simple
+    server.experimental.enable_tasks()
+    # endregion TaskSupport_simple
+
+
+def TaskSupport_custom(server: Server[Any], redis_url: str) -> None:
+    # region TaskSupport_custom
+    server.experimental.enable_tasks(
+        store=RedisTaskStore(redis_url),
+        queue=RedisTaskMessageQueue(redis_url),
+    )
+    # endregion TaskSupport_custom

examples/snippets/docstrings/mcp/server/lowlevel/experimental.py

@@ -0,0 +1,32 @@
+"""Companion examples for src/mcp/server/lowlevel/experimental.py docstrings."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.lowlevel.server import Server
+from mcp.shared.experimental.tasks.message_queue import TaskMessageQueue
+from mcp.shared.experimental.tasks.store import TaskStore
+
+
+class RedisTaskStore(TaskStore):  # type: ignore[abstract]
+    def __init__(self, redis_url: str) -> None: ...
+
+
+class RedisTaskMessageQueue(TaskMessageQueue):  # type: ignore[abstract]
+    def __init__(self, redis_url: str) -> None: ...
+
+
+def ExperimentalHandlers_enable_tasks_simple(server: Server[Any]) -> None:
+    # region ExperimentalHandlers_enable_tasks_simple
+    server.experimental.enable_tasks()
+    # endregion ExperimentalHandlers_enable_tasks_simple
+
+
+def ExperimentalHandlers_enable_tasks_custom(server: Server[Any], redis_url: str) -> None:
+    # region ExperimentalHandlers_enable_tasks_custom
+    server.experimental.enable_tasks(
+        store=RedisTaskStore(redis_url),
+        queue=RedisTaskMessageQueue(redis_url),
+    )
+    # endregion ExperimentalHandlers_enable_tasks_custom