MCP Python SDK

<div align="center">

Model Context Protocol (MCP) 的 Python 实现

</div>

<!-- omit in toc -->

目录

• MCP Python SDK
  • 概述
  • 安装
    • 将 MCP 添加到你的 Python 项目
    • 运行独立的 MCP 开发工具
  • 快速开始
  • 什么是 MCP？
  • 核心概念
    • 服务器
    • 资源
    • 工具
    • 提示词
    • 图像
    • 上下文
  • 运行你的服务器
    • 开发模式
    • Claude Desktop 集成
    • 直接执行
    • 挂载到现有的 ASGI 服务器
  • 示例
    • Echo 服务器
    • SQLite 浏览器
  • 高级用法
    • 底层服务器
    • 编写 MCP 客户端
    • MCP 原语
    • 服务器能力
  • 文档
  • 贡献
  • 许可证

概述

Model Context Protocol 允许应用程序以标准化的方式为 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

# 创建一个 MCP 服务器
mcp = FastMCP("Demo")


# 添加一个加法工具
@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}!"

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

mcp install server.py

或者，你可以使用 MCP Inspector 测试它：

mcp dev server.py

什么是 MCP？

Model Context Protocol (MCP) 允许你构建服务器，以安全、标准化的方式向 LLM 应用程序公开数据和功能。 可以将其视为 Web API，但专门为 LLM 交互而设计。 MCP 服务器可以：

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

核心概念

服务器

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

# 添加生命周期支持，用于具有强类型的启动/关闭
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass

from fake_database import Database  # 替换为你实际的 DB 类型

from mcp.server.fastmcp import Context, FastMCP

# 创建一个命名服务器
mcp = FastMCP("My App")

# 指定部署和开发的依赖项
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])


@dataclass
class AppContext:
    db: Database


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """使用类型安全的上下文管理应用程序生命周期"""
    # 在启动时初始化
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        # 在关闭时清理
        await db.disconnect()


# 将生命周期传递给服务器
mcp = FastMCP("My App", lifespan=app_lifespan)


# 在工具中访问类型安全的生命周期上下文
@mcp.tool()
def query_db(ctx: Context) -> str:
    """使用初始化资源的工具"""
    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:
    """静态配置数据"""
    return "App configuration here"


@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """动态用户数据"""
    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:
    """计算给定体重（公斤）和身高（米）的 BMI"""
    return weight_kg / (height_m**2)


@mcp.tool()
async def fetch_weather(city: str) -> str:
    """获取城市当前天气"""
    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:
    """从图像创建缩略图"""
    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:
    """处理具有进度跟踪的多个文件"""
    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

# 添加依赖项
mcp dev server.py --with pandas --with numpy

# 挂载本地代码
mcp dev server.py --with-editable .

Claude Desktop 集成

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

mcp install server.py

# 自定义名称
mcp install server.py --name "My Analytics Server"

# 环境变量
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")

# 将 SSE 服务器挂载到现有的 ASGI 服务器
app = Starlette(
    routes=[
        Mount('/', app=mcp.sse_app()),
    ]
)

# 或动态挂载为主机
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))

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

示例

Echo 服务器

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

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Echo")


@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    """将消息作为资源回显"""
    return f"Resource echo: {message}"


@mcp.tool()
def echo_tool(message: str) -> str:
    """将消息作为工具回显"""
    return f"Tool echo: {message}"


@mcp.prompt()
def echo_prompt(message: str) -> str:
    """创建一个回显提示词"""
    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:
    """提供数据库模式作为资源"""
    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:
    """安全地执行 SQL 查询"""
    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  # 替换为你实际的 DB 类型

from mcp.server import Server


@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """管理服务器启动和关闭生命周期。"""
    # 在启动时初始化资源
    db = await Database.connect()
    try:
        yield {"db": db}
    finally:
        # 在关闭时清理
        await db.disconnect()


# 将生命周期传递给服务器
server = Server("example-server", lifespan=server_lifespan)


# 在处理程序中访问生命周期上下文
@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

# 创建一个服务器实例
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

# 创建 stdio 连接的服务器参数
server_params = StdioServerParameters(
    command="python",  # 可执行文件
    args=["example_server.py"],  # 可选的命令行参数
    env=None,  # 可选的环境变量
)


# 可选：创建一个采样回调
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:
            # 初始化连接
            await session.initialize()

            # 列出可用的提示词
            prompts = await session.list_prompts()

            # 获取一个提示词
            prompt = await session.get_prompt(
                "example-prompt", arguments={"arg1": "value"}
            )

            # 列出可用的资源
            resources = await session.list_resources()

            # 列出可用的工具
            tools = await session.list_tools()

            # 读取一个资源
            content, mime_type = await session.read_resource("file://some/path")

            # 调用一个工具
            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<br/>listChanged	资源公开和更新
tools	listChanged	工具发现和执行
logging	-	服务器日志配置
completion	-	参数完成建议

文档

• Model Context Protocol 文档
• Model Context Protocol 规范
• 官方支持的服务器

贡献

我们热衷于支持所有经验水平的贡献者，并希望看到你参与到项目中。 请参阅 贡献指南 以开始使用。

许可证

此项目已获得 MIT 许可证的许可 - 有关详细信息，请参阅 LICENSE 文件。