MCP PySDK

MCP Python SDK（MCP Python软件开发工具包）

模型上下文协议（MCP）的Python实现

  ](https://www.python.org/downloads/)   

目录

• MCP Python SDK（MCP Python软件开发工具包）

- 概述 - 安装 - 将MCP添加到您的Python项目中 - 运行独立的MCP开发工具 - 快速入门 - 什么是MCP？ - 核心概念 - 服务器 - 资源 - 工具 - 结构化输出 - 提示 - 图片 - 上下文 - 在函数中获取上下文 - 上下文属性和方法 - 完井作业（或完成作业） - 引出（或激发） - 抽样 - 日志记录和通知 - 认证 - FastMCP 属性 - 会话属性和方法 - 请求上下文属性 - 运行您的服务器 - 开发模式 - Claude 桌面集成 - 直接执行 - 可流式传输的HTTP传输 - 基于浏览器的客户端的CORS配置 - 挂载到现有的ASGI服务器 - 可流式传输的HTTP服务器 - 基本安装 - 基于主机的路由 - 具有路径配置的多台服务器 - 初始化时的路径配置 - SSE服务器 - 高级用法 - 低级服务器 - 结构化输出支持 - 分页（高级） - 编写MCP客户端 - 客户端显示实用程序 - 客户端的OAuth认证 - 解析工具结果 - MCP 原语（或基本操作） - 服务器功能 - 文档 - 贡献；做出贡献 - 许可证

概述

模型上下文协议（Model Context Protocol）允许应用程序以标准化的方式为大型语言模型（LLMs）提供上下文，将提供上下文的功能与实际的LLM交互分离开来。这个Python软件开发工具包（SDK）实现了完整的MCP规范，使得以下操作变得容易：

• 构建能够连接到任何MCP服务器的MCP客户端
• 创建MCP服务器以暴露资源、提示和工具
• 使用标准传输方式，如stdio、SSE和Streamable HTTP
• 处理所有MCP协议消息和生命周期事件

安装

将MCP添加到您的Python项目中

我们建议使用 紫外线 管理您的Python项目。

如果你还没有创建一个使用UV管理的项目，请创建一个：

uv init mcp-server-demo
cd mcp-server-demo

然后将MCP添加到您的项目依赖项中：

uv add "mcp[cli]"

或者，对于使用 pip 管理依赖的项目：

pip install "mcp[cli]"

运行独立的MCP开发工具

使用 uv 运行 mcp 命令：

uv run mcp

快速入门

让我们创建一个简单的MCP服务器，用于提供一个计算器工具和一些数据：

"""
FastMCP quickstart example.

cd to the `examples/snippets/clients` directory and run:
    uv run server fastmcp_quickstart stdio
"""

from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")

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

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

# Add a prompt
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    styles = {
        "friendly": "Please write a warm, friendly greeting",
        "formal": "Please write a formal, professional greeting",
        "casual": "Please write a casual, relaxed greeting",
    }

    return f"{styles.get(style, styles['friendly'])} for someone named {name}."

_完整示例： 示例/代码片段/服务器/fastmcp_quickstart.py_

您可以将此服务器安装在 Claude 桌面版 并立即通过运行来与它进行交互：

uv run mcp install server.py

或者，您可以使用MCP Inspector进行测试：

uv run mcp dev server.py

什么是MCP？

该 模型上下文协议（MCP） 它允许你构建服务器，以安全、标准化的方式向大型语言模型（LLM）应用程序暴露数据和功能。可以将其视为一种网络API，但专为大型语言模型交互而设计。MCP服务器可以：

• 通过……展示数据 资源 （可以把这些想象成类似GET端点；它们用于将信息加载到大型语言模型（LLM）的上下文中）
• 通过……提供功能 工具 （有点像POST端点；它们用于执行代码或产生其他副作用）
• 通过定义交互模式来 提示 （用于大型语言模型（LLM）交互的可重用模板）
• 还有更多！

核心概念

服务器

FastMCP服务器是您与MCP协议进行交互的核心接口。它负责连接管理、协议合规性以及消息路由：

"""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()

_完整示例： 示例/代码片段/服务器/生命周期示例.py_

资源

资源是您向大型语言模型（LLMs）提供数据的方式。它们类似于REST API中的GET端点——它们提供数据，但不应执行大量计算或产生副作用：

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
}"""

_完整示例： 示例/代码片段/服务器/基本资源.py_

工具

工具使大型语言模型（LLMs）能够通过您的服务器执行操作。与资源不同，工具预期会执行计算并产生副作用：

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()}"

