MCP Python SDK 核心概念与实战指南

3.2 运行与测试

MCP 服务通常通过标准输入输出（Stdio）运行。要在 Claude Desktop 中使用此服务：

1. 打开 Claude Desktop 配置文件 (~/Library/Application Support/Claude/claude_desktop_config.json 或 Windows 对应路径)。
2. 重启 Claude Desktop，你现在可以直接在这个 AI 聊天框中让它计算斜边长了。

添加配置：

{
  "mcpServers": {
    "my-math": {
      "command": "python",
      "args": ["/绝对路径/到/math_server.py"]
    }
  }
}

4. 进阶篇：核心原语详解

除了工具 (Tools)，MCP 还定义了 资源 (Resources) 和 提示词 (Prompts)。

4.1 资源 (Resources)

资源类似于 API 中的 GET 请求，用于向 LLM 暴露只读数据。它们通过 URI 寻址。

假设我们需要让 AI 读取系统的日志文件：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("LogServer")

# 定义动态资源
# 这里的 pattern "logs://{system_name}/current" 就像 API 路由
@mcp.resource("logs://{system_name}/current")
def get_system_logs(system_name: str) -> str:
    """获取指定系统的最新日志"""
    # 在实际应用中，这里会读取文件或数据库
    # 假设日志数据生成函数 f(x)
    return f"Log entry for {system_name}: System status implies stability > 99%."

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

4.2 提示词 (Prompts)

提示词是预定义的上下文模板，用于引导用户与 AI 的对话。

@mcp.prompt()
def review_code(code: str) -> str:
    """创建一个代码审查的 Prompt 模板"""
    return f"""
    请作为一名资深架构师审查以下代码。
    重点关注：安全性、性能 (O(n) 复杂度) 和可读性。
    代码内容：{code}
    """

5. 高级篇：底层架构与生命周期

对于需要精细控制（如自定义生命周期、鉴权、复杂的异步流）的场景，不能只用 FastMCP，需要直接操作底层的 Server 类。

5.1 手动构建 Server 与 Session

这里展示如何不使用 FastMCP，而是使用底层的 Server 和 StdioServerTransport。

import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.stdio import StdioServerTransport
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource

# 1. 实例化低级 Server 对象
app = Server("LowLevelApp")

# 2. 注册工具处理器
@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="echo",
            description="Echo back input",
            inputSchema={"type":"object","properties":{"message":{"type":"string"}},"required":["message"]}
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "echo":
        msg = arguments["message"]
        return [TextContent(type="text", text=f"Echo: {msg}")]
    raise ValueError(f"Unknown tool: {name}")

# 3. 显式的主循环控制
async def main():
    # 使用标准输入输出传输层
    async with StdioServerTransport() as transport:
        # 将 Server 连接到 Transport
        await app.run(
            transport.read_messages,
            transport.write_message,
            # 初始化时的 capabilities 设置
            initialization_options=None
        )

if __name__ == "__main__":
    asyncio.run(main())

5.2 异步与并发

MCP SDK 完全基于 Python 的 asyncio。这意味着你可以：

• 在 Tool 内部并发调用外部 API。
• 使用 Context 对象向客户端发送进度通知或日志。

并发执行时总延迟约等于各任务最大耗时：Total Latency ≈ max(T_api1, T_api2)。

@mcp.tool()
async def fetch_concurrent_data(ctx: Context):
    # 向客户端发送日志，不阻塞主流程
    await ctx.info("Starting concurrent fetch...")
    # 假设这里有两个耗时的 IO 操作
    res1, res2 = await asyncio.gather(task1(), task2())
    return f"Result: {res1}, {res2}"

6. 客户端篇：构建 MCP Client

MCP 是双向的。除了构建 Server 给 Claude 用，你也可以构建 Client 来调用别人的 Server（例如连接到本地的 Postgres MCP Server）。

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

async def run_client():
    # 定义要连接的 Server 参数
    server_params = StdioServerParameters(
        command="python",
        args=["my_server.py"],  # 指向你的 Server 脚本
    )
    
    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 初始化
            await session.initialize()
            # 2. 列出可用工具
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")
            # 3. 调用工具
            result = await session.call_tool("add", arguments={"a": 10, "b": 20})
            print(f"Calculation Result: {result.content[0].text}")

if __name__ == "__main__":
    asyncio.run(run_client())

7. 实战案例：SQLite 数据库管理器

我们将结合所有知识，构建一个具备资源读取和工具执行能力的 SQLite MCP Server。

功能设计

1. Resource: sqlite://schema 查看所有表结构。
2. Tool: query_db 执行只读 SQL 查询。

代码实现

import sqlite3
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

# 初始化数据库（演示用）
DB_PATH = "demo.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, role TEXT)")
    conn.execute("INSERT OR IGNORE INTO users (id, name, role) VALUES (1, 'Alice', 'Admin')")
    conn.commit()
    conn.close()

init_db()

mcp = FastMCP("SQLiteManager")

# --- Resource: 暴露数据库 Schema ---
@mcp.resource("sqlite://schema")
def get_schema() -> str:
    """获取数据库当前的 Schema 信息"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("SELECT sql FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    conn.close()
    schema_str = "\n".join([t[0] for t in tables if t[0]])
    return f"Database Schema:\n{schema_str}"

# --- Tool: 执行 SQL 查询 ---
# 使用 Pydantic 定义更加严谨的输入结构
class QueryParams(BaseModel):
    query: str = Field(description="SELECT SQL query to execute")

@mcp.tool()
def query_database(params: QueryParams) -> str:
    """执行只读 SQL 查询 (SELECT only)"""
    query = params.query.strip()
    # 简单的安全检查 (实际生产需更严格的校验)
    if not query.upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries are allowed for safety.")
    
    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.cursor()
        cursor.execute(query)
        results = cursor.fetchall()
        # 将结果转换为易读的字符串格式
        return str(results)
    except Exception as e:
        return f"Error executing query: {str(e)}"
    finally:
        conn.close()

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

8. 最佳实践与注意事项

1. 安全性 (Security):
   • MCP Server 本质上是在允许 LLM 在你的机器上执行代码。务必对 Tool 的输入进行严格校验（如上面的 SQL 检查）。
   • 不要在 Tool 中硬编码敏感密钥，使用环境变量。
2. 错误处理 (Error Handling):
   • 不要让 Server 崩溃。在 Tool 内部捕获异常并返回人类可读的错误信息字符串，这样 LLM 可以尝试自我修正参数并重试。
   • 可以使用 mcp-inspector（需要 npm install -g @modelcontextprotocol/inspector）来可视化调试你的 Python Server。
3. 性能 (Performance):
   • 对于重型计算任务，请务必使用 async def 并避免在主线程中执行阻塞调用。假设任务耗时为 t，若阻塞主线程，吞吐量将降为 1/t。

调试 (Debugging):

npx @modelcontextprotocol/inspector python your_server.py

通过本文档，你应该已经掌握了从最简单的函数包装到构建完整的、基于资源的 MCP 服务的全部技能。