MCP SDK安全深度审计报告: 揭示跨语言实现中的共性安全威胁与成熟度差异

云鼎实验室

发布于 2026-04-09 12:00:32

850

文章被收录于专栏：云鼎实验室的专栏云鼎实验室的专栏

1.引言

1.1 背景和分析范围

Model Context Protocol（MCP）旨在标准化大语言模型与外部系统的上下文交互。当前，关于MCP的核心概念与技术背景已有诸多系统性文献阐述，故本文不再重复赘述。

随着MCP在应用层的广泛落地，Client端与Server端均已暴露出多维度问题。然而，现有讨论多聚焦于功能实现与交互逻辑，鲜少关注支撑其运行的SDK层安全性。

本分析基于云鼎实验室对Python、Java、TypeScript、C#及Kotlin五种官方SDK实现的深度安全审计，结合大量具体代码实例，系统解构MCP的安全设计理念与工程实践，旨在揭示潜在安全风险。

本分析重点关注跨语言实现中的共性安全威胁和特有风险，以及性能特性中的安全相关水位。

1.2 核心安全关切

通过对多语言实现的综合分析，识别出三个关键安全域：

反序列化安全：各语言序列化框架的安全配置和潜在风险。
传输层安全：协议通信中的传输层加密/认证等机制。
访问控制安全：客户端-服务端交互中的认证授权和会话管理机制。

1.3 分析方法

本分析采用多维度安全评估方法：

静态代码分析：深入审查每种语言的具体实现代码（截至20250424版本）。
跨语言对比：识别不同实现间的安全特性差异。
实战场景验证：基于现实真实部署环境的安全评估。

1.4 主要发现

🔴 高风险发现

传输层安全缺失：所有 SDK 均未强制使用 HTTPS，缺乏证书验证机制。
认证机制不统一：仅 TypeScript SDK 提供完整 OAuth 2.0 实现，其他语言依赖外部实现。
资源限制不足：普遍缺乏统一的连接限制、请求量限制，部分语言缺乏背压控制。

🟡 中等风险发现

反序列化安全：虽采用类型安全框架，但在处理嵌套数据时仍存在 DoS 风险。
会话管理不完善：会话攻击防护不足，缺少统一的会话管理机制。
内存管理差异：不同语言的资源回收策略差异较大，主动管理策略不足，可能导致内存泄漏。

🟢 安全优势

所有 SDK 均避免使用原生序列化机制。
采用强类型系统和 Schema 验证。
实现了基本的错误处理和输入验证。

安全成熟度排名

TypeScript SDK (8/10) - 最完整的安全实现
C# SDK (7/10) - 优秀的资源管理
Java SDK (5.5/10) - 结构化良好的实现
Python SDK (4.5/10) - 基础实现，安全特性不足
Kotlin SDK (3/10) - 最简单的实现，依赖外部安全

2.安全设计分析

2.1 反序列化安全深度分析

反序列化风险是MCP协议各语言实现中最关键的安全风险点。

通过深入代码审计，我们发现每种语言都有其独特的安全实现模式和潜在风险。

2.1.1 Python SDK：Pydantic类型安全框架

安全机制分析：

Python SDK 采用 Pydantic v2 作为核心序列化框架，提供了强大的类型安全保护：

# python-sdk/src/mcp/types.py
class JSONRPCMessage(
    RootModel[JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError]
):
    pass

    # 所有模型都包含安全配置
class RequestParams(BaseModel):
    class Meta(BaseModel):
        progressToken: ProgressToken | None = None
        model_config = ConfigDict(extra="allow")  # 允许额外字段但类型受限

关键安全特性：

强类型约束：通过 RootModel 严格限制只能反序列化为四种预定义的JSON-RPC消息类型。
Schema 验证：所有输入都必须通过预定义的 Pydantic 模型验证。
异常隔离：反序列化失败时抛出 ValidationError 而非导致程序崩溃。

潜在安全风险：

在工具参数处理中存在二次 JSON 解析的安全隐患：

# python-sdk/src/mcp/server/fastmcp/utilities/func_metadata.py
def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
    for field_name in self.arg_model.model_fields.keys():
        if isinstance(data[field_name], str):
            try:
                pre_parsed = json.loads(data[field_name])  # 存在二次解析风险
                if isinstance(pre_parsed, str | int | float):
                    continue  # 基本类型跳过，但复杂对象会被处理
                new_data[field_name] = pre_parsed
            except json.JSONDecodeError:
                continue
    return new_data

此处的二次解析可能被攻击者利用，攻击者可以构造嵌套 JSON 字符串绕过第一层验证实现注入攻击。
复杂对象（字典、列表）会被直接接受并解析，可能包含恶意内容。
没有对解析后的数据进行深度验证。

2.1.2 Java SDK：Jackson 多阶段反序列化模式

安全机制分析：

Java SDK 使用 Jackson 2.17.0，采用了两阶段反序列化策略提供安全保护：

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java

public static JSONRPCMessage deserializeJsonRpcMessage(ObjectMapper objectMapper, String jsonText)
        throws IOException {
    // 第一阶段：解析为安全的 Map 结构
    var map = objectMapper.readValue(jsonText, MAP_TYPE_REF);

    // 第二阶段：基于结构特征进行类型判断
    if (map.containsKey("method") && map.containsKey("id")) {
        return objectMapper.convertValue(map, JSONRPCRequest.class);
    }
    else if (map.containsKey("method") && !map.containsKey("id")) {
        return objectMapper.convertValue(map, JSONRPCNotification.class);
    }
    else if (map.containsKey("result") || map.containsKey("error")) {
        return objectMapper.convertValue(map, JSONRPCResponse.class);
    }

    throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
}

// 类型安全的Record定义
@JsonInclude(JsonInclude.Include.NON_ABSENT)
@JsonIgnoreProperties(ignoreUnknown = true)  // 忽略未知属性但保持类型安全
public record JSONRPCRequest(
    @JsonProperty("jsonrpc") String jsonrpc,
    @JsonProperty("method") String method,
    @JsonProperty("id") Object id,
    @JsonProperty("params") Object params) implements JSONRPCMessage {
}

关键安全特性：

两阶段解析：先解析为通用Map，再根据结构进行类型转换、并在转换前检查。
Record类型安全：大量使用 Java Records 提供不可变数据容器确保不可变性和类型安全。
类型边界控制：使用 @JsonSubTypes 明确限制多态反序列化的类型范围。
注解控制：通过 @JsonIgnoreProperties 和 @JsonTypeInfo 严格控制序列化行为。
密封接口保护：使用 Java 17 的密封接口特性限制继承层次。

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java

@JsonTypeInfo(use = JsonTypeInfo.Id.DEDUCTION, include = As.PROPERTY)
@JsonSubTypes({ @JsonSubTypes.Type(value = TextResourceContents.class, name = "text"),
		@JsonSubTypes.Type(value = BlobResourceContents.class, name = "blob") })
public sealed interface ResourceContents permits TextResourceContents, BlobResourceContents {
    // 密封接口限制了可能的实现类型
}

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
@JsonSubTypes({
    @JsonSubTypes.Type(value = TextContent.class, name = "text"),
    @JsonSubTypes.Type(value = ImageContent.class, name = "image")
})
public sealed interface Content permits TextContent, ImageContent {
    // 密封接口限制了可能的实现类型
}

安全配置检查：

