MCP 与 Python SDK 技术分享

一、MCP 概述

模型上下文协议（Model Context Protocol，MCP）允许应用程序以标准化的方式为大语言模型（LLM）提供上下文，将提供上下文的关注点与实际的 LLM 交互分离开来。

1. MCP 的核心能力

MCP 协议定义了服务器可以实现的三个核心能力：

2. MCP 客户端和服务端以及 AI 的交互流程

Untitled diagram-2025-03-19-025827.png

1. 用户发起请求：用户通过 AI 助手（如 Claude）提交请求
2. AI 助手识别任务：AI 助手分析用户需求并决定使用哪些 MCP 工具
3. 工具调用：AI 助手通过 MCP 协议调用相应的工具或资源
4. 服务端处理：MCP 服务端接收请求并执行相应的功能
5. 返回结果：服务端将处理结果返回给客户端
6. 结果呈现：AI 助手将结果以用户友好的方式展示

3. MCP 可实现的功能

这些服务器是官方提供的一些服务工具。由TypeScript 或者 Python SDK 实现：

• AWS 知识库检索 - 使用 Bedrock Agent Runtime 从 AWS 知识库中检索信息
• Brave 搜索 - 使用 Brave 的搜索 API 进行网络和本地搜索
• EverArt - 使用各种模型进行 AI 图像生成
• Everything - 参考/测试服务器，包含提示、资源和工具
• Fetch - 网络内容获取和转换，提高 LLM 使用效率
• 文件系统 - 具有可配置访问控制的安全文件操作
• Git - 用于读取、搜索和操作 Git 仓库的工具
• GitHub - 仓库管理、文件操作和 GitHub API 集成
• GitLab - GitLab API，支持项目管理
• Google Drive - Google Drive 文件访问和搜索功能
• Google 地图 - 位置服务、路线指引和地点详情
• 记忆系统 - 基于知识图谱的持久化记忆系统
• PostgreSQL - 具有架构检查功能的只读数据库访问
• Puppeteer - 浏览器自动化和网页抓取
• Redis - 与 Redis 键值存储交互
• Sentry - 从 Sentry.io 检索和分析问题
• Sequential Thinking - 通过思维序列进行动态和反思性问题解决
• Slack - 频道管理和消息发送功能
• SQLite - 数据库交互和商业智能功能
• 时间 - 时间和时区转换功能

二、MCP Python SDK

这个 Python 软件开发工具包（SDK）实现了完整的 MCP 规范，便于开发者：

1. 构建能够连接到任何 MCP 服务器的 MCP 客户端。
2. 创建能够暴露资源、提示和工具的 MCP 服务器。
3. 使用标准传输方式，如标准输入输出（stdio）和服务器发送事件（SSE）。
4. 处理所有 MCP 协议消息和生命周期事件。

标准输入输出（stdio）是一种基本的输入输出机制，在大多数操作系统和编程语言中都有支持。在 Python 中，sys.stdin 用于读取标准输入，sys.stdout 用于写入标准输出。在 MCP 的场景中，使用 stdio 意味着服务器和客户端通过当前进程的标准输入和标准输出来进行通信。

服务器发送事件（Server-Sent Events，SSE）是一种允许服务器向客户端实时推送数据的 Web API。与传统的请求 - 响应模式不同，SSE 建立了一个单向的 HTTP 连接，服务器可以在有新数据时随时向客户端发送消息，而客户端只需监听服务器的推送。

1. 安装

1.1 将 MCP 添加到 Python 项目中

建议使用 uv 来管理 Python 项目。在由 uv 管理的 Python 项目中，可以通过以下命令将 MCP 添加到依赖项中：

uv add "mcp[cli]"

uv 是一个用于管理 Python 项目的工具，uv 可以帮助你管理 Python 项目的依赖项，借助 uv run 命令，你可以执行项目中的各种工具和脚本。

或者，对于使用 pip 管理依赖项的项目，可以使用以下命令：

pip install mcp

1.2 运行独立的 MCP 开发工具

要使用 uv 运行 mcp 命令，可以执行以下操作：

uv run mcp

2. 快速开始

创建一个简单的 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 install server.py 命令的主要作用是将 MCP 服务器配置到 Claude Desktop 中，以便在 Claude Desktop 中使用该服务器提供的功能。

或者，使用 MCP 检查器对其进行测试：

mcp dev server.py

"greeting://{name}" 是一种自定义的资源 URI（统一资源标识符）模板，用于在 MCP（Model Context Protocol）中标识和管理资源。

URI 模板的格式能够依据需求自定义，不过通常包含以下几个部分：

