MCP Python SDK

https://py.sdk.modelcontextprotocol.io/
Admin
MCP Python SDK is the official Python library for implementing the Model Context Protocol, enabling...
Tokens:1,118,632
Snippets:6,420
Trust Score:9.9
Update:1 week ago
Show doc for...
Tokens:
Raw
### Install with pip

Source: https://py.sdk.modelcontextprotocol.io/get-started/installation

Use `pip install "mcp[cli]"` to install the SDK with the CLI extra. This is the standard installation method for pip users.

```bash
pip install "mcp[cli]"
```

--------------------------------

### install

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/cli/cli

Installs an MCP server with configuration options such as custom name, editable install, additional packages, and environment variables.

```APIDOC
## install

### Description
Installs an MCP server. Allows specifying a custom server name, installing in editable mode, adding extra packages, and setting environment variables.

### Method
CLI Command

### Command
`install`

### Parameters
#### Arguments
- **file_spec** (str) - Required - Python file to run, optionally with :object suffix

#### Options
- **--name, -n** (str | None) - Optional - Custom name for the server (defaults to server's name attribute or file name)
- **--with-editable, -e** (Path | None) - Optional - Directory containing pyproject.toml to install in editable mode. Must exist, be a directory, and resolve path.
- **--with** (list[str]) - Optional - Additional packages to install
- **--env-var, -v** (list[str]) - Optional - Environment variables in KEY=VALUE format
- **--env-file, -f** (Path | None) - Optional - Load environment variables from a .env file. Must exist, be a file, and resolve path.

### Usage Example

install server.py --name my-server --with package1 --env-var KEY=VALUE


```

--------------------------------

### Install with uv

Source: https://py.sdk.modelcontextprotocol.io/get-started/installation

Use `uv add "mcp[cli]"` to install the SDK with the CLI extra. This is the recommended approach when using the uv package manager.

```bash
uv add "mcp[cli]"
```

--------------------------------

### Register server with Claude Desktop (mcp install)

Source: https://py.sdk.modelcontextprotocol.io/run

Registers the server with Claude Desktop. -v KEY=VALUE and -f .env record environment variables in the entry. Claude Desktop starts the server in its own process; your shell's environment is not available.

```bash
uv run mcp install server.py --name "Bookshop"
uv run mcp install server.py -v API_KEY=abc123 -f .env

```

--------------------------------

### Example: using get_display_name in a client

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/shared/metadata_utils

Example showing how to use `get_display_name` in a client to display available tools with human-readable names.

```python
# In a client displaying available tools
tools = await session.list_tools()
for tool in tools.tools:
    display_name = get_display_name(tool)
    print(f"Available tool: {display_name}")

```

--------------------------------

### expand examples

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/shared/uri_template

Examples of expanding URI templates with various operators and value types.

```python
>>> t = UriTemplate.parse("file://docs/{name}")
>>> t.expand({"name": "hello world.txt"})
'file://docs/hello%20world.txt'

>>> t = UriTemplate.parse("file://docs/{+path}")
>>> t.expand({"path": "src/main.py"})
'file://docs/src/main.py'

>>> t = UriTemplate.parse("/search{?q,lang}")
>>> t.expand({"q": "mcp", "lang": "en"})
'/search?q=mcp&lang=en'

>>> t = UriTemplate.parse("/files{/path*}")
>>> t.expand({"path": ["a", "b", "c"]})
'/files/a/b/c'
```

--------------------------------

### Install OpenTelemetry SDK and OTLP exporter

Source: https://py.sdk.modelcontextprotocol.io/run/opentelemetry

Command to install the OpenTelemetry SDK and OTLP exporter, enabling span collection. Without these, spans are no-ops and cost almost nothing.

```bash
uv add opentelemetry-sdk opentelemetry-exporter-otlp

```

--------------------------------

### Run the Apps example client

Source: https://py.sdk.modelcontextprotocol.io/advanced/apps

Runs the client from the examples/stories directory, which negotiates Apps, reads the tool's _meta.ui.resourceUri, fetches the HTML, and calls the tool.

```bash
uv run python -m stories.apps.client
```

--------------------------------

### Add development dependencies with pip (pip install pytest inline-snapshot)

Source: https://py.sdk.modelcontextprotocol.io/get-started/testing

Installs pytest and inline-snapshot as development dependencies using pip. Required to run the test example.

```bash
pip install pytest inline-snapshot

```

--------------------------------

### Server.__init__

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/lowlevel

