Embed code snippets for README from executable examples (#1055)

.github/workflows/shared.yml

@@ -3,6 +3,9 @@ name: Shared Checks
 on:
   workflow_call:
 
+permissions:
+  contents: read
+
 jobs:
   pre-commit:
     runs-on: ubuntu-latest
@@ -46,3 +49,19 @@ jobs:
       - name: Run pytest
         run: uv run --frozen --no-sync pytest
     continue-on-error: true
+
+  readme-snippets:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+          version: 0.7.2
+
+      - name: Install dependencies
+        run: uv sync --frozen --all-extras --python 3.10
+
+      - name: Check README snippets are up to date
+        run: uv run --frozen scripts/update_readme_snippets.py --check

.pre-commit-config.yaml

@@ -23,6 +23,7 @@ repos:
         types: [python]
         language: system
         pass_filenames: false
+        exclude: ^README\.md$
       - id: pyright
         name: pyright
         entry: uv run pyright
@@ -36,3 +37,9 @@ repos:
         language: system
         files: ^(pyproject\.toml|uv\.lock)$
         pass_filenames: false
+      - id: readme-snippets
+        name: Check README snippets are up to date
+        entry: uv run scripts/update_readme_snippets.py --check
+        language: system
+        files: ^(README\.md|examples/.*\.py|scripts/update_readme_snippets\.py)$
+        pass_filenames: false

CONTRIBUTING.md

@@ -40,7 +40,12 @@ uv run ruff check .
 uv run ruff format .
 ```
 
-7. Submit a pull request to the same branch you branched from
+7. Update README snippets if you modified example code:
+```bash
+uv run scripts/update_readme_snippets.py
+```
+
+8. Submit a pull request to the same branch you branched from
 
 ## Code Style
 

README.md

@@ -33,6 +33,8 @@
     - [Context](#context)
     - [Completions](#completions)
     - [Elicitation](#elicitation)
+    - [Sampling](#sampling)
+    - [Logging and Notifications](#logging-and-notifications)
     - [Authentication](#authentication)
   - [Running Your Server](#running-your-server)
     - [Development Mode](#development-mode)
@@ -119,7 +121,7 @@ mcp = FastMCP("Demo")
 
 # Add an addition tool
 @mcp.tool()
-def add(a: int, b: int) -> int:
+def sum(a: int, b: int) -> int:
     """Add two numbers"""
     return a + b
 
@@ -207,48 +209,57 @@ def query_db() -> str:
 
 Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects:
 
+<!-- snippet-source examples/snippets/servers/basic_resource.py -->
 ```python
 from mcp.server.fastmcp import FastMCP
 
-mcp = FastMCP("My App")
+mcp = FastMCP(name="Resource Example")
 
 
-@mcp.resource("config://app", title="Application Configuration")
+@mcp.resource("file://documents/{name}")
-def get_config() -> str:
+def read_document(name: str) -> str:
-    """Static configuration data"""
+    """Read a document by name."""
-    return "App configuration here"
+    # This would normally read from disk
+    return f"Content of {name}"
 
 
-@mcp.resource("users://{user_id}/profile", title="User Profile")
+@mcp.resource("config://settings")
-def get_user_profile(user_id: str) -> str:
+def get_settings() -> str:
-    """Dynamic user data"""
+    """Get application settings."""
-    return f"Profile data for user {user_id}"
+    return """{
+  "theme": "dark",
+  "language": "en",
+  "debug": false
+}"""
 ```
+_Full example: [examples/snippets/servers/basic_resource.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_resource.py)_
+<!-- /snippet-source -->
 
 ### Tools
 
 Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects:
 
+<!-- snippet-source examples/snippets/servers/basic_tool.py -->
 ```python
-import httpx
 from mcp.server.fastmcp import FastMCP
 
-mcp = FastMCP("My App")
+mcp = FastMCP(name="Tool Example")
 
 
-@mcp.tool(title="BMI Calculator")
+@mcp.tool()
-def calculate_bmi(weight_kg: float, height_m: float) -> float:
+def sum(a: int, b: int) -> int:
-    """Calculate BMI given weight in kg and height in meters"""
+    """Add two numbers together."""
-    return weight_kg / (height_m**2)
+    return a + b
 
 
-@mcp.tool(title="Weather Fetcher")
+@mcp.tool()
-async def fetch_weather(city: str) -> str:
+def get_weather(city: str, unit: str = "celsius") -> str:
-    """Fetch current weather for a city"""
+    """Get weather for a city."""
-    async with httpx.AsyncClient() as client:
+    # This would normally call a weather API
-        response = await client.get(f"https://api.weather.com/{city}")
+    return f"Weather in {city}: 22degrees{unit[0].upper()}"
-        return response.text
 ```
+_Full example: [examples/snippets/servers/basic_tool.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_tool.py)_
+<!-- /snippet-source -->
 
 #### Structured Output
 
@@ -375,11 +386,12 @@ def get_temperature(city: str) -> float:
 
 Prompts are reusable templates that help LLMs interact with your server effectively:
 
+<!-- snippet-source examples/snippets/servers/basic_prompt.py -->
 ```python
 from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.prompts import base
 
-mcp = FastMCP("My App")
+mcp = FastMCP(name="Prompt Example")
 
 
 @mcp.prompt(title="Code Review")
@@ -395,6 +407,8 @@ def debug_error(error: str) -> list[base.Message]:
         base.AssistantMessage("I'll help debug that. What have you tried so far?"),
     ]
 ```
+_Full example: [examples/snippets/servers/basic_prompt.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_prompt.py)_
+<!-- /snippet-source -->
 
 ### Images
 
@@ -419,21 +433,31 @@ def create_thumbnail(image_path: str) -> Image:
 
 The Context object gives your tools and resources access to MCP capabilities:
 
+<!-- snippet-source examples/snippets/servers/tool_progress.py -->
 ```python
