mcp sse stateful session in contrib

CODEOWNERS

@@ -365,3 +365,4 @@ extensions/filters/http/oauth2 @derekargueta @mattklein123
 /contrib/network/connection_balance/dlb @mattklein123 @daixiang0
 /contrib/qat/ @giantcroc @soulxu
 /contrib/generic_proxy/ @wbpcode @soulxu @zhaohuabing @rojkov @htuch
+/contrib/mcp_sse_stateful_session/ @jue-yin @UNOWNED

api/BUILD

@@ -76,6 +76,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/http/http_dubbo_transcoder/v3:pkg",
         "//contrib/envoy/extensions/filters/http/golang/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/language/v3alpha:pkg",
+        "//contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/squash/v3:pkg",
         "//contrib/envoy/extensions/filters/http/sxg/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/llm_inference/v3:pkg",
@@ -94,6 +95,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/network/sip_proxy/router/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/sip_proxy/tra/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/sip_proxy/v3alpha:pkg",
+        "//contrib/envoy/extensions/http/mcp_sse_stateful_session/envelope/v3alpha:pkg",
         "//contrib/envoy/extensions/matching/input_matchers/hyperscan/v3alpha:pkg",
         "//contrib/envoy/extensions/network/connection_balance/dlb/v3alpha:pkg",
         "//contrib/envoy/extensions/private_key_providers/cryptomb/v3alpha:pkg",

api/contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha/BUILD

@@ -0,0 +1,12 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "//envoy/config/core/v3:pkg",
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+    ],
+)

api/contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha/mcp_sse_stateful_session.proto

@@ -0,0 +1,46 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.http.mcp_sse_stateful_session.v3alpha;
+
+import "envoy/config/core/v3/extension.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.http.mcp_sse_stateful_session.v3alpha";
+option java_outer_classname = "McpSseStatefulSessionProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Model Context Protocol(MCP) server-side events(SSE) Stateful session filter]
+// MCP SSE Stateful session :ref:`configuration overview <config_http_filters_mcp_sse_stateful_session>`.
+// [#extension: envoy.filters.http.mcp_sse_stateful_session]
+
+//
+message McpSseStatefulSession {
+  // Specifies the implementation of session state. This session state is used to store and retrieve the address of the
+  // upstream host assigned to the session.
+  //
+  // [#extension-category: envoy.http.mcp_sse_stateful_session]
+  config.core.v3.TypedExtensionConfig session_state = 1;
+
+  // Determines whether the HTTP request must be strictly routed to the requested destination. When set to ``true``,
+  // if the requested destination is unavailable, Envoy will return a 503 status code. The default value is ``false``,
+  // which allows Envoy to fall back to its load balancing mechanism. In this case, if the requested destination is not
+  // found, the request will be routed according to the load balancing algorithm.
+  bool strict = 2;
+}
+
+message McpSseStatefulSessionPerRoute {
+  oneof override {
+    option (validate.required) = true;
+
+    // Disable the stateful session filter for this particular vhost or route. If disabled is
+    // specified in multiple per-filter-configs, the most specific one will be used.
+    bool disabled = 1 [(validate.rules).bool = {const: true}];
+
+    // Per-route stateful session configuration that can be served by RDS or static route table.
+    McpSseStatefulSession mcp_sse_stateful_session = 2;
+  }
+}

api/contrib/envoy/extensions/http/mcp_sse_stateful_session/envelope/v3alpha/BUILD

@@ -0,0 +1,9 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = ["@com_github_cncf_udpa//udpa/annotations:pkg"],
+)

api/contrib/envoy/extensions/http/mcp_sse_stateful_session/envelope/v3alpha/envelope.proto