Initializes a new Server instance with the given name and configuration. Accepts optional version, title, description, instructions, website URL, icons, cache hints, lifespan context manager, and various request handler callbacks for tools, resources, prompts, completion, and ping.

```APIDOC
## Server.__init__

### Description
Initializes a new Server instance with the given name and configuration. Accepts optional version, title, description, instructions, website URL, icons, cache hints, lifespan context manager, and various request handler callbacks for tools, resources, prompts, completion, and ping.

### Method
Constructor

### Signature
```python
def __init__(
    self,
    name: str,
    *,
    version: str = "",
    title: str | None = None,
    description: str | None = None,
    instructions: str | None = None,
    website_url: str | None = None,
    icons: list[types.Icon] | None = None,
    cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
    lifespan: Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] = lifespan,
    get_tool_input_schema: Callable[[str], Mapping[str, Any] | None] | None = None,
    on_list_tools: Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListToolsResult]] | None = None,
    on_call_tool: Callable[[ServerRequestContext[LifespanResultT], types.CallToolRequestParams], Awaitable[types.CallToolResult | types.InputRequiredResult]] | None = None,
    on_list_resources: Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourcesResult]] | None = None,
    on_list_resource_templates: Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourceTemplatesResult]] | None = None,
    on_read_resource: Callable[[ServerRequestContext[LifespanResultT], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult | types.InputRequiredResult]] | None = None,
    on_subscribe_resource: Callable[[ServerRequestContext[LifespanResultT], types.SubscribeRequestParams], Awaitable[types.EmptyResult]] | None = None,
    on_unsubscribe_resource: Callable[[ServerRequestContext[LifespanResultT], types.UnsubscribeRequestParams], Awaitable[types.EmptyResult]] | None = None,
    on_subscriptions_listen: Callable[[ServerRequestContext[LifespanResultT], types.SubscriptionsListenRequestParams], Awaitable[types.SubscriptionsListenResult]] | None = None,
    on_list_prompts: Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListPromptsResult]] | None = None,
    on_get_prompt: Callable[[ServerRequestContext[LifespanResultT], types.GetPromptRequestParams], Awaitable[types.GetPromptResult | types.InputRequiredResult]] | None = None,
    on_completion: Callable[[ServerRequestContext[LifespanResultT], types.CompleteRequestParams], Awaitable[types.CompleteResult]] | None = None,
    on_ping: Callable[[ServerRequestContext[LifespanResultT], types.RequestParams | None], Awaitable[types.EmptyResult]] = _ping_handler,
) -> None
```

### Parameters
- **name** (str) - Required - The name of the server.
- **version** (str) - Optional - The version of the server. Defaults to "".
- **title** (str | None) - Optional - The title of the server.
- **description** (str | None) - Optional - A description of the server.
- **instructions** (str | None) - Optional - Instructions for the server.
- **website_url** (str | None) - Optional - The website URL for the server.
- **icons** (list[types.Icon] | None) - Optional - A list of icons for the server.
- **cache_hints** (Mapping[CacheableMethod, CacheHint] | None) - Optional - Cache hints for the server.
- **lifespan** (Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]]) - Optional - A lifespan context manager for the server.
- **get_tool_input_schema** (Callable[[str], Mapping[str, Any] | None] | None) - Optional - A function to get the input schema for a tool.
- **on_list_tools** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListToolsResult]] | None) - Optional - Handler for listing tools.
- **on_call_tool** (Callable[[ServerRequestContext[LifespanResultT], types.CallToolRequestParams], Awaitable[types.CallToolResult | types.InputRequiredResult]] | None) - Optional - Handler for calling a tool.
- **on_list_resources** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourcesResult]] | None) - Optional - Handler for listing resources.
- **on_list_resource_templates** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourceTemplatesResult]] | None) - Optional - Handler for listing resource templates.
- **on_read_resource** (Callable[[ServerRequestContext[LifespanResultT], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult | types.InputRequiredResult]] | None) - Optional - Handler for reading a resource.
- **on_subscribe_resource** (Callable[[ServerRequestContext[LifespanResultT], types.SubscribeRequestParams], Awaitable[types.EmptyResult]] | None) - Optional - Handler for subscribing to a resource.
- **on_unsubscribe_resource** (Callable[[ServerRequestContext[LifespanResultT], types.UnsubscribeRequestParams], Awaitable[types.EmptyResult]] | None) - Optional - Handler for unsubscribing from a resource.
- **on_subscriptions_listen** (Callable[[ServerRequestContext[LifespanResultT], types.SubscriptionsListenRequestParams], Awaitable[types.SubscriptionsListenResult]] | None) - Optional - Handler for listening to subscriptions.
- **on_list_prompts** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListPromptsResult]] | None) - Optional - Handler for listing prompts.
- **on_get_prompt** (Callable[[ServerRequestContext[LifespanResultT], types.GetPromptRequestParams], Awaitable[types.GetPromptResult | types.InputRequiredResult]] | None) - Optional - Handler for getting a prompt.
- **on_completion** (Callable[[ServerRequestContext[LifespanResultT], types.CompleteRequestParams], Awaitable[types.CompleteResult]] | None) - Optional - Handler for completion requests.
- **on_ping** (Callable[[ServerRequestContext[LifespanResultT], types.RequestParams | None], Awaitable[types.EmptyResult]]) - Optional - Handler for ping requests. Defaults to _ping_handler.

### Returns
None
```