-from mcp.server.fastmcp import FastMCP, Context
+from mcp.server.fastmcp import Context, FastMCP
 
-mcp = FastMCP("My App")
+mcp = FastMCP(name="Progress Example")
 
 
 @mcp.tool()
-async def long_task(files: list[str], ctx: Context) -> str:
+async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
-    """Process multiple files with progress tracking"""
+    """Execute a task with progress updates."""
-    for i, file in enumerate(files):
+    await ctx.info(f"Starting: {task_name}")
-        ctx.info(f"Processing {file}")
+
-        await ctx.report_progress(i, len(files))
+    for i in range(steps):
-        data, mime_type = await ctx.read_resource(f"file://{file}")
+        progress = (i + 1) / steps
-    return "Processing complete"
+        await ctx.report_progress(
+            progress=progress,
+            total=1.0,
+            message=f"Step {i + 1}/{steps}",
+        )
+        await ctx.debug(f"Completed step {i + 1}")
+
+    return f"Task '{task_name}' completed"
 ```
+_Full example: [examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/tool_progress.py)_
+<!-- /snippet-source -->
 
 ### Completions
 
@@ -465,8 +489,10 @@ async def use_completion(session: ClientSession):
 ```
 
 Server implementation:
+
+<!-- snippet-source examples/snippets/servers/completion.py -->
 ```python
-from mcp.server import Server
+from mcp.server.fastmcp import FastMCP
 from mcp.types import (
     Completion,
     CompletionArgument,
@@ -475,72 +501,167 @@ from mcp.types import (
     ResourceTemplateReference,
 )
 
-server = Server("example-server")
+mcp = FastMCP(name="Example")
 
 
-@server.completion()
+@mcp.resource("github://repos/{owner}/{repo}")
+def github_repo(owner: str, repo: str) -> str:
+    """GitHub repository resource."""
+    return f"Repository: {owner}/{repo}"
+
+
+@mcp.prompt(description="Code review prompt")
+def review_code(language: str, code: str) -> str:
+    """Generate a code review."""
+    return f"Review this {language} code:\n{code}"
+
+
+@mcp.completion()
 async def handle_completion(
     ref: PromptReference | ResourceTemplateReference,
     argument: CompletionArgument,
     context: CompletionContext | None,
 ) -> Completion | None:
+    """Provide completions for prompts and resources."""
+
+    # Complete programming languages for the prompt
+    if isinstance(ref, PromptReference):
+        if ref.name == "review_code" and argument.name == "language":
+            languages = ["python", "javascript", "typescript", "go", "rust"]
+            return Completion(
+                values=[lang for lang in languages if lang.startswith(argument.value)],
+                hasMore=False,
+            )
+
+    # Complete repository names for GitHub resources
     if isinstance(ref, ResourceTemplateReference):
         if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
-            # Use context to provide owner-specific repos
+            if context and context.arguments and context.arguments.get("owner") == "modelcontextprotocol":
-            if context and context.arguments:
-                owner = context.arguments.get("owner")
-                if owner == "modelcontextprotocol":
                 repos = ["python-sdk", "typescript-sdk", "specification"]
-                    # Filter based on partial input
+                return Completion(values=repos, hasMore=False)
-                    filtered = [r for r in repos if r.startswith(argument.value)]
+
-                    return Completion(values=filtered)
     return None
 ```