_完整示例： 示例/代码片段/服务器/基本工具.py_

工具可以选择性地通过包含一个参数来接收一个上下文（Context）对象 Context 类型注解。此上下文由FastMCP框架自动注入，并提供对MCP功能的访问：

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"

_完整示例： 示例/代码片段/服务器/工具进度.py_

结构化输出

如果工具的返回类型允许，默认情况下，工具将返回结构化结果 注解是兼容的。否则，它们将返回非结构化的结果。

结构化输出支持以下返回类型：

• Pydantic 模型（BaseModel 子类）
• TypedDicts（类型字典）
• 数据类和其他带有类型提示的类
• dict[str, T] （其中T是任何可JSON序列化的类型）
• 基本数据类型（str，int，float，bool，bytes，None）——被封装在 {"result": value}
• 泛型类型（列表、元组、联合类型、可选类型等）——被封装在 {"result": value}

没有类型提示的类无法被序列化以用于结构化输出。仅 带有适当注解属性的类将被转换为 Pydantic 模型 用于模式生成和验证。

结构化结果会自动与输出模式进行验证 由注解生成。这确保了工具返回的是类型良好的（代码/数据）， 客户端可以轻松处理的已验证数据。

注： 为了保持向后兼容性，未结构化结果也同样（被支持/可用） 已返回。为保持向后兼容性，提供了非结构化结果 与MCP规范的先前版本兼容，并且支持特殊特性（quirks）兼容性 在当前SDK版本中，与之前的FastMCP版本相比。

注： 在工具函数的返回类型注解的情况下 导致该工具被归类为结构化工具 _这是不受欢迎的_， 该分类可以通过传递来抑制 structured_output=False 到……的 @tool 装饰器。

"""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}

_完整示例： 示例/代码片段/服务器/结构化输出.py_

提示

提示是可重用的模板，有助于大型语言模型（LLMs）与您的服务器进行有效交互：

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?"),
    ]

_完整示例： 示例/代码片段/服务器/基本提示.py_

图标

MCP服务器可以提供用于用户界面显示的图标。图标可以添加到服务器实现、工具、资源和提示中：

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"

_完整示例： 示例/快速多类分类（fastmcp）/图标演示（icons_demo.py）_

图片

FastMCP提供了一个 Image 自动处理图像数据的类：

"""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")

_完整示例： 示例/代码片段/服务器/images.py_

上下文

\Context\ 对象会自动注入到通过类型提示请求它的工具和资源函数中。它提供了访问 MCP（可能是指某个框架或平台的组件，如“Microservices Communication Protocol”或其他具体含义，根据上下文确定）功能的能力，如日志记录、进度报告、资源读取、用户交互和请求元数据。

在函数中获取上下文

要在工具或资源功能中使用上下文，请添加一个带有 Context 类型注解：

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 对象提供以下功能：

• ctx.request_id - 当前请求的唯一标识符
• ctx.client_id - 如有，请提供客户端ID
• ctx.fastmcp - 访问FastMCP服务器实例（参见 FastMCP 属性)
• ctx.session - 访问底层会话以进行高级通信（见 会话属性和方法)
• ctx.request_context - 访问特定请求的数据和生命周期资源（见 请求上下文属性)
• await ctx.debug(message) - 发送调试日志消息
• await ctx.info(message) - 发送信息日志消息
• await ctx.warning(message) - 发送警告日志消息
• await ctx.error(message) - 发送错误日志消息
• await ctx.log(level, message, logger_name=None) - 发送带有自定义级别的日志
• await ctx.report_progress(progress, total=None, message=None) - 报告操作进度
• await ctx.read_resource(uri) - 通过URI读取资源
• await ctx.elicit(message, schema) - 向用户请求并验证额外信息

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"

_完整示例： 示例/代码片段/服务器/工具进度.py_

完成情况/完工情况

MCP 支持为提示符参数和资源模板参数提供补全建议。通过上下文参数，服务器可以根据之前解析的值提供补全选项：

客户使用情况：