--------------------------------

### Install OpenTelemetry SDK and exporter

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Install the OpenTelemetry SDK and OTLP exporter to start collecting spans. The SDK depends only on opentelemetry-api, so spans are no-ops until you add these packages.

```console
uv add opentelemetry-sdk opentelemetry-exporter-otlp
```

--------------------------------

### func_metadata usage example

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/mcpserver/utilities/func_metadata

Example showing how to use func_metadata to get metadata for a function, validate raw data against the generated argument model, and call the function with the validated arguments.

```python
meta = func_metadata(func)
validated_args = meta.arg_model.model_validate(some_raw_data_dict)
return func(**validated_args.model_dump_one_level())
```

--------------------------------

### IdentityAssertionOAuthProvider client setup

Source: https://py.sdk.modelcontextprotocol.io/client/identity-assertion

Complete example of using IdentityAssertionOAuthProvider with an httpx2.AsyncClient and streamable_http_client. It includes an in-memory token storage, an ID-JAG issuer function, and the main client setup. The provider requires a pre-registered client_id and client_secret, the issuer, and an async assertion_provider callback.

```python
import time
import uuid

import httpx2
import jwt

from mcp import Client
from mcp.client.auth.extensions.identity_assertion import IdentityAssertionOAuthProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken

IDP_SIGNING_KEY = "the-enterprise-idp-signing-key-for-this-demo"


class InMemoryTokenStorage:
    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


def idp_issue_id_jag(subject: str, audience: str, resource: str) -> str:
    now = int(time.time())
    claims = {
        "iss": "https://idp.example.com",
        "sub": subject,
        "aud": audience,
        "client_id": "finance-agent",
        "resource": resource,
        "scope": "notes:read",
        "jti": str(uuid.uuid4()),
        "iat": now,
        "exp": now + 300,
    }
    return jwt.encode(claims, IDP_SIGNING_KEY, algorithm="HS256", headers={"typ": "oauth-id-jag+jwt"})


async def fetch_id_jag(audience: str, resource: str) -> str:
    return idp_issue_id_jag("alice@example.com", audience, resource)


oauth = IdentityAssertionOAuthProvider(
    server_url="http://localhost:8001/mcp",
    storage=InMemoryTokenStorage(),
    client_id="finance-agent",
    client_secret="finance-agent-secret",
    issuer="https://auth.example.com/",
    assertion_provider=fetch_id_jag,
    scope="notes:read",
)


async def main() -> None:
    async with httpx2.AsyncClient(auth=oauth, timeout=httpx2.Timeout(30.0, read=300.0)) as http_client:
        transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

```

--------------------------------

### VS Code: .vscode/mcp.json configuration

Source: https://py.sdk.modelcontextprotocol.io/get-started/real-host

Create `.vscode/mcp.json` in your project root. Two differences from Cursor's file, and they are the only two: the wrapper key is `servers`, not `mcpServers`, and each entry declares its `type`. Confirm the trust prompt, then **MCP: List Servers** in the Command Palette shows `bookshop` running. You need VS Code 1.99 or later with the **GitHub Copilot** extension signed in (Copilot Free is enough), and Copilot Chat must be in **Agent** mode, because no other mode calls tools.

```json
{
  "servers": {
    "bookshop": {
      "type": "stdio",
      "command": "uv",
      "args": ["run", "--with", "mcp[cli]", "mcp", "run", "/absolute/path/to/server.py"]
    }
  }
}
```

--------------------------------

### RFC 9728 Protected Resource Metadata JSON

Source: https://py.sdk.modelcontextprotocol.io/run/authorization