+_Full example: [examples/snippets/servers/completion.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/completion.py)_
+<!-- /snippet-source -->
 ### Elicitation
 
 Request additional information from users during tool execution:
 
+<!-- snippet-source examples/snippets/servers/elicitation.py -->
 ```python
-from mcp.server.fastmcp import FastMCP, Context
-from mcp.server.elicitation import (
-    AcceptedElicitation,
-    DeclinedElicitation,
-    CancelledElicitation,
-)
 from pydantic import BaseModel, Field
 
-mcp = FastMCP("Booking System")
+from mcp.server.fastmcp import Context, FastMCP
+
+mcp = FastMCP(name="Elicitation Example")
+
+
+class BookingPreferences(BaseModel):
+    """Schema for collecting user preferences."""
+
+    checkAlternative: bool = Field(description="Would you like to check another date?")
+    alternativeDate: str = Field(
+        default="2024-12-26",
+        description="Alternative date (YYYY-MM-DD)",
+    )
 
 
 @mcp.tool()
-async def book_table(date: str, party_size: int, ctx: Context) -> str:
+async def book_table(
-    """Book a table with confirmation"""
+    date: str,
-
+    time: str,
-    # Schema must only contain primitive types (str, int, float, bool)
+    party_size: int,
-    class ConfirmBooking(BaseModel):
+    ctx: Context,
-        confirm: bool = Field(description="Confirm booking?")
+) -> str:
-        notes: str = Field(default="", description="Special requests")
+    """Book a table with date availability check."""
-
+    # Check if date is available
+    if date == "2024-12-25":
+        # Date unavailable - ask user for alternative
         result = await ctx.elicit(
-        message=f"Confirm booking for {party_size} on {date}?", schema=ConfirmBooking
+            message=(f"No tables available for {party_size} on {date}. Would you like to try another date?"),
+            schema=BookingPreferences,
         )
 
-    match result:
+        if result.action == "accept" and result.data:
-        case AcceptedElicitation(data=data):
+            if result.data.checkAlternative:
-            if data.confirm:
+                return f"[SUCCESS] Booked for {result.data.alternativeDate}"
-                return f"Booked! Notes: {data.notes or 'None'}"
+            return "[CANCELLED] No booking made"
-            return "Booking cancelled"
+        return "[CANCELLED] Booking cancelled"
-        case DeclinedElicitation():
+
-            return "Booking declined"
+    # Date available
-        case CancelledElicitation():
+    return f"[SUCCESS] Booked for {date} at {time}"
-            return "Booking cancelled"
 ```
+_Full example: [examples/snippets/servers/elicitation.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/elicitation.py)_
+<!-- /snippet-source -->
 
 The `elicit()` method returns an `ElicitationResult` with:
 - `action`: "accept", "decline", or "cancel"
 - `data`: The validated response (only when accepted)
 - `validation_error`: Any validation error message
 