✅ 未使用 enableDefaultTyping() 危险配置
✅ 避免了基于 Id.CLASS 的类型信息处理
✅ 使用sealed interface 限制了可反序列化的类型
✅ 使用了 JsonIgnoreProperties(ignoreUnknown = true) 增强健壮性

潜在安全风险:

虽然Java SDK实现了多层安全机制，但在动态工具调用中的参数解析处理时仍存在潜在的DoS风险：

//java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java
...
private static final TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<>() {};
...

@JsonInclude(JsonInclude.Include.NON_ABSENT)
@JsonIgnoreProperties(ignoreUnknown = true)
public record CallToolRequest(// @formatter:off
@JsonProperty("name") String name,
@JsonProperty("arguments") Map<String, Object> arguments) implements Request {

public CallToolRequest(String name, String jsonArguments) {
this(name, parseJsonArguments(jsonArguments));          
	}
// 第一阶段：解析为Map<String, Object> - ⚠️ 风险点！
private static Map<String, Object> parseJsonArguments(String jsonArguments) {
try {
return OBJECT_MAPPER.readValue(jsonArguments, MAP_TYPE_REF); // ⚠️ 无限制解析
		}
catch (IOException e) {
throw new IllegalArgumentException("Invalid arguments: " + jsonArguments, e);
		}
	}
}
...
// 第二阶段：类型转换 - 已经太晚了！
CallToolRequest request = new CallToolRequest(name, arguments);
...

风险情况：

Map<String, Object> 中的 Object 类型可以容纳任意 Java 对象

此处缺少对嵌套深度、数据大小、数据类型的限制。

问题分析：

第一阶段已经受到攻击: 在到达类型安全检查之前，恶意JSON已经被完全解析到内存中；
资源消耗发生在解析阶段: 不管后续如何处理，大量内存已经被消耗；
Map<String, Object>没有限制: Object可以是任意复杂的嵌套结构；

问题原因：层次问题。安全措施在应用层（类型系统），攻击发生在解析层（JSON处理），两者不在同一个防护层次。

类型边界控制 + 密封接口
- ✅ 有效防止任意类反序列化攻击
- ⚠️ 对数据量攻击的防护有限 - 需要在解析前进行大小限制
Record类型安全
- ✅ 确保数据结构不可变
- ⚠️ Record只保证结构安全，不限制内容大小
注解控制
- ✅ 控制序列化行为
- ⚠️ 不阻止大数据量解析

2.1.3 TypeScript SDK：Zod Schema 运行时验证

安全机制分析：

TypeScript SDK 使用 Zod 提供了严格的运行时类型验证：

// typescript-sdk/src/types.ts
export const JSONRPCMessageSchema = z.union([
  JSONRPCRequestSchema,
  JSONRPCNotificationSchema,
  JSONRPCResponseSchema,
  JSONRPCErrorSchema,
]);

export const JSONRPCRequestSchema = z
  .object({
    jsonrpc: z.literal(JSONRPC_VERSION),
    id: RequestIdSchema,
  })
  .merge(RequestSchema)
  .strict();  // 严格模式：不允许额外属性

  // 工具Schema验证
export const ToolSchema = z
  .object({
    name: z.string(),
    description: z.optional(z.string()),
    inputSchema: z
      .object({
        type: z.literal("object"),
        properties: z.optional(z.object({}).passthrough()),
      })
      .passthrough(),
  })
  .passthrough();

//typescript-sdk/src/server/streamableHttp.ts

const body = await getRawBody(req, {
  limit: MAXIMUM_MESSAGE_SIZE,
  encoding: parsedCt.parameters.charset ?? "utf-8",
});
rawMessage = JSON.parse(body.toString());

关键安全特性：

编译时+运行时验证：结合TypeScript静态检查和Zod运行时验证；
严格模式Schema：使用 .strict() 禁止未定义的额外属性；
联合类型控制：通过联合类型严格限制消息格式提供强类型检查和结构验证；
消息大小限制：限制请求体大小为4MB，防止DoS攻击；

与其他语言SDK相比，依赖Zod验证，相对简单但有效。

2.1.4 C# SDK：源生成序列化器的安全使用

安全机制分析：

C# SDK 使用 System.Text.Json 和源生成器实现了高性能的类型安全序列化：

// csharp-sdk/src/ModelContextProtocol/Utils/Json/McpJsonUtilities.cs

[JsonSourceGenerationOptions(JsonSerializerDefaults.Web,
    DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,  // 安全的默认忽略策略
    NumberHandling = JsonNumberHandling.AllowReadingFromString)]   // 允许字符串到数字转换

// 预定义所有可序列化类型，防止任意类型反序列化
[JsonSerializable(typeof(JsonRpcMessage))]
[JsonSerializable(typeof(JsonRpcRequest))]
[JsonSerializable(typeof(JsonRpcNotification))]
// ... 完整的类型白名单

internal sealed partial class JsonContext : JsonSerializerContext;

public static JsonSerializerOptions DefaultOptions { get; } = CreateDefaultOptions();

private static JsonSerializerOptions CreateDefaultOptions()
{
    JsonSerializerOptions options = new(JsonContext.Default.Options);
    options.TypeInfoResolverChain.Add(AIJsonUtilities.DefaultOptions.TypeInfoResolver!);
    options.MakeReadOnly();  // 锁定配置防止运行时修改
    return options;
}

关键安全特性：

源生成序列化器：编译时生成序列化代码，避免反射漏洞；
类型白名单：通过 JsonSerializable 属性明确定义、严格控制可序列化类型；
只读配置：MakeReadOnly() 防止运行时配置被恶意修改；
无动态类型加载机制：没有用到Type.GetType或Assembly.Load等危险动态加载。

潜在安全风险:

///csharp-sdk/src/ModelContextProtocol/Utils/Json/McpJsonUtilities.cs

internal static bool IsValidMcpToolSchema(JsonElement element)
{
    if (element.ValueKind is not JsonValueKind.Object) return false;

    foreach (JsonProperty property in element.EnumerateObject())
    {
        if (property.NameEquals("type"))
        {
            if (property.Value.ValueKind is not JsonValueKind.String ||
                !property.Value.ValueEquals("object"))
                return false;
            return true; // No need to check other properties
// 此处仅检查type属性就返回，验证不充分
        }
    }
    return false;
}

仅检查type属性就返回，验证不够充分。从注释看开发者认为没必要再检查其他

但实际上，由于验证不充分，可能导致：

无效的工具模式被接受并注册
客户端尝试调用工具时发生运行时错误
潜在的安全漏洞，特别是在处理用户提供的工具定义时

2.1.5 Kotlin SDK：kotlinx.serialization 的类型安全设计

安全机制分析：

Kotlin SDK 使用 kotlinx.serialization，这是专为 Kotlin 设计的类型安全序列化库：

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/Protocol.kt

@OptIn(ExperimentalSerializationApi::class)
public val McpJson: Json by lazy {
    Json {
        ignoreUnknownKeys = true      // 忽略未知键：潜在安全风险
        encodeDefaults = true
        isLenient = true              // 宽松模式：接受格式不严格的JSON
        classDiscriminatorMode = ClassDiscriminatorMode.NONE
        explicitNulls = false
    }
}

// 类型安全的密封类定义
@Serializable(with = JSONRPCMessagePolymorphicSerializer::class)
public sealed interface JSONRPCMessage

@Serializable
public data class JSONRPCRequest(
    val jsonrpc: String = "2.0",
    val id: RequestId,
    val method: String,
    val params: JsonElement? = null,
) : JSONRPCMessage