Example JSON response from GET on the well-known path, built from AuthSettings. Shows resource, authorization_servers, scopes_supported, and bearer_methods_supported.

```json
{
  "resource": "http://127.0.0.1:8000/mcp",
  "authorization_servers": ["https://auth.example.com/"],
  "scopes_supported": ["notes:read"],
  "bearer_methods_supported": ["header"]
}

```

--------------------------------

### v2 low-level Server example (docs_src/whats_new/tutorial001.py)

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Shows the rebuilt low-level Server API in v2. Defines a search_books tool with an input schema, then implements list_tools and call_tool handlers that receive ServerRequestContext and typed params. Raises MCPError with INVALID_PARAMS for unknown tools and formats a result string from the request arguments.

```python
# docs_src/whats_new/tutorial001.py
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    INVALID_PARAMS,
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={  # (1)!
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:  # (2)!
    return ListToolsResult(tools=[SEARCH_BOOKS])  # (3)!


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:  # (4)!
    if params.name != 
```

--------------------------------

### command instance attribute

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/client/stdio

The executable to run to start the server.

```python
command: str
```

--------------------------------

### Context Usage in Tool Function

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/mcpserver/context

Example showing how to use the Context object in a tool function to log messages, report progress, access resources, and get request information.

```APIDOC
## Context Usage in Tool Function

### Description
Demonstrates how to inject and use the Context object in a tool function to log messages, report progress, access resources, and retrieve request information.

### Method
N/A (SDK usage example)

### Endpoint
N/A (SDK usage example)

### Parameters
- **x** (int) - Required - Input value to the tool function.
- **ctx** (Context) - Optional - Context object annotated with Context type, providing access to MCP capabilities.

### Request Example
```python
@server.tool()
async def my_tool(x: int, ctx: Context) -> str:
    # Log messages to the client
    await ctx.info(f"Processing {x}")
    await ctx.debug("Debug info")
    await ctx.warning("Warning message")
    await ctx.error("Error message")

    # Report progress
    await ctx.report_progress(50, 100)

    # Access resources
    data = await ctx.read_resource("resource://data")

    # Get request info
    request_id = ctx.request_id

    return str(x)
```

### Response
#### Success Response
- **str** - The string representation of the input value `x`.

#### Response Example
```python
"42"
```
```

--------------------------------

### Add development dependencies with uv (uv add --dev pytest inline-snapshot)

Source: https://py.sdk.modelcontextprotocol.io/get-started/testing

Installs pytest and inline-snapshot as development dependencies using uv. Required to run the test example.

```bash
uv add --dev pytest inline-snapshot

```

--------------------------------

### Build a server with a UI-bound tool and file resource

Source: https://py.sdk.modelcontextprotocol.io/advanced/apps

Use this to serve HTML from disk for a UI-bound tool. The resource_uri must be ui://... and the tool's resource must be registered. add_resource sets the text/html;profile=mcp-app MIME type if not set, and rejects explicit mismatches.

```python
from pathlib import Path

from mcp.server.apps import Apps
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.resources import FileResource

REPORT_HTML = Path(__file__).parent / "report.html"

apps = Apps()


@apps.tool(resource_uri="ui://report/app.html")
def refresh_report() -> str:
    """Refresh the report data."""
    return "report refreshed"


apps.add_resource(FileResource(uri="ui://report/app.html", name="report", path=REPORT_HTML))

mcp = MCPServer("report", extensions=[apps])
```

--------------------------------

### Example output from calling search_books with 'dune'

Source: https://py.sdk.modelcontextprotocol.io/handlers/context

Shows the output when calling the search_books tool with the query 'dune'. The request ID changes with each call, demonstrating that every request gets its own Context.

```text
[request 3] Found 3 books matching 'dune'.

```

--------------------------------

### connect_sse method for SSE connection setup

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/sse

Handles HTTP requests for SSE, validates request type and security, creates streams and session ID, constructs the client POST URI, and starts the SSE response task.

