1. 项目概述为什么MCP服务器开发是当前的热点最近在开发者社区里MCPModel Context Protocol这个词的热度越来越高。如果你关注AI应用开发尤其是大模型与工具集成的领域那么你很可能已经听说过它。简单来说MCP是一个标准协议它定义了大模型如GPT-4、Claude、DeepSeek等如何安全、结构化地与外部工具、数据源和API进行交互。你可以把它想象成大模型的“USB接口”或“插件系统”——它让模型不再是一个封闭的黑盒而是能连接到你自己的数据库、内部API、文件系统甚至硬件设备从而执行更复杂、更定制化的任务。那么开发一个MCP服务器具体是做什么呢想象一个场景你希望你的AI助手能帮你查询公司内部的销售数据、自动创建Jira工单或者控制你家里的智能灯光。这些能力模型本身并不具备它们存在于你的私有环境里。MCP服务器的角色就是作为这些私有能力和大模型之间的“翻译官”和“安全网关”。它遵循MCP协议将你的内部工具包装成模型可以理解和调用的标准化“工具”同时处理认证、授权、数据格式转换和错误处理。这使得像Cursor、Claude Desktop、Windmill这类支持MCP的AI客户端能够无缝、安全地使用你提供的专属能力。这个项目标题“15分钟从零到生产级部署”之所以吸引人是因为它戳中了开发者的痛点协议听起来很酷但上手门槛高吗部署复杂吗能否快速看到效果本文将带你进行一次高速实战我们将使用Python——这门在AI和自动化领域最流行的语言从零开始构建一个具备实用价值的MCP服务器并完成包括OAuth安全认证在内的、可直接用于生产环境的部署。无论你是想为团队内部打造AI增效工具还是探索AI应用的新形态这都将是一次极具价值的实践。2. 核心设计构建一个生产就绪的MCP服务器架构在动手写代码之前我们先花点时间厘清思路。一个“生产级”的MCP服务器绝不仅仅是一个能跑通的Demo。它需要兼顾功能、安全、可靠性和可维护性。我们的设计目标是在15分钟内搭建起一个具备这些特质雏形的系统框架。2.1 协议理解与工具定义MCP协议的核心抽象是“工具”Tools和“资源”Resources。对于大多数应用场景“工具”是我们关注的重点。一个工具由名称、描述、输入参数schema和具体的执行函数组成。当AI客户端如Cursor需要完成某项任务时它会根据工具描述构造符合schema的参数并通过JSON-RPC over STDIO标准输入输出或HTTP调用我们的服务器。我们的第一个服务器将实现一个经典且实用的功能一个待办事项Todo List管理器。为什么选这个因为它逻辑简单但涵盖了CRUD增删改查核心操作非常适合演示。我们将定义四个工具list_todos: 列出所有待办事项。create_todo: 创建新的待办事项。update_todo_status: 更新某个事项的状态如完成/未完成。delete_todo: 删除一个事项。这些工具将操作一个后端数据存储。为了快速达到“生产级”我们不能用内存变量那样数据无法持久化。我们将选择SQLite数据库它无需额外服务单文件存储非常适合轻量级应用和快速原型。同时我们会为数据库操作设计一个简单的数据访问层以便未来轻松替换为PostgreSQL或MySQL。2.2 安全与认证设计OAuth集成这是“生产级”与“玩具级”最关键的区分点。一个暴露在外的、无需认证的服务器是极其危险的。MCP协议支持在服务器声明中要求客户端进行认证。我们将集成OAuth 2.0授权码流程这是业界标准的API认证方式。流程是这样的我们的MCP服务器在初始化时告诉客户端“调用我的工具需要OAuth认证”。当用户首次在客户端如Cursor中尝试使用我们的工具时客户端会弹出浏览器引导用户跳转到我们指定的授权页面可以是我们自建的也可以是第三方如GitHub、Google的。用户登录并授权后授权服务器会通过回调地址返回一个授权码code。客户端用这个code向授权服务器交换访问令牌access_token。客户端在后续所有调用我们MCP服务器的请求中都会携带这个access_token。我们的服务器收到请求后需要验证这个token的有效性例如向授权服务器的introspection端点发起验证。在本实战中为了在15分钟内完成闭环我们将模拟一个简化但符合流程的OAuth提供方。我们会创建一个简单的Flask应用来扮演授权服务器颁发和验证JWTJSON Web Token令牌。在实际生产中你可以将这个提供方替换为Auth0、Okta或你公司的统一认证中心。2.3 技术栈选型MCP SDK: 我们将使用官方提供的mcpPython库。这是最直接、最兼容的选择它封装了协议通信的底层细节让我们专注于工具逻辑。Web框架用于OAuth服务: 选择Flask。它轻量、灵活适合快速搭建小型HTTP服务。我们的OAuth授权服务器和令牌验证端点都将用它构建。数据库:SQLitesqlite3Python内置。无需安装开箱即用。令牌管理:PyJWT库用于生成和验证JWT格式的访问令牌。进程通信: MCP默认使用STDIO标准输入输出这对于本地集成是最简单可靠的。我们的服务器将作为一个独立的命令行程序启动通过标准流与客户端对话。部署与打包: 我们将使用Docker进行容器化。这是生产部署的黄金标准它能确保环境一致性并方便地在任何支持Docker的服务器上运行。这个选型平衡了开发速度、学习成本和生产可靠性。所有库都是主流且文档丰富遇到问题容易找到解决方案。3. 逐步实现从零编写服务器核心代码现在我们开始动手编码。请确保你的开发环境已安装Python 3.9或更高版本。3.1 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境这是管理Python项目依赖的最佳实践。mkdir mcp-todo-server cd mcp-todo-server python -m venv venv # 在Windows上使用venv\Scripts\activate source venv/bin/activate接下来创建requirements.txt文件列出我们需要的所有依赖。# requirements.txt mcp1.0.0 flask3.0.0 pyjwt2.8.0然后安装它们pip install -r requirements.txt3.2 构建数据层数据库模型与操作在项目根目录下创建database.py文件。这里我们将定义Todo的数据模型和基本的数据库操作类。# database.py import sqlite3 import json from typing import List, Optional, Dict, Any from dataclasses import dataclass, asdict from datetime import datetime dataclass class TodoItem: id: Optional[int] None title: str description: str status: str pending # pending, in_progress, completed created_at: Optional[str] None updated_at: Optional[str] None class TodoDatabase: def __init__(self, db_path: str todos.db): self.db_path db_path self._init_db() def _init_db(self): 初始化数据库表 with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, description TEXT, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建一个触发器在更新时自动更新updated_at字段 conn.execute( CREATE TRIGGER IF NOT EXISTS update_todo_timestamp AFTER UPDATE ON todos BEGIN UPDATE todos SET updated_at CURRENT_TIMESTAMP WHERE id NEW.id; END; ) def get_connection(self): 获取数据库连接并设置返回字典格式的行 conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row # 使查询结果可按列名访问 return conn def create_todo(self, title: str, description: str ) - TodoItem: 创建新的待办事项 with self.get_connection() as conn: cursor conn.cursor() cursor.execute( INSERT INTO todos (title, description) VALUES (?, ?), (title, description) ) todo_id cursor.lastrowid # 获取刚创建的项目 cursor.execute(SELECT * FROM todos WHERE id ?, (todo_id,)) row cursor.fetchone() conn.commit() return self._row_to_todo(row) def list_todos(self, status_filter: Optional[str] None) - List[TodoItem]: 列出所有待办事项可选项按状态过滤 with self.get_connection() as conn: query SELECT * FROM todos params () if status_filter: query WHERE status ? params (status_filter,) query ORDER BY created_at DESC cursor conn.execute(query, params) return [self._row_to_todo(row) for row in cursor.fetchall()] def update_todo_status(self, todo_id: int, new_status: str) - Optional[TodoItem]: 更新待办事项状态 allowed_statuses [pending, in_progress, completed] if new_status not in allowed_statuses: raise ValueError(f状态必须是其中之一{allowed_statuses}) with self.get_connection() as conn: cursor conn.cursor() cursor.execute( UPDATE todos SET status ? WHERE id ?, (new_status, todo_id) ) if cursor.rowcount 0: return None # 没有找到对应ID的项目 cursor.execute(SELECT * FROM todos WHERE id ?, (todo_id,)) row cursor.fetchone() conn.commit() return self._row_to_todo(row) def delete_todo(self, todo_id: int) - bool: 删除一个待办事项 with self.get_connection() as conn: cursor conn.cursor() cursor.execute(DELETE FROM todos WHERE id ?, (todo_id,)) conn.commit() return cursor.rowcount 0 def _row_to_todo(self, row: sqlite3.Row) - TodoItem: 将数据库行转换为TodoItem对象 if row is None: return None return TodoItem( idrow[id], titlerow[title], descriptionrow[description], statusrow[status], created_atrow[created_at], updated_atrow[updated_at] )注意我们使用了sqlite3.Row工厂它允许我们通过列名如row[‘id’]来访问数据这比使用数字索引更清晰、更不易出错。同时我们利用SQLite的触发器自动管理updated_at时间戳这是一个保持数据一致性的好习惯。3.3 实现MCP服务器主逻辑现在创建核心的MCP服务器文件server.py。我们将使用mcpSDK来声明工具并处理请求。# server.py import asyncio import sys import json from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 导入我们刚才写的数据层 from database import TodoDatabase # 初始化数据库 db TodoDatabase() async def handle_list_todos(arguments: dict) - str: 处理列出待办事项的请求 status_filter arguments.get(status_filter) todos db.list_todos(status_filterstatus_filter) if not todos: return 当前没有待办事项。 result_lines [] for todo in todos: result_lines.append(f- [{todo.id}] {todo.title} ({todo.status})) if todo.description: result_lines.append(f 描述{todo.description}) result_lines.append(f 创建于{todo.created_at}) return \n.join(result_lines) async def handle_create_todo(arguments: dict) - str: 处理创建待办事项的请求 title arguments.get(title) if not title: return 错误必须提供待办事项的标题。 description arguments.get(description, ) todo db.create_todo(title, description) return f已成功创建待办事项 [#{todo.id}]{todo.title} async def handle_update_status(arguments: dict) - str: 处理更新状态的请求 todo_id arguments.get(todo_id) new_status arguments.get(new_status) if not todo_id or not new_status: return 错误必须提供待办事项ID和新状态。 try: todo_id_int int(todo_id) except ValueError: return 错误待办事项ID必须是数字。 updated db.update_todo_status(todo_id_int, new_status) if updated: return f已成功将待办事项 [#{updated.id}] 的状态更新为 {updated.status}。 else: return f错误未找到ID为 {todo_id} 的待办事项。 async def handle_delete_todo(arguments: dict) - str: 处理删除待办事项的请求 todo_id arguments.get(todo_id) if not todo_id: return 错误必须提供待办事项ID。 try: todo_id_int int(todo_id) except ValueError: return 错误待办事项ID必须是数字。 success db.delete_todo(todo_id_int) if success: return f已成功删除待办事项 [#{todo_id}]。 else: return f错误未找到ID为 {todo_id} 的待办事项删除失败。 async def main(): # 定义服务器提供的工具列表 # 每个工具需要名称、描述和输入参数的模式schema tools [ { name: list_todos, description: 列出所有的待办事项。可以选择按状态过滤pending, in_progress, completed。, inputSchema: { type: object, properties: { status_filter: { type: string, description: 按状态过滤待办事项。可选值pending, in_progress, completed。, enum: [pending, in_progress, completed] } } } }, { name: create_todo, description: 创建一个新的待办事项。, inputSchema: { type: object, properties: { title: { type: string, description: 待办事项的标题必填。 }, description: { type: string, description: 待办事项的详细描述可选。 } }, required: [title] } }, { name: update_todo_status, description: 更新一个现有待办事项的状态。, inputSchema: { type: object, properties: { todo_id: { type: string, description: 要更新的待办事项的ID数字。 }, new_status: { type: string, description: 新的状态。, enum: [pending, in_progress, completed] } }, required: [todo_id, new_status] } }, { name: delete_todo, description: 删除一个待办事项。, inputSchema: { type: object, properties: { todo_id: { type: string, description: 要删除的待办事项的ID数字。 } }, required: [todo_id] } } ] # 创建服务器参数声明我们需要的OAuth认证 server_params StdioServerParameters( commandsys.executable, # 使用当前Python解释器 args[__file__], # 运行这个脚本本身 # 声明服务器要求OAuth 2.0授权码流程认证 # 这里我们使用一个模拟的授权服务器地址 auth(oauth, { authorization_url: http://localhost:5001/oauth/authorize, token_url: http://localhost:5001/oauth/token, scopes: [read_todos, write_todos], client_id: mcp-todo-client }) ) # 使用stdio_client管理服务器进程和通信 async with stdio_client(server_params) as (read_stream, write_stream): session ClientSession(read_stream, write_stream) # 初始化会话向客户端宣告我们的工具 await session.initialize(toolstools) # 进入主循环监听并处理客户端请求 async for message in session.listen(): if message.method tools/call: # 提取工具调用请求的详细信息 call_id message.params[callId] tool_name message.params[name] arguments message.params.get(arguments, {}) # 根据工具名称路由到对应的处理函数 handler_map { list_todos: handle_list_todos, create_todo: handle_create_todo, update_todo_status: handle_update_status, delete_todo: handle_delete_todo, } handler handler_map.get(tool_name) if handler: try: # 异步执行工具逻辑 result_text await handler(arguments) # 将成功结果发送回客户端 await session.send_tool_call_result(call_id, content[{ type: text, text: result_text }]) except Exception as e: # 如果执行出错返回错误信息 await session.send_tool_call_error(call_id, codeINTERNAL_ERROR, messagestr(e)) else: # 请求了不存在的工具 await session.send_tool_call_error(call_id, codeNOT_FOUND, messagef未知的工具{tool_name}) if __name__ __main__: asyncio.run(main())实操心得在定义工具inputSchema时尽可能详细地描述每个参数。这就像是给大模型的“说明书”描述越清晰模型构造正确参数的几率就越高。enum字段对于状态这类有限选项的参数特别有用它能有效约束模型的输出范围。3.4 搭建简易OAuth授权服务器为了让认证流程跑通我们需要一个简单的OAuth服务器。创建oauth_server.py。# oauth_server.py from flask import Flask, request, redirect, jsonify import jwt import time from datetime import datetime, timedelta app Flask(__name__) # 这是一个非常简化的模拟生产环境请使用成熟的OAuth库如authlib和安全的密钥存储 SECRET_KEY your-super-secret-jwt-key-change-this-in-production # 务必在生产环境中更换 CLIENT_ID mcp-todo-client CLIENT_SECRET simulated-client-secret # 模拟用简化流程中未实际校验 REDIRECT_URI http://localhost:5000/oauth/callback # 假设客户端回调地址 # 模拟一个用户数据库和授权码存储 authorization_codes {} app.route(/oauth/authorize, methods[GET]) def authorize(): OAuth授权端点 - 模拟用户登录和授权 client_id request.args.get(client_id) redirect_uri request.args.get(redirect_uri, REDIRECT_URI) state request.args.get(state, ) if client_id ! CLIENT_ID: return jsonify(errorinvalid_client), 400 # 在实际应用中这里会有一个登录页面和授权确认页面。 # 我们直接模拟用户已同意授权。 import secrets auth_code secrets.token_urlsafe(16) authorization_codes[auth_code] { client_id: client_id, expires_at: time.time() 300 # 5分钟有效期 } # 重定向回客户端附带授权码 return redirect(f{redirect_uri}?code{auth_code}state{state}) app.route(/oauth/token, methods[POST]) def token(): OAuth令牌端点 - 用授权码交换访问令牌 grant_type request.form.get(grant_type) code request.form.get(code) client_id request.form.get(client_id) client_secret request.form.get(client_secret) # 简化流程实际应校验 if grant_type ! authorization_code: return jsonify(errorunsupported_grant_type), 400 if client_id ! CLIENT_ID: return jsonify(errorinvalid_client), 400 # 检查授权码是否有效且未过期 auth_info authorization_codes.get(code) if not auth_info or auth_info[expires_at] time.time(): return jsonify(errorinvalid_grant), 400 # 生成JWT访问令牌 payload { iss: mcp-todo-oauth-server, sub: simulated-user-123, aud: CLIENT_ID, exp: datetime.utcnow() timedelta(hours1), # 1小时过期 iat: datetime.utcnow(), scope: read_todos write_todos, } access_token jwt.encode(payload, SECRET_KEY, algorithmHS256) # 清理已使用的授权码 authorization_codes.pop(code, None) return jsonify({ access_token: access_token, token_type: bearer, expires_in: 3600, scope: read_todos write_todos }) app.route(/oauth/introspect, methods[POST]) def introspect(): 令牌验证端点 - MCP服务器用来验证客户端传来的令牌 # 通常客户端会以表单形式或Bearer Token头传递token token request.form.get(token) or request.headers.get(Authorization, ).replace(Bearer , ) if not token: return jsonify(activeFalse), 200 try: # 验证JWT签名和有效期 decoded jwt.decode(token, SECRET_KEY, algorithms[HS256], audienceCLIENT_ID) return jsonify(activeTrue, client_idCLIENT_ID, scopedecoded.get(scope), subdecoded.get(sub)) except jwt.ExpiredSignatureError: return jsonify(activeFalse, errortoken_expired), 200 except jwt.InvalidTokenError: return jsonify(activeFalse, errorinvalid_token), 200 if __name__ __main__: app.run(port5001, debugTrue)这个服务器模拟了OAuth 2.0授权码流程的核心部分。它提供了授权端点/oauth/authorize、令牌端点/oauth/token和令牌自省端点/oauth/introspect。MCP客户端会使用前两个端点获取令牌而我们的MCP服务器在收到请求时会调用自省端点来验证令牌的有效性。重要安全提示这个示例为了演示极度简化绝对不可直接用于生产。生产环境必须使用强随机密钥并通过环境变量或密钥管理服务注入。实现完整的用户登录、会话管理和授权确认流程。使用HTTPS保护所有端点。妥善存储和校验client_secret。考虑使用像authlib这样的专业库来处理OAuth流程的复杂性。4. 生产级部署容器化与配置管理代码写好了如何让它稳定、安全地运行起来容器化是当前的最佳答案。我们将使用Docker来打包应用并配置环境变量来管理敏感信息。4.1 编写Dockerfile在项目根目录创建Dockerfile。# Dockerfile # 使用官方Python轻量级镜像 FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 设置环境变量防止Python输出被缓冲使日志能实时看到 ENV PYTHONUNBUFFERED1 # 安装系统依赖如果需要例如sqlite3通常已包含 # RUN apt-get update apt-get install -y --no-install-recommends gcc rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 声明容器运行时暴露的端口OAuth服务器用 EXPOSE 5001 # 定义容器启动命令 # 这里我们启动两个进程MCP服务器和OAuth服务器 # 使用一个shell脚本或进程管理器如supervisor来管理多个进程更佳此处为简化使用后台运行 CMD sh -c python oauth_server.py python server.py这个Dockerfile做了几件事基于轻量级Python镜像设置环境安装依赖复制代码并定义启动命令。我们让OAuth服务器在后台运行同时前台运行MCP主服务器。在生产中更推荐使用supervisord或gunicorn对于Web服务等进程管理工具。4.2 创建Docker Compose配置可选但推荐对于多服务应用docker-compose.yml能简化管理。虽然我们目前只有一个容器但良好的习惯从开始培养。创建docker-compose.yml。# docker-compose.yml version: 3.8 services: mcp-todo-server: build: . container_name: mcp-todo-server ports: - 5001:5001 # 将容器的OAuth服务端口映射到主机 environment: - JWT_SECRET_KEY${JWT_SECRET_KEY:-your-dev-secret-key} # 从环境变量读取密钥 - DATABASE_PATH/data/todos.db volumes: - ./data:/data # 将数据库文件持久化到主机避免容器重启数据丢失 stdin_open: true # 保持标准输入打开对某些MCP客户端是必需的 tty: true # 分配一个伪终端 restart: unless-stopped # 设置自动重启策略这个配置允许我们通过一个命令启动所有服务并方便地管理端口映射、环境变量和数据持久化。注意volumes部分它将容器内的/data目录挂载到主机的./data目录这样SQLite数据库文件就不会随着容器销毁而丢失。4.3 环境配置与启动生成安全的JWT密钥在终端执行以下命令生成一个强密钥。python -c import secrets; print(secrets.token_urlsafe(32))复制输出的字符串。创建环境变量文件在项目根目录创建.env文件确保将其加入.gitignore。# .env JWT_SECRET_KEY你刚才生成的密钥构建并运行容器# 使用Docker Compose推荐 docker-compose up --build # 或者直接使用Docker命令 # docker build -t mcp-todo-server . # docker run -p 5001:5001 -v $(pwd)/data:/data --env-file .env mcp-todo-server现在你的MCP服务器和OAuth服务器应该都在容器中运行起来了。MCP服务器通过标准输入输出等待连接OAuth服务器在http://localhost:5001上监听。5. 客户端连接与实战测试服务器部署好了我们如何让AI客户端如Cursor使用它呢这需要一个MCP客户端配置文件。5.1 配置Cursor连接我们的服务器在Cursor中MCP服务器配置通常放在一个全局或项目特定的配置文件中。对于Cursor你可以在其设置中查找MCP配置项或者创建一个配置文件。创建一个名为cursor-mcp-config.json的文件位置可能因Cursor版本而异请参考其文档{ mcpServers: { todo-manager: { command: docker, args: [ run, -i, --rm, -v, /path/to/your/project/data:/data, -e, JWT_SECRET_KEY你的密钥, -p, 5001:5001, mcp-todo-server ], env: { PYTHONUNBUFFERED: 1 }, auth: { type: oauth, oauthConfig: { authorization_url: http://localhost:5001/oauth/authorize, token_url: http://localhost:5001/oauth/token, client_id: mcp-todo-client, scopes: [read_todos, write_todos] } } } } }注意上面的配置直接使用docker run命令。更稳定的做法是使用docker-compose up启动服务然后配置Cursor连接到一个已经运行在后台的服务器进程。或者如果服务器已部署在远程主机command可以是一个SSH命令或直接指向一个网络服务。配置完成后重启Cursor。理论上Cursor会识别到这个配置并在你需要使用Todo工具时自动触发OAuth授权流程弹出浏览器让你“登录”我们的模拟服务器获取令牌然后就可以调用工具了。5.2 在客户端中进行测试你可以在Cursor的聊天框中尝试以下指令“请用我的待办事项工具创建一个标题为‘编写项目报告’的事项。”“列出我所有未完成的待办事项。”“把ID为1的待办事项状态更新为‘in_progress’。”“删除ID为2的待办事项。”如果一切配置正确Cursor会理解你的指令调用对应的MCP工具并返回操作结果。5.3 常见问题与排查技巧在实际操作中你几乎一定会遇到一些问题。下面是一个快速排查指南问题现象可能原因排查步骤Cursor完全无法识别工具MCP服务器配置错误或未加载1. 检查Cursor的MCP配置路径是否正确。2. 查看Cursor日志如果有中关于MCP服务器初始化的信息。3. 尝试在终端手动运行Docker命令看服务器是否能正常启动并打印日志。授权流程卡住或失败OAuth服务器未运行或配置不匹配1. 确保oauth_server.py正在运行端口5001。2. 检查cursor-mcp-config.json中的authorization_url和token_url是否与OAuth服务器地址一致。3. 打开浏览器开发者工具的网络标签查看授权重定向过程中是否有错误。工具调用返回权限错误令牌无效或验证失败1. 检查MCP服务器日志看它在调用/oauth/introspect端点时是否收到错误。2. 确认.env文件中的JWT_SECRET_KEY与OAuth服务器使用的SECRET_KEY完全一致。3. 令牌可能已过期尝试让Cursor重新授权。数据库操作失败文件权限或路径问题1. 检查Docker容器的数据卷挂载 (-v参数) 是否正确主机目录是否存在。2. 进入容器内部 (docker exec -it container_id bash)检查/data/todos.db文件是否可读写。服务器启动后立即退出依赖缺失或Python脚本错误1. 查看Docker构建日志确认requirements.txt中的包是否成功安装。2. 检查server.py和oauth_server.py是否有语法错误。可以在本地用python server.py先测试。3. 确认Dockerfile中的CMD命令能正确启动两个进程。一个关键的调试技巧在开发阶段可以先绕过OAuth快速测试工具逻辑。修改server.py中的StdioServerParameters将auth参数设为None。这样客户端连接时就不需要认证了。当然在测试完成后务必改回来。6. 进阶优化与扩展方向一个能跑通的服务器只是起点。要让其真正具备“生产级”的韧性还需要考虑以下方面6.1 增强安全性使用真正的OAuth提供商将oauth_server.py替换为对接到公司内部的单点登录SSO系统或使用专业的云身份提供商如Auth0, Okta, Keycloak。它们能处理密码哈希、多因素认证、令牌刷新等复杂问题。HTTPS everywhere生产环境必须使用HTTPS。为你的OAuth服务器配置TLS证书可以使用Let‘s Encrypt免费获取。确保MCP客户端配置中的URL都是https://。细粒度权限控制目前的scope比较粗。可以设计更细的权限如todos:read,todos:write,todos:delete并在工具处理函数中检查当前令牌的scope是否包含所需权限。输入验证与清理虽然MCP客户端大模型会尝试遵循schema但仍需在服务器端对输入进行严格的验证和清理防止SQL注入或其他攻击。我们示例中使用参数化查询已经避免了SQL注入但对于其他业务逻辑验证仍需加强。6.2 提升可靠性进程管理使用supervisord或systemd在容器内管理多个进程确保一个进程崩溃后能被自动重启。健康检查在Dockerfile或Compose文件中添加健康检查指令让编排工具如Kubernetes能感知服务状态。# docker-compose.yml 补充 healthcheck: test: [CMD, curl, -f, http://localhost:5001/oauth/introspect] interval: 30s timeout: 10s retries: 3 start_period: 40s日志与监控将应用日志结构化例如使用JSON格式并输出到标准输出stdout方便Docker收集。集成像Prometheus这样的监控系统来收集指标请求数、延迟、错误率。数据库升级当待办事项数据量增大或需要高可用时将SQLite迁移到PostgreSQL或MySQL。只需修改database.py中的连接逻辑即可上层工具函数接口可以保持不变。6.3 功能扩展添加更多工具这个模式可以无限扩展。例如添加search_todos全文搜索、set_todo_due_date设置截止日期、add_todo_comment添加评论等工具。支持“资源”ResourcesMCP协议除了“工具”还有“资源”的概念。你可以将“今天的待办列表”或“某个特定项目下的所有事项”定义为一个资源URL客户端可以读取其内容。这对于提供静态或动态生成的数据块很有用。集成外部系统这才是MCP的威力所在。你可以开发连接内部GitLab的服务器创建MR、查看CI状态、连接财务系统的服务器查询报表、连接物联网平台的服务器控制设备。只需要遵循同样的模式定义工具、实现业务逻辑、处理认证。通过这15分钟的高强度实战我们不仅构建了一个可工作的MCP服务器更搭建了一个具备生产级部署雏形的安全、可扩展的框架。核心在于理解MCP作为桥梁的角色以及如何将你的内部能力通过标准化协议安全地暴露给AI。接下来你可以将这个Todo管理器替换成任何你需要的业务逻辑开启AI赋能工作流的新篇章。 15分钟实战：基于Python与OAuth构建生产级MCP服务器
1. 项目概述为什么MCP服务器开发是当前的热点最近在开发者社区里MCPModel Context Protocol这个词的热度越来越高。如果你关注AI应用开发尤其是大模型与工具集成的领域那么你很可能已经听说过它。简单来说MCP是一个标准协议它定义了大模型如GPT-4、Claude、DeepSeek等如何安全、结构化地与外部工具、数据源和API进行交互。你可以把它想象成大模型的“USB接口”或“插件系统”——它让模型不再是一个封闭的黑盒而是能连接到你自己的数据库、内部API、文件系统甚至硬件设备从而执行更复杂、更定制化的任务。那么开发一个MCP服务器具体是做什么呢想象一个场景你希望你的AI助手能帮你查询公司内部的销售数据、自动创建Jira工单或者控制你家里的智能灯光。这些能力模型本身并不具备它们存在于你的私有环境里。MCP服务器的角色就是作为这些私有能力和大模型之间的“翻译官”和“安全网关”。它遵循MCP协议将你的内部工具包装成模型可以理解和调用的标准化“工具”同时处理认证、授权、数据格式转换和错误处理。这使得像Cursor、Claude Desktop、Windmill这类支持MCP的AI客户端能够无缝、安全地使用你提供的专属能力。这个项目标题“15分钟从零到生产级部署”之所以吸引人是因为它戳中了开发者的痛点协议听起来很酷但上手门槛高吗部署复杂吗能否快速看到效果本文将带你进行一次高速实战我们将使用Python——这门在AI和自动化领域最流行的语言从零开始构建一个具备实用价值的MCP服务器并完成包括OAuth安全认证在内的、可直接用于生产环境的部署。无论你是想为团队内部打造AI增效工具还是探索AI应用的新形态这都将是一次极具价值的实践。2. 核心设计构建一个生产就绪的MCP服务器架构在动手写代码之前我们先花点时间厘清思路。一个“生产级”的MCP服务器绝不仅仅是一个能跑通的Demo。它需要兼顾功能、安全、可靠性和可维护性。我们的设计目标是在15分钟内搭建起一个具备这些特质雏形的系统框架。2.1 协议理解与工具定义MCP协议的核心抽象是“工具”Tools和“资源”Resources。对于大多数应用场景“工具”是我们关注的重点。一个工具由名称、描述、输入参数schema和具体的执行函数组成。当AI客户端如Cursor需要完成某项任务时它会根据工具描述构造符合schema的参数并通过JSON-RPC over STDIO标准输入输出或HTTP调用我们的服务器。我们的第一个服务器将实现一个经典且实用的功能一个待办事项Todo List管理器。为什么选这个因为它逻辑简单但涵盖了CRUD增删改查核心操作非常适合演示。我们将定义四个工具list_todos: 列出所有待办事项。create_todo: 创建新的待办事项。update_todo_status: 更新某个事项的状态如完成/未完成。delete_todo: 删除一个事项。这些工具将操作一个后端数据存储。为了快速达到“生产级”我们不能用内存变量那样数据无法持久化。我们将选择SQLite数据库它无需额外服务单文件存储非常适合轻量级应用和快速原型。同时我们会为数据库操作设计一个简单的数据访问层以便未来轻松替换为PostgreSQL或MySQL。2.2 安全与认证设计OAuth集成这是“生产级”与“玩具级”最关键的区分点。一个暴露在外的、无需认证的服务器是极其危险的。MCP协议支持在服务器声明中要求客户端进行认证。我们将集成OAuth 2.0授权码流程这是业界标准的API认证方式。流程是这样的我们的MCP服务器在初始化时告诉客户端“调用我的工具需要OAuth认证”。当用户首次在客户端如Cursor中尝试使用我们的工具时客户端会弹出浏览器引导用户跳转到我们指定的授权页面可以是我们自建的也可以是第三方如GitHub、Google的。用户登录并授权后授权服务器会通过回调地址返回一个授权码code。客户端用这个code向授权服务器交换访问令牌access_token。客户端在后续所有调用我们MCP服务器的请求中都会携带这个access_token。我们的服务器收到请求后需要验证这个token的有效性例如向授权服务器的introspection端点发起验证。在本实战中为了在15分钟内完成闭环我们将模拟一个简化但符合流程的OAuth提供方。我们会创建一个简单的Flask应用来扮演授权服务器颁发和验证JWTJSON Web Token令牌。在实际生产中你可以将这个提供方替换为Auth0、Okta或你公司的统一认证中心。2.3 技术栈选型MCP SDK: 我们将使用官方提供的mcpPython库。这是最直接、最兼容的选择它封装了协议通信的底层细节让我们专注于工具逻辑。Web框架用于OAuth服务: 选择Flask。它轻量、灵活适合快速搭建小型HTTP服务。我们的OAuth授权服务器和令牌验证端点都将用它构建。数据库:SQLitesqlite3Python内置。无需安装开箱即用。令牌管理:PyJWT库用于生成和验证JWT格式的访问令牌。进程通信: MCP默认使用STDIO标准输入输出这对于本地集成是最简单可靠的。我们的服务器将作为一个独立的命令行程序启动通过标准流与客户端对话。部署与打包: 我们将使用Docker进行容器化。这是生产部署的黄金标准它能确保环境一致性并方便地在任何支持Docker的服务器上运行。这个选型平衡了开发速度、学习成本和生产可靠性。所有库都是主流且文档丰富遇到问题容易找到解决方案。3. 逐步实现从零编写服务器核心代码现在我们开始动手编码。请确保你的开发环境已安装Python 3.9或更高版本。3.1 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境这是管理Python项目依赖的最佳实践。mkdir mcp-todo-server cd mcp-todo-server python -m venv venv # 在Windows上使用venv\Scripts\activate source venv/bin/activate接下来创建requirements.txt文件列出我们需要的所有依赖。# requirements.txt mcp1.0.0 flask3.0.0 pyjwt2.8.0然后安装它们pip install -r requirements.txt3.2 构建数据层数据库模型与操作在项目根目录下创建database.py文件。这里我们将定义Todo的数据模型和基本的数据库操作类。# database.py import sqlite3 import json from typing import List, Optional, Dict, Any from dataclasses import dataclass, asdict from datetime import datetime dataclass class TodoItem: id: Optional[int] None title: str description: str status: str pending # pending, in_progress, completed created_at: Optional[str] None updated_at: Optional[str] None class TodoDatabase: def __init__(self, db_path: str todos.db): self.db_path db_path self._init_db() def _init_db(self): 初始化数据库表 with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, description TEXT, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建一个触发器在更新时自动更新updated_at字段 conn.execute( CREATE TRIGGER IF NOT EXISTS update_todo_timestamp AFTER UPDATE ON todos BEGIN UPDATE todos SET updated_at CURRENT_TIMESTAMP WHERE id NEW.id; END; ) def get_connection(self): 获取数据库连接并设置返回字典格式的行 conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row # 使查询结果可按列名访问 return conn def create_todo(self, title: str, description: str ) - TodoItem: 创建新的待办事项 with self.get_connection() as conn: cursor conn.cursor() cursor.execute( INSERT INTO todos (title, description) VALUES (?, ?), (title, description) ) todo_id cursor.lastrowid # 获取刚创建的项目 cursor.execute(SELECT * FROM todos WHERE id ?, (todo_id,)) row cursor.fetchone() conn.commit() return self._row_to_todo(row) def list_todos(self, status_filter: Optional[str] None) - List[TodoItem]: 列出所有待办事项可选项按状态过滤 with self.get_connection() as conn: query SELECT * FROM todos params () if status_filter: query WHERE status ? params (status_filter,) query ORDER BY created_at DESC cursor conn.execute(query, params) return [self._row_to_todo(row) for row in cursor.fetchall()] def update_todo_status(self, todo_id: int, new_status: str) - Optional[TodoItem]: 更新待办事项状态 allowed_statuses [pending, in_progress, completed] if new_status not in allowed_statuses: raise ValueError(f状态必须是其中之一{allowed_statuses}) with self.get_connection() as conn: cursor conn.cursor() cursor.execute( UPDATE todos SET status ? WHERE id ?, (new_status, todo_id) ) if cursor.rowcount 0: return None # 没有找到对应ID的项目 cursor.execute(SELECT * FROM todos WHERE id ?, (todo_id,)) row cursor.fetchone() conn.commit() return self._row_to_todo(row) def delete_todo(self, todo_id: int) - bool: 删除一个待办事项 with self.get_connection() as conn: cursor conn.cursor() cursor.execute(DELETE FROM todos WHERE id ?, (todo_id,)) conn.commit() return cursor.rowcount 0 def _row_to_todo(self, row: sqlite3.Row) - TodoItem: 将数据库行转换为TodoItem对象 if row is None: return None return TodoItem( idrow[id], titlerow[title], descriptionrow[description], statusrow[status], created_atrow[created_at], updated_atrow[updated_at] )注意我们使用了sqlite3.Row工厂它允许我们通过列名如row[‘id’]来访问数据这比使用数字索引更清晰、更不易出错。同时我们利用SQLite的触发器自动管理updated_at时间戳这是一个保持数据一致性的好习惯。3.3 实现MCP服务器主逻辑现在创建核心的MCP服务器文件server.py。我们将使用mcpSDK来声明工具并处理请求。# server.py import asyncio import sys import json from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 导入我们刚才写的数据层 from database import TodoDatabase # 初始化数据库 db TodoDatabase() async def handle_list_todos(arguments: dict) - str: 处理列出待办事项的请求 status_filter arguments.get(status_filter) todos db.list_todos(status_filterstatus_filter) if not todos: return 当前没有待办事项。 result_lines [] for todo in todos: result_lines.append(f- [{todo.id}] {todo.title} ({todo.status})) if todo.description: result_lines.append(f 描述{todo.description}) result_lines.append(f 创建于{todo.created_at}) return \n.join(result_lines) async def handle_create_todo(arguments: dict) - str: 处理创建待办事项的请求 title arguments.get(title) if not title: return 错误必须提供待办事项的标题。 description arguments.get(description, ) todo db.create_todo(title, description) return f已成功创建待办事项 [#{todo.id}]{todo.title} async def handle_update_status(arguments: dict) - str: 处理更新状态的请求 todo_id arguments.get(todo_id) new_status arguments.get(new_status) if not todo_id or not new_status: return 错误必须提供待办事项ID和新状态。 try: todo_id_int int(todo_id) except ValueError: return 错误待办事项ID必须是数字。 updated db.update_todo_status(todo_id_int, new_status) if updated: return f已成功将待办事项 [#{updated.id}] 的状态更新为 {updated.status}。 else: return f错误未找到ID为 {todo_id} 的待办事项。 async def handle_delete_todo(arguments: dict) - str: 处理删除待办事项的请求 todo_id arguments.get(todo_id) if not todo_id: return 错误必须提供待办事项ID。 try: todo_id_int int(todo_id) except ValueError: return 错误待办事项ID必须是数字。 success db.delete_todo(todo_id_int) if success: return f已成功删除待办事项 [#{todo_id}]。 else: return f错误未找到ID为 {todo_id} 的待办事项删除失败。 async def main(): # 定义服务器提供的工具列表 # 每个工具需要名称、描述和输入参数的模式schema tools [ { name: list_todos, description: 列出所有的待办事项。可以选择按状态过滤pending, in_progress, completed。, inputSchema: { type: object, properties: { status_filter: { type: string, description: 按状态过滤待办事项。可选值pending, in_progress, completed。, enum: [pending, in_progress, completed] } } } }, { name: create_todo, description: 创建一个新的待办事项。, inputSchema: { type: object, properties: { title: { type: string, description: 待办事项的标题必填。 }, description: { type: string, description: 待办事项的详细描述可选。 } }, required: [title] } }, { name: update_todo_status, description: 更新一个现有待办事项的状态。, inputSchema: { type: object, properties: { todo_id: { type: string, description: 要更新的待办事项的ID数字。 }, new_status: { type: string, description: 新的状态。, enum: [pending, in_progress, completed] } }, required: [todo_id, new_status] } }, { name: delete_todo, description: 删除一个待办事项。, inputSchema: { type: object, properties: { todo_id: { type: string, description: 要删除的待办事项的ID数字。 } }, required: [todo_id] } } ] # 创建服务器参数声明我们需要的OAuth认证 server_params StdioServerParameters( commandsys.executable, # 使用当前Python解释器 args[__file__], # 运行这个脚本本身 # 声明服务器要求OAuth 2.0授权码流程认证 # 这里我们使用一个模拟的授权服务器地址 auth(oauth, { authorization_url: http://localhost:5001/oauth/authorize, token_url: http://localhost:5001/oauth/token, scopes: [read_todos, write_todos], client_id: mcp-todo-client }) ) # 使用stdio_client管理服务器进程和通信 async with stdio_client(server_params) as (read_stream, write_stream): session ClientSession(read_stream, write_stream) # 初始化会话向客户端宣告我们的工具 await session.initialize(toolstools) # 进入主循环监听并处理客户端请求 async for message in session.listen(): if message.method tools/call: # 提取工具调用请求的详细信息 call_id message.params[callId] tool_name message.params[name] arguments message.params.get(arguments, {}) # 根据工具名称路由到对应的处理函数 handler_map { list_todos: handle_list_todos, create_todo: handle_create_todo, update_todo_status: handle_update_status, delete_todo: handle_delete_todo, } handler handler_map.get(tool_name) if handler: try: # 异步执行工具逻辑 result_text await handler(arguments) # 将成功结果发送回客户端 await session.send_tool_call_result(call_id, content[{ type: text, text: result_text }]) except Exception as e: # 如果执行出错返回错误信息 await session.send_tool_call_error(call_id, codeINTERNAL_ERROR, messagestr(e)) else: # 请求了不存在的工具 await session.send_tool_call_error(call_id, codeNOT_FOUND, messagef未知的工具{tool_name}) if __name__ __main__: asyncio.run(main())实操心得在定义工具inputSchema时尽可能详细地描述每个参数。这就像是给大模型的“说明书”描述越清晰模型构造正确参数的几率就越高。enum字段对于状态这类有限选项的参数特别有用它能有效约束模型的输出范围。3.4 搭建简易OAuth授权服务器为了让认证流程跑通我们需要一个简单的OAuth服务器。创建oauth_server.py。# oauth_server.py from flask import Flask, request, redirect, jsonify import jwt import time from datetime import datetime, timedelta app Flask(__name__) # 这是一个非常简化的模拟生产环境请使用成熟的OAuth库如authlib和安全的密钥存储 SECRET_KEY your-super-secret-jwt-key-change-this-in-production # 务必在生产环境中更换 CLIENT_ID mcp-todo-client CLIENT_SECRET simulated-client-secret # 模拟用简化流程中未实际校验 REDIRECT_URI http://localhost:5000/oauth/callback # 假设客户端回调地址 # 模拟一个用户数据库和授权码存储 authorization_codes {} app.route(/oauth/authorize, methods[GET]) def authorize(): OAuth授权端点 - 模拟用户登录和授权 client_id request.args.get(client_id) redirect_uri request.args.get(redirect_uri, REDIRECT_URI) state request.args.get(state, ) if client_id ! CLIENT_ID: return jsonify(errorinvalid_client), 400 # 在实际应用中这里会有一个登录页面和授权确认页面。 # 我们直接模拟用户已同意授权。 import secrets auth_code secrets.token_urlsafe(16) authorization_codes[auth_code] { client_id: client_id, expires_at: time.time() 300 # 5分钟有效期 } # 重定向回客户端附带授权码 return redirect(f{redirect_uri}?code{auth_code}state{state}) app.route(/oauth/token, methods[POST]) def token(): OAuth令牌端点 - 用授权码交换访问令牌 grant_type request.form.get(grant_type) code request.form.get(code) client_id request.form.get(client_id) client_secret request.form.get(client_secret) # 简化流程实际应校验 if grant_type ! authorization_code: return jsonify(errorunsupported_grant_type), 400 if client_id ! CLIENT_ID: return jsonify(errorinvalid_client), 400 # 检查授权码是否有效且未过期 auth_info authorization_codes.get(code) if not auth_info or auth_info[expires_at] time.time(): return jsonify(errorinvalid_grant), 400 # 生成JWT访问令牌 payload { iss: mcp-todo-oauth-server, sub: simulated-user-123, aud: CLIENT_ID, exp: datetime.utcnow() timedelta(hours1), # 1小时过期 iat: datetime.utcnow(), scope: read_todos write_todos, } access_token jwt.encode(payload, SECRET_KEY, algorithmHS256) # 清理已使用的授权码 authorization_codes.pop(code, None) return jsonify({ access_token: access_token, token_type: bearer, expires_in: 3600, scope: read_todos write_todos }) app.route(/oauth/introspect, methods[POST]) def introspect(): 令牌验证端点 - MCP服务器用来验证客户端传来的令牌 # 通常客户端会以表单形式或Bearer Token头传递token token request.form.get(token) or request.headers.get(Authorization, ).replace(Bearer , ) if not token: return jsonify(activeFalse), 200 try: # 验证JWT签名和有效期 decoded jwt.decode(token, SECRET_KEY, algorithms[HS256], audienceCLIENT_ID) return jsonify(activeTrue, client_idCLIENT_ID, scopedecoded.get(scope), subdecoded.get(sub)) except jwt.ExpiredSignatureError: return jsonify(activeFalse, errortoken_expired), 200 except jwt.InvalidTokenError: return jsonify(activeFalse, errorinvalid_token), 200 if __name__ __main__: app.run(port5001, debugTrue)这个服务器模拟了OAuth 2.0授权码流程的核心部分。它提供了授权端点/oauth/authorize、令牌端点/oauth/token和令牌自省端点/oauth/introspect。MCP客户端会使用前两个端点获取令牌而我们的MCP服务器在收到请求时会调用自省端点来验证令牌的有效性。重要安全提示这个示例为了演示极度简化绝对不可直接用于生产。生产环境必须使用强随机密钥并通过环境变量或密钥管理服务注入。实现完整的用户登录、会话管理和授权确认流程。使用HTTPS保护所有端点。妥善存储和校验client_secret。考虑使用像authlib这样的专业库来处理OAuth流程的复杂性。4. 生产级部署容器化与配置管理代码写好了如何让它稳定、安全地运行起来容器化是当前的最佳答案。我们将使用Docker来打包应用并配置环境变量来管理敏感信息。4.1 编写Dockerfile在项目根目录创建Dockerfile。# Dockerfile # 使用官方Python轻量级镜像 FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 设置环境变量防止Python输出被缓冲使日志能实时看到 ENV PYTHONUNBUFFERED1 # 安装系统依赖如果需要例如sqlite3通常已包含 # RUN apt-get update apt-get install -y --no-install-recommends gcc rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 声明容器运行时暴露的端口OAuth服务器用 EXPOSE 5001 # 定义容器启动命令 # 这里我们启动两个进程MCP服务器和OAuth服务器 # 使用一个shell脚本或进程管理器如supervisor来管理多个进程更佳此处为简化使用后台运行 CMD sh -c python oauth_server.py python server.py这个Dockerfile做了几件事基于轻量级Python镜像设置环境安装依赖复制代码并定义启动命令。我们让OAuth服务器在后台运行同时前台运行MCP主服务器。在生产中更推荐使用supervisord或gunicorn对于Web服务等进程管理工具。4.2 创建Docker Compose配置可选但推荐对于多服务应用docker-compose.yml能简化管理。虽然我们目前只有一个容器但良好的习惯从开始培养。创建docker-compose.yml。# docker-compose.yml version: 3.8 services: mcp-todo-server: build: . container_name: mcp-todo-server ports: - 5001:5001 # 将容器的OAuth服务端口映射到主机 environment: - JWT_SECRET_KEY${JWT_SECRET_KEY:-your-dev-secret-key} # 从环境变量读取密钥 - DATABASE_PATH/data/todos.db volumes: - ./data:/data # 将数据库文件持久化到主机避免容器重启数据丢失 stdin_open: true # 保持标准输入打开对某些MCP客户端是必需的 tty: true # 分配一个伪终端 restart: unless-stopped # 设置自动重启策略这个配置允许我们通过一个命令启动所有服务并方便地管理端口映射、环境变量和数据持久化。注意volumes部分它将容器内的/data目录挂载到主机的./data目录这样SQLite数据库文件就不会随着容器销毁而丢失。4.3 环境配置与启动生成安全的JWT密钥在终端执行以下命令生成一个强密钥。python -c import secrets; print(secrets.token_urlsafe(32))复制输出的字符串。创建环境变量文件在项目根目录创建.env文件确保将其加入.gitignore。# .env JWT_SECRET_KEY你刚才生成的密钥构建并运行容器# 使用Docker Compose推荐 docker-compose up --build # 或者直接使用Docker命令 # docker build -t mcp-todo-server . # docker run -p 5001:5001 -v $(pwd)/data:/data --env-file .env mcp-todo-server现在你的MCP服务器和OAuth服务器应该都在容器中运行起来了。MCP服务器通过标准输入输出等待连接OAuth服务器在http://localhost:5001上监听。5. 客户端连接与实战测试服务器部署好了我们如何让AI客户端如Cursor使用它呢这需要一个MCP客户端配置文件。5.1 配置Cursor连接我们的服务器在Cursor中MCP服务器配置通常放在一个全局或项目特定的配置文件中。对于Cursor你可以在其设置中查找MCP配置项或者创建一个配置文件。创建一个名为cursor-mcp-config.json的文件位置可能因Cursor版本而异请参考其文档{ mcpServers: { todo-manager: { command: docker, args: [ run, -i, --rm, -v, /path/to/your/project/data:/data, -e, JWT_SECRET_KEY你的密钥, -p, 5001:5001, mcp-todo-server ], env: { PYTHONUNBUFFERED: 1 }, auth: { type: oauth, oauthConfig: { authorization_url: http://localhost:5001/oauth/authorize, token_url: http://localhost:5001/oauth/token, client_id: mcp-todo-client, scopes: [read_todos, write_todos] } } } } }注意上面的配置直接使用docker run命令。更稳定的做法是使用docker-compose up启动服务然后配置Cursor连接到一个已经运行在后台的服务器进程。或者如果服务器已部署在远程主机command可以是一个SSH命令或直接指向一个网络服务。配置完成后重启Cursor。理论上Cursor会识别到这个配置并在你需要使用Todo工具时自动触发OAuth授权流程弹出浏览器让你“登录”我们的模拟服务器获取令牌然后就可以调用工具了。5.2 在客户端中进行测试你可以在Cursor的聊天框中尝试以下指令“请用我的待办事项工具创建一个标题为‘编写项目报告’的事项。”“列出我所有未完成的待办事项。”“把ID为1的待办事项状态更新为‘in_progress’。”“删除ID为2的待办事项。”如果一切配置正确Cursor会理解你的指令调用对应的MCP工具并返回操作结果。5.3 常见问题与排查技巧在实际操作中你几乎一定会遇到一些问题。下面是一个快速排查指南问题现象可能原因排查步骤Cursor完全无法识别工具MCP服务器配置错误或未加载1. 检查Cursor的MCP配置路径是否正确。2. 查看Cursor日志如果有中关于MCP服务器初始化的信息。3. 尝试在终端手动运行Docker命令看服务器是否能正常启动并打印日志。授权流程卡住或失败OAuth服务器未运行或配置不匹配1. 确保oauth_server.py正在运行端口5001。2. 检查cursor-mcp-config.json中的authorization_url和token_url是否与OAuth服务器地址一致。3. 打开浏览器开发者工具的网络标签查看授权重定向过程中是否有错误。工具调用返回权限错误令牌无效或验证失败1. 检查MCP服务器日志看它在调用/oauth/introspect端点时是否收到错误。2. 确认.env文件中的JWT_SECRET_KEY与OAuth服务器使用的SECRET_KEY完全一致。3. 令牌可能已过期尝试让Cursor重新授权。数据库操作失败文件权限或路径问题1. 检查Docker容器的数据卷挂载 (-v参数) 是否正确主机目录是否存在。2. 进入容器内部 (docker exec -it container_id bash)检查/data/todos.db文件是否可读写。服务器启动后立即退出依赖缺失或Python脚本错误1. 查看Docker构建日志确认requirements.txt中的包是否成功安装。2. 检查server.py和oauth_server.py是否有语法错误。可以在本地用python server.py先测试。3. 确认Dockerfile中的CMD命令能正确启动两个进程。一个关键的调试技巧在开发阶段可以先绕过OAuth快速测试工具逻辑。修改server.py中的StdioServerParameters将auth参数设为None。这样客户端连接时就不需要认证了。当然在测试完成后务必改回来。6. 进阶优化与扩展方向一个能跑通的服务器只是起点。要让其真正具备“生产级”的韧性还需要考虑以下方面6.1 增强安全性使用真正的OAuth提供商将oauth_server.py替换为对接到公司内部的单点登录SSO系统或使用专业的云身份提供商如Auth0, Okta, Keycloak。它们能处理密码哈希、多因素认证、令牌刷新等复杂问题。HTTPS everywhere生产环境必须使用HTTPS。为你的OAuth服务器配置TLS证书可以使用Let‘s Encrypt免费获取。确保MCP客户端配置中的URL都是https://。细粒度权限控制目前的scope比较粗。可以设计更细的权限如todos:read,todos:write,todos:delete并在工具处理函数中检查当前令牌的scope是否包含所需权限。输入验证与清理虽然MCP客户端大模型会尝试遵循schema但仍需在服务器端对输入进行严格的验证和清理防止SQL注入或其他攻击。我们示例中使用参数化查询已经避免了SQL注入但对于其他业务逻辑验证仍需加强。6.2 提升可靠性进程管理使用supervisord或systemd在容器内管理多个进程确保一个进程崩溃后能被自动重启。健康检查在Dockerfile或Compose文件中添加健康检查指令让编排工具如Kubernetes能感知服务状态。# docker-compose.yml 补充 healthcheck: test: [CMD, curl, -f, http://localhost:5001/oauth/introspect] interval: 30s timeout: 10s retries: 3 start_period: 40s日志与监控将应用日志结构化例如使用JSON格式并输出到标准输出stdout方便Docker收集。集成像Prometheus这样的监控系统来收集指标请求数、延迟、错误率。数据库升级当待办事项数据量增大或需要高可用时将SQLite迁移到PostgreSQL或MySQL。只需修改database.py中的连接逻辑即可上层工具函数接口可以保持不变。6.3 功能扩展添加更多工具这个模式可以无限扩展。例如添加search_todos全文搜索、set_todo_due_date设置截止日期、add_todo_comment添加评论等工具。支持“资源”ResourcesMCP协议除了“工具”还有“资源”的概念。你可以将“今天的待办列表”或“某个特定项目下的所有事项”定义为一个资源URL客户端可以读取其内容。这对于提供静态或动态生成的数据块很有用。集成外部系统这才是MCP的威力所在。你可以开发连接内部GitLab的服务器创建MR、查看CI状态、连接财务系统的服务器查询报表、连接物联网平台的服务器控制设备。只需要遵循同样的模式定义工具、实现业务逻辑、处理认证。通过这15分钟的高强度实战我们不仅构建了一个可工作的MCP服务器更搭建了一个具备生产级部署雏形的安全、可扩展的框架。核心在于理解MCP作为桥梁的角色以及如何将你的内部能力通过标准化协议安全地暴露给AI。接下来你可以将这个Todo管理器替换成任何你需要的业务逻辑开启AI赋能工作流的新篇章。