docs: add snippet verification for docs/ pages (#2115)

examples/snippets/clients/cancellation.py

@@ -0,0 +1,16 @@
+import mcp.types as types
+from mcp import ClientSession
+
+
+async def cancel_request(session: ClientSession) -> None:
+    """Send a cancellation notification for a previously-issued request."""
+    await session.send_notification(
+        types.ClientNotification(
+            types.CancelledNotification(
+                params=types.CancelledNotificationParams(
+                    requestId="request-id-to-cancel",
+                    reason="User navigated away",
+                )
+            )
+        )
+    )

examples/snippets/clients/logging_client.py

@@ -0,0 +1,13 @@
+from mcp import ClientSession, types
+
+
+async def handle_log(params: types.LoggingMessageNotificationParams) -> None:
+    """Handle log messages from the server."""
+    print(f"[{params.level}] {params.data}")
+
+
+session = ClientSession(
+    read_stream,
+    write_stream,
+    logging_callback=handle_log,
+)

examples/snippets/clients/roots_example.py

@@ -0,0 +1,22 @@
+from mcp import ClientSession, types
+from mcp.shared.context import RequestContext
+
+
+async def handle_list_roots(
+    context: RequestContext[ClientSession, None],
+) -> types.ListRootsResult:
+    """Return the client's workspace roots."""
+    return types.ListRootsResult(
+        roots=[
+            types.Root(uri="file:///home/user/project", name="My Project"),
+            types.Root(uri="file:///home/user/data", name="Data Folder"),
+        ]
+    )
+
+
+# Pass the callback when creating the session
+session = ClientSession(
+    read_stream,
+    write_stream,
+    list_roots_callback=handle_list_roots,
+)

examples/snippets/clients/sse_client.py

@@ -0,0 +1,16 @@
+import asyncio
+
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+
+
+async def main():
+    async with sse_client("http://localhost:8000/sse") as (read_stream, write_stream):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            tools = await session.list_tools()
+            print(f"Available tools: {[t.name for t in tools.tools]}")
+
+
+asyncio.run(main())

examples/snippets/servers/audio_example.py

@@ -0,0 +1,16 @@
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.utilities.types import Audio
+
+mcp = FastMCP("Audio Example")
+
+
+@mcp.tool()
+def get_audio_from_file(file_path: str) -> Audio:
+    """Return audio from a file path (format auto-detected from extension)."""
+    return Audio(path=file_path)
+
+
+@mcp.tool()
+def get_audio_from_bytes(raw_audio: bytes) -> Audio:
+    """Return audio from raw bytes with explicit format."""
+    return Audio(data=raw_audio, format="wav")

examples/snippets/servers/binary_resources.py

@@ -0,0 +1,10 @@
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP("Binary Resource Example")
+
+
+@mcp.resource("images://logo.png", mime_type="image/png")
+def get_logo() -> bytes:
+    """Return a binary image resource."""
+    with open("logo.png", "rb") as f:
+        return f.read()

examples/snippets/servers/elicitation_complete.py

@@ -0,0 +1,15 @@
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.session import ServerSession
+
+mcp = FastMCP("Elicit Complete Example")
+
+
+@mcp.tool()
+async def handle_oauth_callback(elicitation_id: str, ctx: Context[ServerSession, None]) -> str:
+    """Called when OAuth flow completes out-of-band."""
+    # ... process the callback ...
+
+    # Notify the client that the elicitation is done
+    await ctx.session.send_elicit_complete(elicitation_id)
+
+    return "Authorization complete"

examples/snippets/servers/elicitation_enum.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel, Field
+
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.session import ServerSession
+
+mcp = FastMCP("Enum Elicitation Example")
+
+
+class ColorPreference(BaseModel):
+    color: str = Field(
+        description="Pick your favorite color",
+        json_schema_extra={"enum": ["red", "green", "blue", "yellow"]},
+    )
+
+
+@mcp.tool()
+async def pick_color(ctx: Context[ServerSession, None]) -> str:
+    """Ask the user to pick a color from a list."""
+    result = await ctx.elicit(
+        message="Choose a color:",
+        schema=ColorPreference,
+    )
+    if result.action == "accept":
+        return f"You picked: {result.data.color}"
+    return "No color selected"

examples/snippets/servers/embedded_resource_results.py

@@ -0,0 +1,19 @@
+from mcp.server.fastmcp import FastMCP
+from mcp.types import EmbeddedResource, TextResourceContents
+
+mcp = FastMCP("Embedded Resource Example")
+
+
+@mcp.tool()
+def read_config(path: str) -> EmbeddedResource:
+    """Read a config file and return it as an embedded resource."""
+    with open(path) as f:
+        content = f.read()
+    return EmbeddedResource(
+        type="resource",
+        resource=TextResourceContents(
+            uri=f"file://{path}",
+            text=content,
+            mimeType="application/json",
+        ),
+    )

examples/snippets/servers/embedded_resource_results_binary.py

@@ -0,0 +1,21 @@
+import base64
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import BlobResourceContents, EmbeddedResource
+
+mcp = FastMCP("Binary Embedded Resource Example")
+
+
+@mcp.tool()
+def read_binary_file(path: str) -> EmbeddedResource:
+    """Read a binary file and return it as an embedded resource."""
+    with open(path, "rb") as f:
+        data = base64.b64encode(f.read()).decode()
+    return EmbeddedResource(
+        type="resource",
+        resource=BlobResourceContents(
+            uri=f"file://{path}",
+            blob=data,
+            mimeType="application/octet-stream",
+        ),
+    )

examples/snippets/servers/json_schema_example.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, Field
+
+
+class SearchParams(BaseModel):
+    query: str = Field(description="Search query string")
+    max_results: int = Field(default=10, description="Maximum results to return")
+
+
+# Pydantic generates a JSON Schema 2020-12 compatible schema:
+schema = SearchParams.model_json_schema()
+# {
+#     "properties": {
+#         "query": {"description": "Search query string", "type": "string"},
+#         "max_results": {
+#             "default": 10,
+#             "description": "Maximum results to return",
+#             "type": "integer",
+#         },
+#     },
+#     "required": ["query"],
+#     "title": "SearchParams",
+#     "type": "object",
+# }

examples/snippets/servers/prompt_change_notifications.py

@@ -0,0 +1,12 @@
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.session import ServerSession
+
+mcp = FastMCP("Dynamic Prompts")
+
+
+@mcp.tool()
+async def update_prompts(ctx: Context[ServerSession, None]) -> str:
+    """Update available prompts and notify clients."""
+    # ... modify prompts ...
+    await ctx.session.send_prompt_list_changed()
+    return "Prompts updated"

examples/snippets/servers/prompt_embedded_resources.py

@@ -0,0 +1,26 @@
+import mcp.types as types
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.prompts import base
+
+mcp = FastMCP("Embedded Resource Prompt Example")
+
+
+@mcp.prompt()
+def review_file(filename: str) -> list[base.Message]:
+    """Review a file with its contents embedded."""
+    file_content = open(filename).read()
+    return [
+        base.UserMessage(
+            content=types.TextContent(type="text", text=f"Please review {filename}:"),
+        ),
+        base.UserMessage(
+            content=types.EmbeddedResource(
+                type="resource",
+                resource=types.TextResourceContents(
+                    uri=f"file://{filename}",
+                    text=file_content,
+                    mimeType="text/plain",
+                ),
+            ),
+        ),
+    ]

examples/snippets/servers/prompt_image_content.py

@@ -0,0 +1,20 @@
+import mcp.types as types
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.prompts import base
+from mcp.server.fastmcp.utilities.types import Image
+
+mcp = FastMCP("Image Prompt Example")
+
+
+@mcp.prompt()
+def describe_image(image_path: str) -> list[base.Message]:
+    """Prompt that includes an image for analysis."""
+    img = Image(path=image_path)
+    return [
+        base.UserMessage(
+            content=types.TextContent(type="text", text="Describe this image:"),
+        ),
+        base.UserMessage(
+            content=img.to_image_content(),
+        ),
+    ]

examples/snippets/servers/resource_subscriptions.py

@@ -0,0 +1,18 @@
+from mcp.server.lowlevel import Server
+
+server = Server("Subscription Example")
+
+subscriptions: dict[str, set[str]] = {}  # uri -> set of session ids
+
+
+@server.subscribe_resource()
+async def handle_subscribe(uri) -> None:
+    """Handle a client subscribing to a resource."""
+    subscriptions.setdefault(str(uri), set()).add("current_session")
+
+
+@server.unsubscribe_resource()
+async def handle_unsubscribe(uri) -> None:
+    """Handle a client unsubscribing from a resource."""
+    if str(uri) in subscriptions:
+        subscriptions[str(uri)].discard("current_session")

examples/snippets/servers/resource_templates.py

@@ -0,0 +1,9 @@
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP("Template Example")
+
+
+@mcp.resource("users://{user_id}/profile")
+def get_user_profile(user_id: str) -> str:
+    """Read a specific user's profile. The user_id is extracted from the URI."""
+    return f'{{"user_id": "{user_id}", "name": "User {user_id}"}}'

examples/snippets/servers/set_logging_level.py

@@ -0,0 +1,13 @@
+import mcp.types as types
+from mcp.server.lowlevel import Server
+
+server = Server("Logging Level Example")
+
+current_level: types.LoggingLevel = "warning"
+
+
+@server.set_logging_level()
+async def handle_set_level(level: types.LoggingLevel) -> None:
+    """Handle client request to change the logging level."""
+    global current_level
+    current_level = level

examples/snippets/servers/tool_change_notifications.py

@@ -0,0 +1,15 @@
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.session import ServerSession
+
+mcp = FastMCP("Dynamic Tools")
+
+
+@mcp.tool()
+async def register_plugin(name: str, ctx: Context[ServerSession, None]) -> str:
+    """Dynamically register a new tool and notify the client."""
+    # ... register the plugin's tools ...
+
+    # Notify the client that the tool list has changed
+    await ctx.session.send_tool_list_changed()
+
+    return f"Plugin '{name}' registered"