+### Sampling
+
+Tools can interact with LLMs through sampling (generating text):
+
+<!-- snippet-source examples/snippets/servers/sampling.py -->
+```python
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.types import SamplingMessage, TextContent
+
+mcp = FastMCP(name="Sampling Example")
+
+
+@mcp.tool()
+async def generate_poem(topic: str, ctx: Context) -> str:
+    """Generate a poem using LLM sampling."""
+    prompt = f"Write a short poem about {topic}"
+
+    result = await ctx.session.create_message(
+        messages=[
+            SamplingMessage(
+                role="user",
+                content=TextContent(type="text", text=prompt),
+            )
+        ],
+        max_tokens=100,
+    )
+
+    if result.content.type == "text":
+        return result.content.text
+    return str(result.content)
+```
+_Full example: [examples/snippets/servers/sampling.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/sampling.py)_
+<!-- /snippet-source -->
+
+### Logging and Notifications
+
+Tools can send logs and notifications through the context:
+
+<!-- snippet-source examples/snippets/servers/notifications.py -->
+```python
+from mcp.server.fastmcp import Context, FastMCP
+
+mcp = FastMCP(name="Notifications Example")
+
+
+@mcp.tool()
+async def process_data(data: str, ctx: Context) -> str:
+    """Process data with logging."""
+    # Different log levels
+    await ctx.debug(f"Debug: Processing '{data}'")
+    await ctx.info("Info: Starting processing")
+    await ctx.warning("Warning: This is experimental")
+    await ctx.error("Error: (This is just a demo)")
+
+    # Notify about resource changes
+    await ctx.session.send_resource_list_changed()
+
+    return f"Processed: {data}"
+```
+_Full example: [examples/snippets/servers/notifications.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/notifications.py)_
+<!-- /snippet-source -->
+
 ### Authentication
 
 Authentication can be used by servers that want to expose tools accessing protected resources.
@@ -664,8 +785,9 @@ from mcp.server.fastmcp import FastMCP
 mcp = FastMCP(name="EchoServer", stateless_http=True)
 
 
-@mcp.tool(description="A simple echo tool")
+@mcp.tool()
 def echo(message: str) -> str:
+    """A simple echo tool"""
     return f"Echo: {message}"
 ```
 
@@ -676,8 +798,9 @@ from mcp.server.fastmcp import FastMCP
 mcp = FastMCP(name="MathServer", stateless_http=True)
 
 
-@mcp.tool(description="A simple add tool")
+@mcp.tool()
 def add_two(n: int) -> int:
+    """Tool to add two to the input"""
     return n + 2
 ```
 

examples/fastmcp/desktop.py

@@ -20,6 +20,6 @@ def desktop() -> list[str]:
 
 
 @mcp.tool()
-def add(a: int, b: int) -> int:
+def sum(a: int, b: int) -> int:
     """Add two numbers"""
     return a + b

examples/fastmcp/readme-quickstart.py

@@ -6,7 +6,7 @@ mcp = FastMCP("Demo")
 
 # Add an addition tool
 @mcp.tool()
-def add(a: int, b: int) -> int:
+def sum(a: int, b: int) -> int:
     """Add two numbers"""
     return a + b
 

examples/fastmcp/weather_structured.py

@@ -90,7 +90,7 @@ def get_weather_alerts(region: str) -> list[WeatherAlert]:
             WeatherAlert(
                 severity="high",
                 title="Heat Wave Warning",
-                description="Temperatures expected to exceed 40°C",
+                description="Temperatures expected to exceed 40 degrees",
                 affected_areas=["Los Angeles", "San Diego", "Riverside"],
                 valid_until=datetime(2024, 7, 15, 18, 0),
             ),

examples/snippets/pyproject.toml

@@ -0,0 +1,18 @@
+[project]
+name = "mcp-snippets"
+version = "0.1.0"
+description = "MCP Example Snippets"
+requires-python = ">=3.10"
+dependencies = [
+    "mcp",
+]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["servers"]
+
+[project.scripts]
+server = "servers:run_server"

examples/snippets/servers/__init__.py