"""
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()

_完整示例： 示例/代码片段/客户端/完成客户端.py_

引出（信息/反应/需求等）

向用户请求更多信息。此示例展示了在工具调用期间进行的引出（信息获取）过程：

from pydantic import BaseModel, Field

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

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

_完整示例： 示例/代码片段/服务器/elicitation.py_

引出模式（Elicitation schemas）为所有字段类型提供默认值。这些默认值会自动包含在发送给客户端的JSON模式中，使客户端能够预先填充表单。

该 elicit() 该方法返回一个 ElicitationResult 与；带有；和……一起

• action“接受”、“拒绝”或“取消”
• data已验证的响应（仅在被接受时）
• validation_error任何验证错误信息

抽样

工具可以通过采样（生成文本）与大型语言模型（LLMs）进行交互：

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,
    )

    if result.content.type == "text":
        return result.content.text
    return str(result.content)

_完整示例： 示例/代码片段/服务器/采样.py_

日志记录和通知

工具可以通过上下文发送日志和通知：

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}"

_完整示例： 示例/代码片段/服务器/通知.py_

认证

希望提供访问受保护资源工具的服务器可以使用身份验证。

mcp.server.auth 实现了OAuth 2.1资源服务器功能，其中MCP服务器作为资源服务器（RS），用于验证由独立授权服务器（AS）颁发的令牌。这遵循了 MCP授权规范 并实现RFC 9728（受保护资源元数据）用于域服务（AS）发现。

MCP服务器可以通过提供（某种）实现来使用身份验证 TokenVerifier 协议：

"""
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",
    # 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")

_完整示例： 示例/代码片段/服务器/OAuth服务器.py_

要查看一个包含独立实现的授权服务器和资源服务器的完整示例，请参阅 examples/servers/simple-auth/。

建筑：

• 授权服务器 (AS)处理OAuth流程、用户认证和令牌发放
• 资源服务器（RS）您的MCP服务器，用于验证令牌并提供受保护资源
• 客户通过RFC 9728发现AS（认证服务器），获取令牌，并在MCP（可能是指某种客户端或中间件平台）服务器上使用这些令牌

看 TokenVerifier（可译为“令牌验证器”） 有关实现令牌验证的更多详细信息。

FastMCP 属性

可通过以下方式访问的FastMCP服务器实例 ctx.fastmcp 提供服务器配置和元数据的访问权限：

• ctx.fastmcp.name - 初始化时定义的服务器名称
• ctx.fastmcp.instructions - 向客户提供服务器说明/描述
• ctx.fastmcp.website_url - 服务器的可选网站URL
• ctx.fastmcp.icons - 可选的用于UI显示的图标列表
• ctx.fastmcp.settings - 完整的服务器配置对象，包含：

- debug - 调试模式标志 - log_level - 当前日志级别 - host 并且 port - 服务器网络配置 - mount_path， sse_path， streamable_http_path - 交通路径 - stateless_http - 服务器是否以无状态模式运行 - 以及其他配置选项

@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,
    }

会话属性和方法

通过（某种方式）可访问的会话对象 ctx.session 提供对客户端通信的高级控制：

• ctx.session.client_params - 客户端初始化参数和声明的能力
• await ctx.session.send_log_message(level, data, logger) - 发送日志消息，拥有完全控制权
• await ctx.session.create_message(messages, max_tokens) - 请求大型语言模型（LLM）进行采样/完成任务
• await ctx.session.send_progress_notification(token, progress, total, message) - 直接进度更新
• await ctx.session.send_resource_updated(uri) - 通知客户端某个特定资源已更改
• await ctx.session.send_resource_list_changed() - 通知客户资源列表已更改
• await ctx.session.send_tool_list_changed() - 通知客户工具列表已更改
• await ctx.session.send_prompt_list_changed() - 通知客户即时列表已更改

@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"

请求上下文属性

可通过以下方式访问的请求上下文 ctx.request_context 包含请求特定的信息和资源：

• ctx.request_context.lifespan_context - 访问服务器启动时初始化的资源

- 数据库连接、配置对象、共享服务 - 在服务器生命周期函数中定义的资源的安全类型访问

• ctx.request_context.meta - 从客户端请求元数据，包括：

- progressToken - 进度通知的令牌 - 其他由客户端提供的元数据

• ctx.request_context.request - 用于高级处理的原始MCP请求对象
• ctx.request_context.request_id - 此请求的唯一标识符

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

_全生命周期示例： 示例/代码片段/服务器/生命周期示例.py_

运行您的服务器

开发模式

测试和调试服务器的最快方法是使用MCP Inspector：

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 桌面集成

一旦您的服务器准备就绪，请在Claude Desktop中进行安装：

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

直接执行

对于高级场景，如自定义部署：

"""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()

_完整示例： 示例/代码片段/服务器/直接执行.py_

使用以下命令运行：

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

请注意 uv run mcp run 或者 uv run mcp dev 仅支持使用FastMCP的服务器，而不支持低级别的服务器变体。

可流式传输的HTTP传输

注在生产部署中，Streamable HTTP传输正在取代SSE传输。

"""
Run from the repository root:
    uv run examples/snippets/servers/streamable_config.py
