MCP Python SDK

which python
python3 -m venv myenv
source myenv/Scripts/activate
pip install -r requirements.txt
python fast_mcp_server.py

用法

运行客户端

add
    .env
    OPENAI_API_KEY=

运行服务端

uvicorn ne04j_mcp_server:app --host 0.0.0.0 --port 8000

数据准备

插入节点:

CREATE (p1:Person {name: "Tom Hanks", birthYear: 1956})
CREATE (p2:Person {name: "Kevin Bacon", birthYear: 1958})
CREATE (m1:Movie {title: "Forrest Gump", releaseYear: 1994})
CREATE (m2:Movie {title: "Apollo 13", releaseYear: 1995})

插入关系

MATCH (p:Person {name: "Tom Hanks"}), (m:Movie {title: "Forrest Gump"})
CREATE (p)-[:ACTED_IN]->(m)

MATCH (p:Person {name: "Tom Hanks"}), (m:Movie {title: "Apollo 13"})
CREATE (p)-[:ACTED_IN]->(m)

MATCH (p1:Person {name: "Tom Hanks"}), (p2:Person {name: "Kevin Bacon"})
CREATE (p1)-[:FRIENDS_WITH]->(p2)

插入属性

MATCH (p:Person {name: "Tom Hanks"})
SET p.oscarsWon = 2

MATCH (m:Movie {title: "Forrest Gump"})
SET m.genre = "Drama"

MATCH (p:Person {name: "Tom Hanks"})-[r:ACTED_IN]->(m:Movie {title: "Forrest Gump"})
SET r.role = "Forrest Gump"

插入复杂结构

// 创建一个社区，然后匹配人员以将他们添加到社区
CREATE (c1:Community {name: "Hollywood Stars"})
WITH c1
MATCH (p:Person)
WHERE p.name IN ["Tom Hanks", "Kevin Bacon"]
CREATE (p)-[:MEMBER_OF]->(c1)

其他事项 - 样板 - 不清楚 - 暂时忽略。

<div align="center">

Model Context Protocol (MCP) 的 Python 实现

</div>

<!-- omit in toc -->

目录

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

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

什么是 MCP？

Model Context Protocol (MCP) 让你构建服务器，以安全、标准化的方式向 LLM 应用程序公开数据和功能。 可以把它想象成一个 Web API，但专门为 LLM 交互而设计。 MCP 服务器可以：

• 通过 资源 公开数据（可以把它们想象成 GET 端点；它们用于将信息加载到 LLM 的上下文中）
• 通过 工具 提供功能（有点像 POST 端点；它们用于执行代码或产生副作用）
• 通过 提示词 定义交互模式（用于 LLM 交互的可重用模板）
• 还有更多！

核心概念

服务器

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

# 添加对具有强类型的启动/关闭的生命周期支持
from dataclasses import dataclass
from typing import AsyncIterator
from mcp.server.fastmcp import FastMCP

# 创建一个命名服务器
mcp = FastMCP("My App")

# 指定部署和开发的依赖项
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])

@dataclass
class AppContext:
    db: Database  # 替换为你的实际 DB 类型

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """使用类型安全上下文管理应用程序生命周期"""
    try:
        # 在启动时初始化
        await db.connect()
        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()

资源

资源是你如何向 LLM 公开数据。 它们类似于 REST API 中的 GET 端点 - 它们提供数据，但不应执行重要的计算或产生副作用：

@mcp.resource("config://app")
def get_config() -> str:
    """静态配置数据"""
    return "App configuration here"

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """动态用户数据"""
    return f"Profile data for user {user_id}"

工具

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

@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """计算给定体重（公斤）和身高（米）的 BMI"""
    return weight_kg / (height_m ** 2)

@mcp.tool()
async def fetch_weather(city: str) -> str:
    """获取城市当前天气"""
    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:
    """从图像创建缩略图"""
    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:
    """处理具有进度跟踪的多个文件"""
    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

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

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

Claude Desktop 集成

服务器准备就绪后，将其安装在 Claude Desktop 中：

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

直接执行

对于自定义部署等高级场景：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")

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

使用以下命令运行它：

python server.py
# or
mcp run server.py

示例

Echo 服务器

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

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Echo")

@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    """将消息作为资源回显"""
    return f"Resource echo: {message}"

@mcp.tool()
def echo_tool(message: str) -> str:
    """将消息作为工具回显"""
    return f"Tool echo: {message}"

@mcp.prompt()
def echo_prompt(message: str) -> str:
    """创建一个回显提示词"""
    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:
    """将数据库模式作为资源提供"""
    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:
    """安全地执行 SQL 查询"""
    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 typing import AsyncIterator

@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """管理服务器启动和关闭生命周期。"""
    try:
        # 在启动时初始化资源
        await db.connect()
        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 提供：

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

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

# 创建一个服务器实例
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
from mcp.client.stdio import stdio_client

# 为 stdio 连接创建服务器参数
server_params = StdioServerParameters(
    command="python", # 可执行文件
    args=["example_server.py"], # 可选命令行参数
    env=None # 可选环境变量
)

# 可选：创建一个采样回调
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 原语

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

原语	控制	描述	示例用法
提示词	用户控制	由用户选择调用的交互式模板	斜杠命令、菜单选项
资源	应用控制	由客户端应用程序管理的上下文数据	文件内容、API 响应
工具	模型控制	公开给 LLM 以采取行动的函数	API 调用、数据更新

服务器能力

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

能力	功能标志	描述
prompts	listChanged	提示词模板管理
resources	subscribe<br/>listChanged	资源公开和更新
tools	listChanged	工具发现和执行
logging	-	服务器日志记录配置
completion	-	参数完成建议

文档

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

贡献

我们热衷于支持所有经验水平的贡献者，并希望看到你参与到项目中。 请参阅 贡献指南 以开始使用。

许可证

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