MCP Python SDK

Model Context Protocol (MCP) 的 Python 实现

![PyPI][pypi-badge]
![MIT licensed][mit-badge]
![Python Version][python-badge]
![Documentation][docs-badge]
![Specification][spec-badge]
![GitHub Discussions][discussions-badge]

目录

• 概述
• 安装
• 快速开始
• 什么是 MCP?
• 核心概念
  • 服务器
  • 资源
  • 工具
  • 提示
  • 图像
  • 上下文
• 运行您的服务器
  • 开发模式
  • Claude 桌面集成
  • 直接执行
• 示例
  • 回声服务器
  • 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

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

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

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

资源

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

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

工具

工具允许 LLMs 通过您的服务器执行操作。与资源不同，工具预期会执行计算并产生副作用：

@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

提示

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

@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:
    """Create a thumbnail from an image"""
    img = PILImage.open(image_path)
    img.thumbnail((100, 100))
    return Image(data=img.tobytes(), format="png")

上下文

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

from mcp.server.fastmcp import FastMCP, Context

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

示例

回声服务器

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

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 探索器

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

from mcp.server.fastmcp import FastMCP
import sqlite3

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

高级用法

低级服务器

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

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

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

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

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) 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
            resource = 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 响应
工具	模型控制	向 LLMs 暴露以执行操作的功能	API 调用，数据更新

服务器功能

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

功能	特性标志	描述
prompts	listChanged	提示模板管理
resources	subscribelistChanged	资源暴露和更新
tools	listChanged	工具发现和执行
logging	-	服务器日志配置
completion	-	参数补全建议

文档

• 模型上下文协议文档
• 模型上下文协议规范
• 官方支持的服务器

贡献

我们热衷于支持所有经验水平的贡献者，并非常希望您能参与到项目中来。请参阅贡献指南开始。

许可证

此项目根据 MIT 许可证授权 - 详情请见 LICENSE 文件。