@@ -0,0 +1,36 @@
+"""MCP Snippets.
+
+This package contains simple examples of MCP server features.
+Each server demonstrates a single feature and can be run as a standalone server.
+
+To run a server, use the command:
+    uv run server basic_tool sse
+"""
+
+import importlib
+import sys
+from typing import Literal, cast
+
+
+def run_server():
+    """Run a server by name with optional transport.
+
+    Usage: server <server-name> [transport]
+    Example: server basic_tool sse
+    """
+    if len(sys.argv) < 2:
+        print("Usage: server <server-name> [transport]")
+        print("Available servers: basic_tool, basic_resource, basic_prompt, tool_progress,")
+        print("                   sampling, elicitation, completion, notifications")
+        print("Available transports: stdio (default), sse, streamable-http")
+        sys.exit(1)
+
+    server_name = sys.argv[1]
+    transport = sys.argv[2] if len(sys.argv) > 2 else "stdio"
+
+    try:
+        module = importlib.import_module(f".{server_name}", package=__name__)
+        module.mcp.run(cast(Literal["stdio", "sse", "streamable-http"], transport))
+    except ImportError:
+        print(f"Error: Server '{server_name}' not found")
+        sys.exit(1)

examples/snippets/servers/basic_prompt.py

@@ -0,0 +1,18 @@
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.prompts import base
+
+mcp = FastMCP(name="Prompt Example")
+
+
+@mcp.prompt(title="Code Review")
+def review_code(code: str) -> str:
+    return f"Please review this code:\n\n{code}"
+
+
+@mcp.prompt(title="Debug Assistant")
+def debug_error(error: str) -> list[base.Message]:
+    return [
+        base.UserMessage("I'm seeing this error:"),
+        base.UserMessage(error),
+        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
+    ]

examples/snippets/servers/basic_resource.py

@@ -0,0 +1,20 @@
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP(name="Resource Example")
+
+
+@mcp.resource("file://documents/{name}")
+def read_document(name: str) -> str:
+    """Read a document by name."""
+    # This would normally read from disk
+    return f"Content of {name}"
+
+
+@mcp.resource("config://settings")
+def get_settings() -> str:
+    """Get application settings."""
+    return """{
+  "theme": "dark",
+  "language": "en",
+  "debug": false
+}"""

examples/snippets/servers/basic_tool.py

@@ -0,0 +1,16 @@
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP(name="Tool Example")
+
+
+@mcp.tool()
+def sum(a: int, b: int) -> int:
+    """Add two numbers together."""
+    return a + b
+
+
+@mcp.tool()
+def get_weather(city: str, unit: str = "celsius") -> str:
+    """Get weather for a city."""
+    # This would normally call a weather API
+    return f"Weather in {city}: 22degrees{unit[0].upper()}"

examples/snippets/servers/completion.py

@@ -0,0 +1,49 @@
+from mcp.server.fastmcp import FastMCP
+from mcp.types import (
+    Completion,
+    CompletionArgument,
+    CompletionContext,
+    PromptReference,
+    ResourceTemplateReference,
+)
+
+mcp = FastMCP(name="Example")
+
+
+@mcp.resource("github://repos/{owner}/{repo}")
+def github_repo(owner: str, repo: str) -> str:
+    """GitHub repository resource."""
+    return f"Repository: {owner}/{repo}"
+
+
+@mcp.prompt(description="Code review prompt")
+def review_code(language: str, code: str) -> str:
+    """Generate a code review."""
+    return f"Review this {language} code:\n{code}"
+
+
+@mcp.completion()
+async def handle_completion(
+    ref: PromptReference | ResourceTemplateReference,
+    argument: CompletionArgument,
+    context: CompletionContext | None,
+) -> Completion | None:
+    """Provide completions for prompts and resources."""
+
+    # Complete programming languages for the prompt
+    if isinstance(ref, PromptReference):
+        if ref.name == "review_code" and argument.name == "language":
+            languages = ["python", "javascript", "typescript", "go", "rust"]
+            return Completion(
+                values=[lang for lang in languages if lang.startswith(argument.value)],
+                hasMore=False,
+            )
+
+    # Complete repository names for GitHub resources
+    if isinstance(ref, ResourceTemplateReference):
+        if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
+            if context and context.arguments and context.arguments.get("owner") == "modelcontextprotocol":
+                repos = ["python-sdk", "typescript-sdk", "specification"]
+                return Completion(values=repos, hasMore=False)
+
+    return None

examples/snippets/servers/elicitation.py

