MCP Python SDK

模型上下文协议 (MCP) 的 Python 实现

PyPI 麻省理工学院许可 Python 版本 GitHub 讨论

概述

模型上下文协议 (MCP) 允许应用程序以标准化的方式为 LLM 提供上下文，从而将提供上下文的关注点与实际的 LLM 交互分离开来。此 Python SDK 实现了完整的 MCP 规范，从而可以轻松实现以下功能：

构建可连接到任何 MCP 服务器的 MCP 客户端
创建公开资源、提示和工具的 MCP 服务器
使用标准传输，例如 stdio 和 SSE
处理所有 MCP 协议消息和生命周期事件

安装

将 MCP 添加到你的 Python 项目

我们建议使用uv来管理您的 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 服务器，它公开一个计算器工具和一些数据：

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

您可以在Claude Desktop中安装此服务器并通过运行以下命令立即与其交互：

mcp install server.py

或者，您可以使用 MCP 检查器进行测试：

mcp dev server.py

什么是 MCP？

模型上下文协议 (MCP)允许您构建服务器，以安全、标准化的方式向 LLM 应用程序公开数据和功能。您可以将其视为一个 Web API，但专为 LLM 交互而设计。MCP 服务器可以：

通过资源公开数据（可以将其视为 GET 端点；它们用于将信息加载到 LLM 的上下文中）
通过工具提供功能（有点像 POST 端点；它们用于执行代码或产生副作用）
通过提示（LLM 交互的可重复使用模板）定义交互模式
还有更多！

核心概念

服务器

FastMCP 服务器是 MCP 协议的核心接口。它负责连接管理、协议合规性和消息路由：

# Add lifespan support for startup/shutdown with strong typing
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass

from fake_database import Database  # Replace with your actual DB type

from mcp.server.fastmcp import Context, FastMCP

# Create a named server
mcp = FastMCP("My App")

# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])


@dataclass
class AppContext:
    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) -> str:
    """Tool that uses initialized resources"""
    db = ctx.request_context.lifespan_context["db"]
    return db.query()

资源

资源是你向 LLM 公开数据的方式。它们类似于 REST API 中的 GET 端点——它们提供数据，但不应执行大量计算或产生副作用：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.resource("config://app")
def get_config() -> str:
    """Static configuration data"""
    return "App configuration here"


@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Dynamic user data"""
    return f"Profile data for user {user_id}"

工具

工具允许 LLM 通过你的服务器执行操作。与资源不同，工具需要执行计算并产生副作用：

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate BMI given weight in kg and height in meters"""
    return weight_kg / (height_m**2)


@mcp.tool()
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
        return response.text

提示

提示是可重复使用的模板，可帮助 LLM 有效地与您的服务器交互：

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP("My App")


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


@mcp.prompt()
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?"),
    ]

图片

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

from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage

mcp = FastMCP("My App")


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

语境

Context 对象使您的工具和资源能够访问 MCP 功能：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("My App")


@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
    """Process multiple files with progress tracking"""
    for i, file in enumerate(files):
        ctx.info(f"Processing {file}")
        await ctx.report_progress(i, len(files))
        data, mime_type = await ctx.read_resource(f"file://{file}")
    return "Processing complete"

运行您的服务器

开发模式

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

mcp dev server.py

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

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

Claude 桌面集成

服务器准备就绪后，请在 Claude Desktop 中安装它：

mcp install server.py

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

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

直接执行

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

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")

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

使用以下命令运行它：

python server.py
# or
mcp run server.py

挂载到现有的 ASGI 服务器

您可以使用sse_app方法将 SSE 服务器挂载到现有的 ASGI 服务器。这样您就可以将 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()))

有关在 Starlette 中安装应用程序的更多信息，请参阅Starlette 文档。

示例

回显服务器

一个简单的服务器，演示资源、工具和提示：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Echo")


@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    """Echo a message as a resource"""
    return f"Resource echo: {message}"


