Python开发者必看：5分钟用MCP SDK搭建你的第一个AI工具服务器（附完整代码）

作为一名Python开发者，你可能已经习惯了用Flask或FastAPI构建传统Web服务。但当你需要为AI应用构建专用服务端时，MCP（Model Context Protocol）Python SDK提供了一个全新的范式。它专为AI工作流设计，能让你快速暴露工具、资源和提示模板，与大型语言模型无缝集成。

1. 环境准备：安装与基础配置

在开始之前，确保你的开发环境满足以下要求：

• Python 3.10或更高版本
• 包管理工具（推荐使用uv或pip）
• 基本的Python开发环境

安装MCP SDK非常简单，只需运行以下命令：

pip install "mcp[cli]"

如果你使用更现代的uv工具链：

uv add "mcp[cli]"

验证安装是否成功：

mcp --version

提示：建议使用虚拟环境隔离项目依赖，避免与其他项目产生冲突。可以使用python -m venv venv创建虚拟环境。

2. 构建第一个MCP服务器

让我们从一个最简单的"Hello World"示例开始，逐步扩展功能。创建一个名为demo_server.py的文件：

from mcp.server.fastmcp import FastMCP

# 初始化MCP服务器实例
mcp = FastMCP(
    "DemoServer",  # 服务器名称
    description="我的第一个MCP服务器",  # 可选描述
    version="0.1.0"  # 版本号
)

# 添加一个工具
@mcp.tool()
def greet(name: str) -> str:
    """生成个性化问候语"""
    return f"你好，{name}！欢迎使用MCP服务。"

# 添加一个计算器工具
@mcp.tool()
def calculate(a: float, b: float, operation: str) -> float:
    """执行基本数学运算
    
    参数:
        a: 第一个操作数
        b: 第二个操作数
        operation: 运算类型(add/subtract/multiply/divide)
    """
    operations = {
        'add': a + b,
        'subtract': a - b,
        'multiply': a * b,
        'divide': a / b if b != 0 else float('nan')
    }
    return operations.get(operation.lower(), float('nan'))

if __name__ == "__main__":
    # 启动服务器
    mcp.run()

这个简单的服务器暴露了两个工具：

1. greet - 生成个性化问候语
2. calculate - 执行基本数学运算

3. 运行与测试服务器

启动服务器有多种方式，最简单的是直接运行Python文件：

python demo_server.py

或者使用MCP CLI工具：

mcp run demo_server.py

启动后，你会看到类似这样的输出：

INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

3.1 使用MCP Inspector测试

MCP提供了一个交互式测试工具——MCP Inspector。安装并运行它：

npx -y @modelcontextprotocol/inspector

在Inspector中连接到http://localhost:8000/mcp，你将看到：

• 可用工具列表
• 每个工具的详细描述和参数
• 交互式测试界面

3.2 命令行测试

你也可以直接通过curl测试API端点：

curl -X POST http://localhost:8000/mcp/tools/calculate \
  -H "Content-Type: application/json" \
  -d '{"a": 10, "b": 5, "operation": "add"}'

预期响应：

{"result":15}

4. 扩展服务器功能

基础服务器运行起来后，让我们添加更多实用功能。

4.1 添加资源端点

资源是MCP中的只读数据端点，类似于REST API中的GET端点：

# 在demo_server.py中添加

@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: str) -> dict:
    """获取用户简档数据"""
    # 这里可以是数据库查询或其他数据源
    return {
        "id": user_id,
        "name": "示例用户",
        "join_date": "2023-01-01",
        "preferences": {"theme": "dark", "language": "zh"}
    }

4.2 添加提示模板

提示模板帮助标准化LLM交互：

from mcp.server.fastmcp.prompts import base

@mcp.prompt()
def code_review_prompt(code: str, language: str = "python") -> list[base.Message]:
    """生成代码审查提示"""
    return [
        base.UserMessage(f"请审查以下{language}代码，指出潜在问题和改进建议："),
        base.UserMessage(code),
        base.AssistantMessage("我将分析这段代码。首先，我看到...")
    ]