"""

from mcp.server.fastmcp import FastMCP

# Stateful server (maintains session state)
mcp = FastMCP("StatefulServer")

# Other configuration options:
# Stateless server (no session persistence)
# mcp = FastMCP("StatelessServer", stateless_http=True)

# Stateless server (no session persistence, no sse stream with supported client)
# mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True)

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

_完整示例： 示例/代码片段/服务器/可流式传输配置.py_

在Starlette应用程序中，您可以挂载多个FastMCP服务器：

"""
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)

@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)

@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 = "/"

_完整示例： 示例/代码片段/服务器/可流式传输的Starlette挂载.py_

对于具有Streamable HTTP实现的低级别服务器，请参阅：

• 有状态服务器： examples/servers/simple-streamablehttp/
• 无状态服务器： examples/servers/simple-streamablehttp-stateless/

可流式传输的HTTP传输支持：

• 有状态和无状态操作模式
• 使用事件存储实现可重试性
• JSON 或 SSE 响应格式
• 多节点部署具有更好的可扩展性

基于浏览器的客户端的CORS配置

如果您希望您的服务器能够被基于浏览器的MCP客户端访问，您需要配置CORS（跨源资源共享）头部信息 Mcp-Session-Id 头部信息必须对外暴露，以便浏览器客户端能够访问它：

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"],
)

这种配置是必要的，因为：

• MCP可流式HTTP传输使用了 Mcp-Session-Id 用于会话管理的头部信息
• 浏览器限制对响应头的访问，除非通过CORS（跨源资源共享）明确暴露
• 如果没有这个配置，基于浏览器的客户端将无法从初始化响应中读取会话ID

挂载到现有的ASGI服务器

默认情况下，SSE 服务器挂载在 /sse 并且Streamable HTTP服务器已安装在 /mcp您可以使用以下描述的方法自定义这些路径。

有关在Starlette中挂载应用程序的更多信息，请参阅 Starlette 文档.

可流式传输的HTTP服务器

你可以使用（相关方法）将StreamableHTTP服务器挂载到现有的ASGI服务器上 streamable_http_app 方法。这允许您将StreamableHTTP服务器与其他ASGI应用程序集成。

基本安装

"""
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
"""

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

from mcp.server.fastmcp import FastMCP

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

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

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

_完整示例： 示例/代码片段/服务器/可流式传输的HTTP基本挂载.py_

基于主机的路由

"""
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
"""

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

from mcp.server.fastmcp import FastMCP

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

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

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

_完整示例： 示例/代码片段/服务器/可流式传输的HTTP主机挂载.py_

具有路径配置的多台服务器

"""
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
"""

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")
chat_mcp = FastMCP("Chat Server")

@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 = "/"

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

_完整示例： 示例/代码片段/服务器/可流式传输的HTTP多服务器.py_

初始化时的路径配置

"""
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", 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()),
    ]
)

_完整示例： 示例/代码片段/服务器/可流式传输的HTTP路径配置.py_

SSE服务器

注SSE传输方式正被取代 可流式传输的HTTP传输。

你可以将SSE服务器挂载到现有的ASGI服务器上，使用 sse_app 方法。这允许你将SSE服务器与其他ASGI应用程序集成。

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()))

当在不同路径下挂载多个MCP服务器时，您可以通过多种方式配置挂载路径：

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")

要了解有关在Starlette中挂载应用程序的更多信息，请参阅 Starlette 文档。

高级用法

低级别服务器

为了获得更精细的控制，您可以直接使用底层服务器实现。这将使您能够完全访问协议，并允许您自定义服务器的各个方面，包括通过生命周期API进行生命周期管理：

"""
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())

_完整示例： 示例/代码片段/服务器/低级别/生命周期.py_

使用寿命API提供：

• 一种在服务器启动时初始化资源并在服务器停止时清理资源的方法
• 在处理程序中通过请求上下文访问已初始化的资源
• 在生命周期处理器和请求处理器之间进行类型安全的上下文传递

"""
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())

_完整示例： 示例/代码片段/服务器/低级别/基本.py_

警告：该 uv run mcp run 并且 uv run mcp dev 该工具不支持低级别服务器。

结构化输出支持

低级别的服务器支持为工具提供结构化输出，使您能够同时返回人类可读的内容和机器可读的结构化数据。工具可以定义一个 outputSchema 验证其结构化输出：

"""
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())

_完整示例： 示例/代码片段/服务器/低级/结构化输出.py_

工具可以通过三种方式返回数据：

