building-mcp-servers

构建 MCP 服务

Model Context Protocol 服务端开发参考指南（2026 年 1 月）。涵盖使用 Streamable HTTP 传输的 TypeScript、Python 和 C# 实现。

快速参考

语言	包	版本	传输方式
TypeScript	@modelcontextprotocol/sdk	1.25.1	StreamableHTTPServerTransport
Python	mcp	1.25.0	transport="streamable-http"
C#	ModelContextProtocol.AspNetCore	0.6.0-preview	.WithHttpTransport()

传输状态

传输方式	状态	使用场景
stdio	支持	本地/命令行（Claude Desktop、Cursor）
Streamable HTTP	推荐	远程服务、生产环境
SSE	已弃用	仅用于旧版

Streamable HTTP 传输

单一端点取代了双 SSE 端点。支持有状态会话或无状态（无服务器）模式。

协议流程

Client                              Server
  |------ POST /mcp ---------------->|  (JSON-RPC messages)
  |<----- JSON or SSE response ------|
  |------ GET /mcp ----------------->|  (Optional: server-initiated)
  |<----- SSE stream ----------------|

必需的标头

POST /mcp HTTP/1.1
Content-Type: application/json
Accept: application/json, text/event-stream
Mcp-Session-Id: <session-id>  # After initialization

TypeScript 实现

安装

npm install @modelcontextprotocol/sdk zod express
npm install -D @types/express

基础服务端

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { z } from "zod";
import express from "express";

const app = express();
app.use(express.json());

const server = new McpServer({
  name: "my-server",
  version: "1.0.0"
});

// Tool
server.registerTool("add", {
  description: "Add two numbers",
  inputSchema: { a: z.number(), b: z.number() }
}, async ({ a, b }) => ({
  content: [{ type: "text", text: String(a + b) }]
}));

// Resource
server.registerResource(
  "config",
  "config://app",
  { description: "App configuration" },
  async (uri) => ({
    contents: [{ uri: uri.href, text: JSON.stringify({ env: "prod" }) }]
  })
);

// Prompt
server.registerPrompt("review", {
  description: "Code review",
  argsSchema: { code: z.string() }
}, ({ code }) => ({
  messages: [{ role: "user", content: { type: "text", text: `Review:\n${code}` } }]
}));

// Transport
const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: () => crypto.randomUUID()
});

await server.connect(transport);

app.all("/mcp", async (req, res) => {
  await transport.handleRequest(req, res);
});

app.listen(3000);

注意： 顶层 await 需要使用 ES 模块的 Node.js（在 package.json 中设置 "type": "module" 或使用 .mjs 扩展名）。

无状态模式（无服务器）

const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: undefined  // Disables sessions
});

Python 实现

安装

pip install "mcp[cli]"

基础服务端（FastMCP）

from mcp.server.fastmcp import FastMCP, Context
from typing import List

mcp = FastMCP("my-server")

# Tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

# Async tool with progress
@mcp.tool()
async def process_files(files: List[str], ctx: Context) -> str:
    """Process files with progress."""
    for i, f in enumerate(files):
        await ctx.report_progress(i + 1, len(files))
    return f"Processed {len(files)} files"

# Resource
@mcp.resource("config://app")
def get_config() -> dict:
    """App configuration."""
    return {"env": "prod", "version": "1.0.0"}

# Dynamic resource
@mcp.resource("users://{user_id}/profile")
def get_user(user_id: str) -> dict:
    """Get user profile."""
    return {"id": user_id, "name": f"User {user_id}"}

# Prompt
@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """Code review prompt."""
    return f"Review this {language} code:\n\n```{language}\n{code}\n```"

if __name__ == "__main__":
    # Streamable HTTP
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")
    # Or stdio: mcp.run()

FastAPI 集成

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

api = FastAPI()
mcp = FastMCP("api-tools")

@mcp.tool()
def query(sql: str) -> dict:
    return {"result": "data"}