4.3 添加图像处理工具

MCP支持图像数据类型：

from PIL import Image as PILImage
from mcp.server.fastmcp import Image

@mcp.tool()
def resize_image(image_path: str, width: int, height: int) -> Image:
    """调整图像尺寸"""
    img = PILImage.open(image_path)
    img = img.resize((width, height))
    return Image(data=img.tobytes(), format="png")

5. 高级配置与最佳实践

5.1 生命周期管理

对于需要初始化资源的服务器，使用生命周期管理：

from contextlib import asynccontextmanager
from typing import AsyncIterator
from dataclasses import dataclass

@dataclass
class AppState:
    db_connection: Database
    config: dict

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
    """管理服务器生命周期"""
    # 启动时初始化
    db = await Database.connect("postgres://user:pass@localhost/db")
    config = load_config()
    
    try:
        yield AppState(db, config)
    finally:
        # 关闭时清理
        await db.disconnect()

mcp = FastMCP("AdvancedServer", lifespan=lifespan)

5.2 错误处理

为工具添加健壮的错误处理：

@mcp.tool()
async def safe_file_operation(file_path: str, ctx: Context) -> str:
    """安全的文件操作"""
    try:
        if not os.path.exists(file_path):
            await ctx.error(f"文件不存在: {file_path}")
            raise ValueError("文件不存在")
        
        # 执行文件操作...
        return "操作成功"
    except Exception as e:
        await ctx.error(f"文件操作失败: {str(e)}")
        raise

5.3 性能监控

添加性能追踪：

@mcp.tool()
async def expensive_operation(ctx: Context) -> str:
    """耗时操作示例"""
    start_time = time.time()
    
    # 执行操作...
    await ctx.info("操作进行中...")
    
    duration = time.time() - start_time
    await ctx.metric("operation_duration", duration)
    
    return f"操作完成，耗时{duration:.2f}秒"

6. 部署与集成

6.1 生产环境部署

对于生产环境，推荐使用Streamable HTTP传输：

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        stateless_http=True,  # 无状态模式，适合水平扩展
        json_response=True,   # 使用JSON响应
        host="0.0.0.0",      # 监听所有接口
        port=8080            # 使用标准HTTP端口
    )

6.2 集成到Claude Desktop

将服务器安装到Claude Desktop非常简单：

mcp install demo_server.py --name "我的工具服务器"

或者带环境变量：

mcp install demo_server.py -v DB_URL=postgres://user:pass@localhost/db

6.3 挂载到现有ASGI应用

如果你已经有FastAPI或Starlette应用，可以挂载MCP服务器：

from fastapi import FastAPI
from starlette.routing import Mount

app = FastAPI()
app.mount("/mcp", mcp.streamable_http_app())

# 其他FastAPI路由...
@app.get("/health")
def health_check():
    return {"status": "ok"}

7. 完整示例代码

以下是整合了所有功能的完整示例：

from contextlib import asynccontextmanager
from typing import AsyncIterator
from dataclasses import dataclass
from datetime import datetime
import os
import time

from mcp.server.fastmcp import FastMCP, Context, Image
from mcp.server.fastmcp.prompts import base
from PIL import Image as PILImage

# 模拟数据库连接
class Database:
    @classmethod
    async def connect(cls, url):
        print(f"连接到数据库: {url}")
        return cls()
    
    async def disconnect(self):
        print("关闭数据库连接")
    
    async def query(self, sql):
        return f"执行查询: {sql}"

@dataclass
class AppState:
    db: Database
    startup_time: datetime

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
    """管理服务器生命周期"""
    db = await Database.connect("postgres://user:pass@localhost/db")
    try:
        yield AppState(db, datetime.now())
    finally:
        await db.disconnect()