@@ -0,0 +1,75 @@
+syntax = "proto3";
+
+package envoy.extensions.http.mcp_sse_stateful_session.envelope.v3alpha;
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.http.mcp_sse_stateful_session.envelope.v3alpha";
+option java_outer_classname = "EnvelopeProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/contrib/envoy/extensions/http/mcp_sse_stateful_session/envelope/v3alpha";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Model Context Protocol(MCP) server-side events(SSE) stateful session extension]
+
+// The extension implements MCP 241105 spec for SSE-based session tracking.
+// It enables Envoy to handle session context in SSE event streams, allowing session ID
+// and upstream host to be encoded/decoded as required by the protocol.
+//
+// When processing the response from the upstream, Envoy will check if the SSE data stream contains
+// the session context. If the SSE data stream contains the session context, Envoy will join it and
+// the upstream host as new session context using a separator.
+//
+// When processing the request from the downstream, Envoy will check if the url query params contain
+// the session context. If the request contains the session context, Envoy will strip the
+// upstream host from the session context.
+// [#extension: envoy.http.mcp_sse_stateful_session.envelope]
+message EnvelopeSessionState {
+  // The query parameter name used to track the session state in SSE data streams.
+  // If the query parameter specified by this field is present in the SSE data stream,
+  // the upstream host address will be encoded in following format:
+  //
+  // .. code-block:: none
+  //
+  //   sessionId={original_value}.{encoded_host}
+  //
+  // Where {encoded_host} is the Base64Url encoded host address.
+  //
+  // When processing the request from downstream, this extension will:
+  // 1. Split the value at the last dot
+  // 2. Decode the host address for upstream routing
+  // 3. Keep only the original session ID in the request
+  //
+  // For example:
+  //
+  // .. code-block:: none
+  //
+  //   GET /path?sessionId=original_session_id.{encoded_host}
+  //   # after processing:
+  //   GET /path?sessionId=original_session_id
+  //
+  // Note: Uses Base64Url encoding for the host address and '.' as separator.
+  string param_name = 1 [(validate.rules).string = {min_len: 1}];
+
+  // The list of patterns to match the chunk end in the SSE data stream.
+  // Any of these patterns matched will be considered as the end of a chunk.
+  // recommended value is ["\r\n\r\n", "\n\n", "\r\r"]
+  // according to the HTML standard, the end of a server-sent-events' chunk can be
+  // - \r\n\r\n (double Carriage-Return Line-Feed)
+  // - \n\n (double Line-Feed)
+  // - \r\r (double Carriage-Return)
+  // https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream
+  // Customized patterns can be added to match the chunk end pattern.
+  repeated string chunk_end_patterns = 2 [(validate.rules).repeated = {
+    min_items: 1
+    items {string {min_len: 1}}
+  }];
+
+  // The maximum size of the pending chunk.
+  // If the pending chunk size is greater than this value, this filter will be disabled.
+  // This is to prevent the filter from consuming too much memory when the SSE data stream is large.
+  // In normal cases, the sessionId should be the initialize message and be in a small chunk.
+  // The default value is 4KB.
+  int32 max_pending_chunk_size = 3;
+}

api/versioning/BUILD

@@ -16,6 +16,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/http/golang/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/language/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/llm_inference/v3:pkg",
+        "//contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/squash/v3:pkg",
         "//contrib/envoy/extensions/filters/http/sxg/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/client_ssl_auth/v3:pkg",
@@ -33,6 +34,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/network/sip_proxy/router/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/sip_proxy/tra/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/sip_proxy/v3alpha:pkg",
+        "//contrib/envoy/extensions/http/mcp_sse_stateful_session/envelope/v3alpha:pkg",
         "//contrib/envoy/extensions/matching/input_matchers/hyperscan/v3alpha:pkg",
         "//contrib/envoy/extensions/network/connection_balance/dlb/v3alpha:pkg",
         "//contrib/envoy/extensions/private_key_providers/cryptomb/v3alpha:pkg",

contrib/contrib_build_config.bzl

