[v1.x] refactor: import SSEError from httpx_sse public API (#2561)

Co-authored-by: Owen Devereaux <owendevereaux@users.noreply.github.com>
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>

.github/actions/conformance/client.py

@@ -0,0 +1,367 @@
+"""MCP unified conformance test client.
+
+This client is designed to work with the @modelcontextprotocol/conformance npm package.
+It handles all conformance test scenarios via environment variables and CLI arguments.
+
+Contract:
+    - MCP_CONFORMANCE_SCENARIO env var -> scenario name
+    - MCP_CONFORMANCE_CONTEXT env var -> optional JSON (for client-credentials scenarios)
+    - Server URL as last CLI argument (sys.argv[1])
+    - Must exit 0 within 30 seconds
+
+Scenarios:
+    initialize                              - Connect, initialize, list tools, close
+    tools_call                              - Connect, call add_numbers(a=5, b=3), close
+    sse-retry                               - Connect, call test_reconnection, close
+    elicitation-sep1034-client-defaults     - Elicitation with default accept callback
+    auth/client-credentials-jwt             - Client credentials with private_key_jwt
+    auth/client-credentials-basic           - Client credentials with client_secret_basic
+    auth/*                                  - Authorization code flow (default for auth scenarios)
+"""
+
+import asyncio
+import json
+import logging
+import os
+import sys
+from collections.abc import Callable, Coroutine
+from typing import Any, cast
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+from pydantic import AnyUrl
+
+from mcp import ClientSession, types
+from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.auth.extensions.client_credentials import (
+    ClientCredentialsOAuthProvider,
+    PrivateKeyJWTOAuthProvider,
+    SignedJWTParameters,
+)
+from mcp.client.streamable_http import streamable_http_client
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+from mcp.shared.context import RequestContext
+
+# Set up logging to stderr (stdout is for conformance test output)
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    stream=sys.stderr,
+)
+logger = logging.getLogger(__name__)
+
+# Type for async scenario handler functions
+ScenarioHandler = Callable[[str], Coroutine[Any, None, None]]
+
+# Registry of scenario handlers
+HANDLERS: dict[str, ScenarioHandler] = {}
+
+
+def register(name: str) -> Callable[[ScenarioHandler], ScenarioHandler]:
+    """Register a scenario handler."""
+
+    def decorator(fn: ScenarioHandler) -> ScenarioHandler:
+        HANDLERS[name] = fn
+        return fn
+
+    return decorator
+
+
+def get_conformance_context() -> dict[str, Any]:
+    """Load conformance test context from MCP_CONFORMANCE_CONTEXT environment variable."""
+    context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT")
+    if not context_json:
+        raise RuntimeError(
+            "MCP_CONFORMANCE_CONTEXT environment variable not set. "
+            "Expected JSON with client_id, client_secret, and/or private_key_pem."
+        )
+    try:
+        return json.loads(context_json)
+    except json.JSONDecodeError as e:
+        raise RuntimeError(f"Failed to parse MCP_CONFORMANCE_CONTEXT as JSON: {e}") from e
+
+
+class InMemoryTokenStorage(TokenStorage):
+    """Simple in-memory token storage for conformance testing."""
+
+    def __init__(self) -> None:
+        self._tokens: OAuthToken | None = None
+        self._client_info: OAuthClientInformationFull | None = None
+
+    async def get_tokens(self) -> OAuthToken | None:
+        return self._tokens
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        self._tokens = tokens
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        return self._client_info
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        self._client_info = client_info
+
+
+class ConformanceOAuthCallbackHandler:
+    """OAuth callback handler that automatically fetches the authorization URL
+    and extracts the auth code, without requiring user interaction.
+    """
+
+    def __init__(self) -> None:
+        self._auth_code: str | None = None
+        self._state: str | None = None
+
+    async def handle_redirect(self, authorization_url: str) -> None:
+        """Fetch the authorization URL and extract the auth code from the redirect."""
+        logger.debug(f"Fetching authorization URL: {authorization_url}")
+
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                authorization_url,
+                follow_redirects=False,
+            )
+
+            if response.status_code in (301, 302, 303, 307, 308):
+                location = cast(str, response.headers.get("location"))
+                if location:
+                    redirect_url = urlparse(location)
+                    query_params: dict[str, list[str]] = parse_qs(redirect_url.query)
+
+                    if "code" in query_params:
+                        self._auth_code = query_params["code"][0]
+                        state_values = query_params.get("state")
+                        self._state = state_values[0] if state_values else None
+                        logger.debug(f"Got auth code from redirect: {self._auth_code[:10]}...")
+                        return
+                    else:
+                        raise RuntimeError(f"No auth code in redirect URL: {location}")
+                else:
+                    raise RuntimeError(f"No redirect location received from {authorization_url}")
+            else:
+                raise RuntimeError(f"Expected redirect response, got {response.status_code} from {authorization_url}")
+
+    async def handle_callback(self) -> tuple[str, str | None]:
+        """Return the captured auth code and state."""
+        if self._auth_code is None:
+            raise RuntimeError("No authorization code available - was handle_redirect called?")
+        auth_code = self._auth_code
+        state = self._state
+        self._auth_code = None
+        self._state = None
+        return auth_code, state
+
+
+# --- Scenario Handlers ---
+
+
+@register("initialize")
+async def run_initialize(server_url: str) -> None:
+    """Connect, initialize, list tools, close."""
+    async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            logger.debug("Initialized successfully")
+            await session.list_tools()
+            logger.debug("Listed tools successfully")
+
+
+@register("tools_call")
+async def run_tools_call(server_url: str) -> None:
+    """Connect, initialize, list tools, call add_numbers(a=5, b=3), close."""
+    async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            await session.list_tools()
+            result = await session.call_tool("add_numbers", {"a": 5, "b": 3})
+            logger.debug(f"add_numbers result: {result}")
+
+
+@register("sse-retry")
+async def run_sse_retry(server_url: str) -> None:
+    """Connect, initialize, list tools, call test_reconnection, close."""
+    async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            await session.list_tools()
+            result = await session.call_tool("test_reconnection", {})
+            logger.debug(f"test_reconnection result: {result}")
+
+
+async def default_elicitation_callback(
+    context: RequestContext["ClientSession", Any],
+    params: types.ElicitRequestParams,
+) -> types.ElicitResult | types.ErrorData:
+    """Accept elicitation and apply defaults from the schema (SEP-1034)."""
+    content: dict[str, str | int | float | bool | list[str] | None] = {}
+
+    # For form mode, extract defaults from the requested_schema
+    if isinstance(params, types.ElicitRequestFormParams):
+        schema = params.requestedSchema
+        logger.debug(f"Elicitation schema: {schema}")
+        properties = schema.get("properties", {})
+        for prop_name, prop_schema in properties.items():
+            if "default" in prop_schema:
+                content[prop_name] = prop_schema["default"]
+        logger.debug(f"Applied defaults: {content}")
+
+    return types.ElicitResult(action="accept", content=content)
+
+
+@register("elicitation-sep1034-client-defaults")
+async def run_elicitation_defaults(server_url: str) -> None:
+    """Connect with elicitation callback that applies schema defaults."""
+    async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
+        async with ClientSession(
+            read_stream, write_stream, elicitation_callback=default_elicitation_callback
+        ) as session:
+            await session.initialize()
+            await session.list_tools()
+            result = await session.call_tool("test_client_elicitation_defaults", {})
+            logger.debug(f"test_client_elicitation_defaults result: {result}")
+
+
+@register("auth/client-credentials-jwt")
+async def run_client_credentials_jwt(server_url: str) -> None:
+    """Client credentials flow with private_key_jwt authentication."""
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    private_key_pem = context.get("private_key_pem")
+    signing_algorithm = context.get("signing_algorithm", "ES256")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not private_key_pem:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'private_key_pem'")
+
+    jwt_params = SignedJWTParameters(
+        issuer=client_id,
+        subject=client_id,
+        signing_algorithm=signing_algorithm,
+        signing_key=private_key_pem,
+    )
+
+    oauth_auth = PrivateKeyJWTOAuthProvider(
+        server_url=server_url,
+        storage=InMemoryTokenStorage(),
+        client_id=client_id,
+        assertion_provider=jwt_params.create_assertion_provider(),
+    )
+
+    await _run_auth_session(server_url, oauth_auth)
+
+
+@register("auth/client-credentials-basic")
+async def run_client_credentials_basic(server_url: str) -> None:
+    """Client credentials flow with client_secret_basic authentication."""
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    client_secret = context.get("client_secret")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not client_secret:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_secret'")
+
+    oauth_auth = ClientCredentialsOAuthProvider(
+        server_url=server_url,
+        storage=InMemoryTokenStorage(),
+        client_id=client_id,
+        client_secret=client_secret,
+        token_endpoint_auth_method="client_secret_basic",
+    )
+
+    await _run_auth_session(server_url, oauth_auth)
+
+
+async def run_auth_code_client(server_url: str) -> None:
+    """Authorization code flow (default for auth/* scenarios)."""
+    callback_handler = ConformanceOAuthCallbackHandler()
+    storage = InMemoryTokenStorage()
+
+    # Check for pre-registered client credentials from context
+    context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT")
+    if context_json:
+        try:
+            context = json.loads(context_json)
+            client_id = context.get("client_id")
+            client_secret = context.get("client_secret")
+            if client_id:
+                await storage.set_client_info(
+                    OAuthClientInformationFull(
+                        client_id=client_id,
+                        client_secret=client_secret,
+                        redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+                        token_endpoint_auth_method="client_secret_basic" if client_secret else "none",
+                    )
+                )
+                logger.debug(f"Pre-loaded client credentials: client_id={client_id}")
+        except json.JSONDecodeError:
+            logger.exception("Failed to parse MCP_CONFORMANCE_CONTEXT")
+
+    oauth_auth = OAuthClientProvider(
+        server_url=server_url,
+        client_metadata=OAuthClientMetadata(
+            client_name="conformance-client",
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+            grant_types=["authorization_code", "refresh_token"],
+            response_types=["code"],
+        ),
+        storage=storage,
+        redirect_handler=callback_handler.handle_redirect,
+        callback_handler=callback_handler.handle_callback,
+        client_metadata_url="https://conformance-test.local/client-metadata.json",
+    )
+
+    await _run_auth_session(server_url, oauth_auth)
+
+
+async def _run_auth_session(server_url: str, oauth_auth: OAuthClientProvider) -> None:
+    """Common session logic for all OAuth flows."""
+    client = httpx.AsyncClient(auth=oauth_auth, timeout=30.0)
+    async with streamable_http_client(url=server_url, http_client=client) as (read_stream, write_stream, _):
+        async with ClientSession(
+            read_stream, write_stream, elicitation_callback=default_elicitation_callback
+        ) as session:
+            await session.initialize()
+            logger.debug("Initialized successfully")
+
+            tools_result = await session.list_tools()
+            logger.debug(f"Listed tools: {[t.name for t in tools_result.tools]}")
+
+            # Call the first available tool (different tests have different tools)
+            if tools_result.tools:
+                tool_name = tools_result.tools[0].name
+                try:
+                    result = await session.call_tool(tool_name, {})
+                    logger.debug(f"Called {tool_name}, result: {result}")
+                except Exception as e:
+                    logger.debug(f"Tool call result/error: {e}")
+
+    logger.debug("Connection closed successfully")
+
+
+def main() -> None:
+    """Main entry point for the conformance client."""
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)
+        sys.exit(1)
+
+    server_url = sys.argv[1]
+    scenario = os.environ.get("MCP_CONFORMANCE_SCENARIO")
+
+    if scenario:
+        logger.debug(f"Running explicit scenario '{scenario}' against {server_url}")
+        handler = HANDLERS.get(scenario)
+        if handler:
+            asyncio.run(handler(server_url))
+        elif scenario.startswith("auth/"):
+            asyncio.run(run_auth_code_client(server_url))
+        else:
+            print(f"Unknown scenario: {scenario}", file=sys.stderr)
+            sys.exit(1)
+    else:
+        logger.debug(f"Running default auth flow against {server_url}")
+        asyncio.run(run_auth_code_client(server_url))
+
+
+if __name__ == "__main__":
+    main()