```python
@asynccontextmanager
async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
    if scope["type"] != "http":
        logger.error("connect_sse received non-HTTP request")
        raise ValueError("connect_sse can only handle HTTP requests")

    # Validate request headers for DNS rebinding protection
    request = Request(scope, receive)
    error_response = await self._security.validate_request(request, is_post=False)
    if error_response:
        await error_response(scope, receive, send)
        raise ValueError("Request validation failed")

    logger.debug("Setting up SSE connection")

    read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
    write_stream, write_stream_reader = create_context_streams[SessionMessage](0)

    session_id = uuid4()
    user = scope.get("user")
    if isinstance(user, AuthenticatedUser):
        self._session_owners[session_id] = authorization_context(user)
    self._read_stream_writers[session_id] = read_stream_writer
    logger.debug(f"Created new session with ID: {session_id}")

    # Determine the full path for the message endpoint to be sent to the client.
    # scope['root_path'] is the prefix where the current Starlette app
    # instance is mounted.
    # e.g., "" if top-level, or "/api_prefix" if mounted under "/api_prefix".
    root_path = scope.get("root_path", "")

    # self._endpoint is the path *within* this app, e.g., "/messages".
    # Concatenating them gives the full absolute path from the server root.
    # e.g., "" + "/messages" -> "/messages"
    # e.g., "/api_prefix" + "/messages" -> "/api_prefix/messages"
    full_message_path_for_client = root_path.rstrip("/") + self._endpoint

    # This is the URI (path + query) the client will use to POST messages.
    client_post_uri_data = f"{quote(full_message_path_for_client)}?session_id={session_id.hex}"

    sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, Any]](0)

    async def sse_writer():
        logger.debug("Starting SSE writer")
        async with sse_stream_writer, write_stream_reader:
            await sse_stream_writer.send({"event": "endpoint", "data": client_post_uri_data})
            logger.debug(f"Sent endpoint event: {client_post_uri_data}")

            async for session_message in write_stream_reader:
                logger.debug(f"Sending message via SSE: {session_message}")
                await sse_stream_writer.send(
                    {
                        "event": "message",
                        "data": session_message.message.model_dump_json(by_alias=True, exclude_unset=True),
                    }
                )

    try:
        async with anyio.create_task_group() as tg:

            async def response_wrapper(scope: Scope, receive: Receive, send: Send):
                """The EventSourceResponse returning signals a client close / disconnect.
                In this case we close our side of the streams to signal the client that
                the connection has been closed.
                """
                await EventSourceResponse(content=sse_stream_reader, data_sender_callable=sse_writer)(
                    scope, receive, send
                )
                await read_stream_writer.aclose()
                await write_stream_reader.aclose()
                await sse_stream_reader.aclose()
                logging.debug(f"Client session disconnected {session_id}")

            logger.debug("Starting SSE response task")
            tg.start_soon(response_wrapper, scope, receive, send)
```

--------------------------------

### Create an MCP server with Apps extension

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/apps

Sets up an `Apps` instance, registers a tool that returns the current time, adds an HTML resource, and attaches the extension to an `MCPServer`. The tool uses `resource_uri` to point to the UI resource and `client_supports_apps(ctx)` can be used to degrade gracefully for clients that don't support Apps.

```python
apps = Apps()

@apps.tool(resource_uri="ui://clock/app.html", description="Current time")
def get_time(ctx: Context) -> str:
    return datetime.now(timezone.utc).isoformat()

apps.add_html_resource("ui://clock/app.html", CLOCK_HTML)

mcp = MCPServer("clock", extensions=[apps])

```

--------------------------------

### Follow board with client.listen() in Python

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Client-side subscription example that listens for resource updates and tool list changes. The client.listen() call sends the request and waits for acknowledgment, so the stream is live when the block starts; each event is a cue to refetch, not a payload.

```python
# docs_src/subscriptions/tutorial003.py
from mcp import Client
from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged
from mcp.types import TextResourceContents

BOARD = "board://sprint"


async def read_board(client: Client, uri: str = BOARD) -> str:
    [contents] = (await client.read_resource(uri)).contents
    assert isinstance(contents, TextResourceContents)
    return contents.text


async def follow_board(client: Client) -> None:
    async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
        async for event in sub:
            match event:
                case ResourceUpdated(uri=uri):
                    print(await read_board(client, uri))
                case ToolsListChanged():
                    tools = await client.list_tools()
                    print("tools:", [tool.name for tool in tools.tools])
                case _:
                    pass  # kinds the filter did not ask for never arrive


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await follow_board(client)
```

--------------------------------

### Universal launch command using uv run

Source: https://py.sdk.modelcontextprotocol.io/get-started/real-host

The command to start the server on any host. Uses `uv run --with` to resolve the SDK into a fresh environment, so it works from any directory without a project or virtual environment. Replace `uv` with its absolute path if the host's PATH is minimal.

```bash
uv run --with "mcp[cli]" mcp run /absolute/path/to/server.py

```

--------------------------------

### Run launch command manually for troubleshooting

Source: https://py.sdk.modelcontextprotocol.io/get-started/real-host

