使用说明

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 a community and then match persons to add them to the community
CREATE (c1:Community {name: "Hollywood Stars"})
WITH c1
MATCH (p:Person)
WHERE p.name IN ["Tom Hanks", "Kevin Bacon"]
CREATE (p)-[:MEMBER_OF]->(c1)

其他事情-锅炉板-没有线索-暂时忽略。

模型上下文协议（MCP）的Python实现

概述

模型上下文协议允许应用程序以标准化的方式为LLM提供上下文，将提供上下文的关注点与实际的LLM交互分开。此Python SDK实现了完整的MCP规范，使其易于：

构建可以连接到任何MCP服务器的MCP客户端
创建公开资源、提示和工具的MCP服务器
使用标准传输方式，如stdio和SSE
处理所有MCP协议消息和生命周期事件

安装

我们建议使用紫外线管理你的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}!"

您可以在以下位置安装此服务器克劳德桌面并通过运行以下命令立即与之交互：

mcp install server.py

或者，您可以使用MCP检查器进行测试：

mcp dev server.py

什么是MCP？

这模型上下文协议（MCP）允许您构建以安全、标准化的方式向LLM应用程序公开数据和功能的服务器。将其视为一个web API，但它是专门为LLM交互而设计的。MCP服务器可以：

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

核心概念

服务器

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

# Add lifespan support for startup/shutdown with strong typing
from dataclasses import dataclass
from typing import AsyncIterator
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"])

@dataclass
class AppContext:
    db: Database  # Replace with your actual DB type

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context"""
    try:
        # Initialize on startup
        await db.connect()
        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()

资源

资源是您向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, mime_type = await ctx.read_resource(f"file://{file}")
    return "Processing complete"

运行服务器

发展模式

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

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

高级用法

低级服务器

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

from contextlib import asynccontextmanager
from typing import AsyncIterator

@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """Manage server startup and shutdown lifecycle."""
    try:
        # Initialize resources on startup
        await db.connect()
        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的使用寿命提供：

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

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, 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

# 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协议定义了服务器可以实现的三个核心原语：

图元	控件	描述	示例用法
提示	用户控制	用户选择调用的交互式模板	Slash命令、菜单选项
资源	应用程序控制	客户端应用程序管理的上下文数据	文件内容、API响应
工具	模型控制	向LLM公开以执行操作的函数	API调用、数据更新

服务器功能

MCP服务器在初始化过程中声明功能：

功能	功能标志	描述
`prompts`	`listChanged`	快速模板管理
`resources`	`subscribe` `listChanged`	资源暴露和更新
`tools`	`listChanged`	工具发现和执行
`logging`	-	服务器日志记录配置
`completion`	-	论点完成建议

文档

贡献

我们热衷于支持各级经验的贡献者，并希望看到您参与该项目。请参阅贡献指南开始吧。

许可证

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

模型上下文协议SDK

使用说明

MCP Python SDK

用法

运行客户端

运行服务器

数据准备

插入节点：

插入关系

插入属性

插入复杂结构

其他事情-锅炉板-没有线索-暂时忽略。

目录

概述

安装

快速启动

什么是MCP？

核心概念

服务器

资源

工具

鼓励

图像

上下文

运行服务器

发展模式

Claude桌面集成

直接执行

例子

回声服务器

SQLite资源管理器

高级用法

低级服务器

编写MCP客户端

MCP图元

服务器功能

文档

贡献

许可证

使用说明

MCP Python SDK

用法

运行客户端

运行服务器

数据准备

插入节点：

插入关系

插入属性

插入复杂结构

其他事情-锅炉板-没有线索-暂时忽略。

目录

概述

安装

快速启动

什么是MCP？

核心概念

服务器

资源

工具

鼓励

图像

上下文

运行服务器

发展模式

Claude桌面集成

直接执行

例子

回声服务器

SQLite资源管理器

高级用法

低级服务器

编写MCP客户端

MCP图元

服务器功能

文档

贡献

许可证

相关 MCP

代码分析服务器

AI开发框架

Token节约MCP服务

GraphQL API交互服务

内存银行管理协议

AI应用与智能体开发框架