api.mount("/mcp", mcp.streamable_http_app())
# Run: uvicorn app:api --port 8000

C# 实现

安装

dotnet add package ModelContextProtocol.AspNetCore --prerelease

基础服务端

using System.ComponentModel;
using Microsoft.Extensions.DependencyInjection;
using ModelContextProtocol.Server;

var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddMcpServer()
    .WithHttpTransport()
    .WithToolsFromAssembly()
    .WithPromptsFromAssembly()
    .WithResourcesFromAssembly();

var app = builder.Build();
app.MapMcp();  // Maps /, /sse, /messages
app.Run();

// Tools
[McpServerToolType]
public static class MyTools
{
    [McpServerTool, Description("Add two numbers")]
    public static int Add(
        [Description("First number")] int a,
        [Description("Second number")] int b) => a + b;

    [McpServerTool, Description("Get weather")]
    public static async Task<string> GetWeather(
        HttpClient http,  // Injected from DI
        [Description("City name")] string city,
        CancellationToken ct)
    {
        var data = await http.GetStringAsync($"https://api.weather.example/{city}", ct);
        return data;
    }
}

// Prompts
[McpServerPromptType]
public static class MyPrompts
{
    [McpServerPrompt, Description("Code review prompt")]
    public static ChatMessage CodeReview(
        [Description("Code to review")] string code) =>
        new(ChatRole.User, $"Review this code:\n\n{code}");
}

// Resources
[McpServerResourceType]
public static class MyResources
{
    [McpServerResource(Name = "config://app"), Description("App config")]
    public static string Config() => """{"env": "production"}""";

    [McpServerResource(UriTemplate = "docs://{topic}")]
    public static string GetDoc([Description("Topic")] string topic) =>
        $"Documentation for {topic}";
}

Stdio 传输（本地）

var builder = Host.CreateApplicationBuilder(args);

builder.Logging.AddConsole(o => o.LogToStandardErrorThreshold = LogLevel.Trace);

builder.Services
    .AddMcpServer()
    .WithStdioServerTransport()
    .WithToolsFromAssembly();

await builder.Build().RunAsync();

身份验证（OAuth 2.1）

MCP 服务是 OAuth 资源服务（而非授权服务）。主要要求：

• 必须使用 PKCE（S256 方法）
• 必须使用资源指示符（RFC 8707）
• 在 Authorization 标头中使用 Bearer 令牌
• 对每个请求进行 受众验证

发现端点

GET /.well-known/oauth-protected-resource  # Server metadata
GET /.well-known/oauth-authorization-server  # Auth server metadata

服务端对 401 的响应

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer realm="mcp",
  resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"

令牌验证（Python）

import jwt
from jwt import PyJWKClient

class TokenValidator:
    def __init__(self, jwks_uri: str, audience: str):
        self.jwks = PyJWKClient(jwks_uri)
        self.audience = audience

    def validate(self, token: str) -> dict:
        key = self.jwks.get_signing_key_from_jwt(token)
        return jwt.decode(
            token, key.key,
            algorithms=["RS256"],
            audience=self.audience,
            options={"require": ["exp", "aud", "iss"]}
        )

受保护资源元数据响应

{
  "resource": "https://mcp.example.com",
  "authorization_servers": ["https://auth.example.com"],
  "scopes_supported": ["read", "write", "admin"],
  "bearer_methods_supported": ["header"]
}

OAuth 中间件集成（Python/Starlette）

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from mcp.server.fastmcp import FastMCP

class BearerAuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, validator: TokenValidator):
        super().__init__(app)
        self.validator = validator

    async def dispatch(self, request, call_next):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return JSONResponse(
                {"error": "unauthorized"},
                status_code=401,
                headers={"WWW-Authenticate": 'Bearer realm="mcp"'}
            )
        try:
            token = auth.replace("Bearer ", "")
            request.state.auth = self.validator.validate(token)
        except Exception:
            return JSONResponse({"error": "invalid_token"}, status_code=401)
        return await call_next(request)