.github/actions/conformance/expected-failures.yml

@@ -0,0 +1,4 @@
+# Known conformance test failures for v1.x
+# These are tracked and should be removed as they're fixed.
+server: []
+client: []

.github/actions/conformance/run-server.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+set -e
+
+PORT="${PORT:-3001}"
+SERVER_URL="http://localhost:${PORT}/mcp"
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+cd "$SCRIPT_DIR/../../.."
+
+# Start everything-server
+uv run --frozen mcp-everything-server --port "$PORT" &
+SERVER_PID=$!
+trap "kill $SERVER_PID 2>/dev/null || true; wait $SERVER_PID 2>/dev/null || true" EXIT
+
+# Wait for server to be ready
+MAX_RETRIES=30
+RETRY_COUNT=0
+while ! curl -s "$SERVER_URL" > /dev/null 2>&1; do
+    RETRY_COUNT=$((RETRY_COUNT + 1))
+    if [ $RETRY_COUNT -ge $MAX_RETRIES ]; then
+        echo "Server failed to start after ${MAX_RETRIES} retries" >&2
+        exit 1
+    fi
+    sleep 0.5
+done
+
+echo "Server ready at $SERVER_URL"
+
+# Run conformance tests
+npx @modelcontextprotocol/conformance@0.1.13 server --url "$SERVER_URL" "$@"