@@ -0,0 +1,41 @@
+from pydantic import BaseModel, Field
+
+from mcp.server.fastmcp import Context, FastMCP
+
+mcp = FastMCP(name="Elicitation Example")
+
+
+class BookingPreferences(BaseModel):
+    """Schema for collecting user preferences."""
+
+    checkAlternative: bool = Field(description="Would you like to check another date?")
+    alternativeDate: str = Field(
+        default="2024-12-26",
+        description="Alternative date (YYYY-MM-DD)",
+    )
+
+
+@mcp.tool()
+async def book_table(
+    date: str,
+    time: str,
+    party_size: int,
+    ctx: Context,
+) -> str:
+    """Book a table with date availability check."""
+    # Check if date is available
+    if date == "2024-12-25":
+        # Date unavailable - ask user for alternative
+        result = await ctx.elicit(
+            message=(f"No tables available for {party_size} on {date}. Would you like to try another date?"),
+            schema=BookingPreferences,
+        )
+
+        if result.action == "accept" and result.data:
+            if result.data.checkAlternative:
+                return f"[SUCCESS] Booked for {result.data.alternativeDate}"
+            return "[CANCELLED] No booking made"
+        return "[CANCELLED] Booking cancelled"
+
+    # Date available
+    return f"[SUCCESS] Booked for {date} at {time}"

examples/snippets/servers/notifications.py

@@ -0,0 +1,18 @@
+from mcp.server.fastmcp import Context, FastMCP
+
+mcp = FastMCP(name="Notifications Example")
+
+
+@mcp.tool()
+async def process_data(data: str, ctx: Context) -> str:
+    """Process data with logging."""
+    # Different log levels
+    await ctx.debug(f"Debug: Processing '{data}'")
+    await ctx.info("Info: Starting processing")
+    await ctx.warning("Warning: This is experimental")
+    await ctx.error("Error: (This is just a demo)")
+
+    # Notify about resource changes
+    await ctx.session.send_resource_list_changed()
+
+    return f"Processed: {data}"

examples/snippets/servers/sampling.py

@@ -0,0 +1,24 @@
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.types import SamplingMessage, TextContent
+
+mcp = FastMCP(name="Sampling Example")
+
+
+@mcp.tool()
+async def generate_poem(topic: str, ctx: Context) -> str:
+    """Generate a poem using LLM sampling."""
+    prompt = f"Write a short poem about {topic}"
+
+    result = await ctx.session.create_message(
+        messages=[
+            SamplingMessage(
+                role="user",
+                content=TextContent(type="text", text=prompt),
+            )
+        ],
+        max_tokens=100,
+    )
+
+    if result.content.type == "text":
+        return result.content.text
+    return str(result.content)

examples/snippets/servers/tool_progress.py

@@ -0,0 +1,20 @@
+from mcp.server.fastmcp import Context, FastMCP
+
+mcp = FastMCP(name="Progress Example")
+
+
+@mcp.tool()
+async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
+    """Execute a task with progress updates."""
+    await ctx.info(f"Starting: {task_name}")
+
+    for i in range(steps):
+        progress = (i + 1) / steps
+        await ctx.report_progress(
+            progress=progress,
+            total=1.0,
+            message=f"Step {i + 1}/{steps}",
+        )
+        await ctx.debug(f"Completed step {i + 1}")
+
+    return f"Task '{task_name}' completed"

pyproject.toml

@@ -99,13 +99,14 @@ ignore = ["PERF203"]
 [tool.ruff]
 line-length = 120
 target-version = "py310"
+extend-exclude = ["README.md"]
 
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
 "tests/server/fastmcp/test_func_metadata.py" = ["E501"]
 
 [tool.uv.workspace]
-members = ["examples/servers/*"]
+members = ["examples/servers/*", "examples/snippets"]
 
 [tool.uv.sources]
 mcp = { workspace = true }