Before you touch any host config, run the launch command yourself. This is the command that the hosts will use to start the server.

```bash
uv run --with "mcp[cli]" mcp run /absolute/path/to/server.py
```

--------------------------------

### Run MCP server over stdio with stdio_server()

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/stdio

Shows the typical usage of stdio_server() as an async context manager. It yields read and write streams that are passed to a server's run() method. The example is wrapped in anyio.run() to start the async function.

```python
async def run_server():
    async with stdio_server() as (read_stream, write_stream):
        server = await create_my_server()
        await server.run(read_stream, write_stream, init_options)

anyio.run(run_server)

```

--------------------------------

### Client dataclass example with call_tool

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/client

Shows how to connect to an MCP server using a Streamable HTTP URL, create a `Client` context, and call a tool. Calls `asyncio.run(main())` to execute the async function.

```python
import asyncio

from mcp import Client

async def main():
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.call_tool("add", {"a": 1, "b": 2})

asyncio.run(main())


```

--------------------------------

### mcp_requirement function

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/cli/claude

Returns a requirement string pinning spawned environments to the running SDK version. Uses `importlib.metadata.version("mcp")` to get the installed version; if the version contains `.dev` or `+`, or if the package is not found, it returns the unpinned package name. Otherwise returns `package==version`.

```python
def mcp_requirement(package: str = "mcp") -> str:
    """Requirement string pinning spawned environments to the running SDK version.

    `uv run --with mcp` resolves the requirement in a fresh environment, where
    an unpinned `mcp` means the latest stable release — not necessarily the
    version the user installed (pre-releases in particular are never selected
    without an explicit pin). Source builds carry dev/local version segments
    that are not published to PyPI, so they fall back to the unpinned form,
    as does a missing distribution (no metadata to pin from).
    """
    try:
        version = importlib.metadata.version("mcp")
    except importlib.metadata.PackageNotFoundError:
        return package
    if ".dev" in version or "+" in version:
        return package
    return f"{package}=={version}"

```

--------------------------------

### Server constructor with on_list_tools and on_call_tools

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Example of creating a lowlevel Server instance with handler functions passed as on_* keyword arguments, replacing the old decorator-based registration.

```APIDOC
## Server constructor with on_list_tools and on_call_tools

### Description
Creates a lowlevel MCP server with handlers for listing tools and calling tools, passed as constructor keyword arguments.

### Method
Constructor

### Endpoint
`Server(name, on_list_tools=..., on_call_tool=...)`

### Parameters
#### Constructor Parameters
- **name** (str) - Required - The server name.
- **on_list_tools** (callable) - Optional - Async handler for listing tools. Receives `(ctx, params)` where `params` is `PaginatedRequestParams | None`. Returns `ListToolsResult`.
- **on_call_tool** (callable) - Optional - Async handler for calling a tool. Receives `(ctx, params)` where `params` is `CallToolRequestParams`. Returns `CallToolResult`.

### Request Example
```python
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

async def handle_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[Tool(name="my_tool", description="A tool", input_schema={"type": "object"})])

async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    return CallToolResult(
        content=[TextContent(type="text", text=f"Called {params.name}")],
        is_error=False,
    )

server = Server("my-server", on_list_tools=handle_list_tools, on_call_tool=handle_call_tool)
```

### Response
#### Success Response
- **server** (Server) - The configured server instance.

#### Response Example
```python
server = Server("my-server", on_list_tools=handle_list_tools, on_call_tool=handle_call_tool)
```
```

--------------------------------

### Mount MCP app inside Starlette without lifespan (causes error)

Source: https://py.sdk.modelcontextprotocol.io/troubleshooting

Shows the problematic setup where the MCP app is mounted inside a Starlette app without a lifespan, causing the session manager to never start and resulting in a 'Task group is not initialized' error on the first request.

```python
from starlette.applications import Starlette
from starlette.routing import Mount

from mcp.server import MCPServer

mcp = MCPServer("Weather")


@mcp.tool()
def forecast(city: str) -> str:
    """Today's forecast for one city."""
    return f"{city}: Rain."


# The mount works. The MCP app's own lifespan never runs.
app = Starlette(routes=[Mount("/", app=mcp.streamable_http_app())])

```

--------------------------------

### Install MCP server for Claude Desktop with uv run mcp install

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Run this command to have the SDK import server.py, read the server name, and write the launch command into Claude Desktop's config file. It converts the path to an absolute one automatically.

```bash
uv run mcp install server.py
```

--------------------------------

### Run MCP server over stdio with mcp.run()

