MCP Python SDK

<div align="center">

Model Context Protocol (MCP) 的 Python 实现

</div>

<!-- omit in toc -->

目录

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

概述

Model Context Protocol 允许应用程序以标准化的方式为 LLM 提供上下文，从而将提供上下文的关注点与实际的 LLM 交互分离。 这个 Python SDK 实现了完整的 MCP 规范，使其易于：

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

安装

我们建议使用 uv 来管理你的 Python 项目：

uv add "mcp[cli]"

或者：

pip install mcp

快速开始

让我们创建一个简单的 MCP 服务器，它暴露一个计算器工具和一些数据：

# server.py
from mcp.server.fastmcp import FastMCP

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

# 添加一个加法工具
@mcp.tool()
def add(a: int, b: int) -> int:
    """添加两个数字"""
    return a + b

# 添加一个动态问候资源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """获取个性化问候语"""
    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 mcp.server.fastmcp import FastMCP

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

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

资源

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

@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 通过你的服务器执行操作。 与资源不同，工具应执行计算并产生副作用：

@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 有效地与你的服务器交互：

@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[Message]:
    return [
        UserMessage("I'm seeing this error:"),
        UserMessage(error),
        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.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.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 = 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

示例

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 浏览器

一个更复杂的示例，展示了数据库集成：

from mcp.server.fastmcp import FastMCP
import sqlite3

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

高级用法

底层服务器

为了获得更多控制，你可以直接使用底层服务器实现。 这使你可以完全访问协议，并允许你自定义服务器的各个方面：

from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# 创建一个服务器实例
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, Any] | 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
from mcp.client.stdio import stdio_client

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

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) 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()

            # 读取一个资源
            resource = 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 文件。