1. 仅内容返回内容块列表（在2025年6月18日规范修订前的默认行为）
2. 仅限结构化数据返回一个将被序列化为JSON的字典（在规范修订版2025-06-18中引入）
3. 两者都返回一个元组 (content, structured_data)，这是为了向后兼容而优选的选项

当一个 outputSchema 一旦定义好，服务器会自动将结构化输出与模式进行验证。这确保了类型安全，并有助于尽早发现错误。

分页（高级）

对于需要处理大型数据集的服务器，底层服务器提供了列表操作的分页版本。这是一种可选的优化功能——大多数服务器在处理数百或数千个项目时才需要分页功能。

服务器端实现

"""
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 

#### 客户端消费

""" 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())

_完整示例： [示例/代码片段/客户端/分页客户端.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/pagination_client.py)_

#### 要点

- **光标是不透明的字符串** - 服务器定义格式（数字偏移量、时间戳等）
- **返回 `nextCursor=None`** 当没有更多页面时
- **向后兼容** - 不支持分页的客户端仍然可以正常工作（他们将只获取第一页）
- **灵活的页面尺寸** - 每个终端可以根据数据特性定义自己的页面大小

见 [简单分页示例](examples/servers/simple-pagination) 以实现完整功能。

### 编写MCP客户端

该SDK提供了一个高级客户端接口，用于通过各种方式连接到MCP服务器 [交通（方式/运输）](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports)：

""" 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.TextContent): 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()

_完整示例： [示例/代码片段/客户端/stdio_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/stdio_client.py)_

客户也可以使用以下方式连接 [可流式传输的HTTP传输](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http)：

""" 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 streamablehttp_client

async def main(): # Connect to a streamable HTTP server async with streamablehttp_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())

_完整示例： [示例/代码片段/客户端/streamable_basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/streamable_basic.py)_

### 客户端显示实用程序

在构建MCP客户端时，SDK提供了实用工具来帮助显示工具、资源和提示的可读名称：

""" 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()

_完整示例： [示例/代码片段/客户端/显示实用程序.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/display_utilities.py)_

这个（或“它”） `get_display_name()` 该函数实现了显示名称时的正确优先级规则：

- 对于工具： `title` > `annotations.title` > `name`
- 对于其他对象： `title` > `name`

这确保了您的客户端用户界面显示服务器提供的最用户友好的名称。

### 客户端的OAuth认证

该SDK包含 [授权支持](https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization) 用于连接受保护的MCP服务器：

""" 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

from pydantic import AnyUrl

from mcp import ClientSession from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.streamable_http import streamablehttp_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 streamablehttp_client("http://localhost:8001/mcp", auth=oauth_auth) 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()

_完整示例： [示例/代码片段/客户端/OAuth客户端.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/oauth_client.py)_

如需一个完整的可运行示例，请参阅 [`examples/clients/simple-auth-client/`](examples/clients/simple-auth-client/).

### 解析工具结果

当通过MCP调用工具时 `CallToolResult` 对象以结构化格式包含工具的响应。了解如何解析这个结果对于正确处理工具输出至关重要。

"""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())

### MCP 原语（或基本操作）

MCP协议定义了服务器可以实现的三个核心原语：

| 原始类型 | 控制                 | 描述                                               | 示例用法                    |
|-----------|-----------------------|-----------------------------------------------------|------------------------------|
| 提示符   | 用户控制的           | 由用户选择调用的交互式模板           | 斜杠命令，菜单选项 |
资源 | 由应用程序控制 | 客户端应用程序管理的上下文数据 | 文件内容、API响应
工具 | 模型控制 | 暴露给大型语言模型以采取行动的功能 | API调用，数据更新

### 服务器功能

MCP服务器在初始化期间声明其功能：

| 能力   | 功能标志                 | 描述                        |
|--------------|------------------------------|------------------------------------|
| `prompts` | `listChanged` | 提示模板管理
| `resources` | `subscribe`
`listChanged`| 资源暴露与更新 |
| `tools` | `listChanged` | 工具发现与执行
| `logging` | -                            | 服务器日志配置                       |
| `completions`| -                            | 参数补全建议    |

## 文档

- [API 参考](https://modelcontextprotocol.github.io/python-sdk/api/)
- [模型上下文协议文档](https://modelcontextprotocol.io)
- [模型上下文协议规范](https://spec.modelcontextprotocol.io)
- [官方支持的服务器](https://github.com/modelcontextprotocol/servers)

## 贡献

我们热衷于支持各层次经验的贡献者，并期待您能参与到这个项目中来。详见 [贡献指南](CONTRIBUTING.md) 开始吧。

## 许可证

此项目采用MIT许可证授权——详情请参阅LICENSE文件。