关键安全特性：

密封类型控制：广泛使用sealed interface密封类和密封接口限制类型边界；
注解序列化：使用 @Serializable 注解明确标记可序列化类型；
多态序列化器：通过 JsonContentPolymorphicSerializer 安全处理多态类型。

潜在安全风险：

宽松的JSON配置模式和宽松解析可能导致安全问题：

// isLenient = true 允许以下危险格式：
// - 尾随逗号：{"key": "value",}
// - 单引号：{'key': 'value'}
// - 未引号键：{key: "value"}


// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/ReadBuffer.kt
// ignoreUnknownKeys = true 可能忽略安全相关字段
internal fun deserializeMessage(line: String): JSONRPCMessage {
    return McpJson.decodeFromString<JSONRPCMessage>(line)// 宽松解析可能接受恶意格式
}

配置 isLenient = true 提高了代码兼容性

但也可能导致接受格式不严格的 JSON 输入，结合特定业务逻辑则存在被利用的风险。

2.2 传输层安全威胁分析

2.2.1 HTTPS 强制策略的跨语言差异

Python SDK：

# python-sdk/src/mcp/client/sse.py
@asynccontextmanager
async def sse_client(
    url: str,
    headers: dict[str, Any] | None = None,
    timeout: float = 5,
    sse_read_timeout: float = 60 * 5,
):
    ...
    async with anyio.create_task_group() as tg:
        try:
            logger.info(f"Connecting to SSE endpoint: {remove_request_params(url)}")
            async with httpx.AsyncClient(headers=headers) as client:
	...

客户端使用httpx.AsyncClient()默认配置

未显式配置SSL验证参数(verify=True/False)

允许使用 HTTP 进行明文通信

# python-sdk/src/mcp/server/sse.py

class SseServerTransport:
    def __init__(self, endpoint: str) -> None:
        super().__init__()
        self._endpoint = endpoint
        self._read_stream_writers = {}
        logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")

    @asynccontextmanager
    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] != "http":
            logger.error("connect_sse received non-HTTP request")
            raise ValueError("connect_sse can only handle HTTP requests")

服务端仅检查请求类型为HTTP，未区分HTTP/HTTPS

没有协议安全性验证，接受所有HTTP协议的连接

Java SDK：

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/util/Utils.java

public static URI resolveUri(URI baseUrl, String endpointUrl) {
    URI endpointUri = URI.create(endpointUrl);
    if (endpointUri.isAbsolute() && !isUnderBaseUri(baseUrl, endpointUri)) {
        throw new IllegalArgumentException("Absolute endpoint URL does not match the base URL.");
    }
    return baseUrl.resolve(endpointUri);
}

private static boolean isUnderBaseUri(URI baseUri, URI endpointUri) {
    // 验证 scheme 和 authority，但未强制要求 HTTPS
    return baseUri.getScheme().equals(endpointUri.getScheme()) &&
           baseUri.getAuthority().equals(endpointUri.getAuthority());
}

Java SDK 提供了 URI 验证 scheme 和 authority，但未验证请求类型，没有强制要求使用HTTPS。

TypeScript SDK：

// typescript-sdk/src/server/auth/router.ts

export function mcpAuthRouter(options: AuthRouterOptions): RequestHandler {
  const issuer = options.issuerUrl;
  const baseUrl = options.baseUrl;

    if (issuer.protocol !== "https:" && issuer.hostname !== "localhost" && issuer.hostname !== "127.0.0.1") {
    throw new Error("Issuer URL must be HTTPS");
  }
  if (issuer.hash) {
    throw new Error("Issuer URL must not have a fragment");
  }
  if (issuer.search) {
    throw new Error("Issuer URL must not have a query string");
  }
...
}

TypeScript SDK 仅在 OAuth 场景中强制 HTTPS，其他场景不做强制要求。

C# SDK：

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseClientTransportOptions.cs


