# 构建 MCP 服务器模型上下文协议服务器开发参考指南（2026 年 1 月）。涵盖使用 Streamable HTTP 传输的 TypeScript、Python 和 C# 实现。 ## 快速参考语言 |套餐 |版本 |交通 ---|---|---|--- 打字稿 | `@modelcontextprotocol/sdk` | 1.25.1 | `StreamableHTTPServerTransport` 蟒蛇 | `mcp` | 1.25.0 | `transport="streamable-http"` C# | `ModelContextProtocol.AspNetCore` | 0.6.0-预览 | `.WithHttpTransport()` ### 运输状态交通 |状态 |使用案例 ---|---|--- **标准输入输出** |支持 |本地/CLI（Claude 桌面、光标） **流式 HTTP** |推荐|远程服务器、生产 **上交所** |已弃用 |仅旧版 ## 可流式 HTTP 传输 **单端点**取代双 SSE 端点。支持有状态会话或无状态（无服务器）模式。 ### 协议流程客户端服务器 |------ POST /mcp ---------------->| （JSON-RPC 消息） |<----- JSON 或 SSE 响应 ------| |------ GET /mcp ----------------->| （可选：服务器启动） |<----- SSE 流 ----------------| ### 必需的标头 POST /mcp HTTP/1.1 内容类型：application/json 接受：应用程序/json、文本/事件流 Mcp-Session-Id: # 初始化后 ## TypeScript 实现 ### 安装 npm install @modelcontextprotocol/sdk zod express npm install -D @types/express ### 基本服务器从“@modelcontextprotocol/sdk/server/mcp.js”导入{McpServer}；从“@modelcontextprotocol/sdk/server/streamableHttp.js”导入{ StreamableHTTPServerTransport }；从“zod”导入{z}；从“快递”进口快递；常量应用程序 = Express(); app.use(express.json()); 常量服务器 = 新 McpServer({ 名称：“我的服务器”，版本：“1.0.0” }); // 工具 server.registerTool("添加", { 描述：“两个数字相加”， inputSchema: { a: z.number(), b: z.number() } }, 异步 ({ a, b }) => ({ 内容：[{类型：“文本”，文本：字符串（a + b）}] })); // 资源服务器.注册资源( “配置”， “配置://应用程序”， { 描述：“应用程序配置” }, 异步（uri）=>（{ 内容：[{ uri: uri.href, text: JSON.stringify({ env: "prod" }) }] }) ）； // 提示 server.registerPrompt("审核", { 描述：“代码审查”， argsSchema: { 代码: z.string() } }, ({ 代码 }) => ({ 消息：[{ 角色：“用户”，内容：{ 类型：“文本”，文本：`评论：\n${code}` } }] })); // 运输 const 传输 = 新 StreamableHTTPServerTransport({ sessionIdGenerator: () => crypto.randomUUID() }); 等待服务器.connect(传输); app.all("/mcp", async (req, res) => { 等待 Transport.handleRequest(req, res); }); 应用程序.听(3000); > **注意：** 顶级 `await` 需要带有 ES 模块的 Node.js（package.json 中的 `"type": "module"` 或 `.mjs` 扩展名）。 ### 无状态模式（无服务器） const 传输 = 新 StreamableHTTPServerTransport({ sessionIdGenerator: undefined // 禁用会话 }); ## Python 实现 ### 安装 pip 安装“mcp[cli]” ### 基本服务器 (FastMCP) 从 mcp.server.fastmcp 导入 FastMCP，上下文从输入导入列表 mcp = FastMCP(“我的服务器”) # 工具 @mcp.tool() def add(a: int, b: int) -> int: “”“添加两个数字。”“” 返回 a + b # 有进度的异步工具 @mcp.tool() async def process_files(files: List[str], ctx: Context) -> str: """有进度地处理文件。""" 对于枚举（文件）中的 i、f：等待 ctx.report_progress(i + 1, len(文件)) return f“已处理的 {len(files)} 个文件” # 资源 @mcp.resource("config://app") def get_config() -> 字典： """应用程序配置。""" 返回 {"env": "prod", "version": "1.0.0"} # 动态资源 @mcp.resource(“用户://{user_id}/个人资料”) def get_user(user_id: str) -> 字典: """获取用户个人资料。""" 返回 {"id": user_id, "name": f"用户 {user_id}"} # 提示 @mcp.prompt() def code_review(代码: str, 语言: str = "python") -> str: """代码审查提示。""" return f"查看此{语言}代码：\n\n```{语言}\n{代码}\n````" 如果 __name__ == "__main__": # 可传输的 HTTP mcp.run（传输=“streamable-http”，主机=“0.0.0.0”，端口= 8000，路径=“/ mcp”） # 或 stdio: mcp.run() ### FastAPI 集成从 fastapi 导入 FastAPI 从 mcp.server.fastmcp 导入 FastMCP api = FastAPI() mcp = FastMCP("api-tools") @mcp.tool() def 查询(sql: str) -> 字典: 返回{“结果”：“数据”} api.mount("/mcp", mcp.streamable_http_app()) # 运行：uvicorn app:api --port 8000 ## C# 实现 ### 安装 dotnet 添加包 ModelContextProtocol.AspNetCore --prerelease ### 基本服务器使用 System.ComponentModel；使用 Microsoft.Extensions.DependencyInjection；使用 ModelContextProtocol.Server； var builder = WebApplication.CreateBuilder(args); 建造者服务 .AddMcpServer() .WithHttpTransport() .WithToolsFromAssembly() .WithPromptsFromAssembly() .WithResourcesFromAssembly(); var app = builder.Build(); app.MapMcp(); // 地图 /、/sse、/messages 应用程序.运行(); // 工具 [Mcp服务器工具类型] 公共静态类 MyTools { [McpServerTool, Description("添加两个数字")] 公共静态 int 添加（ [描述(“第一个数字”)] int a, [Description("第二个数字")] int b) => a + b; [McpServerTool，描述（“获取天气”）] 公共静态异步任务<字符串> GetWeather( HttpClient http, // 从 DI 注入 [Description("城市名称")] 字符串城市，取消令牌 ct) { var data = wait http.GetStringAsync($"https://api.weather.example/{city}", ct); 返回数据； } } // 提示 [Mcp服务器提示类型] 公共静态类 MyPrompts { [McpServerPrompt, Description("代码审查提示")] 公共静态聊天消息代码审查（ [描述(“要查看的代码”)] 字符串代码) => new(ChatRole.User, $"查看此代码：\n\n{code}"); } // 资源 [Mcp服务器资源类型] 公共静态类 MyResources { [McpServerResource(名称 = "config://app"), 描述("应用程序配置")] public static string Config() => """{"env": "生产"}"""; [McpServerResource(UriTemplate = "docs://{topic}")] 公共静态字符串 GetDoc([描述("主题")] 字符串主题) => $"{topic} 的文档"; } ### Stdio 交通（本地） var builder = Host.CreateApplicationBuilder(args); builder.Logging.AddConsole(o => o.LogToStandardErrorThreshold = LogLevel.Trace); 建造者服务 .AddMcpServer() .WithStdioServerTransport() .WithToolsFromAssembly(); 等待 builder.Build().RunAsync(); ## 身份验证（OAuth 2.1） MCP 服务器是 **OAuth 资源服务器**（不是授权服务器）。关键要求： * **PKCE 强制**（S256 方法） * **需要资源指标** (RFC 8707) * **授权标头中的不记名令牌** * **针对每个请求进行受众验证** ### 发现端点 GET /.well-known/oauth-protected-resource # 服务器元数据 GET /.well-known/oauth-authorization-server # 身份验证服务器元数据 ### 401 服务器响应 HTTP/1.1 401 未经授权 WWW-身份验证：承载领域=“mcp”， resource_metadata =“https://mcp.example.com/.well-known/oauth-protected-resource” ### 令牌验证 (Python) 导入jwt 从 jwt 导入 PyJWKClient 类令牌验证器： def __init__(self, jwks_uri: str, 受众: str): self.jwks = PyJWKClient(jwks_uri) self.audience = 观众 def validate(self, token: str) -> 字典： key = self.jwks.get_signing_key_from_jwt(令牌) 返回 jwt.decode( 令牌、密钥.密钥、算法=["RS256"], 观众=自我观众，选项={"require": ["exp", "aud", "iss"]} ） ### 受保护资源元数据响应 { “资源”：“https://mcp.example.com”， “authorization_servers”：[“https://auth.example.com”]， “scopes_supported”：[“读”，“写”，“管理”]， “bearer_methods_supported”：[“标头”] } ### OAuth 中间件集成 (Python/Starlette) 从 starlette.applications 导入 Starlette 从 starlette.middleware 导入中间件从 starlette.middleware.base 导入 BaseHTTPMiddleware 从 starlette.responses 导入 JSONResponse 从 mcp.server.fastmcp 导入 FastMCP 类 BearerAuthMiddleware(BaseHTTPMiddleware): def __init__(self, 应用程序, 验证器: TokenValidator): 超级().__init__(应用程序) self.validator = 验证器异步 def 调度（自身，请求，call_next）： auth = request.headers.get("授权", "") 如果不是 auth.startswith("Bearer "): 返回 JSON 响应（ {“错误”：“未经授权”}，状态代码=401， headers={"WWW-Authenticate": '承载领域="mcp"'} ）尝试： token = auth.replace("承载者", "") request.state.auth = self.validator.validate(token) 除了例外：返回 JSONResponse({"error": "invalid_token"}, status_code=401) 返回等待call_next（请求） # 设置 mcp = FastMCP(“安全服务器”) 验证器 = TokenValidator( jwks_uri =“https://auth.example.com/.well-known/jwks.json”，受众=“https://mcp.example.com” ）应用程序 = 星星（路线= [安装（“/ mcp”，app = mcp.streamable_http_app（））]，中间件=[中间件(BearerAuthMiddleware, 验证器=验证器)] ） ### OAuth 中间件集成（TypeScript/Express）从“jsonwebtoken”导入jwt；从“jwks-rsa”导入jwksClient； const client = jwksClient({ jwksUri: "https://auth.example.com/.well-known/jwks.json" }); const authMiddleware = async (req, res, next) => { const auth = req.headers.authorization; if (!auth?.startsWith("Bearer ")) { 返回 res.status(401).json({ 错误: "未经授权" }); } 尝试{ const 令牌 = auth.slice(7); const 解码 = jwt.decode(token, { 完整: true }); const key = 等待 client.getSigningKey(decoded.header.kid); req.auth = jwt.verify(token, key.getPublicKey(), { 受众：“https://mcp.example.com”，算法：[“RS256”] }); 下一个（）; } 捕获（错误）{ res.status(401).json({ 错误: "invalid_token" }); } }; app.use("/mcp", authMiddleware); ### 经过身份验证的客户端请求 const 传输 = 新 StreamableHTTPClientTransport( 新网址（“https://mcp.example.com/mcp”）， { 请求初始化：{ headers: { 授权: `Bearer ${accessToken}` } } } ）； ### 安全反模式反模式|修复 ---|--- 令牌直通上游 API |对上游调用使用单独的令牌缺少观众验证 |始终验证“aud”声明 URL 中的令牌 |仅使用授权标头 ### 工具中的范围执行在执行敏感操作之前检查范围： **打字稿：** const authMiddleware = async (req, res, next) => { // ... 令牌验证 ... req.auth = { sub:decoded.sub, 范围:decoded.scope?.split(" ") || []}；下一个（）; }; server.registerTool("delete_user", { /* ... */ }, async ({ userId }, { meta }) => { const 作用域 = meta?.auth?.scopes || []； if (!scopes.includes("admin:write")) { return { isError: true, content: [{ type: "text", text: "范围不足" }] }; } // ...执行删除... }); **Python:** 使用HTTP中间件进行验证，然后签入工具： # 使用 Starlette 中间件（请参阅上面的 OAuth 中间件集成） # 将经过验证的声明存储在 request.state 中，然后签入工具： @mcp.tool() def delete_user(user_id: str) -> str: """删除用户 - 需要 admin:write 范围。""" # 范围强制通过中间件发生在 HTTP 层 # 工具假设请求已经通过身份验证检查 return f"已删除用户 {user_id}" > **注意：** 对于具有每个工具身份验证上下文的高级中间件，请使用提供“Middleware”类和“Context.get_state()”的“fastmcp”包（“pip install fastmcp”）。官方的“mcp”包提供 FastMCP，但具有更简单的中间件选项。 ### 将身份验证上下文传递给工具处理程序 **TypeScript：** 在传输上存储身份验证或使用请求范围的上下文： // 在中间件中：附加到请求 req.auth = { 子: 解码.sub, 电子邮件: 解码.email }; // 通过服务器上下文或闭包传递给工具 const 会话 = new Map(); app.all("/mcp", async (req, res) => { const Transport = new StreamableHTTPServerTransport({ /* ... */ }); session.set(transport.sessionId, { auth: req.auth }); // 通过session.get(sessionId)访问工具 }); **Python（带有 `fastmcp` 包）：** 使用中间件和上下文状态： # pip 安装 fastmcp 从 fastmcp 导入 FastMCP，上下文从 fastmcp.server.middleware 导入中间件，MiddlewareContext mcp = FastMCP(“我的服务器”) 类 AuthMiddleware(中间件): 异步 def on_call_tool(self, context: MiddlewareContext, call_next): # 提取并验证令牌，然后存储在上下文中 context.fastmcp_context.set_state("user_id", "user_123") context.fastmcp_context.set_state("范围", ["读", "写"]) 返回等待 call_next() mcp.add_middleware(AuthMiddleware()) @mcp.工具异步 def get_my_profile(ctx: Context) -> dict: user_id = ctx.get_state("user_id") # 由中间件设置返回 {“user_id”：user_id，“个人资料”：“...”} ### 令牌刷新处理访问令牌的有效期很短（通常为 1 小时）。策略： 1. **客户端刷新：**客户端在过期前刷新token并重新连接 2. **主动服务器刷新：** 后台任务刷新即将过期的令牌 3. **按需刷新：** 返回401，客户端刷新并重试 **推荐模式：** 使用短期访问令牌（5-60 分钟）和刷新令牌。在 401 上： // 客户端重试逻辑异步函数 callWithRefresh(tool: string, args: object) { 尝试{ 返回等待 client.callTool(工具, args); } 捕获（错误）{ if (err.status === 401) { accessToken = 等待refreshAccessToken(refreshToken); Transport.updateHeaders({ 授权: `Bearer ${accessToken}` }); 返回等待 client.callTool(工具, args); } 抛出错误； } } ## 迁移：SSE 到 Streamable HTTP ### 服务器变更 // OLD (SSE) - 两个端点 app.get("/sse", async (req, res) => { const Transport = new SSEServerTransport("/sse/messages", res); 等待服务器.connect(传输); }); app.post("/sse/messages", async (req, res) => { /* ... */ }); // 新（流式 HTTP）- 单端点 app.all("/mcp", async (req, res) => { const Transport = new StreamableHTTPServerTransport(); 等待服务器.connect(传输); 等待 Transport.handleRequest(req, res); }); ### 客户变更 // 旧的从“@modelcontextprotocol/sdk/client/sse.js”导入{SSEClientTransport}； const Transport = new SSEClientTransport(new URL("http://localhost:3000/sse")); // 新从“@modelcontextprotocol/sdk/client/streamableHttp.js”导入{ StreamableHTTPClientTransport }； const Transport = new StreamableHTTPClientTransport(new URL("http://localhost:3000/mcp")); ### 向后兼容的服务器 // 迁移期间支持两种传输 app.all("/mcp", /* 新的 Streamable HTTP 处理程序 */); app.get("/sse", /* 旧版 SSE 处理程序 */); app.post("/sse/messages", /* 旧版消息处理程序 */); ## 客户端配置 ### 克劳德桌面/克劳德代码 (stdio) { “mcp服务器”：{ “我的服务器”：{ “命令”：“节点”， “args”：[“/path/to/server.js”]， “env”：{“API_KEY”：“秘密”} } } } ### 远程服务器（流式 HTTP） { “mcp服务器”：{ “远程”：{ “类型”：“streamable-http”， “url”：“https://mcp.example.com/mcp” } } } ## 错误处理返回工具错误（不是协议错误）以进行模型自校正： **打字稿：** server.registerTool("查询", { /* ... */ }, async ({ sql }) => { if (sql.includes("DROP")) { 返回{ 错误：正确，内容：[{类型：“文本”，文本：“不允许破坏性查询”}] }; } // ... }); **Python：** 在工具中引发异常 - 它们被捕获并作为错误返回： @mcp.tool() def 查询(sql: str) -> 字典: 如果 sql.upper() 中的“DROP”： raise ValueError("不允许破坏性查询") 返回{“结果”：“数据”} ## 常见错误错误|修复 ---|--- 将 SSE 用于新服务器 |使用流式 HTTP 写入stdio 服务器中的stdout |使用 stderr 来记录日志初始化后缺少会话 ID |始终包含 `Mcp-Session-Id` 标头不验证 OAuth 受众 |验证令牌“aud”与您的服务器匹配向上游转发客户端令牌 |对上游 API 使用单独的凭据 ## 官方资源 ### SDK * **TypeScript** ： * **Python**： * **C#** ： ### 文档 * **规范** ： * **运输** ： * **授权** : ### 测试 npx @modelcontextprotocol/inspector 节点路径/to/server.js

构建 Mcp 服务器

包含技能