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 Python SDK
  • 概述
  • 安装
    • 将 MCP 添加到你的 Python 项目中
    • 运行独立的 MCP 开发工具
  • 快速开始
  • 什么是 MCP?
  • 核心概念
    • 服务器
    • 资源
    • 工具
    • 提示
    • 图像
    • 上下文
  • 运行你的服务器
    • 开发模式
    • Claude 桌面集成
    • 直接执行
    • 挂载到现有的 ASGI 服务器
  • 示例
    • 回声服务器
    • SQLite 探索器
  • 高级用法
    • 低级服务器
    • 编写 MCP 客户端
    • MCP 原语
    • 服务器功能
  • 文档
  • 贡献
  • 许可

概述

Model Context Protocol 允许应用程序以标准化的方式为 LLMs 提供上下文，将提供上下文的关注点与实际的 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 Inspector 测试它：

mcp dev server.py

什么是 MCP？

模型上下文协议 (MCP) 允许你以安全、标准化的方式构建服务器，从而向LLM应用程序暴露数据和功能。你可以将其想象成一种专门为LLM交互设计的Web API。MCP服务器可以：

• 通过资源（Resources）暴露数据（类似于GET端点；它们用于将信息加载到LLM的上下文中）
• 通过工具（Tools）提供功能（类似于POST端点；它们用于执行代码或产生其他副作用）
• 通过提示（Prompts）定义交互模式（可重用的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()

资源

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

工具

工具允许LLMs通过你的服务器采取行动。与资源不同的是，工具预期会进行计算并产生副作用：

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

提示

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

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

上下文

上下文对象让你的工具和资源能够访问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响应
工具	模型控制	向大语言模型暴露以执行操作的功能	API调用、数据更新

服务器功能

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

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

文档

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

贡献

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

许可证

该项目采用 MIT 许可证 - 详情请见 LICENSE 文件。