# Introduction

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

!!! tip "Looking for the upcoming v2?"
    See the [v2 development documentation](https://py.sdk.modelcontextprotocol.io/v2/).

The **Model Context Protocol (MCP)** allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction.

This Python SDK implements the full MCP specification, making it easy to:

- **Build MCP servers** that expose resources, prompts, and tools
- **Create MCP clients** that can connect to any MCP server
- **Use standard transports** like stdio, SSE, and Streamable HTTP

If you want to read more about the specification, please visit the [MCP documentation](https://modelcontextprotocol.io).

## Quick Example

Here's a simple MCP server that exposes a tool, resource, and prompt:

```python title="server.py"
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Test Server", json_response=True)


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"


@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    return f"Write a {style} greeting for someone named {name}."


if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

Run the server:

```bash
uv run --with mcp server.py
```

Then open the [MCP Inspector](https://github.com/modelcontextprotocol/inspector) and connect to `http://localhost:8000/mcp`:

```bash
npx -y @modelcontextprotocol/inspector
```

## Getting Started

<!-- TODO(Marcelo): automatically generate the follow references with a header on each of those files. -->
1. **[Install](https://py.sdk.modelcontextprotocol.io/installation/index.md)** the MCP SDK
2. **[Build servers](https://py.sdk.modelcontextprotocol.io/server/index.md)** - tools, resources, prompts, transports, ASGI mounting
3. **[Write clients](https://py.sdk.modelcontextprotocol.io/client/index.md)** - connect to servers, use tools/resources/prompts
4. **[Explore authorization](https://py.sdk.modelcontextprotocol.io/authorization/index.md)** - add security to your servers
5. **[Use low-level APIs](https://py.sdk.modelcontextprotocol.io/low-level-server/index.md)** - for advanced customization
6. **[Protocol features](https://py.sdk.modelcontextprotocol.io/protocol/index.md)** - MCP primitives, server capabilities

## API Reference

Full API documentation is available in the [API Reference](https://py.sdk.modelcontextprotocol.io/api/).

## llms.txt

Reading with an LLM? This documentation is also published in the [llms.txt](https://llmstxt.org/) format:
[llms.txt](https://py.sdk.modelcontextprotocol.io/llms.txt) is an index of the pages, and
[llms-full.txt](https://py.sdk.modelcontextprotocol.io/llms-full.txt) contains every page in a single file.

# Installation

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

The Python SDK is available on PyPI as [`mcp`](https://pypi.org/project/mcp/) so installation is as simple as:

=== "pip"

    ```bash
    pip install mcp
    ```
=== "uv"

    ```bash
    uv add mcp
    ```

The following dependencies are automatically installed:

- [`httpx`](https://pypi.org/project/httpx/): HTTP client to handle HTTP Streamable and SSE transports.
- [`httpx-sse`](https://pypi.org/project/httpx-sse/): HTTP client to handle SSE transport.
- [`pydantic`](https://pypi.org/project/pydantic/): Types, JSON schema generation, data validation, and [more](https://docs.pydantic.dev/latest/).
- [`starlette`](https://pypi.org/project/starlette/): Web framework used to build the HTTP transport endpoints.
- [`python-multipart`](https://pypi.org/project/python-multipart/): Handle HTTP body parsing.
- [`sse-starlette`](https://pypi.org/project/sse-starlette/): Server-Sent Events for Starlette, used to build the SSE transport endpoint.
- [`pydantic-settings`](https://pypi.org/project/pydantic-settings/): Settings management used in FastMCP.
- [`uvicorn`](https://pypi.org/project/uvicorn/): ASGI server used to run the HTTP transport endpoints.
- [`jsonschema`](https://pypi.org/project/jsonschema/): JSON schema validation.
- [`pywin32`](https://pypi.org/project/pywin32/): Windows specific dependencies for the CLI tools.

This package has the following optional groups:

- `cli`: Installs `typer` and `python-dotenv` for the MCP CLI tools.

# Building Servers

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

## Core Concepts

### Server

The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:

<!-- snippet-source examples/snippets/servers/lifespan_example.py -->
```python
"""Example showing lifespan support for startup/shutdown with strong typing."""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession


# Mock database class for example
class Database:
    """Mock database class for example."""

    @classmethod
    async def connect(cls) -> "Database":
        """Connect to database."""
        return cls()

    async def disconnect(self) -> None:
        """Disconnect from database."""
        pass

    def query(self) -> str:
        """Execute a query."""
        return "Query result"


@dataclass
class AppContext:
    """Application context with typed dependencies."""

    db: Database


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context."""
    # Initialize on startup
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        # Cleanup on shutdown
        await db.disconnect()


# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)


# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context[ServerSession, AppContext]) -> str:
    """Tool that uses initialized resources."""
    db = ctx.request_context.lifespan_context.db
    return db.query()
```

_Full example: [examples/snippets/servers/lifespan_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lifespan_example.py)_
<!-- /snippet-source -->

### Resources

Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects:

<!-- snippet-source examples/snippets/servers/basic_resource.py -->
```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Resource Example")


@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
    """Read a document by name."""
    # This would normally read from disk
    return f"Content of {name}"


@mcp.resource("config://settings")
def get_settings() -> str:
    """Get application settings."""
    return """{
  "theme": "dark",
  "language": "en",
  "debug": false
}"""
```

_Full example: [examples/snippets/servers/basic_resource.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/basic_resource.py)_
<!-- /snippet-source -->

#### Resource Templates and Template Reading

Resources with URI parameters (e.g., `{name}`) are registered as templates. When a client reads a templated resource, the URI parameters are extracted and passed to the function:

<!-- snippet-source examples/snippets/servers/resource_templates.py -->
```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Template Example")


@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Read a specific user's profile. The user_id is extracted from the URI."""
    return f'{{"user_id": "{user_id}", "name": "User {user_id}"}}'
```

_Full example: [examples/snippets/servers/resource_templates.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/resource_templates.py)_
<!-- /snippet-source -->

Clients read a template resource by providing a concrete URI:

```python
# Client-side: read a template resource with a concrete URI
content = await session.read_resource("users://alice/profile")
```

Templates with multiple parameters work the same way:

```python
@mcp.resource("repos://{owner}/{repo}/readme")
def get_readme(owner: str, repo: str) -> str:
    """Each URI parameter becomes a function argument."""
    return f"README for {owner}/{repo}"
```

#### Binary Resources

Resources can return binary data by returning `bytes` instead of `str`. Set the `mime_type` to indicate the content type:

<!-- snippet-source examples/snippets/servers/binary_resources.py -->
```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Binary Resource Example")


@mcp.resource("images://logo.png", mime_type="image/png")
def get_logo() -> bytes:
    """Return a binary image resource."""
    with open("logo.png", "rb") as f:
        return f.read()
```

_Full example: [examples/snippets/servers/binary_resources.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/binary_resources.py)_
<!-- /snippet-source -->

Binary content is automatically base64-encoded and returned as `BlobResourceContents` in the MCP response.

#### Resource Subscriptions

Clients can subscribe to resource updates. Use the low-level server API to handle subscription and unsubscription requests:

<!-- snippet-source examples/snippets/servers/resource_subscriptions.py -->
```python
from mcp.server.lowlevel import Server

server = Server("Subscription Example")

subscriptions: dict[str, set[str]] = {}  # uri -> set of session ids


@server.subscribe_resource()
async def handle_subscribe(uri) -> None:
    """Handle a client subscribing to a resource."""
    subscriptions.setdefault(str(uri), set()).add("current_session")


@server.unsubscribe_resource()
async def handle_unsubscribe(uri) -> None:
    """Handle a client unsubscribing from a resource."""
    if str(uri) in subscriptions:
        subscriptions[str(uri)].discard("current_session")
```

_Full example: [examples/snippets/servers/resource_subscriptions.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/resource_subscriptions.py)_
<!-- /snippet-source -->

When a subscribed resource changes, notify clients with `send_resource_updated()`:

```python
from pydantic import AnyUrl

# After modifying resource data:
await session.send_resource_updated(AnyUrl("resource://my-resource"))
```

### Tools

Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects:

<!-- snippet-source examples/snippets/servers/basic_tool.py -->
```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Tool Example")


@mcp.tool()
def sum(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b


@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get weather for a city."""
    # This would normally call a weather API
    return f"Weather in {city}: 22degrees{unit[0].upper()}"
```

_Full example: [examples/snippets/servers/basic_tool.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/basic_tool.py)_
<!-- /snippet-source -->

#### Error Handling

When a tool encounters an error, it should signal this to the client rather than returning a normal result. The MCP protocol uses the `isError` flag on `CallToolResult` to distinguish error responses from successful ones. There are three ways to handle errors:

<!-- snippet-source examples/snippets/servers/tool_errors.py -->
```python
"""Example showing how to handle and return errors from tools."""

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from mcp.types import CallToolResult, TextContent

mcp = FastMCP("Tool Error Handling Example")


# Option 1: Raise ToolError for expected error conditions.
# The error message is returned to the client with isError=True.
@mcp.tool()
def divide(a: float, b: float) -> float:
    """Divide two numbers."""
    if b == 0:
        raise ToolError("Cannot divide by zero")
    return a / b


# Option 2: Unhandled exceptions are automatically caught and
# converted to error responses with isError=True.
@mcp.tool()
def read_config(path: str) -> str:
    """Read a configuration file."""
    # If this raises FileNotFoundError, the client receives an
    # error response like "Error executing tool read_config: ..."
    with open(path) as f:
        return f.read()


# Option 3: Return CallToolResult directly for full control
# over error responses, including custom content.
@mcp.tool()
def validate_input(data: str) -> CallToolResult:
    """Validate input data."""
    errors: list[str] = []
    if len(data) < 3:
        errors.append("Input must be at least 3 characters")
    if not data.isascii():
        errors.append("Input must be ASCII only")

    if errors:
        return CallToolResult(
            content=[TextContent(type="text", text="\n".join(errors))],
            isError=True,
        )
    return CallToolResult(
        content=[TextContent(type="text", text="Validation passed")],
    )
```

_Full example: [examples/snippets/servers/tool_errors.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/tool_errors.py)_
<!-- /snippet-source -->

- **`ToolError`** is the preferred approach for most cases — raise it with a descriptive message and the framework handles the rest.
- **Unhandled exceptions** are caught automatically, so tools won't crash the server. The exception message is forwarded to the client as an error response.
- **`CallToolResult`** with `isError=True` gives full control when you need to customize the error content or include multiple content items.

Tools can optionally receive a Context object by including a parameter with the `Context` type annotation. This context is automatically injected by the FastMCP framework and provides access to MCP capabilities:

<!-- snippet-source examples/snippets/servers/tool_progress.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP(name="Progress Example")


@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
    """Execute a task with progress updates."""
    await ctx.info(f"Starting: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"Step {i + 1}/{steps}",
        )
        await ctx.debug(f"Completed step {i + 1}")

    return f"Task '{task_name}' completed"
```

_Full example: [examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/tool_progress.py)_
<!-- /snippet-source -->

#### Structured Output

Tools will return structured results by default, if their return type
annotation is compatible. Otherwise, they will return unstructured results.

Structured output supports these return types:

- Pydantic models (BaseModel subclasses)
- TypedDicts
- Dataclasses and other classes with type hints
- `dict[str, T]` (where T is any JSON-serializable type)
- Primitive types (str, int, float, bool, bytes, None) - wrapped in `{"result": value}`
- Generic types (list, tuple, Union, Optional, etc.) - wrapped in `{"result": value}`

Classes without type hints cannot be serialized for structured output. Only
classes with properly annotated attributes will be converted to Pydantic models
for schema generation and validation.

Structured results are automatically validated against the output schema
generated from the annotation. This ensures the tool returns well-typed,
validated data that clients can easily process.

**Note:** For backward compatibility, unstructured results are also
returned. Unstructured results are provided for backward compatibility
with previous versions of the MCP specification, and are quirks-compatible
with previous versions of FastMCP in the current version of the SDK.

**Note:** In cases where a tool function's return type annotation
causes the tool to be classified as structured _and this is undesirable_,
the  classification can be suppressed by passing `structured_output=False`
to the `@tool` decorator.

##### Advanced: Direct CallToolResult

For full control over tool responses including the `_meta` field (for passing data to client applications without exposing it to the model), you can return `CallToolResult` directly:

<!-- snippet-source examples/snippets/servers/direct_call_tool_result.py -->
```python
"""Example showing direct CallToolResult return for advanced control."""

from typing import Annotated

from pydantic import BaseModel

from mcp.server.fastmcp import FastMCP
from mcp.types import CallToolResult, TextContent

mcp = FastMCP("CallToolResult Example")


class ValidationModel(BaseModel):
    """Model for validating structured output."""

    status: str
    data: dict[str, int]


@mcp.tool()
def advanced_tool() -> CallToolResult:
    """Return CallToolResult directly for full control including _meta field."""
    return CallToolResult(
        content=[TextContent(type="text", text="Response visible to the model")],
        _meta={"hidden": "data for client applications only"},
    )


@mcp.tool()
def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
    """Return CallToolResult with structured output validation."""
    return CallToolResult(
        content=[TextContent(type="text", text="Validated response")],
        structuredContent={"status": "success", "data": {"result": 42}},
        _meta={"internal": "metadata"},
    )


@mcp.tool()
def empty_result_tool() -> CallToolResult:
    """For empty results, return CallToolResult with empty content."""
    return CallToolResult(content=[])
```

_Full example: [examples/snippets/servers/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/direct_call_tool_result.py)_
<!-- /snippet-source -->

**Important:** `CallToolResult` must always be returned (no `Optional` or `Union`). For empty results, use `CallToolResult(content=[])`. For optional simple types, use `str | None` without `CallToolResult`.

<!-- snippet-source examples/snippets/servers/structured_output.py -->
```python
"""Example showing structured output with tools."""

from typing import TypedDict

from pydantic import BaseModel, Field

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Structured Output Example")


# Using Pydantic models for rich structured data
class WeatherData(BaseModel):
    """Weather information structure."""

    temperature: float = Field(description="Temperature in Celsius")
    humidity: float = Field(description="Humidity percentage")
    condition: str
    wind_speed: float


@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """Get weather for a city - returns structured data."""
    # Simulated weather data
    return WeatherData(
        temperature=22.5,
        humidity=45.0,
        condition="sunny",
        wind_speed=5.2,
    )


# Using TypedDict for simpler structures
class LocationInfo(TypedDict):
    latitude: float
    longitude: float
    name: str


@mcp.tool()
def get_location(address: str) -> LocationInfo:
    """Get location coordinates"""
    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK")


# Using dict[str, Any] for flexible schemas
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    """Get various statistics"""
    return {"mean": 42.5, "median": 40.0, "std_dev": 5.2}


# Ordinary classes with type hints work for structured output
class UserProfile:
    name: str
    age: int
    email: str | None = None

    def __init__(self, name: str, age: int, email: str | None = None):
        self.name = name
        self.age = age
        self.email = email


@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    """Get user profile - returns structured data"""
    return UserProfile(name="Alice", age=30, email="alice@example.com")


# Classes WITHOUT type hints cannot be used for structured output
class UntypedConfig:
    def __init__(self, setting1, setting2):  # type: ignore[reportMissingParameterType]
        self.setting1 = setting1
        self.setting2 = setting2


@mcp.tool()
def get_config() -> UntypedConfig:
    """This returns unstructured output - no schema generated"""
    return UntypedConfig("value1", "value2")


# Lists and other types are wrapped automatically
@mcp.tool()
def list_cities() -> list[str]:
    """Get a list of cities"""
    return ["London", "Paris", "Tokyo"]
    # Returns: {"result": ["London", "Paris", "Tokyo"]}


@mcp.tool()
def get_temperature(city: str) -> float:
    """Get temperature as a simple float"""
    return 22.5
    # Returns: {"result": 22.5}
```

_Full example: [examples/snippets/servers/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/structured_output.py)_
<!-- /snippet-source -->

### Prompts

Prompts are reusable templates that help LLMs interact with your server effectively:

<!-- snippet-source examples/snippets/servers/basic_prompt.py -->
```python
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP(name="Prompt Example")


@mcp.prompt(title="Code Review")
def review_code(code: str) -> str:
    return f"Please review this code:\n\n{code}"


@mcp.prompt(title="Debug Assistant")
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("I'm seeing this error:"),
        base.UserMessage(error),
        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]
```

_Full example: [examples/snippets/servers/basic_prompt.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/basic_prompt.py)_
<!-- /snippet-source -->

#### Prompts with Embedded Resources

Prompts can include embedded resources to provide file contents or data alongside the conversation messages:

<!-- snippet-source examples/snippets/servers/prompt_embedded_resources.py -->
```python
import mcp.types as types
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP("Embedded Resource Prompt Example")


@mcp.prompt()
def review_file(filename: str) -> list[base.Message]:
    """Review a file with its contents embedded."""
    file_content = open(filename).read()
    return [
        base.UserMessage(
            content=types.TextContent(type="text", text=f"Please review {filename}:"),
        ),
        base.UserMessage(
            content=types.EmbeddedResource(
                type="resource",
                resource=types.TextResourceContents(
                    uri=f"file://{filename}",
                    text=file_content,
                    mimeType="text/plain",
                ),
            ),
        ),
    ]
```

_Full example: [examples/snippets/servers/prompt_embedded_resources.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/prompt_embedded_resources.py)_
<!-- /snippet-source -->

#### Prompts with Image Content

Prompts can include images using `ImageContent` or the `Image` helper class:

<!-- snippet-source examples/snippets/servers/prompt_image_content.py -->
```python
import mcp.types as types
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
from mcp.server.fastmcp.utilities.types import Image

mcp = FastMCP("Image Prompt Example")


@mcp.prompt()
def describe_image(image_path: str) -> list[base.Message]:
    """Prompt that includes an image for analysis."""
    img = Image(path=image_path)
    return [
        base.UserMessage(
            content=types.TextContent(type="text", text="Describe this image:"),
        ),
        base.UserMessage(
            content=img.to_image_content(),
        ),
    ]
```

_Full example: [examples/snippets/servers/prompt_image_content.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/prompt_image_content.py)_
<!-- /snippet-source -->

#### Prompt Change Notifications

When your server dynamically adds or removes prompts, notify connected clients:

<!-- snippet-source examples/snippets/servers/prompt_change_notifications.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP("Dynamic Prompts")


@mcp.tool()
async def update_prompts(ctx: Context[ServerSession, None]) -> str:
    """Update available prompts and notify clients."""
    # ... modify prompts ...
    await ctx.session.send_prompt_list_changed()
    return "Prompts updated"
```

_Full example: [examples/snippets/servers/prompt_change_notifications.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/prompt_change_notifications.py)_
<!-- /snippet-source -->

### Icons

MCP servers can provide icons for UI display. Icons can be added to the server implementation, tools, resources, and prompts:

```python
from mcp.server.fastmcp import FastMCP, Icon

# Create an icon from a file path or URL
icon = Icon(
    src="icon.png",
    mimeType="image/png",
    sizes=["64x64"]
)

# Add icons to server
mcp = FastMCP(
    "My Server",
    website_url="https://example.com",
    icons=[icon]
)

# Add icons to tools, resources, and prompts
@mcp.tool(icons=[icon])
def my_tool():
    """Tool with an icon."""
    return "result"

@mcp.resource("demo://resource", icons=[icon])
def my_resource():
    """Resource with an icon."""
    return "content"
```

_Full example: [examples/fastmcp/icons_demo.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/fastmcp/icons_demo.py)_

### Images

FastMCP provides an `Image` class that automatically handles image data:

<!-- snippet-source examples/snippets/servers/images.py -->
```python
"""Example showing image handling with FastMCP."""

from PIL import Image as PILImage

from mcp.server.fastmcp import FastMCP, Image

mcp = FastMCP("Image Example")


@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
    """Create a thumbnail from an image"""
    img = PILImage.open(image_path)
    img.thumbnail((100, 100))
    return Image(data=img.tobytes(), format="png")
```

_Full example: [examples/snippets/servers/images.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/images.py)_
<!-- /snippet-source -->

### Audio

FastMCP provides an `Audio` class for returning audio data from tools, similar to `Image`:

<!-- snippet-source examples/snippets/servers/audio_example.py -->
```python
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.utilities.types import Audio

mcp = FastMCP("Audio Example")


@mcp.tool()
def get_audio_from_file(file_path: str) -> Audio:
    """Return audio from a file path (format auto-detected from extension)."""
    return Audio(path=file_path)


@mcp.tool()
def get_audio_from_bytes(raw_audio: bytes) -> Audio:
    """Return audio from raw bytes with explicit format."""
    return Audio(data=raw_audio, format="wav")
```

_Full example: [examples/snippets/servers/audio_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/audio_example.py)_
<!-- /snippet-source -->

The `Audio` class accepts `path` or `data` (mutually exclusive) and an optional `format` string. Supported formats include `wav`, `mp3`, `ogg`, `flac`, `aac`, and `m4a`. When using a file path, the MIME type is inferred from the file extension.

### Embedded Resource Results

Tools can return `EmbeddedResource` to attach file contents or data inline in the result:

<!-- snippet-source examples/snippets/servers/embedded_resource_results.py -->
```python
from mcp.server.fastmcp import FastMCP
from mcp.types import EmbeddedResource, TextResourceContents

mcp = FastMCP("Embedded Resource Example")


@mcp.tool()
def read_config(path: str) -> EmbeddedResource:
    """Read a config file and return it as an embedded resource."""
    with open(path) as f:
        content = f.read()
    return EmbeddedResource(
        type="resource",
        resource=TextResourceContents(
            uri=f"file://{path}",
            text=content,
            mimeType="application/json",
        ),
    )
```

_Full example: [examples/snippets/servers/embedded_resource_results.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/embedded_resource_results.py)_
<!-- /snippet-source -->

For binary embedded resources, use `BlobResourceContents` with base64-encoded data:

<!-- snippet-source examples/snippets/servers/embedded_resource_results_binary.py -->
```python
import base64

from mcp.server.fastmcp import FastMCP
from mcp.types import BlobResourceContents, EmbeddedResource

mcp = FastMCP("Binary Embedded Resource Example")


@mcp.tool()
def read_binary_file(path: str) -> EmbeddedResource:
    """Read a binary file and return it as an embedded resource."""
    with open(path, "rb") as f:
        data = base64.b64encode(f.read()).decode()
    return EmbeddedResource(
        type="resource",
        resource=BlobResourceContents(
            uri=f"file://{path}",
            blob=data,
            mimeType="application/octet-stream",
        ),
    )
```

_Full example: [examples/snippets/servers/embedded_resource_results_binary.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/embedded_resource_results_binary.py)_
<!-- /snippet-source -->

### Tool Change Notifications

When your server dynamically adds or removes tools at runtime, notify connected clients so they can refresh their tool list:

<!-- snippet-source examples/snippets/servers/tool_change_notifications.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP("Dynamic Tools")


@mcp.tool()
async def register_plugin(name: str, ctx: Context[ServerSession, None]) -> str:
    """Dynamically register a new tool and notify the client."""
    # ... register the plugin's tools ...

    # Notify the client that the tool list has changed
    await ctx.session.send_tool_list_changed()

    return f"Plugin '{name}' registered"
```

_Full example: [examples/snippets/servers/tool_change_notifications.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/tool_change_notifications.py)_
<!-- /snippet-source -->

### Context

The Context object is automatically injected into tool and resource functions that request it via type hints. It provides access to MCP capabilities like logging, progress reporting, resource reading, user interaction, and request metadata.

#### Getting Context in Functions

To use context in a tool or resource function, add a parameter with the `Context` type annotation:

```python
from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP(name="Context Example")


@mcp.tool()
async def my_tool(x: int, ctx: Context) -> str:
    """Tool that uses context capabilities."""
    # The context parameter can have any name as long as it's type-annotated
    return await process_with_context(x, ctx)
```

#### Context Properties and Methods

The Context object provides the following capabilities:

- `ctx.request_id` - Unique ID for the current request
- `ctx.client_id` - Client ID if available
- `ctx.fastmcp` - Access to the FastMCP server instance (see [FastMCP Properties](#fastmcp-properties))
- `ctx.session` - Access to the underlying session for advanced communication (see [Session Properties and Methods](#session-properties-and-methods))
- `ctx.request_context` - Access to request-specific data and lifespan resources (see [Request Context Properties](#request-context-properties))
- `await ctx.debug(message)` - Send debug log message
- `await ctx.info(message)` - Send info log message
- `await ctx.warning(message)` - Send warning log message
- `await ctx.error(message)` - Send error log message
- `await ctx.log(level, message, logger_name=None)` - Send log with custom level
- `await ctx.report_progress(progress, total=None, message=None)` - Report operation progress
- `await ctx.read_resource(uri)` - Read a resource by URI
- `await ctx.elicit(message, schema)` - Request additional information from user with validation

<!-- snippet-source examples/snippets/servers/tool_progress.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP(name="Progress Example")


@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
    """Execute a task with progress updates."""
    await ctx.info(f"Starting: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"Step {i + 1}/{steps}",
        )
        await ctx.debug(f"Completed step {i + 1}")

    return f"Task '{task_name}' completed"
```

_Full example: [examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/tool_progress.py)_
<!-- /snippet-source -->

### Completions

MCP supports providing completion suggestions for prompt arguments and resource template parameters. With the context parameter, servers can provide completions based on previously resolved values:

Client usage:

<!-- snippet-source examples/snippets/clients/completion_client.py -->
```python
"""
cd to the `examples/snippets` directory and run:
    uv run completion-client
"""

import asyncio
import os

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import PromptReference, ResourceTemplateReference

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="uv",  # Using uv to run the server
    args=["run", "server", "completion", "stdio"],  # Server with completion support
    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
)


async def run():
    """Run the completion client example."""
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()

            # List available resource templates
            templates = await session.list_resource_templates()
            print("Available resource templates:")
            for template in templates.resourceTemplates:
                print(f"  - {template.uriTemplate}")

            # List available prompts
            prompts = await session.list_prompts()
            print("\nAvailable prompts:")
            for prompt in prompts.prompts:
                print(f"  - {prompt.name}")

            # Complete resource template arguments
            if templates.resourceTemplates:
                template = templates.resourceTemplates[0]
                print(f"\nCompleting arguments for resource template: {template.uriTemplate}")

                # Complete without context
                result = await session.complete(
                    ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate),
                    argument={"name": "owner", "value": "model"},
                )
                print(f"Completions for 'owner' starting with 'model': {result.completion.values}")

                # Complete with context - repo suggestions based on owner
                result = await session.complete(
                    ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate),
                    argument={"name": "repo", "value": ""},
                    context_arguments={"owner": "modelcontextprotocol"},
                )
                print(f"Completions for 'repo' with owner='modelcontextprotocol': {result.completion.values}")

            # Complete prompt arguments
            if prompts.prompts:
                prompt_name = prompts.prompts[0].name
                print(f"\nCompleting arguments for prompt: {prompt_name}")

                result = await session.complete(
                    ref=PromptReference(type="ref/prompt", name=prompt_name),
                    argument={"name": "style", "value": ""},
                )
                print(f"Completions for 'style' argument: {result.completion.values}")


def main():
    """Entry point for the completion client."""
    asyncio.run(run())


if __name__ == "__main__":
    main()
```

_Full example: [examples/snippets/clients/completion_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/completion_client.py)_
<!-- /snippet-source -->
### Elicitation

Request additional information from users. This example shows an Elicitation during a Tool Call:

<!-- snippet-source examples/snippets/servers/elicitation.py -->
```python
"""Elicitation examples demonstrating form and URL mode elicitation.

Form mode elicitation collects structured, non-sensitive data through a schema.
URL mode elicitation directs users to external URLs for sensitive operations
like OAuth flows, credential collection, or payment processing.
"""

import uuid

from pydantic import BaseModel, Field

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
from mcp.shared.exceptions import UrlElicitationRequiredError
from mcp.types import ElicitRequestURLParams

mcp = FastMCP(name="Elicitation Example")


class BookingPreferences(BaseModel):
    """Schema for collecting user preferences."""

    checkAlternative: bool = Field(description="Would you like to check another date?")
    alternativeDate: str = Field(
        default="2024-12-26",
        description="Alternative date (YYYY-MM-DD)",
    )


@mcp.tool()
async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerSession, None]) -> str:
    """Book a table with date availability check.

    This demonstrates form mode elicitation for collecting non-sensitive user input.
    """
    # Check if date is available
    if date == "2024-12-25":
        # Date unavailable - ask user for alternative
        result = await ctx.elicit(
            message=(f"No tables available for {party_size} on {date}. Would you like to try another date?"),
            schema=BookingPreferences,
        )

        if result.action == "accept" and result.data:
            if result.data.checkAlternative:
                return f"[SUCCESS] Booked for {result.data.alternativeDate}"
            return "[CANCELLED] No booking made"
        return "[CANCELLED] Booking cancelled"

    # Date available
    return f"[SUCCESS] Booked for {date} at {time}"


@mcp.tool()
async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> str:
    """Process a secure payment requiring URL confirmation.

    This demonstrates URL mode elicitation using ctx.elicit_url() for
    operations that require out-of-band user interaction.
    """
    elicitation_id = str(uuid.uuid4())

    result = await ctx.elicit_url(
        message=f"Please confirm payment of ${amount:.2f}",
        url=f"https://payments.example.com/confirm?amount={amount}&id={elicitation_id}",
        elicitation_id=elicitation_id,
    )

    if result.action == "accept":
        # In a real app, the payment confirmation would happen out-of-band
        # and you'd verify the payment status from your backend
        return f"Payment of ${amount:.2f} initiated - check your browser to complete"
    elif result.action == "decline":
        return "Payment declined by user"
    return "Payment cancelled"


@mcp.tool()
async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
    """Connect to a third-party service requiring OAuth authorization.

    This demonstrates the "throw error" pattern using UrlElicitationRequiredError.
    Use this pattern when the tool cannot proceed without user authorization.
    """
    elicitation_id = str(uuid.uuid4())

    # Raise UrlElicitationRequiredError to signal that the client must complete
    # a URL elicitation before this request can be processed.
    # The MCP framework will convert this to a -32042 error response.
    raise UrlElicitationRequiredError(
        [
            ElicitRequestURLParams(
                mode="url",
                message=f"Authorization required to connect to {service_name}",
                url=f"https://{service_name}.example.com/oauth/authorize?elicit={elicitation_id}",
                elicitationId=elicitation_id,
            )
        ]
    )
```

_Full example: [examples/snippets/servers/elicitation.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/elicitation.py)_
<!-- /snippet-source -->

Elicitation schemas support default values for all field types. Default values are automatically included in the JSON schema sent to clients, allowing them to pre-populate forms.

The `elicit()` method returns an `ElicitationResult` with:

- `action`: "accept", "decline", or "cancel"
- `data`: The validated response (only when accepted)

#### Elicitation with Enum Values

To present a dropdown or selection list in elicitation forms, use `json_schema_extra` with an `enum` key on a `str` field. Do not use `Literal` -- use a plain `str` field with the enum constraint in the JSON schema:

<!-- snippet-source examples/snippets/servers/elicitation_enum.py -->
```python
from pydantic import BaseModel, Field

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP("Enum Elicitation Example")


class ColorPreference(BaseModel):
    color: str = Field(
        description="Pick your favorite color",
        json_schema_extra={"enum": ["red", "green", "blue", "yellow"]},
    )


@mcp.tool()
async def pick_color(ctx: Context[ServerSession, None]) -> str:
    """Ask the user to pick a color from a list."""
    result = await ctx.elicit(
        message="Choose a color:",
        schema=ColorPreference,
    )
    if result.action == "accept":
        return f"You picked: {result.data.color}"
    return "No color selected"
```

_Full example: [examples/snippets/servers/elicitation_enum.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/elicitation_enum.py)_
<!-- /snippet-source -->

#### Elicitation Complete Notification

For URL mode elicitations, send a completion notification after the out-of-band interaction finishes. This tells the client that the elicitation is done and it may retry any blocked requests:

<!-- snippet-source examples/snippets/servers/elicitation_complete.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP("Elicit Complete Example")


@mcp.tool()
async def handle_oauth_callback(elicitation_id: str, ctx: Context[ServerSession, None]) -> str:
    """Called when OAuth flow completes out-of-band."""
    # ... process the callback ...

    # Notify the client that the elicitation is done
    await ctx.session.send_elicit_complete(elicitation_id)

    return "Authorization complete"
```

_Full example: [examples/snippets/servers/elicitation_complete.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/elicitation_complete.py)_
<!-- /snippet-source -->

### Sampling

Tools can interact with LLMs through sampling (generating text):

<!-- snippet-source examples/snippets/servers/sampling.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
from mcp.types import SamplingMessage, TextContent

mcp = FastMCP(name="Sampling Example")


@mcp.tool()
async def generate_poem(topic: str, ctx: Context[ServerSession, None]) -> str:
    """Generate a poem using LLM sampling."""
    prompt = f"Write a short poem about {topic}"

    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(type="text", text=prompt),
            )
        ],
        max_tokens=100,
    )

    # Since we're not passing tools param, result.content is single content
    if result.content.type == "text":
        return result.content.text
    return str(result.content)
```

_Full example: [examples/snippets/servers/sampling.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/sampling.py)_
<!-- /snippet-source -->

### Logging and Notifications

Tools can send logs and notifications through the context:

<!-- snippet-source examples/snippets/servers/notifications.py -->
```python
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP(name="Notifications Example")


@mcp.tool()
async def process_data(data: str, ctx: Context[ServerSession, None]) -> str:
    """Process data with logging."""
    # Different log levels
    await ctx.debug(f"Debug: Processing '{data}'")
    await ctx.info("Info: Starting processing")
    await ctx.warning("Warning: This is experimental")
    await ctx.error("Error: (This is just a demo)")

    # Notify about resource changes
    await ctx.session.send_resource_list_changed()

    return f"Processed: {data}"
```

_Full example: [examples/snippets/servers/notifications.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/notifications.py)_
<!-- /snippet-source -->

#### Setting the Logging Level

Clients can request a minimum logging level via `logging/setLevel`. Use the low-level server API to handle this:

<!-- snippet-source examples/snippets/servers/set_logging_level.py -->
```python
import mcp.types as types
from mcp.server.lowlevel import Server

server = Server("Logging Level Example")

current_level: types.LoggingLevel = "warning"


@server.set_logging_level()
async def handle_set_level(level: types.LoggingLevel) -> None:
    """Handle client request to change the logging level."""
    global current_level
    current_level = level
```

_Full example: [examples/snippets/servers/set_logging_level.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/set_logging_level.py)_
<!-- /snippet-source -->

When this handler is registered, the server automatically declares the `logging` capability during initialization.

### Authentication

For OAuth 2.1 server and client authentication, see [Authorization](https://py.sdk.modelcontextprotocol.io/authorization/index.md).

### FastMCP Properties

The FastMCP server instance accessible via `ctx.fastmcp` provides access to server configuration and metadata:

- `ctx.fastmcp.name` - The server's name as defined during initialization
- `ctx.fastmcp.instructions` - Server instructions/description provided to clients
- `ctx.fastmcp.website_url` - Optional website URL for the server
- `ctx.fastmcp.icons` - Optional list of icons for UI display
- `ctx.fastmcp.settings` - Complete server configuration object containing:
  - `debug` - Debug mode flag
  - `log_level` - Current logging level
  - `host` and `port` - Server network configuration
  - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
  - `stateless_http` - Whether the server operates in stateless mode
  - And other configuration options

```python
@mcp.tool()
def server_info(ctx: Context) -> dict:
    """Get information about the current server."""
    return {
        "name": ctx.fastmcp.name,
        "instructions": ctx.fastmcp.instructions,
        "debug_mode": ctx.fastmcp.settings.debug,
        "log_level": ctx.fastmcp.settings.log_level,
        "host": ctx.fastmcp.settings.host,
        "port": ctx.fastmcp.settings.port,
    }
```

### Session Properties and Methods

The session object accessible via `ctx.session` provides advanced control over client communication:

- `ctx.session.client_params` - Client initialization parameters and declared capabilities
- `await ctx.session.send_log_message(level, data, logger)` - Send log messages with full control
- `await ctx.session.create_message(messages, max_tokens)` - Request LLM sampling/completion
- `await ctx.session.send_progress_notification(token, progress, total, message)` - Direct progress updates
- `await ctx.session.send_resource_updated(uri)` - Notify clients that a specific resource changed
- `await ctx.session.send_resource_list_changed()` - Notify clients that the resource list changed
- `await ctx.session.send_tool_list_changed()` - Notify clients that the tool list changed
- `await ctx.session.send_prompt_list_changed()` - Notify clients that the prompt list changed

```python
@mcp.tool()
async def notify_data_update(resource_uri: str, ctx: Context) -> str:
    """Update data and notify clients of the change."""
    # Perform data update logic here

    # Notify clients that this specific resource changed
    await ctx.session.send_resource_updated(AnyUrl(resource_uri))

    # If this affects the overall resource list, notify about that too
    await ctx.session.send_resource_list_changed()

    return f"Updated {resource_uri} and notified clients"
```

### Request Context Properties

The request context accessible via `ctx.request_context` contains request-specific information and resources:

- `ctx.request_context.lifespan_context` - Access to resources initialized during server startup
  - Database connections, configuration objects, shared services
  - Type-safe access to resources defined in your server's lifespan function
- `ctx.request_context.meta` - Request metadata from the client including:
  - `progressToken` - Token for progress notifications
  - Other client-provided metadata
- `ctx.request_context.request` - The original MCP request object for advanced processing
- `ctx.request_context.request_id` - Unique identifier for this request

```python
# Example with typed lifespan context
@dataclass
class AppContext:
    db: Database
    config: AppConfig

@mcp.tool()
def query_with_config(query: str, ctx: Context) -> str:
    """Execute a query using shared database and configuration."""
    # Access typed lifespan context
    app_ctx: AppContext = ctx.request_context.lifespan_context

    # Use shared resources
    connection = app_ctx.db
    settings = app_ctx.config

    # Execute query with configuration
    result = connection.execute(query, timeout=settings.query_timeout)
    return str(result)
```

_Full lifespan example: [examples/snippets/servers/lifespan_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/lifespan_example.py)_

## Running Your Server

### Development Mode

The fastest way to test and debug your server is with the MCP Inspector:

```bash
uv run mcp dev server.py

# Add dependencies
uv run mcp dev server.py --with pandas --with numpy

# Mount local code
uv run mcp dev server.py --with-editable .
```

### Claude Desktop Integration

Once your server is ready, install it in Claude Desktop:

```bash
uv run mcp install server.py

# Custom name
uv run mcp install server.py --name "My Analytics Server"

# Environment variables
uv run mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
uv run mcp install server.py -f .env
```

### Direct Execution

For advanced scenarios like custom deployments:

<!-- snippet-source examples/snippets/servers/direct_execution.py -->
```python
"""Example showing direct execution of an MCP server.

This is the simplest way to run an MCP server directly.
cd to the `examples/snippets` directory and run:
    uv run direct-execution-server
    or
    python servers/direct_execution.py
"""

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.tool()
def hello(name: str = "World") -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"


def main():
    """Entry point for the direct execution server."""
    mcp.run()


if __name__ == "__main__":
    main()
```

_Full example: [examples/snippets/servers/direct_execution.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/direct_execution.py)_
<!-- /snippet-source -->

Run it with:

```bash
python servers/direct_execution.py
# or
uv run mcp run servers/direct_execution.py
```

Note that `uv run mcp run` or `uv run mcp dev` only supports server using FastMCP and not the low-level server variant.

### Streamable HTTP Transport

> **Note**: Streamable HTTP transport is the recommended transport for production deployments. Use `stateless_http=True` and `json_response=True` for optimal scalability.

<!-- snippet-source examples/snippets/servers/streamable_config.py -->
```python
"""
Run from the repository root:
    uv run examples/snippets/servers/streamable_config.py
"""

from mcp.server.fastmcp import FastMCP

# Stateless server with JSON responses (recommended)
mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True)

# Other configuration options:
# Stateless server with SSE streaming responses
# mcp = FastMCP("StatelessServer", stateless_http=True)

# Stateful server with session persistence
# mcp = FastMCP("StatefulServer")


# Add a simple tool to demonstrate the server
@mcp.tool()
def greet(name: str = "World") -> str:
    """Greet someone by name."""
    return f"Hello, {name}!"


# Run server with streamable_http transport
if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

_Full example: [examples/snippets/servers/streamable_config.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_config.py)_
<!-- /snippet-source -->

You can mount multiple FastMCP servers in a Starlette application:

<!-- snippet-source examples/snippets/servers/streamable_starlette_mount.py -->
```python
"""
Run from the repository root:
    uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload
"""

import contextlib

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

from mcp.server.fastmcp import FastMCP

# Create the Echo server
echo_mcp = FastMCP(name="EchoServer", stateless_http=True, json_response=True)


@echo_mcp.tool()
def echo(message: str) -> str:
    """A simple echo tool"""
    return f"Echo: {message}"


# Create the Math server
math_mcp = FastMCP(name="MathServer", stateless_http=True, json_response=True)


@math_mcp.tool()
def add_two(n: int) -> int:
    """Tool to add two to the input"""
    return n + 2


# Create a combined lifespan to manage both session managers
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with contextlib.AsyncExitStack() as stack:
        await stack.enter_async_context(echo_mcp.session_manager.run())
        await stack.enter_async_context(math_mcp.session_manager.run())
        yield


# Create the Starlette app and mount the MCP servers
app = Starlette(
    routes=[
        Mount("/echo", echo_mcp.streamable_http_app()),
        Mount("/math", math_mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)

# Note: Clients connect to http://localhost:8000/echo/mcp and http://localhost:8000/math/mcp
# To mount at the root of each path (e.g., /echo instead of /echo/mcp):
# echo_mcp.settings.streamable_http_path = "/"
# math_mcp.settings.streamable_http_path = "/"
```

_Full example: [examples/snippets/servers/streamable_starlette_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_starlette_mount.py)_
<!-- /snippet-source -->

For low level server with Streamable HTTP implementations, see:

- Stateful server: [`examples/servers/simple-streamablehttp/`](https://github.com/modelcontextprotocol/python-sdk/tree/v1.x/examples/servers/simple-streamablehttp)
- Stateless server: [`examples/servers/simple-streamablehttp-stateless/`](https://github.com/modelcontextprotocol/python-sdk/tree/v1.x/examples/servers/simple-streamablehttp-stateless)

The streamable HTTP transport supports:

- Stateful and stateless operation modes
- Resumability with event stores
- JSON or SSE response formats
- Better scalability for multi-node deployments

#### CORS Configuration for Browser-Based Clients

If you'd like your server to be accessible by browser-based MCP clients, you'll need to configure CORS headers. The `Mcp-Session-Id` header must be exposed for browser clients to access it:

```python
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware

# Create your Starlette app first
starlette_app = Starlette(routes=[...])

# Then wrap it with CORS middleware
starlette_app = CORSMiddleware(
    starlette_app,
    allow_origins=["*"],  # Configure appropriately for production
    allow_methods=["GET", "POST", "DELETE"],  # MCP streamable HTTP methods
    expose_headers=["Mcp-Session-Id"],
)
```

This configuration is necessary because:

- The MCP streamable HTTP transport uses the `Mcp-Session-Id` header for session management
- Browsers restrict access to response headers unless explicitly exposed via CORS
- Without this configuration, browser-based clients won't be able to read the session ID from initialization responses

### Mounting to an Existing ASGI Server

By default, SSE servers are mounted at `/sse` and Streamable HTTP servers are mounted at `/mcp`. You can customize these paths using the methods described below.

For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).

#### StreamableHTTP servers

You can mount the StreamableHTTP server to an existing ASGI server using the `streamable_http_app` method. This allows you to integrate the StreamableHTTP server with other ASGI applications.

##### Basic mounting

<!-- snippet-source examples/snippets/servers/streamable_http_basic_mounting.py -->
```python
"""
Basic example showing how to mount StreamableHTTP server in Starlette.

Run from the repository root:
    uvicorn examples.snippets.servers.streamable_http_basic_mounting:app --reload
"""

import contextlib

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

from mcp.server.fastmcp import FastMCP

# Create MCP server
mcp = FastMCP("My App", json_response=True)


@mcp.tool()
def hello() -> str:
    """A simple hello tool"""
    return "Hello from MCP!"


# Create a lifespan context manager to run the session manager
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with mcp.session_manager.run():
        yield


# Mount the StreamableHTTP server to the existing ASGI server
app = Starlette(
    routes=[
        Mount("/", app=mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)
```

_Full example: [examples/snippets/servers/streamable_http_basic_mounting.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_http_basic_mounting.py)_
<!-- /snippet-source -->

##### Host-based routing

<!-- snippet-source examples/snippets/servers/streamable_http_host_mounting.py -->
```python
"""
Example showing how to mount StreamableHTTP server using Host-based routing.

Run from the repository root:
    uvicorn examples.snippets.servers.streamable_http_host_mounting:app --reload
"""

import contextlib

from starlette.applications import Starlette
from starlette.routing import Host

from mcp.server.fastmcp import FastMCP

# Create MCP server
mcp = FastMCP("MCP Host App", json_response=True)


@mcp.tool()
def domain_info() -> str:
    """Get domain-specific information"""
    return "This is served from mcp.acme.corp"


# Create a lifespan context manager to run the session manager
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with mcp.session_manager.run():
        yield


# Mount using Host-based routing
app = Starlette(
    routes=[
        Host("mcp.acme.corp", app=mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)
```

_Full example: [examples/snippets/servers/streamable_http_host_mounting.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_http_host_mounting.py)_
<!-- /snippet-source -->

##### Multiple servers with path configuration

<!-- snippet-source examples/snippets/servers/streamable_http_multiple_servers.py -->
```python
"""
Example showing how to mount multiple StreamableHTTP servers with path configuration.

Run from the repository root:
    uvicorn examples.snippets.servers.streamable_http_multiple_servers:app --reload
"""

import contextlib

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

from mcp.server.fastmcp import FastMCP

# Create multiple MCP servers
api_mcp = FastMCP("API Server", json_response=True)
chat_mcp = FastMCP("Chat Server", json_response=True)


@api_mcp.tool()
def api_status() -> str:
    """Get API status"""
    return "API is running"


@chat_mcp.tool()
def send_message(message: str) -> str:
    """Send a chat message"""
    return f"Message sent: {message}"


# Configure servers to mount at the root of each path
# This means endpoints will be at /api and /chat instead of /api/mcp and /chat/mcp
api_mcp.settings.streamable_http_path = "/"
chat_mcp.settings.streamable_http_path = "/"


# Create a combined lifespan to manage both session managers
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with contextlib.AsyncExitStack() as stack:
        await stack.enter_async_context(api_mcp.session_manager.run())
        await stack.enter_async_context(chat_mcp.session_manager.run())
        yield


# Mount the servers
app = Starlette(
    routes=[
        Mount("/api", app=api_mcp.streamable_http_app()),
        Mount("/chat", app=chat_mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)
```

_Full example: [examples/snippets/servers/streamable_http_multiple_servers.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_http_multiple_servers.py)_
<!-- /snippet-source -->

##### Path configuration at initialization

<!-- snippet-source examples/snippets/servers/streamable_http_path_config.py -->
```python
"""
Example showing path configuration during FastMCP initialization.

Run from the repository root:
    uvicorn examples.snippets.servers.streamable_http_path_config:app --reload
"""

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

from mcp.server.fastmcp import FastMCP

# Configure streamable_http_path during initialization
# This server will mount at the root of wherever it's mounted
mcp_at_root = FastMCP(
    "My Server",
    json_response=True,
    streamable_http_path="/",
)


@mcp_at_root.tool()
def process_data(data: str) -> str:
    """Process some data"""
    return f"Processed: {data}"


# Mount at /process - endpoints will be at /process instead of /process/mcp
app = Starlette(
    routes=[
        Mount("/process", app=mcp_at_root.streamable_http_app()),
    ]
)
```

_Full example: [examples/snippets/servers/streamable_http_path_config.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/streamable_http_path_config.py)_
<!-- /snippet-source -->

#### SSE servers

> **Note**: SSE transport is being superseded by [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#streamable-http).

You can mount the SSE server to an existing ASGI server using the `sse_app` method. This allows you to integrate the SSE server with other ASGI applications.

```python
from starlette.applications import Starlette
from starlette.routing import Mount, Host
from mcp.server.fastmcp import FastMCP


mcp = FastMCP("My App")

# Mount the SSE server to the existing ASGI server
app = Starlette(
    routes=[
        Mount('/', app=mcp.sse_app()),
    ]
)

# or dynamically mount as host
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
```

When mounting multiple MCP servers under different paths, you can configure the mount path in several ways:

```python
from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.fastmcp import FastMCP

# Create multiple MCP servers
github_mcp = FastMCP("GitHub API")
browser_mcp = FastMCP("Browser")
curl_mcp = FastMCP("Curl")
search_mcp = FastMCP("Search")

# Method 1: Configure mount paths via settings (recommended for persistent configuration)
github_mcp.settings.mount_path = "/github"
browser_mcp.settings.mount_path = "/browser"

# Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting)
# This approach doesn't modify the server's settings permanently

# Create Starlette app with multiple mounted servers
app = Starlette(
    routes=[
        # Using settings-based configuration
        Mount("/github", app=github_mcp.sse_app()),
        Mount("/browser", app=browser_mcp.sse_app()),
        # Using direct mount path parameter
        Mount("/curl", app=curl_mcp.sse_app("/curl")),
        Mount("/search", app=search_mcp.sse_app("/search")),
    ]
)

# Method 3: For direct execution, you can also pass the mount path to run()
if __name__ == "__main__":
    search_mcp.run(transport="sse", mount_path="/search")
```

For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).

## Advanced Usage

For the low-level server API, pagination, and direct handler registration, see [Low-Level Server](https://py.sdk.modelcontextprotocol.io/low-level-server/index.md).

# Writing Clients

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

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](https://py.sdk.modelcontextprotocol.io/authorization/index.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())
```

# Protocol Features

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

This page covers cross-cutting MCP protocol features.

## MCP Primitives

The MCP protocol defines three core primitives that servers can implement:

| Primitive | Control               | Description                                         | Example Use                  |
|-----------|-----------------------|-----------------------------------------------------|------------------------------|
| Prompts   | User-controlled       | Interactive templates invoked by user choice        | Slash commands, menu options |
| Resources | Application-controlled| Contextual data managed by the client application   | File contents, API responses |
| Tools     | Model-controlled      | Functions exposed to the LLM to take actions        | API calls, data updates      |

## Server Capabilities

MCP servers declare capabilities during initialization:

| Capability   | Feature Flag                 | Description                        |
|--------------|------------------------------|------------------------------------|
| `prompts`    | `listChanged`                | Prompt template management         |
| `resources`  | `subscribe`<br/>`listChanged`| Resource exposure and updates      |
| `tools`      | `listChanged`                | Tool discovery and execution       |
| `logging`    | -                            | Server logging configuration       |
| `completions`| -                            | Argument completion suggestions    |

## Ping

Both clients and servers can send ping requests to check that the other side is responsive:

```python
# From a client
result = await session.send_ping()

# From a server (via ServerSession)
result = await server_session.send_ping()
```

Both return an `EmptyResult` on success. If the remote side does not respond within the session timeout, an exception is raised.

## Cancellation

Either side can cancel a previously-issued request by sending a `CancelledNotification`:

<!-- snippet-source examples/snippets/clients/cancellation.py -->
```python
import mcp.types as types
from mcp import ClientSession


async def cancel_request(session: ClientSession) -> None:
    """Send a cancellation notification for a previously-issued request."""
    await session.send_notification(
        types.ClientNotification(
            types.CancelledNotification(
                params=types.CancelledNotificationParams(
                    requestId="request-id-to-cancel",
                    reason="User navigated away",
                )
            )
        )
    )
```

_Full example: [examples/snippets/clients/cancellation.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/cancellation.py)_
<!-- /snippet-source -->

The `CancelledNotificationParams` fields:

- `requestId` (optional): The ID of the request to cancel. Required for non-task cancellations.
- `reason` (optional): A human-readable string describing why the request was cancelled.

## Capability Negotiation

During initialization, the client and server exchange capability declarations. The Python SDK automatically declares capabilities based on which callbacks and handlers are registered:

**Client capabilities** (auto-declared when callbacks are provided):

- `sampling` -- declared when `sampling_callback` is passed to `ClientSession`
- `roots` -- declared when `list_roots_callback` is passed to `ClientSession`
- `elicitation` -- declared when `elicitation_callback` is passed to `ClientSession`

**Server capabilities** (auto-declared when handlers are registered):

- `prompts` -- declared when a `list_prompts` handler is registered
- `resources` -- declared when a `list_resources` handler is registered
- `tools` -- declared when a `list_tools` handler is registered
- `logging` -- declared when a `set_logging_level` handler is registered
- `completions` -- declared when a `completion` handler is registered

After initialization, clients can inspect server capabilities:

```python
capabilities = session.get_server_capabilities()
if capabilities and capabilities.tools:
    tools = await session.list_tools()
```

## Protocol Version Negotiation

The SDK defines `LATEST_PROTOCOL_VERSION` and `SUPPORTED_PROTOCOL_VERSIONS` in `mcp.shared.version`:

```python
from mcp.shared.version import LATEST_PROTOCOL_VERSION, SUPPORTED_PROTOCOL_VERSIONS

# LATEST_PROTOCOL_VERSION is the version the SDK advertises during initialization
# SUPPORTED_PROTOCOL_VERSIONS lists all versions the SDK can work with
```

During initialization, the client sends `LATEST_PROTOCOL_VERSION`. If the server responds with a version not in `SUPPORTED_PROTOCOL_VERSIONS`, the client raises a `RuntimeError`. This ensures both sides agree on a compatible protocol version before exchanging messages.

## JSON Schema (2020-12)

MCP uses [JSON Schema 2020-12](https://json-schema.org/draft/2020-12) for tool input schemas, output schemas, and elicitation schemas. When using Pydantic models, schemas are generated automatically via `model_json_schema()`:

<!-- snippet-source examples/snippets/servers/json_schema_example.py -->
```python
from pydantic import BaseModel, Field


class SearchParams(BaseModel):
    query: str = Field(description="Search query string")
    max_results: int = Field(default=10, description="Maximum results to return")


# Pydantic generates a JSON Schema 2020-12 compatible schema:
schema = SearchParams.model_json_schema()
# {
#     "properties": {
#         "query": {"description": "Search query string", "type": "string"},
#         "max_results": {
#             "default": 10,
#             "description": "Maximum results to return",
#             "type": "integer",
#         },
#     },
#     "required": ["query"],
#     "title": "SearchParams",
#     "type": "object",
# }
```

_Full example: [examples/snippets/servers/json_schema_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/servers/json_schema_example.py)_
<!-- /snippet-source -->

For FastMCP tools, input schemas are derived automatically from function signatures. For structured output, the output schema is derived from the return type annotation.

## Pagination

For pagination details, see:

- Server-side implementation: [Low-Level Server - Pagination](https://py.sdk.modelcontextprotocol.io/low-level-server/index.md#pagination-advanced)
- Client-side consumption: [Low-Level Server - Client-side Consumption](https://py.sdk.modelcontextprotocol.io/low-level-server/index.md#client-side-consumption)

# Low-Level Server

Source: https://py.sdk.modelcontextprotocol.io/low-level-server/

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.

## 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.

# Authorization

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

This page covers OAuth 2.1 authentication for both MCP servers and clients.

## 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).

# Testing

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

If you call yourself a developer, you will want to test your MCP server.
The Python SDK offers the `create_connected_server_and_client_session` function to create a session
using an in-memory transport. I know, I know, the name is too long... We are working on improving it.

Anyway, let's assume you have a simple server with a single tool:

```python title="server.py"
from mcp.server import FastMCP

app = FastMCP("Calculator")

@app.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""  # (1)!
    return a + b
```

1. The docstring is automatically added as the description of the tool.

To run the below test, you'll need to install the following dependencies:

=== "pip"
    ```bash
    pip install inline-snapshot pytest
    ```

=== "uv"
    ```bash
    uv add inline-snapshot pytest
    ```

!!! info
    I think [`pytest`](https://docs.pytest.org/en/stable/) is a pretty standard testing framework,
    so I won't go into details here.

    The [`inline-snapshot`](https://15r10nk.github.io/inline-snapshot/latest/) is a library that allows
    you to take snapshots of the output of your tests. Which makes it easier to create tests for your
    server - you don't need to use it, but we are spreading the word for best practices.

```python title="test_server.py"
from collections.abc import AsyncGenerator

import pytest
from inline_snapshot import snapshot
from mcp.client.session import ClientSession
from mcp.shared.memory import create_connected_server_and_client_session
from mcp.types import CallToolResult, TextContent

from server import app


@pytest.fixture
def anyio_backend():  # (1)!
    return "asyncio"


@pytest.fixture
async def client_session() -> AsyncGenerator[ClientSession]:
    async with create_connected_server_and_client_session(app, raise_exceptions=True) as _session:
        yield _session


@pytest.mark.anyio
async def test_call_add_tool(client_session: ClientSession):
    result = await client_session.call_tool("add", {"a": 1, "b": 2})
    assert result == snapshot(
        CallToolResult(
            content=[TextContent(type="text", text="3")],
            structuredContent={"result": 3},
        )
    )
```

1. If you are using `trio`, you should set `"trio"` as the `anyio_backend`. Check more information in the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on).

There you go! You can now extend your tests to cover more scenarios.

# Overview

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

!!! warning "Deprecated"

    The experimental tasks API is deprecated and will be removed in mcp 2.0.
    Tasks (SEP-1686) were removed from the MCP specification and are expected
    to return as a separate MCP extension in a future release.

This section documents experimental features in the MCP Python SDK. These features
are deprecated and remain available on the 1.x line only for existing users.

## Available Experimental Features

### [Tasks](https://py.sdk.modelcontextprotocol.io/experimental/tasks/index.md)

Tasks enable asynchronous execution of MCP operations. Instead of waiting for a
long-running operation to complete, the server returns a task reference immediately.
Clients can then poll for status updates and retrieve results when ready.

Tasks are useful for:

- **Long-running computations** that would otherwise block
- **Batch operations** that process many items
- **Interactive workflows** that require user input (elicitation) or LLM assistance (sampling)

## Using Experimental APIs

Experimental features are accessed via the `.experimental` property:

```python
# Server-side
@server.experimental.get_task()
async def handle_get_task(request: GetTaskRequest) -> GetTaskResult:
    ...

# Client-side
result = await session.experimental.call_tool_as_task("tool_name", {"arg": "value"})
```

Accessing the `.experimental` properties emits a `DeprecationWarning`.

## Providing Feedback

If you rely on these features and have feedback on their deprecation or the planned
MCP extension, please open an issue on the
[python-sdk repository](https://github.com/modelcontextprotocol/python-sdk/issues).

# Introduction

Source: https://py.sdk.modelcontextprotocol.io/experimental/tasks/

!!! warning "Deprecated"

    The experimental tasks API is deprecated and will be removed in mcp 2.0.
    Tasks (SEP-1686) were removed from the MCP specification and are expected
    to return as a separate MCP extension in a future release.

Tasks enable asynchronous request handling in MCP. Instead of blocking until an operation completes, the receiver creates a task, returns immediately, and the requestor polls for the result.

## When to Use Tasks

Tasks are designed for operations that:

- Take significant time (seconds to minutes)
- Need progress updates during execution
- Require user input mid-execution (elicitation, sampling)
- Should run without blocking the requestor

Common use cases:

- Long-running data processing
- Multi-step workflows with user confirmation
- LLM-powered operations requiring sampling
- OAuth flows requiring user browser interaction

## Task Lifecycle

```text
                    ┌─────────────┐
                    │   working   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
     ┌────────────┐  ┌───────────┐  ┌───────────┐
     │ completed  │  │  failed   │  │ cancelled │
     └────────────┘  └───────────┘  └───────────┘
              ▲
              │
     ┌────────┴────────┐
     │ input_required  │◄──────┐
     └────────┬────────┘       │
              │                │
              └────────────────┘
```

| Status | Description |
|--------|-------------|
| `working` | Task is being processed |
| `input_required` | Receiver needs input from requestor (elicitation/sampling) |
| `completed` | Task finished successfully |
| `failed` | Task encountered an error |
| `cancelled` | Task was cancelled by requestor |

Terminal states (`completed`, `failed`, `cancelled`) are final—tasks cannot transition out of them.

## Bidirectional Flow

Tasks work in both directions:

**Client → Server** (most common):

```text
Client                              Server
  │                                    │
  │── tools/call (task) ──────────────>│ Creates task
  │<── CreateTaskResult ───────────────│
  │                                    │
  │── tasks/get ──────────────────────>│
  │<── status: working ────────────────│
  │                                    │ ... work continues ...
  │── tasks/get ──────────────────────>│
  │<── status: completed ──────────────│
  │                                    │
  │── tasks/result ───────────────────>│
  │<── CallToolResult ─────────────────│
```

**Server → Client** (for elicitation/sampling):

```text
Server                              Client
  │                                    │
  │── elicitation/create (task) ──────>│ Creates task
  │<── CreateTaskResult ───────────────│
  │                                    │
  │── tasks/get ──────────────────────>│
  │<── status: working ────────────────│
  │                                    │ ... user interaction ...
  │── tasks/get ──────────────────────>│
  │<── status: completed ──────────────│
  │                                    │
  │── tasks/result ───────────────────>│
  │<── ElicitResult ───────────────────│
```

## Key Concepts

### Task Metadata

When augmenting a request with task execution, include `TaskMetadata`:

```python
from mcp.types import TaskMetadata

task = TaskMetadata(ttl=60000)  # TTL in milliseconds
```

The `ttl` (time-to-live) specifies how long the task and result are retained after completion.

### Task Store

Servers persist task state in a `TaskStore`. The SDK provides `InMemoryTaskStore` for development:

```python
from mcp.shared.experimental.tasks import InMemoryTaskStore

store = InMemoryTaskStore()
```

For production, implement `TaskStore` with a database or distributed cache.

### Capabilities

Both servers and clients declare task support through capabilities:

**Server capabilities:**

- `tasks.requests.tools.call` - Server accepts task-augmented tool calls

**Client capabilities:**

- `tasks.requests.sampling.createMessage` - Client accepts task-augmented sampling
- `tasks.requests.elicitation.create` - Client accepts task-augmented elicitation

The SDK manages these automatically when you enable task support.

## Quick Example

**Server** (simplified API):

```python
from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.types import CallToolResult, TextContent, TASK_REQUIRED

server = Server("my-server")
server.experimental.enable_tasks()  # One-line setup

@server.call_tool()
async def handle_tool(name: str, arguments: dict):
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    async def work(task: ServerTaskContext):
        await task.update_status("Processing...")
        # ... do work ...
        return CallToolResult(content=[TextContent(type="text", text="Done!")])

    return await ctx.experimental.run_task(work)
```

**Client:**

```python
from mcp.client.session import ClientSession
from mcp.types import CallToolResult

async with ClientSession(read, write) as session:
    await session.initialize()

    # Call tool as task
    result = await session.experimental.call_tool_as_task("my_tool", {"arg": "value"})
    task_id = result.task.taskId

    # Poll until done
    async for status in session.experimental.poll_task(task_id):
        print(f"Status: {status.status}")

    # Get result
    final = await session.experimental.get_task_result(task_id, CallToolResult)
```

## Next Steps

- [Server Implementation](https://py.sdk.modelcontextprotocol.io/experimental/tasks-server/index.md) - Build task-supporting servers
- [Client Usage](https://py.sdk.modelcontextprotocol.io/experimental/tasks-client/index.md) - Call and poll tasks from clients

# Server Implementation

Source: https://py.sdk.modelcontextprotocol.io/experimental/tasks-server/

!!! warning "Deprecated"

    The experimental tasks API is deprecated and will be removed in mcp 2.0.
    Tasks (SEP-1686) were removed from the MCP specification and are expected
    to return as a separate MCP extension in a future release.

This guide covers implementing task support in MCP servers, from basic setup to advanced patterns like elicitation and sampling within tasks.

## Quick Start

The simplest way to add task support:

```python
from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.types import CallToolResult, CreateTaskResult, TextContent, Tool, ToolExecution, TASK_REQUIRED

server = Server("my-server")
server.experimental.enable_tasks()  # Registers all task handlers automatically

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="process_data",
            description="Process data asynchronously",
            inputSchema={"type": "object", "properties": {"input": {"type": "string"}}},
            execution=ToolExecution(taskSupport=TASK_REQUIRED),
        )
    ]

@server.call_tool()
async def handle_tool(name: str, arguments: dict) -> CallToolResult | CreateTaskResult:
    if name == "process_data":
        return await handle_process_data(arguments)
    return CallToolResult(content=[TextContent(type="text", text=f"Unknown: {name}")], isError=True)

async def handle_process_data(arguments: dict) -> CreateTaskResult:
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    async def work(task: ServerTaskContext) -> CallToolResult:
        await task.update_status("Processing...")
        result = arguments.get("input", "").upper()
        return CallToolResult(content=[TextContent(type="text", text=result)])

    return await ctx.experimental.run_task(work)
```

That's it. `enable_tasks()` automatically:

- Creates an in-memory task store
- Registers handlers for `tasks/get`, `tasks/result`, `tasks/list`, `tasks/cancel`
- Updates server capabilities

## Task Visibility

Task IDs generated by `run_task()` embed an opaque marker identifying the session that
created the task, and the default handlers use it to restrict each session to its own
tasks: `tasks/get`, `tasks/result`, and `tasks/cancel` respond with "task not found" for
another session's task, and `tasks/list` returns only the requesting session's tasks. A
client that reconnects gets a new session and can no longer reach tasks it created on the
previous one.

A task ID has no session marker when it was passed to `run_task()` explicitly, when the
task was created directly through the `TaskStore`, or when the server runs in stateless
mode (each request gets a fresh session, so tasks must remain reachable across requests).
Such tasks are accessible to any requestor that presents the exact task ID, and are never
included in `tasks/list` responses because the server cannot tell which session they
belong to. Treat these task IDs as capabilities: generate them with enough entropy that
they cannot be guessed, share them only with the intended recipient, and prefer short
TTLs. Passing an explicit `task_id` to `run_task()` is deprecated for this reason.

To scope tasks to something other than the session — for example a user identity from your
authorization layer — register your own handlers with `@server.experimental.get_task()`,
`@server.experimental.get_task_result()`, `@server.experimental.list_tasks()`, and
`@server.experimental.cancel_task()` instead of relying on the defaults.

## Tool Declaration

Tools declare task support via the `execution.taskSupport` field:

```python
from mcp.types import Tool, ToolExecution, TASK_REQUIRED, TASK_OPTIONAL, TASK_FORBIDDEN

Tool(
    name="my_tool",
    inputSchema={"type": "object"},
    execution=ToolExecution(taskSupport=TASK_REQUIRED),  # or TASK_OPTIONAL, TASK_FORBIDDEN
)
```

| Value | Meaning |
|-------|---------|
| `TASK_REQUIRED` | Tool **must** be called as a task |
| `TASK_OPTIONAL` | Tool supports both sync and task execution |
| `TASK_FORBIDDEN` | Tool **cannot** be called as a task (default) |

Validate the request matches your tool's requirements:

```python
@server.call_tool()
async def handle_tool(name: str, arguments: dict):
    ctx = server.request_context

    if name == "required_task_tool":
        ctx.experimental.validate_task_mode(TASK_REQUIRED)  # Raises if not task mode
        return await handle_as_task(arguments)

    elif name == "optional_task_tool":
        if ctx.experimental.is_task:
            return await handle_as_task(arguments)
        else:
            return handle_sync(arguments)
```

## The run_task Pattern

`run_task()` is the recommended way to execute task work:

```python
async def handle_my_tool(arguments: dict) -> CreateTaskResult:
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    async def work(task: ServerTaskContext) -> CallToolResult:
        # Your work here
        return CallToolResult(content=[TextContent(type="text", text="Done")])

    return await ctx.experimental.run_task(work)
```

**What `run_task()` does:**

1. Creates a task in the store
2. Spawns your work function in the background
3. Returns `CreateTaskResult` immediately
4. Auto-completes the task when your function returns
5. Auto-fails the task if your function raises

**The `ServerTaskContext` provides:**

- `task.task_id` - The task identifier
- `task.update_status(message)` - Update progress
- `task.complete(result)` - Explicitly complete (usually automatic)
- `task.fail(error)` - Explicitly fail
- `task.is_cancelled` - Check if cancellation requested

## Status Updates

Keep clients informed of progress:

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Starting...")

    for i, item in enumerate(items):
        await task.update_status(f"Processing {i+1}/{len(items)}")
        await process_item(item)

    await task.update_status("Finalizing...")
    return CallToolResult(content=[TextContent(type="text", text="Complete")])
```

Status messages appear in `tasks/get` responses, letting clients show progress to users.

## Elicitation Within Tasks

Tasks can request user input via elicitation. This transitions the task to `input_required` status.

### Form Elicitation

Collect structured data from the user:

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Waiting for confirmation...")

    result = await task.elicit(
        message="Delete these files?",
        requestedSchema={
            "type": "object",
            "properties": {
                "confirm": {"type": "boolean"},
                "reason": {"type": "string"},
            },
            "required": ["confirm"],
        },
    )

    if result.action == "accept" and result.content.get("confirm"):
        # User confirmed
        return CallToolResult(content=[TextContent(type="text", text="Files deleted")])
    else:
        # User declined or cancelled
        return CallToolResult(content=[TextContent(type="text", text="Cancelled")])
```

### URL Elicitation

Direct users to external URLs for OAuth, payments, or other out-of-band flows:

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Waiting for OAuth...")

    result = await task.elicit_url(
        message="Please authorize with GitHub",
        url="https://github.com/login/oauth/authorize?client_id=...",
        elicitation_id="oauth-github-123",
    )

    if result.action == "accept":
        # User completed OAuth flow
        return CallToolResult(content=[TextContent(type="text", text="Connected to GitHub")])
    else:
        return CallToolResult(content=[TextContent(type="text", text="OAuth cancelled")])
```

## Sampling Within Tasks

Tasks can request LLM completions from the client:

```python
from mcp.types import SamplingMessage, TextContent

async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Generating response...")

    result = await task.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(type="text", text="Write a haiku about coding"),
            )
        ],
        max_tokens=100,
    )

    haiku = result.content.text if isinstance(result.content, TextContent) else "Error"
    return CallToolResult(content=[TextContent(type="text", text=haiku)])
```

Sampling supports additional parameters:

```python
result = await task.create_message(
    messages=[...],
    max_tokens=500,
    system_prompt="You are a helpful assistant",
    temperature=0.7,
    stop_sequences=["\n\n"],
    model_preferences=ModelPreferences(hints=[ModelHint(name="claude-3")]),
)
```

## Cancellation Support

Check for cancellation in long-running work:

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    for i in range(1000):
        if task.is_cancelled:
            # Clean up and exit
            return CallToolResult(content=[TextContent(type="text", text="Cancelled")])

        await task.update_status(f"Step {i}/1000")
        await process_step(i)

    return CallToolResult(content=[TextContent(type="text", text="Complete")])
```

The SDK's default cancel handler updates the task status. Your work function should check `is_cancelled` periodically.

## Custom Task Store

For production, implement `TaskStore` with persistent storage:

```python
from mcp.shared.experimental.tasks.store import TaskStore
from mcp.types import Task, TaskMetadata, Result

class RedisTaskStore(TaskStore):
    def __init__(self, redis_client):
        self.redis = redis_client

    async def create_task(self, metadata: TaskMetadata, task_id: str | None = None) -> Task:
        # Create and persist task
        ...

    async def get_task(self, task_id: str) -> Task | None:
        # Retrieve task from Redis
        ...

    async def update_task(self, task_id: str, status: str | None = None, ...) -> Task:
        # Update and persist
        ...

    async def store_result(self, task_id: str, result: Result) -> None:
        # Store result in Redis
        ...

    async def get_result(self, task_id: str) -> Result | None:
        # Retrieve result
        ...

    # ... implement remaining methods
```

Use your custom store:

```python
store = RedisTaskStore(redis_client)
server.experimental.enable_tasks(store=store)
```

## Complete Example

A server with multiple task-supporting tools:

```python
from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.types import (
    CallToolResult, CreateTaskResult, TextContent, Tool, ToolExecution,
    SamplingMessage, TASK_REQUIRED,
)

server = Server("task-demo")
server.experimental.enable_tasks()


@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="confirm_action",
            description="Requires user confirmation",
            inputSchema={"type": "object", "properties": {"action": {"type": "string"}}},
            execution=ToolExecution(taskSupport=TASK_REQUIRED),
        ),
        Tool(
            name="generate_text",
            description="Generate text via LLM",
            inputSchema={"type": "object", "properties": {"prompt": {"type": "string"}}},
            execution=ToolExecution(taskSupport=TASK_REQUIRED),
        ),
    ]


async def handle_confirm_action(arguments: dict) -> CreateTaskResult:
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    action = arguments.get("action", "unknown action")

    async def work(task: ServerTaskContext) -> CallToolResult:
        result = await task.elicit(
            message=f"Confirm: {action}?",
            requestedSchema={
                "type": "object",
                "properties": {"confirm": {"type": "boolean"}},
                "required": ["confirm"],
            },
        )

        if result.action == "accept" and result.content.get("confirm"):
            return CallToolResult(content=[TextContent(type="text", text=f"Executed: {action}")])
        return CallToolResult(content=[TextContent(type="text", text="Cancelled")])

    return await ctx.experimental.run_task(work)


async def handle_generate_text(arguments: dict) -> CreateTaskResult:
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    prompt = arguments.get("prompt", "Hello")

    async def work(task: ServerTaskContext) -> CallToolResult:
        await task.update_status("Generating...")

        result = await task.create_message(
            messages=[SamplingMessage(role="user", content=TextContent(type="text", text=prompt))],
            max_tokens=200,
        )

        text = result.content.text if isinstance(result.content, TextContent) else "Error"
        return CallToolResult(content=[TextContent(type="text", text=text)])

    return await ctx.experimental.run_task(work)


@server.call_tool()
async def handle_tool(name: str, arguments: dict) -> CallToolResult | CreateTaskResult:
    if name == "confirm_action":
        return await handle_confirm_action(arguments)
    elif name == "generate_text":
        return await handle_generate_text(arguments)
    return CallToolResult(content=[TextContent(type="text", text=f"Unknown: {name}")], isError=True)
```

## Error Handling in Tasks

Tasks handle errors automatically, but you can also fail explicitly:

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    try:
        result = await risky_operation()
        return CallToolResult(content=[TextContent(type="text", text=result)])
    except PermissionError:
        await task.fail("Access denied - insufficient permissions")
        raise
    except TimeoutError:
        await task.fail("Operation timed out after 30 seconds")
        raise
```

When `run_task()` catches an exception, it automatically:

1. Marks the task as `failed`
2. Sets `statusMessage` to the exception message
3. Propagates the exception (which is caught by the task group)

For custom error messages, call `task.fail()` before raising.

## HTTP Transport Example

For web applications, use the Streamable HTTP transport:

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount

from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import (
    CallToolResult, CreateTaskResult, TextContent, Tool, ToolExecution, TASK_REQUIRED,
)


server = Server("http-task-server")
server.experimental.enable_tasks()


@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="long_operation",
            description="A long-running operation",
            inputSchema={"type": "object", "properties": {"duration": {"type": "number"}}},
            execution=ToolExecution(taskSupport=TASK_REQUIRED),
        )
    ]


async def handle_long_operation(arguments: dict) -> CreateTaskResult:
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    duration = arguments.get("duration", 5)

    async def work(task: ServerTaskContext) -> CallToolResult:
        import anyio
        for i in range(int(duration)):
            await task.update_status(f"Step {i+1}/{int(duration)}")
            await anyio.sleep(1)
        return CallToolResult(content=[TextContent(type="text", text=f"Completed after {duration}s")])

    return await ctx.experimental.run_task(work)


@server.call_tool()
async def handle_tool(name: str, arguments: dict) -> CallToolResult | CreateTaskResult:
    if name == "long_operation":
        return await handle_long_operation(arguments)
    return CallToolResult(content=[TextContent(type="text", text=f"Unknown: {name}")], isError=True)


def create_app():
    session_manager = StreamableHTTPSessionManager(app=server)

    @asynccontextmanager
    async def lifespan(app: Starlette) -> AsyncIterator[None]:
        async with session_manager.run():
            yield

    return Starlette(
        routes=[Mount("/mcp", app=session_manager.handle_request)],
        lifespan=lifespan,
    )


if __name__ == "__main__":
    uvicorn.run(create_app(), host="127.0.0.1", port=8000)
```

## Testing Task Servers

Test task functionality with the SDK's testing utilities:

```python
import pytest
import anyio
from mcp.client.session import ClientSession
from mcp.types import CallToolResult


@pytest.mark.anyio
async def test_task_tool():
    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream(10)
    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream(10)

    async def run_server():
        await server.run(
            client_to_server_receive,
            server_to_client_send,
            server.create_initialization_options(),
        )

    async def run_client():
        async with ClientSession(server_to_client_receive, client_to_server_send) as session:
            await session.initialize()

            # Call the tool as a task
            result = await session.experimental.call_tool_as_task("my_tool", {"arg": "value"})
            task_id = result.task.taskId
            assert result.task.status == "working"

            # Poll until complete
            async for status in session.experimental.poll_task(task_id):
                if status.status in ("completed", "failed"):
                    break

            # Get result
            final = await session.experimental.get_task_result(task_id, CallToolResult)
            assert len(final.content) > 0

    async with anyio.create_task_group() as tg:
        tg.start_soon(run_server)
        tg.start_soon(run_client)
```

## Best Practices

### Keep Work Functions Focused

```python
# Good: focused work function
async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Validating...")
    validate_input(arguments)

    await task.update_status("Processing...")
    result = await process_data(arguments)

    return CallToolResult(content=[TextContent(type="text", text=result)])
```

### Check Cancellation in Loops

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    results = []
    for item in large_dataset:
        if task.is_cancelled:
            return CallToolResult(content=[TextContent(type="text", text="Cancelled")])

        results.append(await process(item))

    return CallToolResult(content=[TextContent(type="text", text=str(results))])
```

### Use Meaningful Status Messages

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    await task.update_status("Connecting to database...")
    db = await connect()

    await task.update_status("Fetching records (0/1000)...")
    for i, record in enumerate(records):
        if i % 100 == 0:
            await task.update_status(f"Processing records ({i}/1000)...")
        await process(record)

    await task.update_status("Finalizing results...")
    return CallToolResult(content=[TextContent(type="text", text="Done")])
```

### Handle Elicitation Responses

```python
async def work(task: ServerTaskContext) -> CallToolResult:
    result = await task.elicit(message="Continue?", requestedSchema={...})

    match result.action:
        case "accept":
            # User accepted, process content
            return await process_accepted(result.content)
        case "decline":
            # User explicitly declined
            return CallToolResult(content=[TextContent(type="text", text="User declined")])
        case "cancel":
            # User cancelled the elicitation
            return CallToolResult(content=[TextContent(type="text", text="Cancelled")])
```

## Next Steps

- [Client Usage](https://py.sdk.modelcontextprotocol.io/experimental/tasks-client/index.md) - Learn how clients interact with task servers
- [Tasks Overview](https://py.sdk.modelcontextprotocol.io/experimental/tasks/index.md) - Review lifecycle and concepts

# Client Usage

Source: https://py.sdk.modelcontextprotocol.io/experimental/tasks-client/

!!! warning "Deprecated"

    The experimental tasks API is deprecated and will be removed in mcp 2.0.
    Tasks (SEP-1686) were removed from the MCP specification and are expected
    to return as a separate MCP extension in a future release.

This guide covers calling task-augmented tools from clients, handling the `input_required` status, and advanced patterns like receiving task requests from servers.

## Quick Start

Call a tool as a task and poll for the result:

```python
from mcp.client.session import ClientSession
from mcp.types import CallToolResult

async with ClientSession(read, write) as session:
    await session.initialize()

    # Call tool as task
    result = await session.experimental.call_tool_as_task(
        "process_data",
        {"input": "hello"},
        ttl=60000,
    )
    task_id = result.task.taskId

    # Poll until complete
    async for status in session.experimental.poll_task(task_id):
        print(f"Status: {status.status} - {status.statusMessage or ''}")

    # Get result
    final = await session.experimental.get_task_result(task_id, CallToolResult)
    print(f"Result: {final.content[0].text}")
```

## Calling Tools as Tasks

Use `call_tool_as_task()` to invoke a tool with task augmentation:

```python
result = await session.experimental.call_tool_as_task(
    "my_tool",           # Tool name
    {"arg": "value"},    # Arguments
    ttl=60000,           # Time-to-live in milliseconds
    meta={"key": "val"}, # Optional metadata
)

task_id = result.task.taskId
print(f"Task: {task_id}, Status: {result.task.status}")
```

The response is a `CreateTaskResult` containing:

- `task.taskId` - Unique identifier for polling
- `task.status` - Initial status (usually `"working"`)
- `task.pollInterval` - Suggested polling interval (milliseconds)
- `task.ttl` - Time-to-live for results
- `task.createdAt` - Creation timestamp

## Polling with poll_task

The `poll_task()` async iterator polls until the task reaches a terminal state:

```python
async for status in session.experimental.poll_task(task_id):
    print(f"Status: {status.status}")
    if status.statusMessage:
        print(f"Progress: {status.statusMessage}")
```

It automatically:

- Respects the server's suggested `pollInterval`
- Stops when status is `completed`, `failed`, or `cancelled`
- Yields each status for progress display

### Handling input_required

When a task needs user input (elicitation), it transitions to `input_required`. You must call `get_task_result()` to receive and respond to the elicitation:

```python
async for status in session.experimental.poll_task(task_id):
    print(f"Status: {status.status}")

    if status.status == "input_required":
        # This delivers the elicitation and waits for completion
        final = await session.experimental.get_task_result(task_id, CallToolResult)
        break
```

The elicitation callback (set during session creation) handles the actual user interaction.

## Elicitation Callbacks

To handle elicitation requests from the server, provide a callback when creating the session:

```python
from mcp.types import ElicitRequestParams, ElicitResult

async def handle_elicitation(context, params: ElicitRequestParams) -> ElicitResult:
    # Display the message to the user
    print(f"Server asks: {params.message}")

    # Collect user input (this is a simplified example)
    response = input("Your response (y/n): ")
    confirmed = response.lower() == "y"

    return ElicitResult(
        action="accept",
        content={"confirm": confirmed},
    )

async with ClientSession(
    read,
    write,
    elicitation_callback=handle_elicitation,
) as session:
    await session.initialize()
    # ... call tasks that may require elicitation
```

## Sampling Callbacks

Similarly, handle sampling requests with a callback:

```python
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent

async def handle_sampling(context, params: CreateMessageRequestParams) -> CreateMessageResult:
    # In a real implementation, call your LLM here
    prompt = params.messages[-1].content.text if params.messages else ""

    # Return a mock response
    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=f"Response to: {prompt}"),
        model="my-model",
    )

async with ClientSession(
    read,
    write,
    sampling_callback=handle_sampling,
) as session:
    # ...
```

## Retrieving Results

Once a task completes, retrieve the result:

```python
if status.status == "completed":
    result = await session.experimental.get_task_result(task_id, CallToolResult)
    for content in result.content:
        if hasattr(content, "text"):
            print(content.text)

elif status.status == "failed":
    print(f"Task failed: {status.statusMessage}")

elif status.status == "cancelled":
    print("Task was cancelled")
```

The result type matches the original request:

- `tools/call` → `CallToolResult`
- `sampling/createMessage` → `CreateMessageResult`
- `elicitation/create` → `ElicitResult`

## Cancellation

Cancel a running task:

```python
cancel_result = await session.experimental.cancel_task(task_id)
print(f"Cancelled, status: {cancel_result.status}")
```

Note: Cancellation is cooperative—the server must check for and handle cancellation.

## Listing Tasks

View all tasks on the server:

```python
result = await session.experimental.list_tasks()
for task in result.tasks:
    print(f"{task.taskId}: {task.status}")

# Handle pagination
while result.nextCursor:
    result = await session.experimental.list_tasks(cursor=result.nextCursor)
    for task in result.tasks:
        print(f"{task.taskId}: {task.status}")
```

## Advanced: Client as Task Receiver

Servers can send task-augmented requests to clients. This is useful when the server needs the client to perform async work (like complex sampling or user interaction).

### Declaring Client Capabilities

Register task handlers to declare what task-augmented requests your client accepts:

```python
from mcp.client.experimental.task_handlers import ExperimentalTaskHandlers
from mcp.types import (
    CreateTaskResult, GetTaskResult, GetTaskPayloadResult,
    TaskMetadata, ElicitRequestParams,
)
from mcp.shared.experimental.tasks import InMemoryTaskStore

# Client-side task store
client_store = InMemoryTaskStore()

async def handle_augmented_elicitation(context, params: ElicitRequestParams, task_metadata: TaskMetadata):
    """Handle task-augmented elicitation from server."""
    # Create a task for this elicitation
    task = await client_store.create_task(task_metadata)

    # Start async work (e.g., show UI, wait for user)
    async def complete_elicitation():
        # ... do async work ...
        result = ElicitResult(action="accept", content={"confirm": True})
        await client_store.store_result(task.taskId, result)
        await client_store.update_task(task.taskId, status="completed")

    context.session._task_group.start_soon(complete_elicitation)

    # Return task reference immediately
    return CreateTaskResult(task=task)

async def handle_get_task(context, params):
    """Handle tasks/get from server."""
    task = await client_store.get_task(params.taskId)
    return GetTaskResult(
        taskId=task.taskId,
        status=task.status,
        statusMessage=task.statusMessage,
        createdAt=task.createdAt,
        lastUpdatedAt=task.lastUpdatedAt,
        ttl=task.ttl,
        pollInterval=100,
    )

async def handle_get_task_result(context, params):
    """Handle tasks/result from server."""
    result = await client_store.get_result(params.taskId)
    return GetTaskPayloadResult.model_validate(result.model_dump())

task_handlers = ExperimentalTaskHandlers(
    augmented_elicitation=handle_augmented_elicitation,
    get_task=handle_get_task,
    get_task_result=handle_get_task_result,
)

async with ClientSession(
    read,
    write,
    experimental_task_handlers=task_handlers,
) as session:
    # Client now accepts task-augmented elicitation from server
    await session.initialize()
```

This enables flows where:

1. Client calls a task-augmented tool
2. Server's tool work calls `task.elicit_as_task()`
3. Client receives task-augmented elicitation
4. Client creates its own task, does async work
5. Server polls client's task
6. Eventually both tasks complete

## Complete Example

A client that handles all task scenarios:

```python
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
from mcp.types import CallToolResult, ElicitRequestParams, ElicitResult


async def elicitation_callback(context, params: ElicitRequestParams) -> ElicitResult:
    print(f"\n[Elicitation] {params.message}")
    response = input("Confirm? (y/n): ")
    return ElicitResult(action="accept", content={"confirm": response.lower() == "y"})


async def main():
    async with stdio_client(command="python", args=["server.py"]) as (read, write):
        async with ClientSession(
            read,
            write,
            elicitation_callback=elicitation_callback,
        ) as session:
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Tools:", [t.name for t in tools.tools])

            # Call a task-augmented tool
            print("\nCalling task tool...")
            result = await session.experimental.call_tool_as_task(
                "confirm_action",
                {"action": "delete files"},
            )
            task_id = result.task.taskId
            print(f"Task created: {task_id}")

            # Poll and handle input_required
            async for status in session.experimental.poll_task(task_id):
                print(f"Status: {status.status}")

                if status.status == "input_required":
                    final = await session.experimental.get_task_result(task_id, CallToolResult)
                    print(f"Result: {final.content[0].text}")
                    break

            if status.status == "completed":
                final = await session.experimental.get_task_result(task_id, CallToolResult)
                print(f"Result: {final.content[0].text}")


if __name__ == "__main__":
    anyio.run(main)
```

## Error Handling

Handle task errors gracefully:

```python
from mcp.shared.exceptions import McpError

try:
    result = await session.experimental.call_tool_as_task("my_tool", args)
    task_id = result.task.taskId

    async for status in session.experimental.poll_task(task_id):
        if status.status == "failed":
            raise RuntimeError(f"Task failed: {status.statusMessage}")

    final = await session.experimental.get_task_result(task_id, CallToolResult)

except McpError as e:
    print(f"MCP error: {e.error.message}")
except Exception as e:
    print(f"Error: {e}")
```

## Next Steps

- [Server Implementation](https://py.sdk.modelcontextprotocol.io/experimental/tasks-server/index.md) - Build task-supporting servers
- [Tasks Overview](https://py.sdk.modelcontextprotocol.io/experimental/tasks/index.md) - Review lifecycle and concepts