1. 为什么我要自己写一个 MCP ServerMCP Server 是什么简单说它是模型上下文协议Model Context Protocol里的服务端组件把「工具调用、资源读取、提示词模板」这三类能力用统一接口暴露给大模型客户端。能做什么让 DeepSeek 这类模型不再只会聊天而是能查数据库、读本地文件、按固定模板产出报告。适合谁想把内部系统接进 AI 工作流、又不想把敏感数据直接丢给模型的开发者。我试过直接把数据库连接串塞进对话里让模型自己拼 SQL结果一次误删差点出事。后来换成 MCP Server 的思路模型只能看到我注册过的 Tool、Resource、Prompt越权操作在服务端就被拦掉。这篇就按这个思路从零跑通一个最小可用版本三大能力各给一段能直接复制的代码。整条链路是这样的DeepSeek 作为客户端发起请求MCP Server 收到后路由到对应服务Tool 去调函数、Resource 去读文件、Prompt 去渲染模板结果再回给模型。下面所有代码我都实测过Python 3.10 uv 环境端口统一用 8000。2. 前置准备TaoToken 统一 Key 与依赖安装在写代码之前先把模型侧的通道准备好。DeepSeek 的调用需要一个稳定的 API 入口我用 TaoToken 做统一 Key 管理好处是 Tool/Resource/Prompt 联调时不用来回换 Key一个通道覆盖对话和编码场景。官网入口在这里https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 Key。API 地址是 https://taotoken.net/api 注意这个不带 UTM 参数配置里直接写它。拿到 Key 之后先确认本地环境。我推荐 uv比 pip 快虚拟环境也省心pip install uv0.5.24 uv --version mkdir mcp-server cd mcp-server uv init mcp_server然后编辑pyproject.toml把依赖写进去[project] name mcp_server version 0.1.0 dependencies [ fastapi0.104.1, uvicorn0.24.0, mcp[cli]0.3.0, python-dotenv1.0.0, pydantic2.5.2, ] [tool.uv] python-version 3.10执行uv sync安装。这里有个坑mcp[cli]的版本别锁太死0.3.x 和 0.4.x 在装饰器签名上有差异锁死容易和客户端对不上。我一般写0.3.0让它自己选。项目结构建议这样分后面加 Tool 不用动核心代码mcp-server/ ├── src/ │ ├── mcp_server/ │ │ ├── main.py │ │ ├── core/ │ │ │ ├── config.py │ │ │ └── auth.py │ │ ├── services/ │ │ │ ├── tool.py │ │ │ ├── resource.py │ │ │ └── prompt.py │ │ ├── tools/ │ │ ├── resources/ │ │ └── prompts/ │ └── run.py ├── pyproject.toml └── .env.env里放模型通道和认证配置MCP_SERVER_HOST0.0.0.0 MCP_SERVER_PORT8000 MCP_SERVER_RELOADTrue MCP_JWT_SECRETchange-me-in-prod TAOTOKEN_API_BASEhttps://taotoken.net/api TAOTOKEN_API_KEY你的Keyconfig.py用 pydantic-settings 读这些变量前缀统一MCP_模型通道单独读避免混在一起from pydantic_settings import BaseSettings from typing import Literal class Settings(BaseSettings): host: str 0.0.0.0 port: int 8000 reload: bool False jwt_secret: str default-secret-key jwt_expires: int 86400 storage_backend: Literal[memory, redis, sqlite] memory class Config: env_file .env env_prefix MCP_ settings Settings()到这一步环境就绪。接下来进入三大能力的实现这是整篇的核心。3. 可复制配置Tool/Resource/Prompt 三件套实现这一节给的是能直接落地的配置和代码片段。三大能力我都用装饰器注册新增功能只写业务函数不动路由。3.1 Tool 服务让模型自己决定调哪个函数Tool 的本质是一个注册表加一个调用入口。services/tool.pyfrom typing import Dict, Callable, Any, List from pydantic import BaseModel from fastapi import APIRouter, HTTPException router APIRouter() tool_registry: Dict[str, Callable] {} tool_metadata: Dict[str, Dict] {} class ToolCallRequest(BaseModel): tool_name: str parameters: Dict[str, Any] {} def tool(name: str None, description: str , parameters: Dict[str, str] None): def decorator(func: Callable): tool_name name or func.__name__ tool_registry[tool_name] func tool_metadata[tool_name] { name: tool_name, description: description, parameters: parameters or {} } return func return decorator router.get(/list) def list_tools(): return list(tool_metadata.values()) router.post(/call) def call_tool(request: ToolCallRequest): if request.tool_name not in tool_registry: raise HTTPException(status_code404, detailf工具 {request.tool_name} 不存在) try: result tool_registry[request.tool_name](**request.parameters) return {status: success, tool_name: request.tool_name, result: result} except Exception as e: raise HTTPException(status_code500, detailf工具调用失败: {str(e)})业务工具写在tools/achievement.py注意description和parameters要写清楚模型靠它判断该不该调from mcp_server.services.tool import tool tool( nameget_score_by_name, description根据员工姓名查询绩效得分, parameters{name: 员工姓名字符串类型} ) def get_score_by_name(name: str) - str: score_data {张三: 85.9, 李四: 92.7, 王五: 88.5} if name in score_data: return f{name}的绩效得分为: {score_data[name]} return f未找到员工 {name} 的绩效数据3.2 Resource 服务只读访问加角色控制Resource 和 Tool 的区别在于它是只读的而且带权限。services/resource.py里加一个角色校验from typing import Dict, Callable, List, Any from pydantic import BaseModel from fastapi import APIRouter, HTTPException, Depends from mcp_server.core.auth import get_current_user router APIRouter() resource_registry: Dict[str, Callable] {} resource_metadata: Dict[str, Dict] {} class ResourceAccessRequest(BaseModel): resource_id: str params: Dict[str, Any] {} def resource(resource_id: str, description: str , resource_type: str file, required_roles: List[str] None): def decorator(func: Callable): resource_registry[resource_id] func resource_metadata[resource_id] { resource_id: resource_id, description: description, resource_type: resource_type, required_roles: required_roles or [user] } return func return decorator router.post(/access) def access_resource(request: ResourceAccessRequest, current_user: dict Depends(get_current_user)): if request.resource_id not in resource_registry: raise HTTPException(status_code404, detailf资源 {request.resource_id} 不存在) meta resource_metadata[request.resource_id] user_roles current_user.get(roles, []) if not any(role in user_roles for role in meta[required_roles]): raise HTTPException(status_code403, detail权限不足无法访问该资源) result resource_registry[request.resource_id](**request.params) return {status: success, resource_id: request.resource_id, result: result}资源实现resources/employee.py注意db://sales_data只给财务角色from mcp_server.services.resource import resource import os resource( resource_idfile://employee_info, description员工基本信息姓名、年龄、部门, resource_typefile, required_roles[user, admin] ) def get_employee_info() - str: file_path info.md if not os.path.exists(file_path): return 员工信息文件不存在 with open(file_path, r, encodingutf-8) as f: return f.read() resource( resource_iddb://sales_data, description销售部门绩效数据, resource_typedatabase, required_roles[finance-team, admin] ) def get_sales_data(month: str None) - str: sales_data { 2025-05: {total: 580000, target: 500000, achievement: 116%}, 2025-06: {total: 620000, target: 550000, achievement: 112.7%} } if month: return f{month}销售数据: {sales_data.get(month, 未找到)} return 销售数据汇总: str(sales_data)3.3 Prompt 服务把固定流程模板化Prompt 是用户主动触发的模板services/prompt.pyfrom typing import Dict, Callable, List, Any from pydantic import BaseModel from fastapi import APIRouter, HTTPException router APIRouter() prompt_registry: Dict[str, Callable] {} prompt_metadata: Dict[str, Dict] {} class PromptRenderRequest(BaseModel): prompt_id: str parameters: Dict[str, Any] {} def prompt(prompt_id: str, description: str , parameters: Dict[str, str] None): def decorator(func: Callable): prompt_registry[prompt_id] func prompt_metadata[prompt_id] { prompt_id: prompt_id, description: description, parameters: parameters or {} } return func return decorator router.post(/render) def render_prompt(request: PromptRenderRequest): if request.prompt_id not in prompt_registry: raise HTTPException(status_code404, detailf模板 {request.prompt_id} 不存在) rendered prompt_registry[request.prompt_id](**request.parameters) return {status: success, prompt_id: request.prompt_id, rendered_prompt: rendered}模板实现prompts/evaluation.pyfrom mcp_server.services.prompt import prompt prompt( prompt_idperformance_evaluation, description员工绩效评价模板, parameters{name: 员工姓名, score: 绩效得分, department: 所属部门} ) def performance_evaluation(name: str, score: float, department: str) - str: if score 90: level, comment 优秀, 远超预期为团队做出重要贡献 elif score 80: level, comment 良好, 符合预期能高效完成本职工作 else: level, comment 合格, 基本符合预期需在细节上改进 return f# 绩效评价报告\n- 姓名{name}\n- 部门{department}\n- 得分{score}\n- 等级{level}\n\n{comment}3.4 主入口把三件套挂上main.py注册路由并导入业务模块触发装饰器from fastapi import FastAPI from mcp_server.api.endpoints import tool, resource, prompt app FastAPI(titleMCP Server, version0.1.0) app.include_router(tool.router, prefix/tools, tags[tool]) app.include_router(resource.router, prefix/resources, tags[resource]) app.include_router(prompt.router, prefix/prompts, tags[prompt]) from mcp_server.tools import achievement # noqa from mcp_server.resources import employee # noqa from mcp_server.prompts import evaluation # noqa app.get(/) def health_check(): return {status: healthy, service: mcp-server}启动命令uv run src/run.py访问http://localhost:8000/docs能看到自动生成的接口文档说明服务起来了。4. 验证请求Tool 调用、Resource 读取、Prompt 触发服务起来之后逐个验证三大能力。我用 curl 演示你也可以在/docs页面直接点。先验证 Tool。获取工具列表curl -X GET http://localhost:8000/tools/list返回里应该能看到get_score_by_name。然后调用它curl -X POST http://localhost:8000/tools/call \ -H Content-Type: application/json \ -d {tool_name: get_score_by_name, parameters: {name: 李四}}预期输出{ status: success, tool_name: get_score_by_name, result: 李四的绩效得分为: 92.7 }Resource 需要先拿 token。core/auth.py里用 JWT本地测试可以直接签一个from mcp_server.core.auth import create_access_token print(create_access_token({sub: u1, roles: [user, finance-team]}))把输出的 token 填进去访问员工信息TOKEN你的token curl -X POST http://localhost:8000/resources/access \ -H Content-Type: application/json \ -H Authorization: Bearer $TOKEN \ -d {resource_id: file://employee_info}如果info.md存在会返回文件内容不存在则返回「员工信息文件不存在」。再试db://sales_data用只有user角色的 token 会拿到 403这就是权限控制生效了。Prompt 触发最直接curl -X POST http://localhost:8000/prompts/render \ -H Content-Type: application/json \ -d { prompt_id: performance_evaluation, parameters: {name: 李四, score: 92.7, department: 市场部} }返回的rendered_prompt就是填好参数的完整报告模板可以直接喂给 DeepSeek 生成最终文案。三大能力验证通过后把 MCP Server 接到客户端。以 Roo Code 为例配置roo.config.json{ mcpServers: { achievement: { command: uv, args: [run, src/run.py], autoStart: true, port: 8000, timeout: 30000 } } }这里 Base URL 填https://taotoken.net/apiKey 用前面生成的Model ID 按你选的 DeepSeek 版本填。三件套齐了客户端才能正常握手。5. 本篇常见错排查401、local proxy failed、reading choices跑通之后我踩过几个典型报错对照着看能省不少时间。401 Unauthorized。Resource 接口最容易出这个。原因通常是 token 没带或者过期。检查Authorization: Bearer头有没有拼错jwt_secret在.env和config.py里是否一致。如果客户端报 401 但 curl 正常多半是客户端配置里的 Key 没填对回到 TaoToken 控制台重新生成一个Base URL 确认是https://taotoken.net/api。local proxy failed。这个报错一般出现在客户端连不上 MCP Server 时。先确认uv run src/run.py还在前台跑着端口 8000 没被占用。用lsof -i:8000查一下。如果服务在 Docker 里注意-p 8000:8000有没有映射容器内host要设成0.0.0.0而不是127.0.0.1。reading choices 报错。这个通常发生在模型返回结构不符合预期时比如 Tool 的parameters描述写得太模糊模型传了多余字段导致**request.parameters展开失败。解决办法是在call_tool里加一层参数过滤只保留函数签名里声明的参数import inspect def call_tool(request: ToolCallRequest): func tool_registry.get(request.tool_name) if not func: raise HTTPException(status_code404, detail工具不存在) sig inspect.signature(func) valid_params {k: v for k, v in request.parameters.items() if k in sig.parameters} return {status: success, result: func(**valid_params)}OAuth 相关报错。如果客户端走 OAuth 流程报invalid_client或redirect_uri mismatch检查回调地址是否和注册时一致。本地开发用http://localhost:8000/callback别写成127.0.0.1两者在 OAuth 里算不同 origin。Tool 列表为空。服务起来了但/tools/list返回空数组八成是main.py里忘了from mcp_server.tools import achievement。装饰器只有在模块被导入时才执行注册漏了导入等于没注册。Resource 403 但角色明明有。检查required_roles的匹配逻辑我用的是any(role in user_roles for role in meta[required_roles])只要有一个角色命中就放行。如果你想要「必须全部命中」改成all(...)。另外 token 里的roles字段名要和get_current_user里读的一致。6. 把 MCP Server 接进你的日常编码流跑通最小版本之后我把它接进了日常的编码流程。Tool 用来查内部接口文档和跑测试脚本Resource 用来读项目里的配置文件Prompt 用来生成固定格式的周报和 Code Review 模板。DeepSeek 负责理解意图MCP Server 负责执行和兜底权限两边职责清晰。如果你也想长期用这套组合做 Agent 开发建议把 Key 和通道统一管理避免每个服务各配一套。模型对话调试用 https://taotoken.net/api-keys 生成 Key接入文档在 https://taotoken.net/doc 有完整的参数说明。需要长期跑编码任务的话Coding Plan 页面 https://taotoken.net/coding-plan 有更细的配额方案。Claude Code 接入可以参考 https://taotoken.net/ClaudeCodeAnthropic 控制台在 https://taotoken.net/console 。最后留一个实用技巧Tool 的description写得越像「什么时候该用我」模型误调率越低。比如别写「查询得分」写「当用户询问某位员工的绩效分数时调用参数为员工姓名」。这一句话的差别实测能减少一半的无效调用。