Source: https://py.sdk.modelcontextprotocol.io/run

Defines an MCP server named "Bookshop" with a single tool and runs it synchronously over stdio. The `if __name__ == "__main__":` guard prevents the server from starting when the file is imported by tools like `mcp dev`, `mcp run`, or `mcp install`.

```python
from mcp.server import MCPServer

mcp = MCPServer("Bookshop")


@mcp.tool()
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."


if __name__ == "__main__":
    mcp.run()

```

--------------------------------

### Server.__init__

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/lowlevel/server

Initializes a new Server instance with the given name and optional configuration. Accepts request handler callbacks for tools, resources, and resource templates.

```APIDOC
## Server.__init__

### Description
Initializes a new Server instance with the given name and optional configuration. Accepts request handler callbacks for tools, resources, and resource templates.

### Method
Constructor

### Signature
```python
@overload
def __init__(
    self,
    name: str,
    *,
    version: str = "",
    title: str | None = None,
    description: str | None = None,
    instructions: str | None = None,
    website_url: str | None = None,
    icons: list[types.Icon] | None = None,
    cache_hints: Mapping[CacheableMethod, CacheHint] | None = None,
    lifespan: Callable[
        [Server[LifespanResultT]],
        AbstractAsyncContextManager[LifespanResultT],
    ] = lifespan,
    get_tool_input_schema: Callable[[str], Mapping[str, Any] | None] | None = None,
    on_list_tools: Callable[
        [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
        Awaitable[types.ListToolsResult],
    ] | None = None,
    on_call_tool: Callable[
        [ServerRequestContext[LifespanResultT], types.CallToolRequestParams],
        Awaitable[types.CallToolResult | types.InputRequiredResult],
    ] | None = None,
    on_list_resources: Callable[
        [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
        Awaitable[types.ListResourcesResult],
    ] | None = None,
    on_list_resource_templates: Callable[
        [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
        Awaitable[types.ListResourceTemplatesResult],
    ] | None = None,
    on_read_resource: Callable[
        [ServerRequestContext[LifespanResultT], types.ReadResourceRequestParams],
        Awaitable[types.ReadResourceResult | types.InputRequiredResult],
    ] | None = None,
    on_subscribe_resource: Callable[
        [ServerRequestContext[LifespanResultT], types.SubscribeRequestParams],
        Awaitable[types.EmptyResult],
    ] | None = None,
)
```

### Parameters
#### Positional Parameters
- **name** (str) - Required - The name of the server.

#### Keyword Parameters
- **version** (str) - Optional - The version of the server. Defaults to empty string.
- **title** (str | None) - Optional - The title of the server.
- **description** (str | None) - Optional - A description of the server.
- **instructions** (str | None) - Optional - Instructions for the server.
- **website_url** (str | None) - Optional - The website URL of the server.
- **icons** (list[types.Icon] | None) - Optional - A list of icons for the server.
- **cache_hints** (Mapping[CacheableMethod, CacheHint] | None) - Optional - Cache hints for the server.
- **lifespan** (Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]]) - Optional - A lifespan context manager factory. Defaults to the `lifespan` function.
- **get_tool_input_schema** (Callable[[str], Mapping[str, Any] | None] | None) - Optional - A function to get the input schema for a tool.
- **on_list_tools** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListToolsResult]] | None) - Optional - Handler for listing tools.
- **on_call_tool** (Callable[[ServerRequestContext[LifespanResultT], types.CallToolRequestParams], Awaitable[types.CallToolResult | types.InputRequiredResult]] | None) - Optional - Handler for calling a tool.
- **on_list_resources** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourcesResult]] | None) - Optional - Handler for listing resources.
- **on_list_resource_templates** (Callable[[ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None], Awaitable[types.ListResourceTemplatesResult]] | None) - Optional - Handler for listing resource templates.
- **on_read_resource** (Callable[[ServerRequestContext[LifespanResultT], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult | types.InputRequiredResult]] | None) - Optional - Handler for reading a resource.
- **on_subscribe_resource** (Callable[[ServerRequestContext[LifespanResultT], types.SubscribeRequestParams], Awaitable[types.EmptyResult]] | None) - Optional - Handler for subscribing to a resource.

### Returns
None
```

--------------------------------

### replay_events_after: message router and stream setup

Source: https://py.sdk.modelcontextprotocol.io/api/mcp/server/streamable_http

Sets up the internal streams and starts a message router that distributes session messages to request streams. It stores events in the event store when available, handles JSON-RPC responses and errors, and yields the read and write streams. On exit, it cleans up memory streams and clears the request stream registry.

```python
# Store the streams
        self._read_stream_writer = read_stream_writer
        self._read_stream = read_stream
        self._write_stream_reader = write_stream_reader
        self._write_stream = write_stream

        # Start a task group for message routing
        async with anyio.create_task_group() as tg:
            # Create a message router that distributes messages to request streams
            async def message_router():
                try:
                    async for session_message in write_stream_reader:  # pragma: no branch
                        # Determine which request stream(s) should receive this message
                        message = session_message.message
                        target_request_id = None
                        # Check if this is a response with a known request id.
                        # Null-id errors (e.g., parse errors) fall through to
                        # the GET stream since they can't be correlated.
                        if isinstance(message, JSONRPCResponse | JSONRPCError) and message.id is not None:
                            target_request_id = str(message.id)
                        # Extract related_request_id from meta if it exists
                        elif (
                            session_message.metadata is not None
                            and isinstance(
                                session_message.metadata,
                                ServerMessageMetadata,
                            )
                            and session_message.metadata.related_request_id is not None
                        ):
                            related_request_id = session_message.metadata.related_request_id
                            if self.is_json_response_enabled:
                                # A JSON body carries only the response: this message
                                # has no wire form (nor a replay), so drop it before
                                # storing or queueing rather than park it (#1764).
                                logger.debug(f"Dropped message related to request {related_request_id} in JSON mode")
                                continue
                            target_request_id = str(related_request_id)

                        request_stream_id = target_request_id if target_request_id is not None else GET_STREAM_KEY

                        # Store the event if we have an event store,
                        # regardless of whether a client is connected
                        # messages will be replayed on the re-connect
                        event_id = None
                        if self._event_store:
                            event_id = await self._event_store.store_event(request_stream_id, message)
                            logger.debug(f"Stored {event_id} from {request_stream_id}")

                        if request_stream_id in self._request_streams:
                            try:
                                # Send both the message and the event ID
                                await self._request_streams[request_stream_id][0].send(EventMessage(message, event_id))
                            except (anyio.BrokenResourceError, anyio.ClosedResourceError):  # pragma: no cover
                                # Stream might be closed, remove from registry
                                self._request_streams.pop(request_stream_id, None)
                        else:
                            logger.debug(
                                f"""Request stream {request_stream_id} not found
                                for message. Still processing message as the client
                                might reconnect and replay."""
                            )
                except anyio.ClosedResourceError:
                    if self._terminated:  # pragma: lax no cover
                        logger.debug("Read stream closed by client")
                    else:
                        logger.exception("Unexpected closure of read stream in message router")
                except Exception:  # pragma: lax no cover
                    logger.exception("Error in message router")

            # Start the message router
            tg.start_soon(message_router)

            try:
                # Yield the streams for the caller to use
                yield read_stream, write_stream
            finally:
                for stream_id in list(self._request_streams.keys()):
                    await self._clean_up_memory_streams(stream_id)
                self._request_streams.clear()
```

--------------------------------

### Install server into Claude Desktop with mcp install

Source: https://py.sdk.modelcontextprotocol.io/get-started/real-host

Command that imports the server file, finds Claude Desktop's config, and writes the launch command. It converts the path to absolute automatically.

```bash
uv run mcp install server.py

```

--------------------------------

### Run stdio server with StdioServerParameters and Client (client.py)

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Use this snippet to launch a stdio server as a subprocess from a Python client. The process starts when the async with block is entered and shuts down when it exits; stderr is inherited unless you build the transport yourself with stdio_client and pass errlog. The example sets the command to uv with args run server.py and an environment variable BOOKSHOP_API_KEY.

```python
# docs_src/client_transports/tutorial004.py
from mcp import Client, StdioServerParameters

server = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
    env={"BOOKSHOP_API_KEY": "secret"},
)


async def main() -> None:
    async with Client(server) as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])
```

--------------------------------

### Low-level Server: search_books tool (server.py)

Source: https://py.sdk.modelcontextprotocol.io/llms-full.txt

Implements the search_books tool using the low-level Server class instead of the @mcp.tool() decorator. Use this when you need to emit an exact schema, control result fields like _meta/is_error/structured_content, or handle methods MCP doesn't define. The example is the same tool shown in the Tools guide, written without the convenience layer.

```python
# docs_src/lowlevel/tutorial001.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search for books by title or author",
    inputSchema={
        "type": "object",
        "properties": {
            
```
Tokens:
Raw