.github/dependabot.yml

@@ -0,0 +1,10 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: monthly
+    groups:
+      github-actions:
+        patterns:
+          - "*"

.github/workflows/conformance.yml

@@ -0,0 +1,44 @@
+name: Conformance Tests
+
+on:
+  push:
+    branches: [v1.x]
+  pull_request:
+    branches: [v1.x]
+  workflow_dispatch:
+
+concurrency:
+  group: conformance-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  server-conformance:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - uses: astral-sh/setup-uv@803947b9bd8e9f986429fa0c5a41c367cd732b41 # v7.2.1
+        with:
+          enable-cache: true
+          version: 0.9.5
+      - uses: actions/setup-node@6044e13b5dc448c55e2357c09f80417699197238 # v6.2.0
+        with:
+          node-version: 24
+      - run: uv sync --frozen --all-extras --package mcp-everything-server
+      - run: ./.github/actions/conformance/run-server.sh
+
+  client-conformance:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - uses: astral-sh/setup-uv@803947b9bd8e9f986429fa0c5a41c367cd732b41 # v7.2.1
+        with:
+          enable-cache: true
+          version: 0.9.5
+      - uses: actions/setup-node@6044e13b5dc448c55e2357c09f80417699197238 # v6.2.0
+        with:
+          node-version: 24
+      - run: uv sync --frozen --all-extras --package mcp
+      - run: npx @modelcontextprotocol/conformance@0.1.13 client --command 'uv run --frozen python .github/actions/conformance/client.py' --suite all --expected-failures .github/actions/conformance/expected-failures.yml

.github/workflows/pull-request-checks.yml

@@ -6,3 +6,12 @@ on:
 jobs:
   checks:
     uses: ./.github/workflows/shared.yml
+
+  all-green:
+    if: always()
+    needs: [checks]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: re-actors/alls-green@05ac9388f0aebcb5727afa17fcccfecd6f8ec5fe # v1.2.2
+        with:
+          jobs: ${{ toJSON(needs) }}

.github/workflows/shared.yml

@@ -61,7 +61,7 @@ jobs:
           uv run --frozen --no-sync coverage combine
           uv run --frozen --no-sync coverage report
 
-  readme-snippets:
+  doc-snippets:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v5
@@ -74,5 +74,5 @@ jobs:
       - name: Install dependencies
         run: uv sync --frozen --all-extras --python 3.10
 
-      - name: Check README snippets are up to date
+      - name: Check doc snippets are up to date
-        run: uv run --frozen scripts/update_readme_snippets.py --check
+        run: uv run --frozen scripts/update_doc_snippets.py --check

.github/workflows/weekly-lockfile-update.yml

@@ -0,0 +1,40 @@
+name: Weekly Lockfile Update
+
+on:
+  workflow_dispatch:
+  schedule:
+    # Every Thursday at 8:00 UTC
+    - cron: "0 8 * * 4"
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  update-lockfile:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
+
+      - uses: astral-sh/setup-uv@803947b9bd8e9f986429fa0c5a41c367cd732b41 # v7.2.1
+        with:
+          version: 0.9.5
+
+      - name: Update lockfile
+        run: |
+          echo '## Updated Dependencies' > pr_body.md
+          echo '' >> pr_body.md
+          echo '```' >> pr_body.md
+          uv lock --upgrade 2>&1 | tee -a pr_body.md
+          echo '```' >> pr_body.md
+
+      - name: Create pull request
+        uses: peter-evans/create-pull-request@c0f553fe549906ede9cf27b5156039d195d2ece0 # v7
+        with:
+          commit-message: "chore: update uv.lock with latest dependencies"
+          title: "chore: weekly dependency update"
+          body-path: pr_body.md
+          branch: weekly-lockfile-update-v1x
+          delete-branch: true
+          add-paths: uv.lock
+          labels: dependencies

.pre-commit-config.yaml

@@ -55,9 +55,9 @@ repos:
         language: system
         files: ^(pyproject\.toml|uv\.lock)$
         pass_filenames: false
-      - id: readme-snippets
+      - id: doc-snippets
-        name: Check README snippets are up to date
+        name: Check doc snippets are up to date
-        entry: uv run --frozen python scripts/update_readme_snippets.py --check
+        entry: uv run --frozen python scripts/update_doc_snippets.py --check
         language: system
-        files: ^(README\.md|examples/.*\.py|scripts/update_readme_snippets\.py)$
+        files: ^(README\.md|docs/.*\.md|examples/.*\.py|scripts/update_doc_snippets\.py)$
         pass_filenames: false

CLAUDE.md

@@ -24,6 +24,11 @@ This document contains critical information about working with this codebase. Fo
    - Coverage: test edge cases and errors
    - New features require tests
    - Bug fixes require regression tests
+   - Avoid `anyio.sleep()` with a fixed duration to wait for async operations. Instead:
+     - Use `anyio.Event` — set it in the callback/handler, `await event.wait()` in the test
+     - For stream messages, use `await stream.receive()` instead of `sleep()` + `receive_nowait()`
+     - Exception: `sleep()` is appropriate when testing time-based features (e.g., timeouts)
+   - Wrap indefinite waits (`event.wait()`, `stream.receive()`) in `anyio.fail_after(5)` to prevent hangs
 
 - For commits fixing bugs or adding features based on user reports add:
 

CONTRIBUTING.md

@@ -64,6 +64,28 @@ pre-commit run --all-files
 
 9. Submit a pull request to the same branch you branched from
 