public record SseClientTransportOptions
{
...
if (value.Scheme != Uri.UriSchemeHttp && value.Scheme != Uri.UriSchemeHttps)
	{
throw new ArgumentException("Endpoint must use HTTP or HTTPS scheme.", nameof(value));
	}
...

必须使用HTTP或HTTPS协议，但不强制HTTPS。

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseClientTransport.cs

public sealed class SseClientTransport : IClientTransport, IAsyncDisposable
{
...
    public SseClientTransport(SseClientTransportOptions transportOptions, ILoggerFactory? loggerFactory = null)
        : this(transportOptions, new HttpClient(), loggerFactory, ownsHttpClient: true)
    {
    }
...

使用标准HttpClient，意味着继承 .NET 框架的所有默认行为

包括默认的证书验证、TLS协议协商、安全头处理等。

Kotlin SDK：

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/SSEServerTransport.kt

public suspend fun handlePostMessage(call: ApplicationCall) {
    ...
// 只检查内容类型，不检查连接安全性
    val ct = call.request.contentType()
    if (ct != ContentType.Application.Json) {
        error("Unsupported content-type: $ct")
    }
...

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/SSEClientTransport.kt

public class SseClientTransport(
    private val client: HttpClient,
    private val urlString: String?, // 接受任意协议
    private val reconnectionTime: Duration? = null,
    private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
) : AbstractTransport() {

...
"endpoint" -> {
try {
val eventData = event.data ?: ""

// 简单地拼接URL，没有协议验证
val maybeEndpoint = Url(baseUrl + eventData)
...

不但没有HTTP/HTTPS相关检查，也没有协议验证，可以是非HTTP协议，错误情况交由后续代码处理。

2.2.2 证书校验机制的普遍缺失

所有五种语言实现，都缺乏明确的证书校验机制，如以下特性：

未实现证书固定（Certificate Pinning）
缺少自定义 CA 证书支持和验证
未限制 TLS 版本
未限制密码套件可选范围
没有证书吊销列表（CRL）检查
依赖底层库的默认行为，可能接受自签名证书

2.3 认证授权与会话管理安全分析

首先需要明确，OAuth（认证与授权）和会话管理的本质区别:

OAuth（认证与授权）目的：确定"你是谁"以及"你能做什么。

// OAuth处理的问题：
// 1. 客户端如何证明自己的身份？
// 2. 用户如何安全地授权客户端？
// 3. 客户端如何获得访问权限？

// OAuth流程示例
const authUrl = await startAuthorization(serverUrl, {
    clientId: "mcp-client-123",
    scope: "read:resources write:tools",
    codeChallenge: "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
});

// 用户授权后，交换访问令牌
const tokens = await exchangeAuthorization(serverUrl, {
    authorizationCode: "auth_code_xyz",
    codeVerifier: "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
});

会话管理（状态维护）目的：在认证之后，维护用户的连接状态和上下文。

// 会话管理处理的问题：
// 1. 如何跟踪已认证用户的活动？
// 2. 如何处理会话超时？
// 3. 如何防止会话劫持？

public class HttpMcpSession {
    public string SessionId { get; }
    public DateTime LastActivity { get; set; }
    public ClaimsPrincipal User { get; }
    public bool IsExpired => DateTime.Now - LastActivity > TimeSpan.FromHours(2);

    public bool ValidateOwnership(ClaimsPrincipal currentUser) {
        return User.FindFirst(ClaimTypes.NameIdentifier)?.Value == 
               currentUser.FindFirst(ClaimTypes.NameIdentifier)?.Value;
    }
}

2.3.1 OAuth 实现的安全分析

TypeScript SDK：最完整的 OAuth 2.0 实现

授权码实现

// typescript-sdk/src/server/auth/handlers/authorize.ts
await provider.authorize(client, {
        state,
        scopes: requestedScopes,
        redirectUri: redirect_uri,
        codeChallenge: code_challenge,
      }, res);
    } catch (error) {
      // Post-redirect errors - redirect with error parameters
      if (error instanceof OAuthError) {
        res.redirect(302, createErrorRedirect(redirect_uri, error, state));
      } else {
        console.error("Unexpected error during authorization:", error);
        const serverError = new ServerError("Internal Server Error");
        res.redirect(302, createErrorRedirect(redirect_uri, serverError, state));
      }
    }
  });

令牌交换实现

// typescript-sdk/src/server/auth/handlers/token.ts
switch (grant_type) {
  case "authorization_code": {
    const parseResult = AuthorizationCodeGrantSchema.safeParse(req.body);
    if (!parseResult.success) {
      throw new InvalidRequestError(parseResult.error.message);
    }

      const { code, code_verifier } = parseResult.data;

      const skipLocalPkceValidation = provider.skipLocalPkceValidation;

    // Perform local PKCE validation unless explicitly skipped 
    // (e.g. to validate code_verifier in upstream server)
    if (!skipLocalPkceValidation) {
      const codeChallenge = await provider.challengeForAuthorizationCode(client, code);
      if (!(await verifyChallenge(code_verifier, codeChallenge))) {
        throw new InvalidGrantError("code_verifier does not match the challenge");
      }
    }

    // Passes the code_verifier to the provider if PKCE validation didn't occur locally
    const tokens = await provider.exchangeAuthorizationCode(client, code, skipLocalPkceValidation ? code_verifier : undefined);
    res.status(200).json(tokens);
    break;
  }

令牌刷新实现

// typescript-sdk/src/server/auth/handlers/token.ts
case "refresh_token": {
  const parseResult = RefreshTokenGrantSchema.safeParse(req.body);
  if (!parseResult.success) {
    throw new InvalidRequestError(parseResult.error.message);
  }

  const { refresh_token, scope } = parseResult.data;

  const scopes = scope?.split(" ");
  const tokens = await provider.exchangeRefreshToken(client, refresh_token, scopes);
  res.status(200).json(tokens);
  break;
}

// typescript-sdk/src/client/auth.ts
export async function refreshAuthorization(
...
): Promise<OAuthTokens> {
  const grantType = "refresh_token";

  let tokenUrl: URL;
  if (metadata) {
    tokenUrl = new URL(metadata.token_endpoint);

    if (
      metadata.grant_types_supported &&
      !metadata.grant_types_supported.includes(grantType)
    ) {
      throw new Error(
        `Incompatible auth server: does not support grant type ${grantType}`,
      );
    }
  } else {
    tokenUrl = new URL("/token", serverUrl);
  }
...

令牌撤销实现

// typescript-sdk/src/server/auth/handlers/revoke.ts
router.post("/", async (req, res) => {
  res.setHeader('Cache-Control', 'no-store');

  try {
    const parseResult = OAuthTokenRevocationRequestSchema.safeParse(req.body);
    if (!parseResult.success) {
      throw new InvalidRequestError(parseResult.error.message);
    }

    const client = req.client;
    if (!client) {
      // This should never happen
      console.error("Missing client information after authentication");
      throw new ServerError("Internal Server Error");
    }

    await provider.revokeToken!(client, parseResult.data);
    res.status(200).json({});
  } catch (error) {
    if (error instanceof OAuthError) {
      const status = error instanceof ServerError ? 500 : 400;
      res.status(status).json(error.toResponseObject());
    } else {
      console.error("Unexpected error revoking token:", error);
      const serverError = new ServerError("Internal Server Error");
      res.status(500).json(serverError.toResponseObject());
    }
  }
});

客户端动态注册实现

// typescript-sdk/src/server/auth/handlers/register.ts
router.post("/", async (req, res) => {
  res.setHeader('Cache-Control', 'no-store');

    try {
    const parseResult = OAuthClientMetadataSchema.safeParse(req.body);
    if (!parseResult.success) {
      throw new InvalidClientMetadataError(parseResult.error.message);
    }

    const clientMetadata = parseResult.data;
    const isPublicClient = clientMetadata.token_endpoint_auth_method === 'none'

    // Generate client credentials
    const clientId = crypto.randomUUID();
    const clientSecret = isPublicClient
      ? undefined
      : crypto.randomBytes(32).toString('hex');
    const clientIdIssuedAt = Math.floor(Date.now() / 1000);

    // Calculate client secret expiry time
    const clientsDoExpire = clientSecretExpirySeconds > 0
    const secretExpiryTime = clientsDoExpire ? clientIdIssuedAt + clientSecretExpirySeconds : 0
    const clientSecretExpiresAt = isPublicClient ? undefined : secretExpiryTime

    let clientInfo: OAuthClientInformationFull = {
      ...clientMetadata,
      client_id: clientId,
      client_secret: clientSecret,
      client_id_issued_at: clientIdIssuedAt,
      client_secret_expires_at: clientSecretExpiresAt,
    };

    clientInfo = await clientsStore.registerClient!(clientInfo);
    res.status(201).json(clientInfo);

中间件安全验证

// typescript-sdk/src/server/auth/middleware/bearerAuth.ts

// Check if token has the required scopes (if any)
if (requiredScopes.length > 0) {
  const hasAllScopes = requiredScopes.every(scope =>
    authInfo.scopes.includes(scope)
  );

  if (!hasAllScopes) {
    throw new InsufficientScopeError("Insufficient scope");
  }
}

// Check if the token is expired
if (!!authInfo.expiresAt && authInfo.expiresAt < Date.now() / 1000) {
  throw new InvalidTokenError("Token has expired");
}

完整的OAuth路由

// typescript-sdk/src/server/auth/router.ts

export function mcpAuthRouter(options: AuthRouterOptions): RequestHandler {
  const issuer = options.issuerUrl;
  const baseUrl = options.baseUrl;

  // Technically RFC 8414 does not permit a localhost HTTPS exemption, but this will be necessary for ease of testing
  if (issuer.protocol !== "https:" && issuer.hostname !== "localhost" && issuer.hostname !== "127.0.0.1") {
    throw new Error("Issuer URL must be HTTPS");
  }
  if (issuer.hash) {
    throw new Error("Issuer URL must not have a fragment");
  }
  if (issuer.search) {
    throw new Error("Issuer URL must not have a query string");
  }

  const authorization_endpoint = "/authorize";
  const token_endpoint = "/token";
  const registration_endpoint = options.provider.clientsStore.registerClient ? "/register" : undefined;
  const revocation_endpoint = options.provider.revokeToken ? "/revoke" : undefined;

  const metadata = {
    issuer: issuer.href,
    service_documentation: options.serviceDocumentationUrl?.href,

    authorization_endpoint: new URL(authorization_endpoint, baseUrl || issuer).href,
    response_types_supported: ["code"],
    code_challenge_methods_supported: ["S256"],

    token_endpoint: new URL(token_endpoint, baseUrl || issuer).href,
    token_endpoint_auth_methods_supported: ["client_secret_post"],
    grant_types_supported: ["authorization_code", "refresh_token"],

    revocation_endpoint: revocation_endpoint ? new URL(revocation_endpoint, baseUrl || issuer).href : undefined,
    revocation_endpoint_auth_methods_supported: revocation_endpoint ? ["client_secret_post"] : undefined,

    registration_endpoint: registration_endpoint ? new URL(registration_endpoint, baseUrl || issuer).href : undefined,
  };

  const router = express.Router();

  router.use(
    authorization_endpoint,
    authorizationHandler({ provider: options.provider, ...options.authorizationOptions })
  );

  router.use(
    token_endpoint,
    tokenHandler({ provider: options.provider, ...options.tokenOptions })
  );

  router.use("/.well-known/oauth-authorization-server", metadataHandler(metadata));

  if (registration_endpoint) {
    router.use(
      registration_endpoint,
      clientRegistrationHandler({
        clientsStore: options.provider.clientsStore,
        ...options,
      })
    );
  }

  if (revocation_endpoint) {
    router.use(
      revocation_endpoint,
      revocationHandler({ provider: options.provider, ...options.revocationOptions })
    );
  }

  return router;
}

潜在安全风险

// typescript-sdk/src/server/auth/handlers/token.ts
export function tokenHandler({ provider, rateLimit: rateLimitConfig }: TokenHandlerOptions): RequestHandler {
...
// Configure CORS to allow any origin, to make accessible to web-based MCP clients
router.use(cors());
//- 风险：可能遭受跨域攻击
//- 设计考虑：为支持Web-based MCP客户端的便利性
//- 建议：生产环境应配置具体的允许源列表

// typescript-sdk/src/server/auth/handlers/register.ts
export function clientRegistrationHandler({
...
// Configure CORS to allow any origin, to make accessible to web-based MCP clients
router.use(cors());


// typescript-sdk/src/server/auth/handlers/revoke.ts
export function revocationHandler({ provider, rateLimit: rateLimitConfig }: RevocationHandlerOptions): RequestHandler {
...
// Configure CORS to allow any origin, to make accessible to web-based MCP clients
router.use(cors());

CORS 过度开放：cors() 无参数调用允许任意源访问。
CSRF/CSP 未实现：未见 CSRF/CSP 的相关实现。也没有预留接口，需要使用者在上层应用中自行实现。
OAuth 高级特性缺失：令牌绑定机制（绑定设备/会话等）、强制令牌加密存储等高级特性未实现。

其他语言的认证缺失

Python SDK：无内置统一认证机制，依赖传输层安全；

Java SDK：无内置统一认证机制，依赖外部实现提供认证支持；

C# SDK：无内置统一认证机制，依赖外部HttpClient配置；

Kotlin SDK：无内置统一认证机制，需要在应用层实现认证逻辑。

2.3.2 会话管理安全分析

Python SDK：简单的实现

基础会话管理

# python-sdk/src/mcp/shared/session.py
class BaseSession(Generic[...]):
    def __init__(self, ...):
        self._response_streams = {}
        self._request_id = 0
        self._in_flight = {}  # 正在处理的请求

    async def send_request(self, request: SendRequestT, result_type: type[ReceiveResultT]) -> ReceiveResultT:
        request_id = self._request_id
        self._request_id = request_id + 1

        # 创建响应流
        response_stream, response_stream_reader = anyio.create_memory_object_stream[JSONRPCResponse | JSONRPCError](1)
        self._response_streams[request_id] = response_stream

        # 发送请求
        jsonrpc_request = JSONRPCRequest(
            jsonrpc="2.0",
            id=request_id,
            **request.model_dump(by_alias=True, mode="json", exclude_none=True),
        )
        await self._write_stream.send(JSONRPCMessage(jsonrpc_request))

请求取消机制

# 处理取消通知
if isinstance(notification.root, CancelledNotification):
    cancelled_id = notification.root.params.requestId
    if cancelled_id in self._in_flight:
        await self._in_flight[cancelled_id].cancel()

潜在安全问题

缺少认证机制：Python SDK没有内置的认证机制；
请求ID可预测：使用简单的递增计数器，容易被预测；
没有会话隔离：多个会话之间没有明确的隔离机制；
缺少加密传输：依赖传输层安全，没有应用层加密。

Java SDK：更加结构化

会话管理器实现

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransportProvider.java
public class HttpServletSseServerTransportProvider extends HttpServlet implements McpServerTransportProvider {
    // 活跃会话映射
    private final Map<String, McpServerSession> sessions = new ConcurrentHashMap<>();

    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response) {
        String sessionId = UUID.randomUUID().toString();

        // 创建新会话
        McpServerSession session = sessionFactory.create(sessionTransport);
        this.sessions.put(sessionId, session);

        // 发送端点事件
        this.sendEvent(writer, ENDPOINT_EVENT_TYPE, 
            this.baseUrl + this.messageEndpoint + "?sessionId=" + sessionId);
    }

    @Override
    protected void doPost(HttpServletRequest request, HttpServletResponse response) {
        String sessionId = request.getParameter("sessionId");
        if (sessionId == null) {
            response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
            String jsonError = objectMapper.writeValueAsString(
                new McpError("Session ID missing in message endpoint"));
            writer.write(jsonError);
            return;
        }

        McpServerSession session = sessions.get(sessionId);
        if (session == null) {
            response.setStatus(HttpServletResponse.SC_NOT_FOUND);
            String jsonError = objectMapper.writeValueAsString(
                new McpError("Session not found: " + sessionId));
            writer.write(jsonError);
            return;
        }
    }
}

会话状态管理

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java
public class McpServerSession implements McpSession {
    private static final int STATE_UNINITIALIZED = 0;
    private static final int STATE_INITIALIZING = 1;
    private static final int STATE_INITIALIZED = 2;

    private final AtomicInteger state = new AtomicInteger(STATE_UNINITIALIZED);
    private final AtomicReference<McpSchema.ClientCapabilities> clientCapabilities = new AtomicReference<>();

    private Mono<McpSchema.JSONRPCResponse> handleIncomingRequest(McpSchema.JSONRPCRequest request) {
        if (McpSchema.METHOD_INITIALIZE.equals(request.method())) {
            this.state.lazySet(STATE_INITIALIZING);
            // 处理初始化
        } else {
            if (this.state.get() != STATE_INITIALIZED) {
                throw new RuntimeException("Received request before initialization was complete");
            }
        }
    }
}

潜在安全问题

会话ID通过URL参数传递：容易被日志记录或泄露；
缺少会话清理机制：没有自动清理过期会话；
并发安全问题：虽然使用了ConcurrentHashMap，但会话状态更新可能存在竞态条件；
没有防CSRF机制：缺少CSRF token验证。

TypeScript SDK：实现多种会话认证机制

OAuth认证机制

// typescript-sdk/src/server/auth/handlers/token.ts
// Token处理器中的PKCE验证
if (!skipLocalPkceValidation) {
    const codeChallenge = await provider.challengeForAuthorizationCode(client, code);
    if (!(await verifyChallenge(code_verifier, codeChallenge))) {
        throw new InvalidGrantError("code_verifier does not match the challenge");
    }
}

Bearer Token验证

// typescript-sdk/src/server/auth/middleware/bearerAuth.ts
export function requireBearerAuth({ provider, requiredScopes = [] }: BearerAuthMiddlewareOptions): RequestHandler {
    return async (req, res, next) => {
        const authHeader = req.headers.authorization;
        if (!authHeader) {
            throw new InvalidTokenError("Missing Authorization header");
        }

        const [type, token] = authHeader.split(' ');
        if (type.toLowerCase() !== 'bearer' || !token) {
            throw new InvalidTokenError("Invalid Authorization header format, expected 'Bearer TOKEN'");
        }

        const authInfo = await provider.verifyAccessToken(token);

        // 检查token是否过期
        if (!!authInfo.expiresAt && authInfo.expiresAt < Date.now() / 1000) {
            throw new InvalidTokenError("Token has expired");
        }
    };
}

HTTP会话管理

// typescript-sdk/src/server/streamableHttp.ts
// 会话ID验证
private validateSession(req: IncomingMessage, res: ServerResponse): boolean {
    const sessionId = req.headers["mcp-session-id"];

    if (!sessionId) {
        res.writeHead(400).end(JSON.stringify({
            jsonrpc: "2.0",
            error: {
                code: -32000,
                message: "Bad Request: Mcp-Session-Id header is required"
            }
        }));
        return false;
    }

    if (sessionId !== this.sessionId) {
        res.writeHead(404).end(JSON.stringify({
            jsonrpc: "2.0",
            error: {
                code: -32001,
                message: "Session not found"
            }
        }));
        return false;
    }

    return true;
}

潜在安全问题

会话固定攻击风险：会话ID使用UUID生成，但没有在认证后重新生成；
缺少会话超时机制：没有实现会话空闲超时或绝对超时；
Token存储安全性：客户端Token存储依赖于实现者，没有强制安全存储；
缺少防重放攻击机制：没有使用nonce或时间戳验证。

C# SDK：高度依赖ASP.NET Core框架

会话管理与用户关联

// csharp-sdk/src/ModelContextProtocol.AspNetCore/HttpMcpSession.cs
internal sealed class HttpMcpSession<TTransport> : IAsyncDisposable {
    public string Id { get; } = sessionId;
    public (string Type, string Value, string Issuer)? UserIdClaim { get; } = GetUserIdClaim(user);
    public long LastActivityTicks { get; private set; } = timeProvider.GetTimestamp();

    public bool HasSameUserId(ClaimsPrincipal user)
        => UserIdClaim == GetUserIdClaim(user);

    private static (string Type, string Value, string Issuer)? GetUserIdClaim(ClaimsPrincipal user) {
        if (user?.Identity?.IsAuthenticated != true) {
            return null;
        }

        var claim = user.FindFirst(ClaimTypes.NameIdentifier) 
            ?? user.FindFirst("sub") 
            ?? user.FindFirst(ClaimTypes.Upn);

        return claim is { } idClaim 
            ? (idClaim.Type, idClaim.Value, idClaim.Issuer) 
            : null;
    }
}

会话超时和清理

// csharp-sdk/src/ModelContextProtocol.AspNetCore/IdleTrackingBackgroundService.cs
protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
    var idleTimeoutTicks = options.Value.IdleTimeout.Ticks;
    var maxIdleSessionCount = options.Value.MaxIdleSessionCount;

    while (!stoppingToken.IsCancellationRequested && await timer.WaitForNextTickAsync(stoppingToken)) {
        var idleActivityCutoff = timeProvider.GetTimestamp() - idleTimeoutTicks;

        foreach (var (_, session) in handler.Sessions) {
            if (session.LastActivityTicks < idleActivityCutoff) {
                RemoveAndCloseSession(session.Id);
                continue;
            }

            idleSessions.Add((session.LastActivityTicks, session.Id));

            if (idleSessions.Count > maxIdleSessionCount) {
                // 清理最老的会话
                var sessionsToPrune = idleSessions.ToArray()[..^maxIdleSessionCount];
                foreach (var (_, id) in sessionsToPrune) {
                    RemoveAndCloseSession(id);
                }
            }
        }
    }
}

安全的会话ID生成

// csharp-sdk/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs
internal static string MakeNewSessionId() {
    Span<byte> buffer = stackalloc byte[16];
    RandomNumberGenerator.Fill(buffer);
    return WebEncoders.Base64UrlEncode(buffer);
}

潜在安全问题

相对完善但仍有改进空间：虽然实现了用户验证和会话超时，但缺少会话撤销机制；
会话固定攻击防护不足：认证后没有重新生成会话ID；
缺少会话加密：会话数据在内存中以明文存储；
没有实现会话并发控制：同一用户可以创建多个会话。

Kotlin SDK：实现协程操作

SSE会话管理

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/SSEServerTransport.kt
class SseServerTransport(
    private val endpoint: String,
    private val session: ServerSSESession,
) : AbstractTransport() {
    @OptIn(ExperimentalUuidApi::class)
    public val sessionId: String = Uuid.random().toString()

    private val initialized: AtomicBoolean = AtomicBoolean(false)

    override suspend fun start() {
        if (!initialized.compareAndSet(false, true)) {
            throw error("SSEServerTransport already started!")
        }

        // 发送端点事件
        session.send(
            event = "endpoint",
            data = "${endpoint.encodeURLPath()}?$SESSION_ID_PARAM=${sessionId}",
        )
    }
}

会话存储

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt
internal fun ServerSSESession.mcpSseTransport(
    postEndpoint: String,
    transports: ConcurrentMap<String, SseServerTransport>,
): SseServerTransport {
    val transport = SseServerTransport(postEndpoint, this)
    transports[transport.sessionId] = transport

    logger.info { "New SSE connection established with sessionId: ${transport.sessionId}" }
    return transport
}

internal suspend fun RoutingContext.mcpPostEndpoint(
    transports: ConcurrentMap<String, SseServerTransport>,
) {
    val sessionId: String = call.request.queryParameters["sessionId"]
        ?: run {
            call.respond(HttpStatusCode.BadRequest, "sessionId query parameter is not provided")
            return
        }

    val transport = transports[sessionId]
    if (transport == null) {
        logger.warn { "Session not found for sessionId: $sessionId" }
        call.respond(HttpStatusCode.NotFound, "Session not found")
        return
    }
}

潜在安全问题

会话ID暴露在查询参数中：与Java类似的问题；
缺少会话验证：没有验证会话的有效性或所有权；
没有会话超时：会话可能永久存在内存中；
WebSocket子协议验证不足：只检查协议名称，没有版本验证。

整体评价

第1名：TypeScript SDK

安全评分：8/10

优势：

✅ 完整的OAuth 2.0实现
✅ PKCE防护
✅ Bearer Token验证
✅ 客户端认证
✅ 速率限制
✅ 令牌撤销机制
✅ 作用域管理

不足：

❌ 缺少会话超时机制
❌ 没有会话清理

第2名：C# SDK

安全评分：7/10

优势：

✅ 会话生命周期管理
✅ 用户身份绑定
✅ 自动会话清理
✅ 安全的会话ID生成
✅ 防止会话劫持

不足：

❌ 没有内置OAuth支持
❌ 依赖外部认证
❌ 缺少令牌管理

第3名：Java SDK

安全评分：5.5/10

优势：

✅ 会话状态管理：明确的初始化状态机制
✅ 并发安全：使用ConcurrentHashMap存储会话
✅ 请求超时：支持配置请求超时时间
✅ 结构化设计：清晰的会话生命周期管理

不足：

❌ 会话ID通过URL参数传递（严重安全隐患）
❌ 没有会话清理机制
❌ 缺少用户认证集成
❌ 没有防CSRF保护

第4名：Kotlin SDK

安全评分：4.5/10

优势：

✅ 原子操作：使用原子布尔值防止重复初始化
✅ 协程安全：利用Kotlin协程特性
✅ WebSocket子协议验证：检查MCP协议支持

不足：

❌ 会话ID在查询参数中暴露
❌ 没有会话验证机制
❌ 缺少会话超时
❌ 没有用户关联
❌ 会话可能永久驻留内存

第五名：Python SDK

安全评分：3/10

优势：

✅ 请求取消机制：支持取消正在处理的请求
✅ 简洁的设计：代码清晰易懂

不足：

❌ 完全没有认证机制
❌ 请求ID可预测：简单递增容易被猜测
❌ 没有会话隔离
❌ 缺少任何安全特性
❌ 依赖传输层安全

3.性能与安全的关联分析

3.1 连接资源管理

Python SDK - 依赖 ASGI 服务器

# python-sdk/src/mcp/server/sse.py
class SseServerTransport:
    def __init__(self, endpoint: str):
        self._read_stream_writers: dict[
            UUID, MemoryObjectSendStream[types.JSONRPCMessage | Exception]
        ] = {}
        # 没有连接数限制，可无限创建会话

Java SDK - 依赖Spring Boot 配置

// java-sdk/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxSseServerTransportProvider.java
@Component
public class WebFluxSseServerTransportProvider {
    private final Map<String, McpServerSession> sessions = new ConcurrentHashMap<>();

    // 依赖 Spring Boot 的配置，如：
    // server.tomcat.max-connections=10000
    // server.tomcat.accept-count=100
    // 但没有 MCP 特定的限制
}

TypeScript SDK - 可配置的速率限制

// typescript-sdk/src/server/auth/handlers/token.ts
import { rateLimit, Options as RateLimitOptions } from "express-rate-limit";

if (rateLimitConfig !== false) {
  router.use(rateLimit({
    windowMs: 15 * 60 * 1000, // 15 分钟窗口
    max: 50, // 每个窗口最多 50 个请求
    standardHeaders: true,
    legacyHeaders: false,
    message: new TooManyRequestsError('You have exceeded the rate limit for token requests').toResponseObject(),
    ...rateLimitConfig
  }));
}

C# SDK - 最大会话数限制

// csharp-sdk/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs
public class HttpServerTransportOptions
{
    /// <summary>
    /// The maximum number of idle sessions to keep open.
    /// </summary>
    public int MaxIdleSessionCount { get; set; } = 100;

    /// <summary>
    /// The maximum time a session can be idle before being closed.
    /// </summary>
    public TimeSpan MaxSessionIdleTime { get; set; } = TimeSpan.FromMinutes(30);
}

// csharp-sdk/src/ModelContextProtocol.AspNetCore/IdleTrackingBackgroundService.cs
if (idleSessions.Count > maxIdleSessionCount)
{
    LogMaxSessionIdleCountExceeded(maxIdleSessionCount);

    // 当会话总数超过最大限制后，开始移除最旧的空闲会话
    while (idleSessions.Count > maxIdleSessionCount)
    {
        var (_, id) = idleSessions.Min;
        idleSessions.Remove(idleSessions.Min);
        RemoveAndCloseSession(id);
    }
}

Kotlin SDK - 无限制

// kotlin-sdk/samples/kotlin-mcp-server/src/main/kotlin/Main.kt
val servers = mutableMapOf<String, Server>()

// 简单的 Map 存储，没有任何限制
servers[transport.sessionId] = server

3.2 背压控制机制

Python SDK - 无缓冲区的最强同步背压

一种"要么完全同步，要么完全阻塞"的极端方案，没有用户配置选项。

# python-sdk/src/mcp/server/sse.py
class SseServerTransport:
    @asynccontextmanager
    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):

# 使用内存流，但没有背压控制
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
# anyio.create_memory_object_stream(0)的参数0，表示无缓冲
# 是一种"要么完全同步，要么完全阻塞"的极端方案
# 完全避免数据积压，内存使用可控，可以认为是最严格的背压
# 但可能影响吞吐量，在高并发场景下性能受限

Java SDK - Reactor 背压

使用Spring WebFlux的Reactor框架，内置完善的背压控制，提供缓冲溢出保护。

// java-sdk/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxSseServerTransportProvider.java
// 使用 Reactor 的背压机制
return session.handle(message).flatMap(response -> ServerResponse.ok().build())
    .onErrorResume(error -> {
        logger.error("Error processing message: {}", error.getMessage());
        return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
            .bodyValue(new McpError(error.getMessage()));
    });

TypeScript SDK - 批量消息资风险

无数组大小限制，无并发限制，可导致资源耗尽。

// typescript-sdk/src/server/streamableHttp.ts
// 批量消息处理，可能导致资源耗尽
if (Array.isArray(rawMessage)) {
  messages = rawMessage.map(msg => JSONRPCMessageSchema.parse(msg));
  // 没有对数组大小的限制
} else {
  messages = [JSONRPCMessageSchema.parse(rawMessage)];
}

// 对每个消息进行处理，无并发限制
for (const message of messages) {
  await this.onmessage?.(message);
}

C# SDK - 最完善的自实现

有界通道限制内存，信号量同步控制并发，会话限制自动清理空闲会话。

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseWriter.cs
internal sealed class SseWriter(string? messageEndpoint = null, BoundedChannelOptions? channelOptions = null) : IAsyncDisposable
{	// 有界通道配置
    private readonly Channel<SseItem<JsonRpcMessage?>> _messages = Channel.CreateBounded<SseItem<JsonRpcMessage?>>(channelOptions ?? new BoundedChannelOptions(1)
    {
        SingleReader = true,
        SingleWriter = false,
    });
    private Utf8JsonWriter? _jsonWriter;
    private Task? _writeTask;
    private CancellationToken? _writeCancellationToken;
// 信号量同步
    private readonly SemaphoreSlim _disposeLock = new(1, 1);
...
using var _ = await _disposeLock.LockAsync(cancellationToken).ConfigureAwait(false);


// csharp-sdk/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs
public class HttpServerTransportOptions
{
...
// 会话限制
public int MaxIdleSessionCount { get; set; } = 100;
public TimeSpan MaxSessionIdleTime { get; set; } = TimeSpan.FromMinutes(30);
...
}

Kotlin SDK -缺乏机制

缺乏背压机制，Channel.UNLIMITED可能导致内存溢出。

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt
public class StdioServerTransport(
...
    private val readChannel = Channel<ByteArray>(Channel.UNLIMITED)
    private val writeChannel = Channel<JSONRPCMessage>(Channel.UNLIMITED)

3.3 对象生命周期管理

Python SDK

# python-sdk/src/mcp/server/sse.py
@asynccontextmanager
async def sse_server(scope: Scope, receive: Receive, send: Send):
    try:
        # 资源初始化
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

        yield (read_stream, write_stream)
    finally:
        # 确保资源清理
        await read_stream_writer.aclose()
        await write_stream_reader.aclose()

Java SDK

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java
try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
    while (!isClosing.get()) {
        String line = reader.readLine();
        // 处理消息
    }
} catch (Exception e) {
    logIfNotClosing("Error in inbound processing", e);
} finally {
//释放资源
    isClosing.set(true);
    if (session != null) {
        session.close();
    }
    inboundSink.tryEmitComplete();
}

TypeScript SDK

// typescript-sdk/src/server/streamableHttp.ts
export class StreamableHTTPServerTransport {
  private _eventSource?: EventSource;

  async close(): Promise<void> {
    this._eventSource?.close();
    // 但可能遗漏其他资源，如定时器、监听器等
  }
}

// typescript-sdk/src/client/sse.ts
// 事件监听器泄漏风险
this._eventSource.addEventListener("endpoint", (event: Event) => {
  // 处理事件
});
// 没有在 close() 中移除监听器

C# SDK

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/StreamServerTransport.cs
public abstract class StreamServerTransport : BaseTransport, IAsyncDisposable
{
    protected virtual async ValueTask DisposeAsyncCore()
    {
        if (Interlocked.Exchange(ref _disposed, 1) == 0)
        {
            SetConnected(false);

            // 取消所有操作
            await _shutdownCts.CancelAsync().ConfigureAwait(false);

            // 等待读取任务完成
            if (_readTask != null)
            {
                try
                {
                    await _readTask.ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    // 预期的取消
                }
            }

            // 清理资源
            _shutdownCts.Dispose();
        }
    }
}

Kotlin SDK：

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/Protocol.kt
class Protocol : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = job + Dispatchers.Default

    suspend fun close() {
        job.cancelAndJoin()
        // 自动取消所有子协程
    }
}

3.4 内存资源回收

Python SDK - 依赖运行时自动内存管理

字典容器管理资源

# python-sdk/src/mcp/server/fastmcp/resources/resource_manager.py
class ResourceManager:
    def __init__(self, warn_on_duplicate_resources: bool = True):
        self._resources: dict[str, Resource] = {}
        self._templates: dict[str, ResourceTemplate] = {}

异常时的资源管理

# python-sdk/src/mcp/server/fastmcp/resources/resource_manager.py
async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
    # 资源创建后立即被dict引用，引用计数+1
    if resource := self._resources.get(uri_str):
        return resource

    # 模板创建临时对象
    for template in self._templates.values():
        if params := template.matches(uri_str):
            try:
                return await template.create_resource(uri_str, params)
            except Exception as e:
                # 异常时对象立即释放
                raise ValueError(f"Error creating resource from template: {e}")

FastMCP中的异常处理

# python-sdk/src/mcp/server/fastmcp/server.py
async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
    resource = await self._resource_manager.get_resource(uri)
    if not resource:
        raise ResourceError(f"Unknown resource: {uri}")

    try:
        content = await resource.read()
        return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
    except Exception as e:
        logger.error(f"Error reading resource {uri}: {e}")
        raise ResourceError(str(e))

引用计数提供确定性释放时机
异常处理机制确保资源不泄漏
GIL简化并发内存管理

Java SDK - 对象池化和重用策略 服务器端资源池化

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java
private final CopyOnWriteArrayList<McpServerFeatures.AsyncToolSpecification> tools = new CopyOnWriteArrayList<>();
private final CopyOnWriteArrayList<McpSchema.ResourceTemplate> resourceTemplates = new CopyOnWriteArrayList<>();
private final ConcurrentHashMap<String, McpServerFeatures.AsyncResourceSpecification> resources = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, McpServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();

ConcurrentHashMap池化存储

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java
public class McpServerSession implements McpSession {
    private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
    private final Map<String, RequestHandler<?>> requestHandlers;
    private final Map<String, NotificationHandler> notificationHandlers;
}

线程池重用

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/client/transport/StdioClientTransport.java
public StdioClientTransport(ServerParameters params, ObjectMapper objectMapper) {
    this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
    this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();

    // 线程池重用
    this.inboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "inbound");
    this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "outbound");
    this.errorScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "error");
}

ConcurrentHashMap提供线程安全的对象复用
流式处理减少对象创建
高并发下可能产生hash冲突

TypeScript SDK - 依赖运行时自动内存管理

显式资源清理

// typescript-sdk/src/server/streamableHttp.ts
async close(): Promise<void> {
    // 显式清理引用，帮助GC识别
    this._streamMapping.forEach((response) => {
      response.end();
    });
    this._streamMapping.clear();

    // Clear any pending responses
    this._requestResponseMap.clear();
    this.onclose?.();
}

资源注册与移除

// typescript-sdk/src/server/mcp.ts
resource(name: string, uriOrTemplate: string | ResourceTemplate) {
    const registeredResource: RegisteredResource = {
        // 闭包可能形成循环引用
        remove: () => {
            delete this._registeredResources[uriOrTemplate];
            this.sendResourceListChanged();
        },
        update: (updates) => {
            // 动态更新引用
            if (typeof updates.uri !== "undefined" && updates.uri !== uriOrTemplate) {
                delete this._registeredResources[uriOrTemplate];
                if (updates.uri) this._registeredResources[updates.uri] = registeredResource;
            }
        }
    };
    return registeredResource;
}

协议层连接清理

// typescript-sdk/src/shared/protocol.ts
private _onclose(): void {
    const responseHandlers = this._responseHandlers;
    this._responseHandlers = new Map();
    this._progressHandlers.clear();
    this._transport = undefined;
    this.onclose?.();

    const error = new McpError(ErrorCode.ConnectionClosed, "Connection closed");
    for (const handler of responseHandlers.values()) {
        handler(error);
    }
}

增量GC减少停顿时间
自动处理循环引用
事件循环模型适合I/O密集型场景

C# SDK - 使用stackalloc减少堆分配

栈分配避免堆压力

// csharp-sdk/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs
internal static string MakeNewSessionId()
{
    Span<byte> buffer = stackalloc byte[16];  // 栈分配，避免堆分配
    RandomNumberGenerator.Fill(buffer);
    return WebEncoders.Base64UrlEncode(buffer);
}

对象池复用

// csharp-sdk/src/Common/Polyfills/System/IO/StreamExtensions.cs
static async ValueTask WriteAsyncCore(Stream stream, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)
{
    byte[] array = ArrayPool<byte>.Shared.Rent(buffer.Length);
    try
    {
        buffer.Span.CopyTo(array);
        await stream.WriteAsync(array, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
    }
    finally
    {
        ArrayPool<byte>.Shared.Return(array);  // 对象池回收
    }
}

引用计数 + 异步清理

// csharp-sdk/src/ModelContextProtocol/Shared/NotificationHandlers.cs
private int _refCount = 1;
private TaskCompletionSource<bool>? _disposeTcs;

public async ValueTask DisposeAsync()
{
    lock (_handlers.SyncObj)
    {
        if (--_refCount != 0)
        {
            _disposeTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
        }
    }
}

栈分配零GC压力，避免内存碎片
对象池减少频繁分配/释放开销
精确的引用计数防止过早回收

Kotlin SDK - 协程轻量级并发模型

协程作用域控制

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/WebSocketMcpTransport.kt
@OptIn(ExperimentalAtomicApi::class)
public abstract class WebSocketMcpTransport : AbstractTransport() {
    private val scope by lazy {
        CoroutineScope(session.coroutineContext + SupervisorJob())
    }
    private val initialized: AtomicBoolean = AtomicBoolean(false)
}

协程自动资源管理

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/WebSocketMcpTransport.kt
override suspend fun start() {
    scope.launch(CoroutineName("WebSocketMcpTransport.collect#${hashCode()}")) {
        while (true) {
            val message = try {
                session.incoming.receive()
            } catch (_: ClosedReceiveChannelException) {
                return@launch  // 协程自动清理
            }
            // 处理消息...
        }
    }
}

结构化取消机制

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt
override suspend fun close() {
    if (!initialized.compareAndSet(expectedValue = true, newValue = false)) return

    // 结构化取消
    readingJob?.cancel()
    sendingJob?.cancel()

    readChannel.close()   // 通道自动清理
    writeChannel.close()
    readBuffer.clear()
}