# Setup
mcp = FastMCP("secure-server")
validator = TokenValidator(
    jwks_uri="https://auth.example.com/.well-known/jwks.json",
    audience="https://mcp.example.com"
)

app = Starlette(
    routes=[Mount("/mcp", app=mcp.streamable_http_app())],
    middleware=[Middleware(BearerAuthMiddleware, validator=validator)]
)

OAuth 中间件集成（TypeScript/Express）

import jwt from "jsonwebtoken";
import jwksClient from "jwks-rsa";

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 ")) {
    return res.status(401).json({ error: "unauthorized" });
  }
  try {
    const token = auth.slice(7);
    const decoded = jwt.decode(token, { complete: true });
    const key = await client.getSigningKey(decoded.header.kid);
    req.auth = jwt.verify(token, key.getPublicKey(), {
      audience: "https://mcp.example.com",
      algorithms: ["RS256"]
    });
    next();
  } catch (err) {
    res.status(401).json({ error: "invalid_token" });
  }
};

app.use("/mcp", authMiddleware);

已验证身份的客户端请求

const transport = new StreamableHTTPClientTransport(
  new URL("https://mcp.example.com/mcp"),
  {
    requestInit: {
      headers: { Authorization: `Bearer ${accessToken}` }
    }
  }
);

安全反模式

反模式	修正方法
将令牌直接传递给上游 API	为上游调用使用单独的令牌
未进行受众验证	始终验证 aud 声明
将令牌放在 URL 中	仅使用 Authorization 标头

工具中的作用域强制执行

在执行敏感操作前检查作用域：

TypeScript：

const authMiddleware = async (req, res, next) => {
  // ... token validation ...
  req.auth = { sub: decoded.sub, scopes: decoded.scope?.split(" ") || [] };
  next();
};

server.registerTool("delete_user", { /* ... */ }, async ({ userId }, { meta }) => {
  const scopes = meta?.auth?.scopes || [];
  if (!scopes.includes("admin:write")) {
    return { isError: true, content: [{ type: "text", text: "Insufficient scope" }] };
  }
  // ... perform deletion ...
});

Python： 使用 HTTP 中间件进行验证，然后在工具中检查：

# With Starlette middleware (see OAuth Middleware Integration above)
# Store validated claims in request.state, then check in tool:

@mcp.tool()
def delete_user(user_id: str) -> str:
    """Delete user - requires admin:write scope."""
    # Scope enforcement happens at HTTP layer via middleware
    # Tool assumes request already passed auth checks
    return f"Deleted user {user_id}"

注意： 如需支持按工具提供身份验证上下文的高级中间件，请使用 fastmcp 包（pip install fastmcp），该包提供 Middleware 类和 Context.get_state()。官方 mcp 包提供 FastMCP，但中间件选项较为简单。

将身份验证上下文传递给工具处理程序

TypeScript： 将身份验证信息存储在传输层，或使用请求作用域上下文：

// In middleware: attach to request
req.auth = { sub: decoded.sub, email: decoded.email };

// Pass to tool via server context or closure
const sessions = new Map();
app.all("/mcp", async (req, res) => {
  const transport = new StreamableHTTPServerTransport({ /* ... */ });
  sessions.set(transport.sessionId, { auth: req.auth });
  // Tools access via sessions.get(sessionId)
});

Python（使用 fastmcp 包）： 使用中间件和上下文状态：

# pip install fastmcp
from fastmcp import FastMCP, Context
from fastmcp.server.middleware import Middleware, MiddlewareContext

mcp = FastMCP("my-server")

class AuthMiddleware(Middleware):
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        # Extract and validate token, then store in context
        context.fastmcp_context.set_state("user_id", "user_123")
        context.fastmcp_context.set_state("scopes", ["read", "write"])
        return await call_next()

