mcp_transcoder: add response streaming option.

api/envoy/extensions/filters/http/mcp_json_rest_bridge/v3/mcp_json_rest_bridge.proto

@@ -227,6 +227,21 @@ message ToolConfig {
   // Config for this tool's entry in a local tools/list response. Used when tool_list_local is set
   // in the ServerToolConfig.
   ToolsListSpecificConfig tool_list_config = 3;
+
+  // Enables streaming transcoding for unstructured text responses (``content`` field of a result).
+  //
+  // When enabled, the response body is streamed directly to the client without buffering. Each
+  // chunk is JSON escaped as it arrives and wrapped with a pre-built JSON-RPC prefix and suffix.
+  //
+  // Streaming flow:
+  //
+  // .. code-block:: text
+  //
+  //   input:  [chunk1] → [chunk2] → [chunk3]
+  //   output: [prefix+escaped_chunk1] → [escaped_chunk2] → [escaped_chunk3+suffix]
+  //
+  // Disabled by default.
+  bool text_content_streaming_enabled = 4;
 }
 
 // Defines the schema of the JSON-RPC to REST mapping. It specifies how the "arguments"

source/extensions/filters/http/mcp_json_rest_bridge/mcp_json_rest_bridge_filter.cc

@@ -6,6 +6,8 @@
 #include "envoy/http/filter.h"
 #include "envoy/http/header_map.h"
 
+#include "source/common/buffer/buffer_impl.h"
+#include "source/common/common/json_escape_string.h"
 #include "source/common/http/headers.h"
 #include "source/common/protobuf/utility.h"
 #include "source/extensions/filters/common/mcp/constants.h"