@mcp.tool()
def echo_tool(message: str) -> str:
    """Echo a message as a tool"""
    return f"Tool echo: {message}"


@mcp.prompt()
def echo_prompt(message: str) -> str:
    """Create an echo prompt"""
    return f"Please process this message: {message}"

SQLite 资源管理器

显示数据库集成的更复杂示例：

import sqlite3

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("SQLite Explorer")


@mcp.resource("schema://main")
def get_schema() -> str:
    """Provide the database schema as a resource"""
    conn = sqlite3.connect("database.db")
    schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
    return "\n".join(sql[0] for sql in schema if sql[0])


@mcp.tool()
def query_data(sql: str) -> str:
    """Execute SQL queries safely"""
    conn = sqlite3.connect("database.db")
    try:
        result = conn.execute(sql).fetchall()
        return "\n".join(str(row) for row in result)
    except Exception as e:
        return f"Error: {str(e)}"

高级用法

低级服务器

为了获得更多控制权，您可以直接使用低级服务器实现。这赋予您完全的协议访问权限，并允许您自定义服务器的各个方面，包括通过生命周期 API 进行生命周期管理：

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fake_database import Database  # Replace with your actual DB type

from mcp.server import Server


@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """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)


# Access lifespan context in handlers
@server.call_tool()
async def query_db(name: str, arguments: dict) -> list:
    ctx = server.request_context
    db = ctx.lifespan_context["db"]
    return await db.query(arguments["query"])

生命周期 API 提供：

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

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]:
    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:
    if name != "example-prompt":
        raise ValueError(f"Unknown prompt: {name}")

    return types.GetPromptResult(
        description="Example prompt",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(type="text", text="Example prompt text"),
            )
        ],
    )


async def run():
    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__":
    import asyncio

    asyncio.run(run())

编写 MCP 客户端

SDK 提供了用于连接 MCP 服务器的高级客户端接口：

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

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="python",  # Executable
    args=["example_server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)


# Optional: create a sampling callback
async def handle_sampling_message(
    message: types.CreateMessageRequestParams,
) -> types.CreateMessageResult:
    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()

            # Get a prompt
            prompt = await session.get_prompt(
                "example-prompt", arguments={"arg1": "value"}
            )

            # List available resources
            resources = await session.list_resources()

            # List available tools
            tools = await session.list_tools()

            # Read a resource
            content, mime_type = await session.read_resource("file://some/path")

            # Call a tool
            result = await session.call_tool("tool-name", arguments={"arg1": "value"})


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

MCP 原语

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

原始	控制	描述	使用示例
提示	用户控制	由用户选择调用的交互式模板	斜线命令、菜单选项
资源	应用程序控制	客户端应用程序管理的上下文数据	文件内容、API 响应
工具	模型控制	向 LLM 公开的函数用于执行操作	API 调用、数据更新

服务器功能

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

能力	功能标志	描述
`prompts`	`listChanged`	提示模板管理
`resources`	`subscribe` `listChanged`	资源公开和更新
`tools`	`listChanged`	工具发现和执行
`logging`	-	服务器日志配置
`completion`	-	论证完成建议

文档

贡献

我们热衷于支持各种经验水平的贡献者，并期待您参与到项目中来。请参阅贡献指南，开始您的贡献之旅。

执照

该项目根据 MIT 许可证获得许可 - 有关详细信息，请参阅 LICENSE 文件。

MCP Python SDK

MCP Python SDK

目录

概述

安装

将 MCP 添加到你的 Python 项目

运行独立的 MCP 开发工具

快速入门

什么是 MCP？

核心概念

服务器

资源

工具

提示

图片

语境

运行您的服务器

开发模式

Claude 桌面集成

直接执行

挂载到现有的 ASGI 服务器

示例

回显服务器

SQLite 资源管理器

高级用法

低级服务器

编写 MCP 客户端

MCP 原语

服务器功能

文档

贡献

执照

Resources

Looking for Admin?

Appeared in Searches

Latest Blog Posts

MCP directory API