mcp.add_middleware(AuthMiddleware())

@mcp.tool
async def get_my_profile(ctx: Context) -> dict:
    user_id = ctx.get_state("user_id")  # Set by middleware
    return {"user_id": user_id, "profile": "..."}

令牌刷新处理

访问令牌的有效期较短（通常为 1 小时）。可采用以下策略：

1. 客户端刷新： 客户端在令牌过期前刷新，并重新连接
2. 服务端主动刷新： 后台任务刷新即将过期的令牌
3. 按需刷新： 返回 401，由客户端刷新后重试

推荐模式： 使用有效期较短的访问令牌（5-60 分钟）和刷新令牌。遇到 401 时：

// Client retry logic
async function callWithRefresh(tool: string, args: object) {
  try {
    return await client.callTool(tool, args);
  } catch (err) {
    if (err.status === 401) {
      accessToken = await refreshAccessToken(refreshToken);
      transport.updateHeaders({ Authorization: `Bearer ${accessToken}` });
      return await client.callTool(tool, args);
    }
    throw err;
  }
}

迁移：从 SSE 迁移到 Streamable HTTP

服务端变更

// OLD (SSE) - Two endpoints
app.get("/sse", async (req, res) => {
  const transport = new SSEServerTransport("/sse/messages", res);
  await server.connect(transport);
});
app.post("/sse/messages", async (req, res) => { /* ... */ });

// NEW (Streamable HTTP) - Single endpoint
app.all("/mcp", async (req, res) => {
  const transport = new StreamableHTTPServerTransport();
  await server.connect(transport);
  await transport.handleRequest(req, res);
});

客户端变更

// OLD
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
const transport = new SSEClientTransport(new URL("http://localhost:3000/sse"));

// NEW
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const transport = new StreamableHTTPClientTransport(new URL("http://localhost:3000/mcp"));

向后兼容的服务端

// Support both transports during migration
app.all("/mcp", /* new Streamable HTTP handler */);
app.get("/sse", /* legacy SSE handler */);
app.post("/sse/messages", /* legacy message handler */);

客户端配置

Claude Desktop / Claude Code（stdio）

{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["/path/to/server.js"],
      "env": { "API_KEY": "secret" }
    }
  }
}

远程服务（Streamable HTTP）

{
  "mcpServers": {
    "remote": {
      "type": "streamable-http",
      "url": "https://mcp.example.com/mcp"
    }
  }
}

错误处理

返回工具错误（而非协议错误），以便模型自行纠正：

TypeScript：

server.registerTool("query", { /* ... */ }, async ({ sql }) => {
  if (sql.includes("DROP")) {
    return {
      isError: true,
      content: [{ type: "text", text: "Destructive queries not allowed" }]
    };
  }
  // ...
});

Python： 在工具中抛出异常——异常会被捕获并作为错误返回：

@mcp.tool()
def query(sql: str) -> dict:
    if "DROP" in sql.upper():
        raise ValueError("Destructive queries not allowed")
    return {"result": "data"}

常见错误

错误	修正方法
为新服务使用 SSE	使用 Streamable HTTP
在 stdio 服务中向 stdout 写入内容	使用 stderr 记录日志
初始化后缺少会话 ID	始终包含 Mcp-Session-Id 标头
未验证 OAuth 受众	验证令牌 aud 是否与您的服务匹配
将客户端令牌转发到上游	为上游 API 使用单独的凭据

官方资源

SDK

• TypeScript: https://github.com/modelcontextprotocol/typescript-sdk
• Python: https://github.com/modelcontextprotocol/python-sdk
• C#: https://github.com/modelcontextprotocol/csharp-sdk

文档

• 规范: https://modelcontextprotocol.io/specification/2025-11-25
• 传输方式: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports
• 授权: https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization

测试

npx @modelcontextprotocol/inspector node path/to/server.js