• 协议部分：例如 greeting://，这部分是自定义的协议标识，用于区分不同类型的资源。你可以根据实际情况定义不同的协议，像 config://、users:// 等。
• 参数部分：例如 {name}，这部分是参数占位符，用于在请求资源时传入具体的参数值。你可以在 URI 模板中定义多个参数，比如 users://{user_id}/profile。

三、MCP Python SDK 核心概念

1. 服务器（Server）

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

# Add lifespan support for startup/shutdown with strong typing
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator

from fake_database import Database  # Replace with your actual DB type

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]:
    """Manage application lifecycle with type-safe context"""
    # 启动时初始化
    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:
    """Tool that uses initialized resources"""
    db = ctx.request_context.lifespan_context["db"]
    return db.query()

2. 资源（Resources）

资源是向 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}"

3. 工具（Tools）

工具允许 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

4. 提示（Prompts）

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

from mcp.server.fastmcp import FastMCP, types

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[types.Message]:
    return [
        types.UserMessage("I'm seeing this error:"),
        types.UserMessage(error),
        types.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]

Prompts 是可复用的模板，能帮助 LLMs 更有效地与服务器交互。它们就像预先定义好的指令或对话模式，当用户输入某些关键词或触发特定条件时，服务器可以使用这些模板来生成更详细的指令或上下文，从而引导 LLMs 执行特定任务。例如：如果用户只输入一个关键词，如 "崩溃率"，服务器可以根据预先定义的 Prompts 生成更详细的指令。

简单来说就是用户可以获取服务所提示的功能列表（Prompts），并通过这些 Prompts 实现客户端与服务端之间的交互。

5. 图像（Images）

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

6. 上下文（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:
    """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 服务器

1. 开发模式

测试和调试服务器的最快方法是使用 MCP 检查器：

mcp dev server.py

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

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

2. Claude 桌面集成

一旦服务器准备就绪，可以将其安装到 Claude 桌面应用中：

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

3. 直接执行

对于像自定义部署这样的高级场景：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")

if __name__ == "__main__":
    mcp.run()

使用以下命令运行：

python server.py
# 或者
mcp run server.py

五、MCP Python SDK 示例

1. 回声服务器（Echo Server）

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

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

2. SQLite 资源管理器（SQLite Explorer）

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

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

六、高级用法

1. 低级服务器（Low-Level Server）

为了获得更多控制权，可以直接使用低级服务器实现。这可以完全访问协议，并允许自定义服务器的各个方面，包括通过生命周期 API 进行生命周期管理：

from contextlib import asynccontextmanager
from typing 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."""
    # 启动时初始化资源
    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 提供：

• 一种在服务器启动时初始化资源并在停止时清理资源的方法。
• 通过处理程序中的请求上下文访问初始化的资源。
• 在生命周期和请求处理程序之间进行类型安全的上下文传递。

Server 提供了对 MCP 协议的底层访问，能让开发者完全掌控协议的各个方面，包括生命周期管理等。它给予开发者很高的灵活性，允许自定义服务器的每个细节。

FastMCP 是一个更符合人体工程学的接口，它简化了服务器的创建和管理过程，提供了更便捷的装饰器来注册工具、资源和提示等。FastMCP 适合快速开发，减少了样板代码，提高了开发效率。

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

2. 编写 MCP 客户端

SDK 提供了一个高级客户端接口，用于连接到 MCP 服务器：

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

# 创建用于stdio连接的服务器参数
server_params = StdioServerParameters(
    command="python",  # Executable
    args=["example_server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)

# 可选：创建一个采样回调
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（模型上下文协议）代表了大语言模型与外部服务交互的新范式，通过标准化的接口将 AI 模型连接到丰富的外部数据源和功能。本文介绍了 MCP 的核心概念及其 Python SDK 的实现，展示了如何构建 MCP 服务以增强 AI 应用的能力。

通过 MCP，我们可以克服 LLM 的知识截止和封闭性限制，让 AI 能够：

• 获取实时数据和最新信息
• 执行复杂计算和业务操作
• 与现有业务系统无缝集成
• 提供个性化、上下文相关的响应

Python MCP SDK 提供了便捷的工具，使开发者能够快速构建功能丰富的 MCP 服务，无论是简单的计算工具、数据查询接口，还是复杂的业务系统集成。SDK 的设计理念注重易用性和灵活性，既可以通过高级 API 快速开发，也可以通过低级接口实现精细控制。

随着 AI 应用在企业和日常生活中的普及，MCP 将成为连接 AI 与现实世界的重要桥梁。掌握 MCP 及其 SDK 的开发方法，将帮助开发者构建更加强大、实用的智能应用，真正释放大语言模型的潜力。

未来，我们期待看到更多基于 MCP 的创新应用，以及更丰富的预置服务和工具，进一步简化 AI 应用的开发流程，推动 AI 技术的普及应用。