+## Dependency Update Policy
+
+See [DEPENDENCY_POLICY.md](DEPENDENCY_POLICY.md) for the full dependency update policy.
+
+When bumping a dependency version manually, update the constraint in `pyproject.toml` then run `uv lock --resolution lowest-direct` (see [RELEASE.md](RELEASE.md)).
+
+Security-relevant dependency updates (P0) are applied within 7 days of public disclosure and backported to active release branches.
+
+The SDK currently supports Python 3.10 through 3.13. New CPython releases are supported within one minor SDK release of their stable release date.
+
+## Triage Process
+
+New issues are triaged by a maintainer within 2 business days. Triage means adding an appropriate label and determining whether the issue is valid.
+
+Issues are labeled per the [SDK Tiering System](https://modelcontextprotocol.io/community/sdk-tiers):
+
+- **Type** (pick one): `bug`, `enhancement`, `question`
+- **Status** (pick one): `needs confirmation`, `needs repro`, `ready for work`, `good first issue`, `help wanted`
+- **Priority** (if actionable): `P0`, `P1`, `P2`, `P3`
+
+P0 issues are security vulnerabilities (CVSS ≥ 7.0) or core functionality failures that prevent basic MCP operations (connection establishment, message exchange, or use of core primitives). P0 issues must be resolved within 7 days.
+
 ## Code Style
 
 - We use `ruff` for linting and formatting

DEPENDENCY_POLICY.md

@@ -0,0 +1,30 @@
+# Dependency Policy
+
+As a library consumed by downstream projects, the MCP Python SDK takes a conservative approach to dependency updates. Dependencies are kept stable unless there is a specific reason to update, such as a security vulnerability, a bug fix, or a need for new functionality.
+
+## Update Triggers
+
+Dependencies are updated when:
+
+- A **security vulnerability** is disclosed (via GitHub security alerts or PyPI advisories) in a dependency that directly affects the SDK's functionality or its consumers.
+- A bug in a dependency directly affects the SDK.
+- A new dependency feature is needed for SDK development.
+- A dependency drops support for a Python version the SDK still targets.
+
+Routine version bumps without a clear motivation are avoided to minimize churn for downstream consumers.
+
+## What We Don't Do
+
+The SDK does not run ad-hoc version bumps for PyPI dependencies. Updating a dependency can force downstream consumers to adopt that update transitively, which can be disruptive for projects with strict dependency policies.
+
+Dependencies are only updated when there is a concrete reason, not simply because a newer version is available.
+
+## Automated Tooling
+
+- **Lockfile refresh**: The lockfile is updated automatically every Thursday at 08:00 UTC by the [`weekly-lockfile-update.yml`](.github/workflows/weekly-lockfile-update.yml) workflow, which runs `uv lock --upgrade` and opens a PR. This does not alter the minimum or maximum versions for dependencies of the `mcp` package itself.
+- **GitHub security updates** are enabled at the repository level and automatically open pull requests for packages with known vulnerabilities. This is a GitHub repo setting, separate from the `dependabot.yml` configuration.
+- **GitHub Actions versions** are kept up to date via Dependabot on a monthly schedule (see `.github/dependabot.yml`).
+
+## Pinning and Ranges
+
+Production dependencies use compatible-release specifiers (`~=`) or lower-bound constraints (`>=`) to allow compatible updates. Exact versions are pinned only when necessary to work around a specific issue. The lockfile (`uv.lock`) records exact resolved versions for reproducible installs.

ROADMAP.md

@@ -0,0 +1,22 @@
+# Roadmap
+
+## Spec Implementation Tracking
+
+The SDK tracks implementation of MCP spec components via GitHub Projects, with a dedicated project board for each spec revision. For example, see the [2025-11-25 spec revision board](https://github.com/orgs/modelcontextprotocol/projects/26).
+
+## Current Focus Areas
+
+### Next Spec Revision
+
+The next MCP specification revision is being developed in the [protocol repository](https://github.com/modelcontextprotocol/modelcontextprotocol). Key areas expected in the next revision include extensions and stateless transports.
+
+The SDK has historically implemented spec changes promptly as they are finalized, with dedicated project boards tracking component-level progress for each revision.
+
+### v2
+
+A major version of the SDK is in active development, tracked via [GitHub Project](https://github.com/orgs/modelcontextprotocol/projects/31). Target milestones:
+
+- **Alpha**: ~mid-March 2026
+- **Beta**: ~May 2026
+
+The v2 release is planned to align with the next spec release, expected around mid-2026.

VERSIONING.md

@@ -0,0 +1,40 @@
+# Versioning Policy
+
+The MCP Python SDK (`mcp`) follows [Semantic Versioning 2.0.0](https://semver.org/).
+
+## Version Format
+
+`MAJOR.MINOR.PATCH`
+
+- **MAJOR**: Incremented for breaking changes (see below).
+- **MINOR**: Incremented for new features that are backward-compatible.
+- **PATCH**: Incremented for backward-compatible bug fixes.
+
+## What Constitutes a Breaking Change
+
+The following changes are considered breaking and require a major version bump:
+
+- Removing or renaming a public API export (class, function, type, or constant).
+- Changing the signature of a public function or method in a way that breaks existing callers (removing parameters, changing required/optional status, changing types).
+- Removing or renaming a public type or dataclass/TypedDict field.
+- Changing the behavior of an existing API in a way that breaks documented contracts.
+- Dropping support for a Python version that is still receiving security updates.
+- Removing support for a transport type.
+- Changes to the MCP protocol version that require client/server code changes.
+
+The following are **not** considered breaking:
+
+- Adding new optional parameters to existing functions.
+- Adding new exports, types, or classes.
+- Adding new optional fields to existing types.
+- Bug fixes that correct behavior to match documented intent.
+- Internal refactoring that does not affect the public API.
+- Adding support for new MCP spec features.
+- Changes to dev dependencies or build tooling.
+
+## How Breaking Changes Are Communicated
+
+1. **Changelog**: All breaking changes are documented in the GitHub release notes with migration instructions.
+2. **Deprecation**: When feasible, APIs are deprecated for at least one minor release before removal using `warnings.warn()` with `DeprecationWarning`, which surfaces warnings at runtime and through static analysis tooling.
+3. **Migration guide**: Major version releases include a migration guide describing what changed and how to update.
+4. **PR labels**: Pull requests containing breaking changes are labeled with `breaking change`.

docs/authorization.md

@@ -1,5 +1,177 @@
 # Authorization
 
-!!! warning "Under Construction"
+This page covers OAuth 2.1 authentication for both MCP servers and clients.
 
-    This page is currently being written. Check back soon for complete documentation.
+## Server-Side Authentication
+
+Authentication can be used by servers that want to expose tools accessing protected resources.
+
+`mcp.server.auth` implements OAuth 2.1 resource server functionality, where MCP servers act as Resource Servers (RS) that validate tokens issued by separate Authorization Servers (AS). This follows the [MCP authorization specification](https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization) and implements RFC 9728 (Protected Resource Metadata) for AS discovery.
+
+MCP servers can use authentication by providing an implementation of the `TokenVerifier` protocol:
+
+<!-- snippet-source examples/snippets/servers/oauth_server.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/oauth_server.py
+"""
+
+from pydantic import AnyHttpUrl
+
+from mcp.server.auth.provider import AccessToken, TokenVerifier
+from mcp.server.auth.settings import AuthSettings
+from mcp.server.fastmcp import FastMCP
+
+
+class SimpleTokenVerifier(TokenVerifier):
+    """Simple token verifier for demonstration."""
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        pass  # This is where you would implement actual token validation
+
+
+# Create FastMCP instance as a Resource Server
+mcp = FastMCP(
+    "Weather Service",
+    json_response=True,
+    # Token verifier for authentication
+    token_verifier=SimpleTokenVerifier(),
+    # Auth settings for RFC 9728 Protected Resource Metadata
+    auth=AuthSettings(
+        issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
+        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
+        required_scopes=["user"],
+    ),
+)
+
+
+@mcp.tool()
+async def get_weather(city: str = "London") -> dict[str, str]:
+    """Get weather data for a city"""
+    return {
+        "city": city,
+        "temperature": "22",
+        "condition": "Partly cloudy",
+        "humidity": "65%",
+    }
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")
+```
+
+_Full example: [examples/snippets/servers/oauth_server.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/oauth_server.py)_
+<!-- /snippet-source -->
+
+For a complete example with separate Authorization Server and Resource Server implementations, see [`examples/servers/simple-auth/`](https://github.com/modelcontextprotocol/python-sdk/tree/v1.x/examples/servers/simple-auth).
+
+**Architecture:**
+
+- **Authorization Server (AS)**: Handles OAuth flows, user authentication, and token issuance
+- **Resource Server (RS)**: Your MCP server that validates tokens and serves protected resources
+- **Client**: Discovers AS through RFC 9728, obtains tokens, and uses them with the MCP server
+
+See [TokenVerifier](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/src/mcp/server/auth/provider.py) for more details on implementing token validation.
+
+## Client-Side Authentication
+
+The SDK includes [authorization support](https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization) for connecting to protected MCP servers:
+
+<!-- snippet-source examples/snippets/clients/oauth_client.py -->
+```python
+"""
+Before running, specify running MCP RS server URL.
+To spin up RS server locally, see
+    examples/servers/simple-auth/README.md
+
+cd to the `examples/snippets` directory and run:
+    uv run oauth-client
+"""
+
+import asyncio
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+from pydantic import AnyUrl
+
+from mcp import ClientSession
+from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.streamable_http import streamable_http_client
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+
+
+class InMemoryTokenStorage(TokenStorage):
+    """Demo In-memory token storage implementation."""
+
+    def __init__(self):
+        self.tokens: OAuthToken | None = None
+        self.client_info: OAuthClientInformationFull | None = None
+
+    async def get_tokens(self) -> OAuthToken | None:
+        """Get stored tokens."""
+        return self.tokens
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        """Store tokens."""
+        self.tokens = tokens
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        """Get stored client information."""
+        return self.client_info
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        """Store client information."""
+        self.client_info = client_info
+
+
+async def handle_redirect(auth_url: str) -> None:
+    print(f"Visit: {auth_url}")
+
+
+async def handle_callback() -> tuple[str, str | None]:
+    callback_url = input("Paste callback URL: ")
+    params = parse_qs(urlparse(callback_url).query)
+    return params["code"][0], params.get("state", [None])[0]
+
+
+async def main():
+    """Run the OAuth client example."""
+    oauth_auth = OAuthClientProvider(
+        server_url="http://localhost:8001",
+        client_metadata=OAuthClientMetadata(
+            client_name="Example MCP Client",
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+            grant_types=["authorization_code", "refresh_token"],
+            response_types=["code"],
+            scope="user",
+        ),
+        storage=InMemoryTokenStorage(),
+        redirect_handler=handle_redirect,
+        callback_handler=handle_callback,
+    )
+
+    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+        async with streamable_http_client("http://localhost:8001/mcp", http_client=custom_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+
+                tools = await session.list_tools()
+                print(f"Available tools: {[tool.name for tool in tools.tools]}")
+
+                resources = await session.list_resources()
+                print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+
+def run():
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    run()
+```
+
+_Full example: [examples/snippets/clients/oauth_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/oauth_client.py)_
+<!-- /snippet-source -->
+
+For a complete working example, see [`examples/clients/simple-auth-client/`](https://github.com/modelcontextprotocol/python-sdk/tree/v1.x/examples/clients/simple-auth-client).

docs/client.md

@@ -0,0 +1,410 @@
+# Writing MCP Clients
+
+The SDK provides a high-level client interface for connecting to MCP servers using various [transports](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports):
+
+<!-- snippet-source examples/snippets/clients/stdio_client.py -->
+```python
+"""
+cd to the `examples/snippets/clients` directory and run:
+    uv run client
+"""
+
+import asyncio
+import os
+
+from pydantic import AnyUrl
+
+from mcp import ClientSession, StdioServerParameters, types
+from mcp.client.stdio import stdio_client
+from mcp.shared.context import RequestContext
+
+# Create server parameters for stdio connection
+server_params = StdioServerParameters(
+    command="uv",  # Using uv to run the server
+    args=["run", "server", "fastmcp_quickstart", "stdio"],  # We're already in snippets dir
+    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
+)
+
+
+# Optional: create a sampling callback
+async def handle_sampling_message(
+    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
+) -> types.CreateMessageResult:
+    print(f"Sampling request: {params.messages}")
+    return types.CreateMessageResult(
+        role="assistant",
+        content=types.TextContent(
+            type="text",
+            text="Hello, world! from model",
+        ),
+        model="gpt-3.5-turbo",
+        stopReason="endTurn",
+    )
+
+
+async def run():
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
+            # Initialize the connection
+            await session.initialize()
+
+            # List available prompts
+            prompts = await session.list_prompts()
+            print(f"Available prompts: {[p.name for p in prompts.prompts]}")
+
+            # Get a prompt (greet_user prompt from fastmcp_quickstart)
+            if prompts.prompts:
+                prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"})
+                print(f"Prompt result: {prompt.messages[0].content}")
+
+            # List available resources
+            resources = await session.list_resources()
+            print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+            # List available tools
+            tools = await session.list_tools()
+            print(f"Available tools: {[t.name for t in tools.tools]}")
+
+            # Read a resource (greeting resource from fastmcp_quickstart)
+            resource_content = await session.read_resource(AnyUrl("greeting://World"))
+            content_block = resource_content.contents[0]
+            if isinstance(content_block, types.TextResourceContents):
+                print(f"Resource content: {content_block.text}")
+
+            # Call a tool (add tool from fastmcp_quickstart)
+            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
+            result_unstructured = result.content[0]
+            if isinstance(result_unstructured, types.TextContent):
+                print(f"Tool result: {result_unstructured.text}")
+            result_structured = result.structuredContent
+            print(f"Structured tool result: {result_structured}")
+
+
+def main():
+    """Entry point for the client script."""
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    main()
+```
+
+_Full example: [examples/snippets/clients/stdio_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/stdio_client.py)_
+<!-- /snippet-source -->
+
+Clients can also connect using [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#streamable-http):
+
+<!-- snippet-source examples/snippets/clients/streamable_basic.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/clients/streamable_basic.py
+"""
+
+import asyncio
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+
+async def main():
+    # Connect to a streamable HTTP server
+    async with streamable_http_client("http://localhost:8000/mcp") as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        # Create a session using the client streams
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the connection
+            await session.initialize()
+            # List available tools
+            tools = await session.list_tools()
+            print(f"Available tools: {[tool.name for tool in tools.tools]}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+_Full example: [examples/snippets/clients/streamable_basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/streamable_basic.py)_
+<!-- /snippet-source -->
+
+## Client Display Utilities
+
+When building MCP clients, the SDK provides utilities to help display human-readable names for tools, resources, and prompts:
+
+<!-- snippet-source examples/snippets/clients/display_utilities.py -->
+```python
+"""
+cd to the `examples/snippets` directory and run:
+    uv run display-utilities-client
+"""
+
+import asyncio
+import os
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from mcp.shared.metadata_utils import get_display_name
+
+# Create server parameters for stdio connection
+server_params = StdioServerParameters(
+    command="uv",  # Using uv to run the server
+    args=["run", "server", "fastmcp_quickstart", "stdio"],
+    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
+)
+
+
+async def display_tools(session: ClientSession):
+    """Display available tools with human-readable names"""
+    tools_response = await session.list_tools()
+
+    for tool in tools_response.tools:
+        # get_display_name() returns the title if available, otherwise the name
+        display_name = get_display_name(tool)
+        print(f"Tool: {display_name}")
+        if tool.description:
+            print(f"   {tool.description}")
+
+
+async def display_resources(session: ClientSession):
+    """Display available resources with human-readable names"""
+    resources_response = await session.list_resources()
+
+    for resource in resources_response.resources:
+        display_name = get_display_name(resource)
+        print(f"Resource: {display_name} ({resource.uri})")
+
+    templates_response = await session.list_resource_templates()
+    for template in templates_response.resourceTemplates:
+        display_name = get_display_name(template)
+        print(f"Resource Template: {display_name}")
+
+
+async def run():
+    """Run the display utilities example."""
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            # Initialize the connection
+            await session.initialize()
+
+            print("=== Available Tools ===")
+            await display_tools(session)
+
+            print("\n=== Available Resources ===")
+            await display_resources(session)
+
+
+def main():
+    """Entry point for the display utilities client."""
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    main()
+```
+
+_Full example: [examples/snippets/clients/display_utilities.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/display_utilities.py)_
+<!-- /snippet-source -->
+
+The `get_display_name()` function implements the proper precedence rules for displaying names:
+
+- For tools: `title` > `annotations.title` > `name`
+- For other objects: `title` > `name`
+
+This ensures your client UI shows the most user-friendly names that servers provide.
+
+## OAuth Authentication
+
+For OAuth 2.1 client authentication, see [Authorization](authorization.md#client-side-authentication).
+
+## Roots
+
+### Listing Roots
+
+Clients can provide a `list_roots_callback` so that servers can discover the client's workspace roots (directories, project folders, etc.):
+
+<!-- snippet-source examples/snippets/clients/roots_example.py -->
+```python
+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,
+)
+```
+
+_Full example: [examples/snippets/clients/roots_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/roots_example.py)_
+<!-- /snippet-source -->
+
+When a `list_roots_callback` is provided, the client automatically declares the `roots` capability (with `listChanged=True`) during initialization.
+
+### Roots Change Notifications
+
+When the client's workspace roots change (e.g., a folder is added or removed), notify the server:
+
+```python
+# After roots change, notify the server
+await session.send_roots_list_changed()
+```
+
+## SSE Transport (Legacy)
+
+For servers that use the older SSE transport, use `sse_client()` from `mcp.client.sse`:
+
+<!-- snippet-source examples/snippets/clients/sse_client.py -->
+```python
+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())
+```
+
+_Full example: [examples/snippets/clients/sse_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/sse_client.py)_
+<!-- /snippet-source -->
+
+The `sse_client()` function accepts optional `headers`, `timeout`, `sse_read_timeout`, and `auth` parameters. The SSE transport is considered legacy; prefer [Streamable HTTP](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#streamable-http) for new servers.
+
+## Ping
+
+Send a ping to verify the server is responsive:
+
+```python
+# After session.initialize()
+result = await session.send_ping()
+# Returns EmptyResult on success; raises on timeout
+```
+
+## Logging
+
+### Receiving Log Messages
+
+Pass a `logging_callback` to receive log messages from the server:
+
+<!-- snippet-source examples/snippets/clients/logging_client.py -->
+```python
+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,
+)
+```
+
+_Full example: [examples/snippets/clients/logging_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/logging_client.py)_
+<!-- /snippet-source -->
+
+### Setting the Server Log Level
+
+Request that the server change its minimum log level:
+
+```python
+await session.set_logging_level("debug")
+```
+
+The `level` parameter is a `LoggingLevel` string: `"debug"`, `"info"`, `"notice"`, `"warning"`, `"error"`, `"critical"`, `"alert"`, or `"emergency"`.
+
+## Parsing Tool Results
+
+When calling tools through MCP, the `CallToolResult` object contains the tool's response in a structured format. Understanding how to parse this result is essential for properly handling tool outputs.
+
+```python
+"""examples/snippets/clients/parsing_tool_results.py"""
+
+import asyncio
+
+from mcp import ClientSession, StdioServerParameters, types
+from mcp.client.stdio import stdio_client
+
+
+async def parse_tool_results():
+    """Demonstrates how to parse different types of content in CallToolResult."""
+    server_params = StdioServerParameters(
+        command="python", args=["path/to/mcp_server.py"]
+    )
+
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+
+            # Example 1: Parsing text content
+            result = await session.call_tool("get_data", {"format": "text"})
+            for content in result.content:
+                if isinstance(content, types.TextContent):
+                    print(f"Text: {content.text}")
+
+            # Example 2: Parsing structured content from JSON tools
+            result = await session.call_tool("get_user", {"id": "123"})
+            if hasattr(result, "structuredContent") and result.structuredContent:
+                # Access structured data directly
+                user_data = result.structuredContent
+                print(f"User: {user_data.get('name')}, Age: {user_data.get('age')}")
+
+            # Example 3: Parsing embedded resources
+            result = await session.call_tool("read_config", {})
+            for content in result.content:
+                if isinstance(content, types.EmbeddedResource):
+                    resource = content.resource
+                    if isinstance(resource, types.TextResourceContents):
+                        print(f"Config from {resource.uri}: {resource.text}")
+                    elif isinstance(resource, types.BlobResourceContents):
+                        print(f"Binary data from {resource.uri}")
+
+            # Example 4: Parsing image content
+            result = await session.call_tool("generate_chart", {"data": [1, 2, 3]})
+            for content in result.content:
+                if isinstance(content, types.ImageContent):
+                    print(f"Image ({content.mimeType}): {len(content.data)} bytes")
+
+            # Example 5: Handling errors
+            result = await session.call_tool("failing_tool", {})
+            if result.isError:
+                print("Tool execution failed!")
+                for content in result.content:
+                    if isinstance(content, types.TextContent):
+                        print(f"Error: {content.text}")
+
+
+async def main():
+    await parse_tool_results()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```

docs/concepts.md

@@ -1,13 +0,0 @@
-# Concepts
-
-!!! warning "Under Construction"
-
-    This page is currently being written. Check back soon for complete documentation.
-
-<!--
-  - Server vs Client
-  - Three primitives (tools, resources, prompts)
-  - Transports (stdio, SSE, streamable HTTP)
-  - Context and sessions
-  - Lifecycle and state
- -->

docs/index.md

@@ -58,9 +58,11 @@ npx -y @modelcontextprotocol/inspector
 
 <!-- TODO(Marcelo): automatically generate the follow references with a header on each of those files. -->
 1. **[Install](installation.md)** the MCP SDK
-2. **[Learn concepts](concepts.md)** - understand the three primitives and architecture
+2. **[Build servers](server.md)** - tools, resources, prompts, transports, ASGI mounting
-3. **[Explore authorization](authorization.md)** - add security to your servers
+3. **[Write clients](client.md)** - connect to servers, use tools/resources/prompts
-4. **[Use low-level APIs](low-level-server.md)** - for advanced customization
+4. **[Explore authorization](authorization.md)** - add security to your servers
+5. **[Use low-level APIs](low-level-server.md)** - for advanced customization
+6. **[Protocol features](protocol.md)** - MCP primitives, server capabilities
 
 ## API Reference
 

docs/low-level-server.md

@@ -1,5 +1,490 @@
 # Low-Level Server
 
-!!! warning "Under Construction"
+For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API.
 
-    This page is currently being written. Check back soon for complete documentation.
+## Lifespan
+
+<!-- snippet-source examples/snippets/servers/lowlevel/lifespan.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/lowlevel/lifespan.py
+"""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+
+# Mock database class for example
+class Database:
+    """Mock database class for example."""
+
+    @classmethod
+    async def connect(cls) -> "Database":
+        """Connect to database."""
+        print("Database connected")
+        return cls()
+
+    async def disconnect(self) -> None:
+        """Disconnect from database."""
+        print("Database disconnected")
+
+    async def query(self, query_str: str) -> list[dict[str, str]]:
+        """Execute a query."""
+        # Simulate database query
+        return [{"id": "1", "name": "Example", "query": query_str}]
+
+
+@asynccontextmanager
+async def server_lifespan(_server: Server) -> AsyncIterator[dict[str, Any]]:
+    """Manage server startup and shutdown lifecycle."""
+    # Initialize resources on startup
+    db = await Database.connect()
+    try:
+        yield {"db": db}
+    finally:
+        # Clean up on shutdown
+        await db.disconnect()
+
+
+# Pass lifespan to server
+server = Server("example-server", lifespan=server_lifespan)
+
+
+@server.list_tools()
+async def handle_list_tools() -> list[types.Tool]:
+    """List available tools."""
+    return [
+        types.Tool(
+            name="query_db",
+            description="Query the database",
+            inputSchema={
+                "type": "object",
+                "properties": {"query": {"type": "string", "description": "SQL query to execute"}},
+                "required": ["query"],
+            },
+        )
+    ]
+
+
+@server.call_tool()
+async def query_db(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
+    """Handle database query tool call."""
+    if name != "query_db":
+        raise ValueError(f"Unknown tool: {name}")
+
+    # Access lifespan context
+    ctx = server.request_context
+    db = ctx.lifespan_context["db"]
+
+    # Execute query
+    results = await db.query(arguments["query"])
+
+    return [types.TextContent(type="text", text=f"Query results: {results}")]
+
+
+async def run():
+    """Run the server with lifespan management."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="example-server",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/servers/lowlevel/lifespan.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lowlevel/lifespan.py)_
+<!-- /snippet-source -->
+
+The lifespan API provides:
+
+- A way to initialize resources when the server starts and clean them up when it stops
+- Access to initialized resources through the request context in handlers
+- Type-safe context passing between lifespan and request handlers
+
+## Basic Example
+
+<!-- snippet-source examples/snippets/servers/lowlevel/basic.py -->
+```python
+"""
+Run from the repository root:
+uv run examples/snippets/servers/lowlevel/basic.py
+"""
+
+import asyncio
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+# Create a server instance
+server = Server("example-server")
+
+
+@server.list_prompts()
+async def handle_list_prompts() -> list[types.Prompt]:
+    """List available prompts."""
+    return [
+        types.Prompt(
+            name="example-prompt",
+            description="An example prompt template",
+            arguments=[types.PromptArgument(name="arg1", description="Example argument", required=True)],
+        )
+    ]
+
+
+@server.get_prompt()
+async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> types.GetPromptResult:
+    """Get a specific prompt by name."""
+    if name != "example-prompt":
+        raise ValueError(f"Unknown prompt: {name}")
+
+    arg1_value = (arguments or {}).get("arg1", "default")
+
+    return types.GetPromptResult(
+        description="Example prompt",
+        messages=[
+            types.PromptMessage(
+                role="user",
+                content=types.TextContent(type="text", text=f"Example prompt text with argument: {arg1_value}"),
+            )
+        ],
+    )
+
+
+async def run():
+    """Run the basic low-level server."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="example",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/servers/lowlevel/basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lowlevel/basic.py)_
+<!-- /snippet-source -->
+
+Caution: The `uv run mcp run` and `uv run mcp dev` tool doesn't support low-level server.
+
+## Structured Output Support
+
+The low-level server supports structured output for tools, allowing you to return both human-readable content and machine-readable structured data. Tools can define an `outputSchema` to validate their structured output:
+
+<!-- snippet-source examples/snippets/servers/lowlevel/structured_output.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/lowlevel/structured_output.py
+"""
+
+import asyncio
+from typing import Any
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+server = Server("example-server")
+
+
+@server.list_tools()
+async def list_tools() -> list[types.Tool]:
+    """List available tools with structured output schemas."""
+    return [
+        types.Tool(
+            name="get_weather",
+            description="Get current weather for a city",
+            inputSchema={
+                "type": "object",
+                "properties": {"city": {"type": "string", "description": "City name"}},
+                "required": ["city"],
+            },
+            outputSchema={
+                "type": "object",
+                "properties": {
+                    "temperature": {"type": "number", "description": "Temperature in Celsius"},
+                    "condition": {"type": "string", "description": "Weather condition"},
+                    "humidity": {"type": "number", "description": "Humidity percentage"},
+                    "city": {"type": "string", "description": "City name"},
+                },
+                "required": ["temperature", "condition", "humidity", "city"],
+            },
+        )
+    ]
+
+
+@server.call_tool()
+async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Handle tool calls with structured output."""
+    if name == "get_weather":
+        city = arguments["city"]
+
+        # Simulated weather data - in production, call a weather API
+        weather_data = {
+            "temperature": 22.5,
+            "condition": "partly cloudy",
+            "humidity": 65,
+            "city": city,  # Include the requested city
+        }
+
+        # low-level server will validate structured output against the tool's
+        # output schema, and additionally serialize it into a TextContent block
+        # for backwards compatibility with pre-2025-06-18 clients.
+        return weather_data
+    else:
+        raise ValueError(f"Unknown tool: {name}")
+
+
+async def run():
+    """Run the structured output server."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="structured-output-example",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/servers/lowlevel/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lowlevel/structured_output.py)_
+<!-- /snippet-source -->
+
+Tools can return data in four ways:
+
+1. **Content only**: Return a list of content blocks (default behavior before spec revision 2025-06-18)
+2. **Structured data only**: Return a dictionary that will be serialized to JSON (Introduced in spec revision 2025-06-18)
+3. **Both**: Return a tuple of (content, structured_data) preferred option to use for backwards compatibility
+4. **Direct CallToolResult**: Return `CallToolResult` directly for full control (including `_meta` field)
+
+When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.
+
+### Returning CallToolResult Directly
+
+For full control over the response including the `_meta` field (for passing data to client applications without exposing it to the model), return `CallToolResult` directly:
+
+<!-- snippet-source examples/snippets/servers/lowlevel/direct_call_tool_result.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py
+"""
+
+import asyncio
+from typing import Any
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+server = Server("example-server")
+
+
+@server.list_tools()
+async def list_tools() -> list[types.Tool]:
+    """List available tools."""
+    return [
+        types.Tool(
+            name="advanced_tool",
+            description="Tool with full control including _meta field",
+            inputSchema={
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+                "required": ["message"],
+            },
+        )
+    ]
+
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult:
+    """Handle tool calls by returning CallToolResult directly."""
+    if name == "advanced_tool":
+        message = str(arguments.get("message", ""))
+        return types.CallToolResult(
+            content=[types.TextContent(type="text", text=f"Processed: {message}")],
+            structuredContent={"result": "success", "message": message},
+            _meta={"hidden": "data for client applications only"},
+        )
+
+    raise ValueError(f"Unknown tool: {name}")
+
+
+async def run():
+    """Run the server."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="example",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/servers/lowlevel/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lowlevel/direct_call_tool_result.py)_
+<!-- /snippet-source -->
+
+**Note:** When returning `CallToolResult`, you bypass the automatic content/structured conversion. You must construct the complete response yourself.
+
+## Pagination (Advanced)
+
+For servers that need to handle large datasets, the low-level server provides paginated versions of list operations. This is an optional optimization - most servers won't need pagination unless they're dealing with hundreds or thousands of items.
+
+### Server-side Implementation
+
+<!-- snippet-source examples/snippets/servers/pagination_example.py -->
+```python
+"""
+Example of implementing pagination with MCP server decorators.
+"""
+
+from pydantic import AnyUrl
+
+import mcp.types as types
+from mcp.server.lowlevel import Server
+
+# Initialize the server
+server = Server("paginated-server")
+
+# Sample data to paginate
+ITEMS = [f"Item {i}" for i in range(1, 101)]  # 100 items
+
+
+@server.list_resources()
+async def list_resources_paginated(request: types.ListResourcesRequest) -> types.ListResourcesResult:
+    """List resources with pagination support."""
+    page_size = 10
+
+    # Extract cursor from request params
+    cursor = request.params.cursor if request.params is not None else None
+
+    # Parse cursor to get offset
+    start = 0 if cursor is None else int(cursor)
+    end = start + page_size
+
+    # Get page of resources
+    page_items = [
+        types.Resource(uri=AnyUrl(f"resource://items/{item}"), name=item, description=f"Description for {item}")
+        for item in ITEMS[start:end]
+    ]
+
+    # Determine next cursor
+    next_cursor = str(end) if end < len(ITEMS) else None
+
+    return types.ListResourcesResult(resources=page_items, nextCursor=next_cursor)
+```
+
+_Full example: [examples/snippets/servers/pagination_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/pagination_example.py)_
+<!-- /snippet-source -->
+
+### Client-side Consumption
+
+<!-- snippet-source examples/snippets/clients/pagination_client.py -->
+```python
+"""
+Example of consuming paginated MCP endpoints from a client.
+"""
+
+import asyncio
+
+from mcp.client.session import ClientSession
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp.types import PaginatedRequestParams, Resource
+
+
+async def list_all_resources() -> None:
+    """Fetch all resources using pagination."""
+    async with stdio_client(StdioServerParameters(command="uv", args=["run", "mcp-simple-pagination"])) as (
+        read,
+        write,
+    ):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+
+            all_resources: list[Resource] = []
+            cursor = None
+
+            while True:
+                # Fetch a page of resources
+                result = await session.list_resources(params=PaginatedRequestParams(cursor=cursor))
+                all_resources.extend(result.resources)
+
+                print(f"Fetched {len(result.resources)} resources")
+
+                # Check if there are more pages
+                if result.nextCursor:
+                    cursor = result.nextCursor
+                else:
+                    break
+
+            print(f"Total resources: {len(all_resources)}")
+
+
+if __name__ == "__main__":
+    asyncio.run(list_all_resources())
+```
+
+_Full example: [examples/snippets/clients/pagination_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/pagination_client.py)_
+<!-- /snippet-source -->
+
+### Key Points
+
+- **Cursors are opaque strings** - the server defines the format (numeric offsets, timestamps, etc.)
+- **Return `nextCursor=None`** when there are no more pages
+- **Backward compatible** - clients that don't support pagination will still work (they'll just get the first page)
+- **Flexible page sizes** - Each endpoint can define its own page size based on data characteristics
+
+See the [simple-pagination example](https://github.com/modelcontextprotocol/python-sdk/tree/v1.x/examples/servers/simple-pagination) for a complete implementation.