scripts/update_readme_snippets.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+Update README.md with live code snippets from example files.
+
+This script finds specially marked code blocks in README.md and updates them
+with the actual code from the referenced files.
+
+Usage:
+    python scripts/update_readme_snippets.py
+    python scripts/update_readme_snippets.py --check  # Check mode for CI
+"""
+
+import argparse
+import re
+import sys
+from pathlib import Path
+
+
+def get_github_url(file_path: str) -> str:
+    """Generate a GitHub URL for the file.
+
+    Args:
+        file_path: Path to the file relative to repo root
+
+    Returns:
+        GitHub URL
+    """
+    base_url = "https://github.com/modelcontextprotocol/python-sdk/blob/main"
+    return f"{base_url}/{file_path}"
+
+
+def process_snippet_block(match: re.Match, check_mode: bool = False) -> str:
+    """Process a single snippet-source block.
+
+    Args:
+        match: The regex match object
+        check_mode: If True, return original if no changes needed
+
+    Returns:
+        The updated block content
+    """
+    full_match = match.group(0)
+    indent = match.group(1)
+    file_path = match.group(2)
+
+    try:
+        # Read the entire file
+        file = Path(file_path)
+        if not file.exists():
+            print(f"Warning: File not found: {file_path}")
+            return full_match
+
+        code = file.read_text().rstrip()
+        github_url = get_github_url(file_path)
+
+        # Build the replacement block
+        indented_code = code.replace("\n", f"\n{indent}")
+        replacement = f"""{indent}<!-- snippet-source {file_path} -->
+{indent}```python
+{indent}{indented_code}
+{indent}```
+{indent}_Full example: [{file_path}]({github_url})_
+{indent}<!-- /snippet-source -->"""
+
+        # In check mode, only check if code has changed
+        if check_mode:
+            # Extract existing code from the match
+            existing_content = match.group(3)
+            if existing_content is not None:
+                existing_lines = existing_content.strip().split("\n")
+                # Find code between ```python and ```
+                code_lines = []
+                in_code = False
+                for line in existing_lines:
+                    if line.strip() == "```python":
+                        in_code = True
+                    elif line.strip() == "```":
+                        break
+                    elif in_code:
+                        code_lines.append(line)
+                existing_code = "\n".join(code_lines).strip()
+                # Compare with the indented version we would generate
+                expected_code = code.replace("\n", f"\n{indent}").strip()
+                if existing_code == expected_code:
+                    return full_match
+
+        return replacement
+
+    except Exception as e:
+        print(f"Error processing {file_path}: {e}")
+        return full_match
+
+
+def update_readme_snippets(readme_path: Path = Path("README.md"), check_mode: bool = False) -> bool:
+    """Update code snippets in README.md with live code from source files.
+
+    Args:
+        readme_path: Path to the README file
+        check_mode: If True, only check if updates are needed without modifying
+
+    Returns:
+        True if file is up to date or was updated, False if check failed
+    """
+    if not readme_path.exists():
+        print(f"Error: README file not found: {readme_path}")
+        return False
+
+    content = readme_path.read_text()
+    original_content = content
+
+    # Pattern to match snippet-source blocks
+    # Matches: <!-- snippet-source path/to/file.py -->
+    #          ... any content ...
+    #          <!-- /snippet-source -->
+    pattern = r"^(\s*)<!-- snippet-source ([^\s]+) -->\n" r"(.*?)" r"^\1<!-- /snippet-source -->"
+
+    # Process all snippet-source blocks
+    updated_content = re.sub(
+        pattern, lambda m: process_snippet_block(m, check_mode), content, flags=re.MULTILINE | re.DOTALL
+    )
+
+    if check_mode:
+        if updated_content != original_content:
+            print(
+                f"Error: {readme_path} has outdated code snippets. "
+                "Run 'python scripts/update_readme_snippets.py' to update."
+            )
+            return False
+        else:
+            print(f"✓ {readme_path} code snippets are up to date")
+            return True
+    else:
+        if updated_content != original_content:
+            readme_path.write_text(updated_content)
+            print(f"✓ Updated {readme_path}")
+        else:
+            print(f"✓ {readme_path} already up to date")
+        return True
+
+
+def main():
+    """Main entry point."""
+    parser = argparse.ArgumentParser(description="Update README code snippets from source files")
+    parser.add_argument(
+        "--check", action="store_true", help="Check mode - verify snippets are up to date without modifying"
+    )
+    parser.add_argument("--readme", default="README.md", help="Path to README file (default: README.md)")
+
+    args = parser.parse_args()
+
+    success = update_readme_snippets(Path(args.readme), check_mode=args.check)
+
+    if not success:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()