# 初始化MCP服务器
mcp = FastMCP(
    "FullFeaturedServer",
    description="功能完整的MCP服务器示例",
    version="1.0.0",
    lifespan=lifespan
)

# 工具示例
@mcp.tool()
async def process_data(ctx: Context, input_data: str) -> str:
    """处理输入数据"""
    # 访问生命周期资源
    db = ctx.request_context.lifespan_context.db
    result = await db.query(f"SELECT * FROM data WHERE id = '{input_data}'")
    
    await ctx.info(f"处理数据: {input_data}")
    await ctx.metric("data_processed", 1)
    
    return f"处理结果: {result}"

# 资源示例
@mcp.resource("system://status")
async def system_status(ctx: Context) -> dict:
    """获取系统状态"""
    state = ctx.request_context.lifespan_context
    return {
        "status": "running",
        "uptime": str(datetime.now() - state.startup_time),
        "db_connected": True
    }

# 图像处理工具
@mcp.tool()
def process_image(image_path: str, operation: str) -> Image:
    """处理图像"""
    img = PILImage.open(image_path)
    
    if operation == "grayscale":
        img = img.convert("L")
    elif operation == "thumbnail":
        img.thumbnail((100, 100))
    
    return Image(data=img.tobytes(), format="png")

# 提示模板
@mcp.prompt()
def analysis_prompt(data: str, context: str) -> list[base.Message]:
    """数据分析提示"""
    return [
        base.UserMessage("请分析以下数据，考虑提供的上下文："),
        base.UserMessage(f"上下文: {context}"),
        base.UserMessage(f"数据: {data}"),
        base.AssistantMessage("基于这些信息，我的初步观察是...")
    ]

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        stateless_http=True,
        host="0.0.0.0",
        port=8000
    )

这个完整示例展示了：

• 生命周期管理
• 数据库集成
• 工具、资源和提示模板
• 图像处理
• 生产环境配置

8. 常见问题排查

8.1 工具不可见

如果工具没有出现在MCP Inspector中：

1. 检查工具装饰器@mcp.tool()是否正确应用
2. 确保工具函数有类型注解
3. 验证函数有文档字符串（用作工具描述）

8.2 连接问题

常见连接问题解决方法：

问题现象	可能原因	解决方案
连接被拒绝	服务器未运行	检查服务器进程是否启动
404错误	路径不正确	确保访问/mcp端点
500错误	服务器内部错误	查看服务器日志

8.3 性能优化

对于性能敏感的应用：

• 使用stateless_http=True减少内存占用
• 启用json_response=True简化响应处理
• 考虑使用异步工具处理IO密集型操作

@mcp.tool()
async def async_operation(ctx: Context):
    """异步工具示例"""
    await some_io_operation()

9. 安全注意事项

构建MCP服务器时，安全至关重要：

1. 输入验证：始终验证工具参数

   @mcp.tool()
   def safe_operation(input: str):
       if not input.isalnum():
           raise ValueError("输入包含非法字符")
   

2. 访问控制：实现认证中间件

   mcp = FastMCP("SecureServer", auth=AuthSettings(...))
   

3. 错误处理：避免泄露敏感信息

   try:
       # 敏感操作
   except Exception:
       raise ValueError("操作失败") from None
   

10. 下一步学习路径

掌握了基础MCP服务器开发后，你可以探索：

1. 高级主题：

   • 自定义传输协议
   • 低级服务器API
   • 协议扩展

2. 集成方案：

   • 与现有Web框架集成
   • 微服务架构中的MCP
   • Kubernetes部署

3. 客户端开发：

   from mcp.client import ClientSession
   async with ClientSession(...) as session:
       await session.call_tool("tool_name", {...})
   

MCP为Python开发者提供了一个强大的工具集，用于构建AI原生服务。通过将传统后端开发经验与MCP的专有特性结合，你可以创建出既强大又灵活的AI集成解决方案。