@@ -7,7 +7,8 @@ CONTRIB_EXTENSIONS = {
     "envoy.filters.http.dynamo":                                "//contrib/dynamo/filters/http/source:config",
     "envoy.filters.http.http_dubbo_transcoder":                 "//contrib/http_dubbo_transcoder/filters/http/source:config",
     "envoy.filters.http.golang":                                "//contrib/golang/filters/http/source:config",
-    "envoy.filters.http.language":                              "//contrib/language/filters/http/source:config_lib",
+    "envoy.filters.http.language":                              "//contrib/language/filters/http/source:config_lib"
+    "envoy.filters.http.mcp_sse_stateful_session":              "//contrib/mcp_sse_stateful_session/filters/http/source:config",,
     "envoy.filters.http.squash":                                "//contrib/squash/filters/http/source:config",
     "envoy.filters.http.sxg":                                   "//contrib/sxg/filters/http/source:config",
     "envoy.filters.http.llm_inference":                         "//contrib/llm_inference/filters/http/source:config",
@@ -92,4 +93,10 @@ CONTRIB_EXTENSIONS = {
     #
 
     "envoy.router.cluster_specifier_plugin.golang":             "//contrib/golang/router/cluster_specifier/source:config",
+
+    #
+    # mcp sse stateful session
+    #
+
+    "envoy.http.mcp_sse_stateful_session.envelope":             "//contrib/mcp_sse_stateful_session/http/source:config",
 }

contrib/extensions_metadata.yaml

@@ -145,3 +145,18 @@ envoy.router.cluster_specifier_plugin.golang:
   - envoy.router.cluster_specifier_plugin
   security_posture: requires_trusted_downstream_and_upstream
   status: alpha
+envoy.filters.http.mcp_sse_stateful_session:
+  categories:
+  - envoy.filters.http
+  security_posture: requires_trusted_downstream_and_upstream
+  status: alpha
+  type_urls:
+  - envoy.extensions.filters.http.mcp_sse_stateful_session.v3alpha.McpSseStatefulSession
+  - envoy.extensions.filters.http.mcp_sse_stateful_session.v3alpha.McpSseStatefulSessionPerRoute
+envoy.http.mcp_sse_stateful_session.envelope:
+  categories:
+  - envoy.http.mcp_sse_stateful_session
+  security_posture: unknown
+  status: alpha
+  type_urls:
+  - envoy.extensions.http.mcp_sse_stateful_session.envelope.v3alpha.EnvelopeSessionState

contrib/mcp_sse_stateful_session/filters/http/source/BUILD

@@ -0,0 +1,44 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_contrib_extension",
+    "envoy_cc_library",
+    "envoy_contrib_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_contrib_package()
+
+envoy_cc_library(
+    name = "mcp_sse_stateful_session_lib",
+    srcs = ["mcp_sse_stateful_session.cc"],
+    hdrs = ["mcp_sse_stateful_session.h"],
+    deps = [
+        "//envoy/http:filter_interface",
+        "//envoy/http:mcp_sse_stateful_session_interface",
+        "//envoy/server:filter_config_interface",
+        "//envoy/upstream:load_balancer_interface",
+        "//source/common/config:utility_lib",
+        "//source/common/http:headers_lib",
+        "//source/common/http:utility_lib",
+        "//source/common/protobuf:utility_lib",
+        "//source/common/upstream:load_balancer_lib",
+        "//source/extensions/filters/http:well_known_names",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "@envoy_api//contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_contrib_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        "//contrib/mcp_sse_stateful_session/filters/http/source:mcp_sse_stateful_session_lib",
+        "//envoy/http:mcp_sse_stateful_session_interface",
+        "//envoy/registry",
+        "//source/common/protobuf:utility_lib",
+        "//source/extensions/filters/http/common:factory_base_lib",
+        "@envoy_api//contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha:pkg_cc_proto",
+    ],
+)

contrib/mcp_sse_stateful_session/filters/http/source/config.cc

@@ -0,0 +1,36 @@
+#include "contrib/mcp_sse_stateful_session/filters/http/source/config.h"
+
+#include <memory>
+
+#include "envoy/registry/registry.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace McpSseStatefulSession {
+
+Envoy::Http::FilterFactoryCb McpSseStatefulSessionFactoryConfig::createFilterFactoryFromProtoTyped(
+    const ProtoConfig& proto_config, const std::string&,
+    Server::Configuration::FactoryContext& context) {
+
+  auto filter_config(std::make_shared<McpSseStatefulSessionConfig>(proto_config, context));
+  return [filter_config](Envoy::Http::FilterChainFactoryCallbacks& callbacks) -> void {
+    callbacks.addStreamFilter(
+        Envoy::Http::StreamFilterSharedPtr{new McpSseStatefulSession(filter_config)});
+  };
+}
+
+Router::RouteSpecificFilterConfigConstSharedPtr
+McpSseStatefulSessionFactoryConfig::createRouteSpecificFilterConfigTyped(
+    const PerRouteProtoConfig& proto_config, Server::Configuration::ServerFactoryContext& context,
+    ProtobufMessage::ValidationVisitor&) {
+  return std::make_shared<PerRouteMcpSseStatefulSession>(proto_config, context);
+}
+
+REGISTER_FACTORY(McpSseStatefulSessionFactoryConfig,
+                 Server::Configuration::NamedHttpFilterConfigFactory);
+
+} // namespace McpSseStatefulSession
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

contrib/mcp_sse_stateful_session/filters/http/source/config.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include "source/extensions/filters/http/common/factory_base.h"
+
+#include "contrib/envoy/extensions/filters/http/mcp_sse_stateful_session/v3alpha/mcp_sse_stateful_session.pb.h"
+#include "contrib/mcp_sse_stateful_session/filters/http/source/mcp_sse_stateful_session.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace McpSseStatefulSession {
+
+/**
+ * Config registration for the stateful session filter. @see NamedHttpFilterConfigFactory.
+ */
+class McpSseStatefulSessionFactoryConfig
+    : public Common::FactoryBase<ProtoConfig, PerRouteProtoConfig> {
+public:
+  McpSseStatefulSessionFactoryConfig()
+      : FactoryBase("envoy.filters.http.mcp_sse_stateful_session") {}
+
+private:
+  Envoy::Http::FilterFactoryCb
+  createFilterFactoryFromProtoTyped(const ProtoConfig& proto_config,
+                                    const std::string& stats_prefix,
+                                    Server::Configuration::FactoryContext& context) override;
+  Router::RouteSpecificFilterConfigConstSharedPtr
+  createRouteSpecificFilterConfigTyped(const PerRouteProtoConfig& proto_config,
+                                       Server::Configuration::ServerFactoryContext& context,
+                                       ProtobufMessage::ValidationVisitor& visitor) override;
+};
+
+} // namespace McpSseStatefulSession
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy