MCP Python SDK

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

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

概述

模型上下文协议 (MCP) 允许应用程序以标准化的方式为 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 检查器进行测试：

mcp dev server.py

什么是 MCP？

模型上下文协议 (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"])

资源

资源是你向 LLM 公开数据的方式。它们类似于 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}"

工具

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

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

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

语境

Context 对象使您的工具和资源能够访问 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 响应
工具	模型控制	向 LLM 公开的函数用于执行操作	API 调用、数据更新

服务器功能

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

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

文档

贡献

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

执照

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

FastMCP