MCP服务端开发与多客户端调用

示例代码

📎 附件：mcp-demo.zip（zip, 3KB）

原文链接：https://www.yuque.com/attachments/yuque/0/2025/zip/2424104/1753931389964-1a3dece0-253c-4af4-90d1-743822b722fb.zip

MCP服务端开发

接下来我们介绍 Model Context Protocol（MCP）Python SDK 的使用方法和核心概念。MCP 是一种用于将外部数据源和工具与大型语言模型（LLM）应用集成的开放协议。通过 MCP，开发者可以构建 MCP 客户端连接到任何 MCP 服务器，创建暴露资源、提示和工具的 MCP 服务器，并处理所有 MCP 协议消息和生命周期事件。本文将涵盖 MCP 的安装、快速入门、核心概念、服务器运行方式、示例以及高级用法等内容。

一、概述

MCP Python SDK 实现了完整的 MCP 规范，使开发者能够轻松地构建 MCP 客户端和服务器。它支持标准传输如 stdio、SSE 和 Streamable HTTP，并且能够处理所有 MCP 协议消息和生命周期事件。

二、安装

（一）添加 MCP 到 Python 项目

安装 uv 命令

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

推荐使用 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

# 创建一个 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

四、核心概念

（一）服务器

FastMCP 服务器是与 MCP 协议交互的核心接口。它处理连接管理、协议合规性和消息路由。以下是一个示例：

# 添加生命周期支持以启用启动/关闭和强类型
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass

from fake_database import Database  # 替换为实际的数据库类型

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]:
    """使用类型安全的上下文管理应用程序生命周期"""
    # 启动时初始化
    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:
    """使用初始化资源的工具"""
    db = ctx.request_context.lifespan_context.db
    return db.query()