@@ -61,7 +63,7 @@ json translateJsonRestResponseToJsonRpc(absl::string_view tool_call_response,
       {McpConstants::RESULT_FIELD,
        {
            {McpConstants::CONTENT_FIELD,
-            json::array({{{McpConstants::TYPE_FIELD, "text"},
+            json::array({{{McpConstants::TYPE_FIELD, McpConstants::TEXT_FIELD},
                           {McpConstants::TEXT_FIELD, tool_call_response}}})},
            {McpConstants::IS_ERROR_FIELD, is_error},
        }},
@@ -142,19 +144,27 @@ McpJsonRestBridgeFilterConfig::McpJsonRestBridgeFilterConfig(
       max_response_body_size_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(proto_config_, max_response_body_size,
                                                               DEFAULT_MAX_RESPONSE_BODY_SIZE)) {
   for (const auto& tool : proto_config.tool_config().tools()) {
-    tool_to_http_rule_[tool.name()] = tool.http_rule();
+    tool_entries_[tool.name()] = {tool.http_rule(), tool.text_content_streaming_enabled()};
   }
   ENVOY_LOG(debug, "Received MCP JSON REST Bridge config: {}", proto_config_.DebugString());
 }
 
 absl::StatusOr<envoy::extensions::filters::http::mcp_json_rest_bridge::v3::HttpRule>
 McpJsonRestBridgeFilterConfig::getHttpRule(absl::string_view tool_name) const {
-  auto it = tool_to_http_rule_.find(tool_name);
-  if (it == tool_to_http_rule_.end()) {
+  auto it = tool_entries_.find(tool_name);
+  if (it == tool_entries_.end()) {
     return absl::InvalidArgumentError(
         fmt::format("Failed to find http rule for tool_name: {}", tool_name));
   }
-  return it->second;
+  return it->second.http_rule;
+}
+
+bool McpJsonRestBridgeFilterConfig::textContentStreamingEnabled(absl::string_view tool_name) const {
+  auto it = tool_entries_.find(tool_name);
+  if (it == tool_entries_.end()) {
+    return false;
+  }
+  return it->second.text_content_streaming_enabled;
 }
 
 absl::StatusOr<envoy::extensions::filters::http::mcp_json_rest_bridge::v3::HttpRule>
@@ -248,8 +258,8 @@ Http::FilterDataStatus McpJsonRestBridgeFilter::decodeData(Buffer::Instance& dat
   return Http::FilterDataStatus::Continue;
 }
 
-Http::FilterHeadersStatus McpJsonRestBridgeFilter::encodeHeaders(Http::ResponseHeaderMap&,
-                                                                 bool end_stream) {
+Http::FilterHeadersStatus
+McpJsonRestBridgeFilter::encodeHeaders(Http::ResponseHeaderMap& response_headers, bool end_stream) {
   switch (mcp_operation_) {
   case McpOperation::Unspecified:
   case McpOperation::Undecided:
@@ -262,6 +272,17 @@ Http::FilterHeadersStatus McpJsonRestBridgeFilter::encodeHeaders(Http::ResponseH
     break;
   }
 
+  // Streaming mode: pre-build the JSON-RPC prefix/suffix, strip Content-Length
+  // (final size is unknown), and let the headers flow through immediately so
+  // the client can start receiving data without waiting for the full body.
+  if (mcp_operation_ == McpOperation::ToolsCall && text_content_streaming_enabled_) {
+    buildStreamingPrefixAndSuffix(getResponseCode(response_headers) >=
+                                  static_cast<int>(Http::Code::BadRequest));
+    response_headers.removeContentLength();
+    response_headers.setContentType(Http::Headers::get().ContentTypeValues.Json);
+    return Http::FilterHeadersStatus::Continue;
+  }
+
   // TODO(guoyilin42): Handle headers-only upstream responses (e.g., 204 No Content).
   // Currently, these cases bypass transcoding, which can cause MCP SDKs to timeout
   // or throw exceptions because they expect a valid JSON-RPC response with a
@@ -280,6 +301,45 @@ Http::FilterDataStatus McpJsonRestBridgeFilter::encodeData(Buffer::Instance& dat
     return Http::FilterDataStatus::Continue;
   }
 
+  // Streaming fast-path for tools/call: JSON-escape each chunk on-the-fly without
+  // buffering the full response body.
+  if (!streaming_json_prefix_.empty()) {
+    uint64_t len = data.length();
+    // Note: An empty chunk can arrive when the upstream uses the body + trailer pattern (end_stream
+    // is false on the last data frame). It is a no-op here; the suffix will be appended in
+    // encodeTrailers.
+    absl::string_view chunk(static_cast<const char*>(data.linearize(len)), len);
+    // TODO(guoyilin42): Consider adding text/event-stream backend response support and explore if
+    // it needs buffering.
+    std::string escaped_chunk = JsonEscaper::escapeString(chunk, JsonEscaper::extraSpace(chunk));
+
+    data.drain(len);
+    // Note: UTF-8 structural validation (i.e., utf8_range::IsStructurallyValid) is omitted
+    // in the streaming fast-path due to the stateless nature of chunk processing (which lacks
+    // a stateful UTF-8 validator to track multi-byte character boundaries across chunk limits).
+    // If the upstream backend returns invalid UTF-8, it will be streamed to the client as-is,
+    // which may cause the client to fail parsing the final JSON.
+    if (is_first_streaming_chunk_) {
+      ENVOY_STREAM_LOG(debug,
+                       "Streaming: emitting prefix + first chunk ({} raw bytes, {} escaped bytes).",
+                       *encoder_callbacks_, len, escaped_chunk.size());
+      data.add(streaming_json_prefix_);
+      is_first_streaming_chunk_ = false;
+    } else {
+      ENVOY_STREAM_LOG(debug, "Streaming: forwarding chunk ({} raw bytes, {} escaped bytes).",
+                       *encoder_callbacks_, len, escaped_chunk.size());
+    }
+    data.add(escaped_chunk);
+    // TODO(guoyilin42): There will be a case that end_stream is not set in the encodeData call.
+    // This is body + trailer case where encodeTrailer call represents the end of response body.
+    // In that case, we should add the streaming_json_suffix at encodeTrailer call.
+    if (end_stream) {
+      ENVOY_STREAM_LOG(debug, "Streaming: appending suffix, stream complete.", *encoder_callbacks_);
+      data.add(streaming_json_suffix_);
+    }
+    return Http::FilterDataStatus::Continue;
+  }
+
   const uint32_t max_response_body_size = config_->maxResponseBodySize();
   if (max_response_body_size > 0 &&
       (response_body_.length() + data.length()) > max_response_body_size) {
@@ -321,6 +381,32 @@ Http::FilterTrailersStatus McpJsonRestBridgeFilter::encodeTrailers(Http::Respons
   return Http::FilterTrailersStatus::Continue;
 }
 
+void McpJsonRestBridgeFilter::buildStreamingPrefixAndSuffix(bool is_error) {
+  // Build a reference JSON-RPC envelope with an empty text placeholder.
+  json ref = {
+      {McpConstants::JSONRPC_FIELD, McpConstants::JSONRPC_VERSION},
+      {McpConstants::ID_FIELD, *session_id_},
+      {McpConstants::RESULT_FIELD,
+       {
+           {McpConstants::CONTENT_FIELD,
+            json::array({{{McpConstants::TYPE_FIELD, McpConstants::TEXT_FIELD},
+                          {McpConstants::TEXT_FIELD, ""}}})},
+           {McpConstants::IS_ERROR_FIELD, is_error},
+       }},
+  };
+  std::string ref_json = ref.dump();
+
+  // Locate the empty-string placeholder for the text value: `"text":""`.
+  std::string marker = absl::StrCat("\"", McpConstants::TEXT_FIELD, "\":\"\"");
+  size_t pos = ref_json.find(marker);
+  if (pos == std::string::npos) {
+    IS_ENVOY_BUG("JSON-RPC streaming marker not found in serialized envelope");
+    return;
+  }
+  streaming_json_prefix_ = ref_json.substr(0, pos + marker.size() - 1);
+  streaming_json_suffix_ = ref_json.substr(pos + marker.size() - 1);
+}
+
 void McpJsonRestBridgeFilter::handleMcpMethod(const nlohmann::json& json_rpc,
                                               Http::RequestHeaderMapOptRef request_headers) {
   ENVOY_STREAM_LOG(debug, "Handling MCP JSON-RPC: {}", *decoder_callbacks_, json_rpc.dump());
@@ -526,6 +612,9 @@ void McpJsonRestBridgeFilter::mapMcpToolToApiBackend(const nlohmann::json& json_
     return;
   }
 
+  // Set the per-request streaming flag based on the tool's config.
+  text_content_streaming_enabled_ = config_->textContentStreamingEnabled(tool_name);
+
   const auto arguments_it = params.find(McpConstants::ARGUMENTS_FIELD);
   if (arguments_it != params.end() && !arguments_it->is_object()) {
     ENVOY_STREAM_LOG(error, "The arguments of the tool call request must be an object.",

source/extensions/filters/http/mcp_json_rest_bridge/mcp_json_rest_bridge_filter.h

@@ -50,10 +50,14 @@ class McpJsonRestBridgeFilterConfig : public Logger::Loggable<Logger::Id::config
     return proto_config_.request_storage_mode();
   }
 
+  bool textContentStreamingEnabled(absl::string_view tool_name) const;
+
 private:
-  absl::flat_hash_map<std::string,
-                      envoy::extensions::filters::http::mcp_json_rest_bridge::v3::HttpRule>
-      tool_to_http_rule_;
+  struct ToolEntry {
+    envoy::extensions::filters::http::mcp_json_rest_bridge::v3::HttpRule http_rule;
+    bool text_content_streaming_enabled;
+  };
+  absl::flat_hash_map<std::string, ToolEntry> tool_entries_;
   envoy::extensions::filters::http::mcp_json_rest_bridge::v3::McpJsonRestBridge proto_config_;
   std::string fallback_protocol_version_;
   uint32_t max_request_body_size_;
@@ -103,6 +107,9 @@ class McpJsonRestBridgeFilter : public Http::PassThroughFilter,
   // Sets dynamic metadata for the filter based on the MCP request method and parameters.
   void setDynamicMetadata(absl::string_view method, const nlohmann::json& json_rpc);
 
+  // Builds streaming_json_prefix_ and streaming_json_suffix_ for the tools/call streaming path.
+  void buildStreamingPrefixAndSuffix(bool is_error);
+
   enum class McpOperation {
     Unspecified = 0,
     // Received the "/mcp" URL but has not parsed the request body yet.
@@ -126,6 +133,16 @@ class McpJsonRestBridgeFilter : public Http::PassThroughFilter,
   Buffer::OwnedImpl response_body_;
   std::string response_body_str_;
 
+  // Per-request streaming flag, set during tool lookup in mapMcpToolToApiBackend.
+  bool text_content_streaming_enabled_ = false;
+
+  // Streaming state for text_content_streaming_enabled.
+  // prefix/suffix are pre-built once in encodeHeaders; an empty prefix signals
+  // that the non-streaming (buffered) path is active.
+  std::string streaming_json_prefix_;
+  std::string streaming_json_suffix_;
+  bool is_first_streaming_chunk_ = true;
+
   McpJsonRestBridgeFilterConfigSharedPtr config_;
 };
 

test/extensions/filters/http/mcp_json_rest_bridge/mcp_json_rest_bridge_filter_test.cc

@@ -1311,6 +1311,130 @@ TEST_F(McpJsonRestBridgeFilterTest, DynamicMetadataStoredWhenConfigured) {
             Http::FilterDataStatus::Continue);
 }
 
+class McpJsonRestBridgeStreamingFilterTest : public testing::Test {
+public:
+  void SetUp() override {
+    envoy::extensions::filters::http::mcp_json_rest_bridge::v3::McpJsonRestBridge proto_config =
+        ParseTextProtoOrDie(R"pb(
+      tool_config {
+        tools {
+          name: "get_api_key"
+          http_rule: { get: "/v1/apiKeys" }
+          text_content_streaming_enabled: true
+        }
+      }
+    )pb");
+    config_ = std::make_shared<McpJsonRestBridgeFilterConfig>(proto_config);
+    filter_ = std::make_unique<McpJsonRestBridgeFilter>(config_);
+    filter_->setDecoderFilterCallbacks(decoder_callbacks_);
+    filter_->setEncoderFilterCallbacks(encoder_callbacks_);
+    EXPECT_CALL(decoder_callbacks_, requestHeaders())
+        .WillRepeatedly(Return(Http::RequestHeaderMapOptRef(request_headers_)));
+    EXPECT_CALL(encoder_callbacks_, responseHeaders())
+        .WillRepeatedly(Return(Http::ResponseHeaderMapOptRef(response_headers_)));
+  }
+
+  void sendToolsCallRequest() {
+    request_headers_ = {{":method", "POST"}, {":path", "/mcp"}};
+    ASSERT_EQ(filter_->decodeHeaders(request_headers_, /*end_stream=*/false),
+              Http::FilterHeadersStatus::StopIteration);
+    EXPECT_CALL(decoder_callbacks_.downstream_callbacks_, clearRouteCache());
+    Buffer::OwnedImpl req(
+        R"json({"jsonrpc":"2.0","id":123,"method":"tools/call","params":{"name":"get_api_key"}})json");
+    ASSERT_EQ(filter_->decodeData(req, /*end_stream=*/true), Http::FilterDataStatus::Continue);
+  }
+
+  McpJsonRestBridgeFilterConfigSharedPtr config_;
+  std::unique_ptr<McpJsonRestBridgeFilter> filter_;
+  NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks_;
+  NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks_;
+  Http::TestRequestHeaderMapImpl request_headers_;
+  Http::TestResponseHeaderMapImpl response_headers_;
+};
+
+TEST_F(McpJsonRestBridgeStreamingFilterTest, SingleChunkReturnsFullJsonRpcResponse) {
+  sendToolsCallRequest();
+
+  response_headers_ = {
+      {":status", "200"}, {"content-type", "text/plain"}, {"content-length", "11"}};
+  EXPECT_EQ(filter_->encodeHeaders(response_headers_, /*end_stream=*/false),
+            Http::FilterHeadersStatus::Continue);
+  EXPECT_THAT(response_headers_.getContentTypeValue(), StrEq("application/json"));
+  EXPECT_FALSE(response_headers_.has(Http::Headers::get().ContentLength));
+
+  Buffer::OwnedImpl chunk("hello world");
+  EXPECT_EQ(filter_->encodeData(chunk, /*end_stream=*/true), Http::FilterDataStatus::Continue);
+  EXPECT_EQ(
+      nlohmann::json::parse(chunk.toString()),
+      nlohmann::json::parse(
+          R"json({"id":123,"jsonrpc":"2.0","result":{"content":[{"text":"hello world","type":"text"}],"isError":false}})json"));
+}
+
+TEST_F(McpJsonRestBridgeStreamingFilterTest, MultipleChunksAreStreamedCorrectly) {
+  sendToolsCallRequest();
+
+  response_headers_ = {{":status", "200"}, {"content-length", "100"}};
+  EXPECT_EQ(filter_->encodeHeaders(response_headers_, /*end_stream=*/false),
+            Http::FilterHeadersStatus::Continue);
+
+  Buffer::OwnedImpl chunk1("part1");
+  EXPECT_EQ(filter_->encodeData(chunk1, /*end_stream=*/false), Http::FilterDataStatus::Continue);
+  // First chunk contains the prefix + escaped "part1".
+  EXPECT_THAT(chunk1.toString(), testing::StartsWith("{\"id\":"));
+
+  Buffer::OwnedImpl chunk2("part2");
+  EXPECT_EQ(filter_->encodeData(chunk2, /*end_stream=*/false), Http::FilterDataStatus::Continue);
+  // Middle chunk is just "part2" (no JSON wrapper).
+  EXPECT_THAT(chunk2.toString(), StrEq("part2"));
+
+  Buffer::OwnedImpl chunk3("part3");
+  EXPECT_EQ(filter_->encodeData(chunk3, /*end_stream=*/true), Http::FilterDataStatus::Continue);
+  // Last chunk contains "part3" + the closing suffix.
+  EXPECT_THAT(chunk3.toString(), testing::EndsWith("}}"));
+
+  // Reassemble and verify the full JSON-RPC response.
+  const std::string full = chunk1.toString() + chunk2.toString() + chunk3.toString();
+  EXPECT_EQ(
+      nlohmann::json::parse(full),
+      nlohmann::json::parse(
+          R"json({"id":123,"jsonrpc":"2.0","result":{"content":[{"text":"part1part2part3","type":"text"}],"isError":false}})json"));
+}
+
+TEST_F(McpJsonRestBridgeStreamingFilterTest, SpecialCharactersAreEscaped) {
+  sendToolsCallRequest();
+
+  response_headers_ = {{":status", "200"}};
+  EXPECT_EQ(filter_->encodeHeaders(response_headers_, /*end_stream=*/false),
+            Http::FilterHeadersStatus::Continue);
+
+  // Content contains double-quotes, a backslash, a newline, and a tab.
+  Buffer::OwnedImpl chunk("{\"key\":\"val\\path\"\n\t}");
+  EXPECT_EQ(filter_->encodeData(chunk, /*end_stream=*/true), Http::FilterDataStatus::Continue);
+
+  const nlohmann::json response = nlohmann::json::parse(chunk.toString());
+  EXPECT_EQ(response["id"], 123);
+  EXPECT_EQ(response["result"]["isError"], false);
+  EXPECT_EQ(response["result"]["content"][0]["text"].get<std::string>(),
+            "{\"key\":\"val\\path\"\n\t}");
+}
+
+TEST_F(McpJsonRestBridgeStreamingFilterTest, ErrorResponseSetsIsErrorTrue) {
+  sendToolsCallRequest();
+
+  response_headers_ = {{":status", "500"}, {"content-length", "21"}};
+  EXPECT_EQ(filter_->encodeHeaders(response_headers_, /*end_stream=*/false),
+            Http::FilterHeadersStatus::Continue);
+  EXPECT_THAT(response_headers_.getContentTypeValue(), StrEq("application/json"));
+  EXPECT_FALSE(response_headers_.has(Http::Headers::get().ContentLength));
+
+  Buffer::OwnedImpl chunk("Internal Server Error");
+  EXPECT_EQ(filter_->encodeData(chunk, /*end_stream=*/true), Http::FilterDataStatus::Continue);
+  EXPECT_EQ(
+      nlohmann::json::parse(chunk.toString()),
+      nlohmann::json::parse(
+          R"json({"id":123,"jsonrpc":"2.0","result":{"content":[{"text":"Internal Server Error","type":"text"}],"isError":true}})json"));
+}
+
 class McpHttpMethodFilterTest : public testing::TestWithParam<std::string> {
 public:
   void SetUp() override {