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

# python-sdk/src/mcp/types.pyclass JSONRPCMessage(    RootModel[JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError]):    pass    # 所有模型都包含安全配置class RequestParams(BaseModel):    class Meta(BaseModel):        progressToken: ProgressToken | None = None        model_config = ConfigDict(extra="allow")  # 允许额外字段但类型受限

# python-sdk/src/mcp/server/fastmcp/utilities/func_metadata.pydef 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

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.javapublic 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 {}

// 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 {    // 密封接口限制了可能的实现类型}

//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);...

// typescript-sdk/src/types.tsexport 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.tsconst body = await getRawBody(req, {  limit: MAXIMUM_MESSAGE_SIZE,  encoding: parsedCt.parameters.charset ?? "utf-8",});rawMessage = JSON.parse(body.toString());

// 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();privatestatic JsonSerializerOptions CreateDefaultOptions(){    JsonSerializerOptions options = new(JsonContext.Default.Options);    options.TypeInfoResolverChain.Add(AIJsonUtilities.DefaultOptions.TypeInfoResolver!);    options.MakeReadOnly();  // 锁定配置防止运行时修改    return options;}

///csharp-sdk/src/ModelContextProtocol/Utils/Json/McpJsonUtilities.csinternalstaticboolIsValidMcpToolSchema(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;}

// 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@Serializablepublic data class JSONRPCRequest(    val jsonrpc: String = "2.0",    val id: RequestId,    val method: String,    val params: JsonElement? = null,) : JSONRPCMessage

// isLenient = true 允许以下危险格式：// - 尾随逗号：{"key": "value",}// - 单引号：{'key': 'value'}// - 未引号键：{key: "value"}// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/ReadBuffer.kt// ignoreUnknownKeys = true 可能忽略安全相关字段internal fundeserializeMessage(line: String): JSONRPCMessage {    return McpJson.decodeFromString<JSONRPCMessage>(line)// 宽松解析可能接受恶意格式}

# python-sdk/src/mcp/client/sse.py@asynccontextmanagerasync 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:...

# python-sdk/src/mcp/server/sse.pyclass 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")

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/util/Utils.javapublic 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());}

// typescript-sdk/src/server/auth/router.tsexport 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");  }...}

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseClientTransportOptions.cspublic record SseClientTransportOptions{...if (value.Scheme != Uri.UriSchemeHttp && value.Scheme != Uri.UriSchemeHttps){throw new ArgumentException("Endpoint must use HTTP or HTTPS scheme.", nameof(value));}...

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseClientTransport.cspublic sealed class SseClientTransport : IClientTransport, IAsyncDisposable{...    public SseClientTransport(SseClientTransportOptions transportOptions, ILoggerFactory? loggerFactory = null)        : this(transportOptions, new HttpClient(), loggerFactory, ownsHttpClient: true)    {    }...

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/SSEServerTransport.ktpublic suspend funhandlePostMessage(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.ktpublic 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)...

// 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);    publicboolValidateOwnership(ClaimsPrincipal currentUser) {        return User.FindFirst(ClaimTypes.NameIdentifier)?.Value ==                currentUser.FindFirst(ClaimTypes.NameIdentifier)?.Value;    }}

// typescript-sdk/src/server/auth/handlers/authorize.tsawait 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.tsswitch (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.tscase "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.tsexport 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.tsrouter.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.tsrouter.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 expiredif (!!authInfo.expiresAt && authInfo.expiresAt < Date.now() / 1000) {  throw new InvalidTokenError("Token has expired");}

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

# python-sdk/src/mcp/shared/session.pyclass 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()

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransportProvider.javapublic 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.javapublic 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");            }        }    }}

// 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");    }}

// typescript-sdk/src/server/auth/middleware/bearerAuth.tsexport 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");        }    };}

// 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;}

// csharp-sdk/src/ModelContextProtocol.AspNetCore/HttpMcpSession.csinternal 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();    publicboolHasSameUserId(ClaimsPrincipal user)        => UserIdClaim == GetUserIdClaim(user);    privatestatic (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.csprotectedoverrideasync 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);                }            }        }    }}

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

// kotlin-sdk/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/SSEServerTransport.ktclass 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 funstart() {        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.ktinternal 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    }}

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

// java-sdk/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxSseServerTransportProvider.java@Componentpublic 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/src/server/auth/handlers/token.tsimport { 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  }));}

// csharp-sdk/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cspublic 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.csif (idleSessions.Count > maxIdleSessionCount){    LogMaxSessionIdleCountExceeded(maxIdleSessionCount);    // 当会话总数超过最大限制后，开始移除最旧的空闲会话    while (idleSessions.Count > maxIdleSessionCount)    {        var (_, id) = idleSessions.Min;        idleSessions.Remove(idleSessions.Min);        RemoveAndCloseSession(id);    }}

// kotlin-sdk/samples/kotlin-mcp-server/src/main/kotlin/Main.ktval servers = mutableMapOf<String, Server>()// 简单的 Map 存储，没有任何限制servers[transport.sessionId] = server

# python-sdk/src/mcp/server/sse.pyclass 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/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/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);}

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/SseWriter.csinternalsealedclassSseWriter(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.cspublic class HttpServerTransportOptions{...// 会话限制public int MaxIdleSessionCount { get; set; } = 100;public TimeSpan MaxSessionIdleTime { get; set; } = TimeSpan.FromMinutes(30);...}

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

# python-sdk/src/mcp/server/sse.py@asynccontextmanagerasync 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/mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.javatry (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/src/server/streamableHttp.tsexport 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() 中移除监听器

// csharp-sdk/src/ModelContextProtocol/Protocol/Transport/StreamServerTransport.cspublic abstract class StreamServerTransport : BaseTransport, IAsyncDisposable{    protectedvirtualasync 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/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/Protocol.ktclass Protocol : CoroutineScope {    private val job = SupervisorJob()    override val coroutineContext = job + Dispatchers.Default    suspend funclose() {        job.cancelAndJoin()        // 自动取消所有子协程    }}

# python-sdk/src/mcp/server/fastmcp/resources/resource_manager.pyclass 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.pyasync 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}")

# python-sdk/src/mcp/server/fastmcp/server.pyasync 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))

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.javaprivate 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<>();

// java-sdk/mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.javapublic 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.javapublic 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");}

// typescript-sdk/src/server/streamableHttp.tsasync 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.tsresource(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.tsprivate _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);    }}

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

// csharp-sdk/src/Common/Polyfills/System/IO/StreamExtensions.csstaticasync 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.csprivate int _refCount = 1;private TaskCompletionSource<bool>? _disposeTcs;publicasync ValueTask DisposeAsync(){    lock (_handlers.SyncObj)    {        if (--_refCount != 0)        {            _disposeTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);        }    }}

// 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.ktoverride suspend funstart() {    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.ktoverride suspend funclose() {    if (!initialized.compareAndSet(expectedValue = true, newValue = false)) return    // 结构化取消    readingJob?.cancel()    sendingJob?.cancel()    readChannel.close()   // 通道自动清理    writeChannel.close()    readBuffer.clear()}