Fix(mcp-parser): parse streaming responses on Configurable plugins

MCP tool responses (tools/list, tools/call) over the Streamable HTTP
transport are text/event-stream, dispatched to plugins via
StreamingResponder.OnResponseFrame. In abctl these calls showed the
request parsed (observe) but the response blank (no result, no
invocation) — the response was never parsed.

Root cause: plugins.Build wraps every Configurable plugin in
pipeline.configuredPlugin (to surface raw config on /v1/pipeline). That
wrapper embeds the Plugin INTERFACE, and Go does not promote method-set
membership through an embedded interface — so the wrapped plugin's
OnResponseFrame is not promoted, and the wrapper fails the
StreamingResponder type-assertion in Pipeline.RunResponseFrame /
HasStreamingResponders. mcp-parser implements Configure (Configurable),
so it gets wrapped and silently loses response-frame dispatch;
inference-parser has no Configure, isn't wrapped, and worked. The
listener then takes a path that records the response event with no
parser activity, leaving result/invocations empty.

Fix: WrapConfigured returns a StreamingResponder-preserving wrapper
(configuredStreamingPlugin) when, and only when, the inner plugin
implements StreamingResponder — forwarding OnResponseFrame to it. Using a
distinct type (rather than an unconditional no-op OnResponseFrame on
configuredPlugin) keeps HasStreamingResponders exact, so the listener's
streaming-vs-buffered path selection is unchanged for non-streaming
pipelines.

Also harden mcp-parser's OnResponseFrame to parse via the SSE-aware
parseMCPResponse instead of a bare json.Unmarshal, so the buffered
whole-body dispatch (which can hand it a raw "data: {...}" SSE blob)
records the result instead of silently dropping it.

Tests: pipeline regression (a Configurable+StreamingResponder plugin
stays a StreamingResponder after wrapping; a non-streaming Configurable
plugin does not become one); mcp-parser raw-SSE-blob frame records an
observe; forward-proxy integration test drives a tools/list SSE response
through the proxy with a WrapConfigured'd mcp-parser and asserts the
response is parsed + observed. Verified e2e on Kind: the tools/list
response now records result + mcp-parser observe.

Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com>
Signed-off-by: Hai Huang <huang195@gmail.com>

Author: huang195

authbridge/authlib/listener/forwardproxy/mcp_sse_repro_test.go

@@ -0,0 +1,106 @@
+package forwardproxy
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/inferenceparser"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/mcpparser"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
+)
+
+// TestForwardProxy_MCP_SSEResponse_RecordsObserve reproduces the live
+// weather-tool-mcp scenario: an MCP tools/list call whose response is
+// text/event-stream (Streamable HTTP). The request is parsed (observe), and
+// the RESPONSE must also be parsed into a result + recorded as an mcp-parser
+// observe on the response event. The live cluster showed result=N,
+// invocations=[] on the response — this test pins whether the proxy+parser
+// path reproduces that.
+func TestForwardProxy_MCP_SSEResponse_RecordsObserve(t *testing.T) {
+	// Upstream mimics weather-tool-mcp: tools/list -> SSE with one data frame
+	// carrying the JSON-RPC result, exactly as FastMCP emits it.
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		io.WriteString(w, "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"tools\":[{\"name\":\"get_weather\"}]}}\n\n")
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+	}))
+	defer upstream.Close()
+
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Production builds Configurable plugins through plugins.Build, which
+	// wraps them via pipeline.WrapConfigured. mcp-parser is Configurable +
+	// StreamingResponder, so it MUST stay a StreamingResponder after wrapping
+	// — that's the bug this test guards (a bare wrapper drops OnResponseFrame
+	// and RunResponseFrame skips it, leaving the SSE response unparsed).
+	// inference-parser is not Configurable, so it's added raw, as in production.
+	pipe, err := pipeline.New([]pipeline.Plugin{
+		pipeline.WrapConfigured(mcpparser.NewMCPParser(), nil),
+		inferenceparser.NewInferenceParser(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv, err := NewServer(pipeline.NewHolder(pipe), store, nil)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	proxy := httptest.NewServer(srv.Handler())
+	defer proxy.Close()
+
+	reqBody := `{"jsonrpc":"2.0","id":2,"method":"tools/list"}`
+	req, _ := http.NewRequest("POST", upstream.URL+"/mcp", bytes.NewReader([]byte(reqBody)))
+	req.Header.Set("Content-Type", "application/json")
+	proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}}
+	resp, err := proxyClient.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	_, _ = io.ReadAll(resp.Body)
+	resp.Body.Close()
+
+	v := store.View(session.DefaultSessionID)
+	if v == nil {
+		t.Fatal("no session recorded")
+	}
+	var reqEv, respEv *pipeline.SessionEvent
+	for i := range v.Events {
+		e := &v.Events[i]
+		if e.MCP == nil || e.MCP.Method != "tools/list" {
+			continue
+		}
+		switch e.Phase {
+		case pipeline.SessionRequest:
+			reqEv = e
+		case pipeline.SessionResponse:
+			respEv = e
+		}
+	}
+	if reqEv == nil || respEv == nil {
+		t.Fatalf("missing req/resp events: req=%v resp=%v", reqEv != nil, respEv != nil)
+	}
+
+	// Request side parsed (sanity).
+	if reqEv.Invocations == nil || len(reqEv.Invocations.Outbound) == 0 {
+		t.Errorf("request event has no mcp-parser invocation: %+v", reqEv.Invocations)
+	}
+
+	// THE ASSERTION: the response was parsed into a result and recorded an
+	// observe. Live cluster fails both.
+	if respEv.MCP.Result == nil {
+		t.Errorf("response MCP.Result not populated (parser never saw the SSE result)")
+	}
+	respObserved := respEv.Invocations != nil && len(respEv.Invocations.Outbound) > 0
+	if !respObserved {
+		t.Errorf("response event has NO mcp-parser invocation — reproduces the bug. invocations=%+v", respEv.Invocations)
+	}
+}

authbridge/authlib/pipeline/configured.go

@@ -28,6 +28,13 @@ type RawConfigProvider interface {
 // implements each of those four interfaces explicitly and forwards
 // conditionally; see the methods below.
 //
+// StreamingResponder is handled differently — see configuredStreamingPlugin
+// and WrapConfigured. It must NOT be forwarded unconditionally here, because
+// listeners type-assert StreamingResponder to choose the streaming-vs-buffered
+// response path (Pipeline.HasStreamingResponders); making every wrapped plugin
+// a no-op StreamingResponder would silently flip that path selection for
+// non-streaming pipelines.
+//
 // Side effect of unconditional forwarding: every wrapped (i.e., every
 // Configurable) plugin satisfies all four optional interfaces regardless
 // of what the inner plugin actually implements. Callers that distinguish
@@ -41,6 +48,29 @@ type configuredPlugin struct {
 	raw json.RawMessage
 }
 
+// configuredStreamingPlugin is the StreamingResponder-preserving variant of
+// configuredPlugin, returned by WrapConfigured only when the wrapped plugin
+// implements StreamingResponder. The base wrapper embeds the Plugin interface,
+// which does not promote OnResponseFrame — so without this a Configurable +
+// StreamingResponder plugin (e.g. mcp-parser, which has Configure) would lose
+// its response-frame dispatch once wrapped: RunResponseFrame type-asserts
+// StreamingResponder, fails on the bare wrapper, and skips the plugin, leaving
+// MCP/SSE responses unparsed.
+//
+// Using a distinct type (instead of an unconditional no-op OnResponseFrame on
+// configuredPlugin) keeps HasStreamingResponders exact: only plugins that
+// genuinely stream are reported as streaming responders.
+type configuredStreamingPlugin struct {
+	*configuredPlugin
+}
+
+// OnResponseFrame forwards to the wrapped plugin's StreamingResponder
+// implementation. WrapConfigured constructs this wrapper only when the inner
+// plugin implements StreamingResponder, so the assertion always holds.
+func (c *configuredStreamingPlugin) OnResponseFrame(ctx context.Context, pctx *Context, frame []byte, last bool) Action {
+	return c.Plugin.(StreamingResponder).OnResponseFrame(ctx, pctx, frame, last)
+}
+
 // WrapConfigured returns a Plugin whose dynamic type retains the raw
 // config bytes the plugin was built from, so the session API can
 // surface them on /v1/pipeline. Callers (plugins.Build) invoke this
@@ -55,7 +85,14 @@ type configuredPlugin struct {
 // and on the rare hot-reload path.
 func WrapConfigured(p Plugin, raw json.RawMessage) Plugin {
 	cp := append(json.RawMessage(nil), raw...)
-	return &configuredPlugin{Plugin: p, raw: cp}
+	base := &configuredPlugin{Plugin: p, raw: cp}
+	// Preserve StreamingResponder: the embedded Plugin interface does not
+	// promote OnResponseFrame, so a Configurable + StreamingResponder plugin
+	// (e.g. mcp-parser) would otherwise lose response-frame dispatch.
+	if _, ok := p.(StreamingResponder); ok {
+		return &configuredStreamingPlugin{configuredPlugin: base}
+	}
+	return base
 }
 
 // RawConfig returns a defensive copy of the raw config bytes the

authbridge/authlib/pipeline/configured_test.go

@@ -15,7 +15,7 @@ type fakePlugin struct {
 	responses int
 }
 
-func (f *fakePlugin) Name() string                  { return f.name }
+func (f *fakePlugin) Name() string                     { return f.name }
 func (f *fakePlugin) Capabilities() PluginCapabilities { return f.caps }
 func (f *fakePlugin) OnRequest(ctx context.Context, pctx *Context) Action {
 	f.requests++
@@ -210,3 +210,50 @@ func TestConfiguredPluginReadyDefaultsTrueForNonReadier(t *testing.T) {
 		t.Fatal("non-Readier wrapped plugin should default to ready=true")
 	}
 }
+
+// streamingFakePlugin is a Plugin that also implements StreamingResponder,
+// modeling a Configurable + streaming plugin like mcp-parser.
+type streamingFakePlugin struct {
+	fakePlugin
+	frames int
+}
+
+func (s *streamingFakePlugin) OnResponseFrame(_ context.Context, _ *Context, _ []byte, _ bool) Action {
+	s.frames++
+	return Action{Type: Continue}
+}
+
+// TestWrapConfigured_PreservesStreamingResponder is the regression test for the
+// mcp-parser "tools/list response not parsed" bug: a Configurable plugin that
+// also implements StreamingResponder must STILL satisfy StreamingResponder
+// after WrapConfigured, and OnResponseFrame must forward to the inner plugin.
+// Before the fix, the wrapper embedded only the Plugin interface, so
+// OnResponseFrame was not promoted and RunResponseFrame skipped the plugin.
+func TestWrapConfigured_PreservesStreamingResponder(t *testing.T) {
+	inner := &streamingFakePlugin{fakePlugin: fakePlugin{name: "mcp-parser"}}
+	cp := WrapConfigured(inner, json.RawMessage(`{}`))
+
+	sr, ok := cp.(StreamingResponder)
+	if !ok {
+		t.Fatal("wrapped Configurable+StreamingResponder plugin must still satisfy StreamingResponder")
+	}
+	sr.OnResponseFrame(context.Background(), nil, []byte("frame"), true)
+	if inner.frames != 1 {
+		t.Fatalf("OnResponseFrame did not forward to inner plugin: frames=%d, want 1", inner.frames)
+	}
+	// The config is still surfaced through the streaming wrapper.
+	if rc, ok := cp.(RawConfigProvider); !ok || string(rc.RawConfig()) != `{}` {
+		t.Fatalf("streaming wrapper lost RawConfig: ok=%v", ok)
+	}
+}
+
+// TestWrapConfigured_NonStreamingNotPromoted confirms the fix preserves
+// HasStreamingResponders semantics: a Configurable plugin that does NOT
+// implement StreamingResponder must NOT become one through the wrapper, so the
+// listener's streaming-vs-buffered path selection is unchanged.
+func TestWrapConfigured_NonStreamingNotPromoted(t *testing.T) {
+	cp := WrapConfigured(&fakePlugin{name: "token-exchange"}, json.RawMessage(`{}`))
+	if _, ok := cp.(StreamingResponder); ok {
+		t.Fatal("wrapping a non-streaming plugin must not make it a StreamingResponder")
+	}
+}

authbridge/authlib/plugins/mcpparser/plugin.go

@@ -304,18 +304,22 @@ func (p *MCPParser) OnResponseFrame(_ context.Context, pctx *pipeline.Context, f
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 
-	var rpc jsonRPCResponse
-	if err := json.Unmarshal(frame, &rpc); err != nil {
-		// A malformed JSON-RPC message in a stream is unusual but
-		// recoverable — skip it and keep going. Don't return Reject
-		// because the listener is forwarding bytes regardless of what
-		// we say (record-only contract).
-		slog.Debug("mcp-parser: malformed frame, skipping", "error", err, "frameLen", len(frame))
-		return pipeline.Action{Type: pipeline.Continue}
-	}
-	if rpc.Result == nil && rpc.Error == nil {
-		// Notifications, heartbeats, or other JSON-RPC shapes without
-		// a result/error envelope — silently skip (no observation).
+	// Parse SSE-aware: a frame is normally one SSE event's pre-stripped
+	// JSON-RPC payload (streaming path), but the buffered dispatch hands the
+	// WHOLE response body as a single last=true frame — and for a Streamable
+	// HTTP server that body is a raw `data: {...}` SSE blob. A bare
+	// json.Unmarshal fails on that blob and silently drops the result (the
+	// tools/list-response-not-recorded bug). parseMCPResponse tries a direct
+	// decode first (clean frame), then scans `data:` lines (raw SSE blob), so
+	// both shapes record correctly.
+	rpc, ok := parseMCPResponse(frame)
+	if !ok {
+		// Notifications, heartbeats, malformed frames, or other shapes with
+		// no JSON-RPC result/error envelope — nothing to record per-frame.
+		// The end-of-stream last=true call records a Skip if no envelope was
+		// ever seen, so the response row still pairs with its request.
+		slog.Debug("mcp-parser: response frame carried no JSON-RPC result/error",
+			"frameLen", len(frame), "frame", parsercommon.Truncate(string(frame), parsercommon.DebugBodyMax))
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 

authbridge/authlib/plugins/mcpparser/streaming_test.go

@@ -97,6 +97,32 @@ func TestMCPParser_OnResponseFrame_EmptyStreamRecordsSkip(t *testing.T) {
 	}
 }
 
+// TestMCPParser_OnResponseFrame_RawSSEBlob reproduces the live tools/list
+// bug: the forward proxy's buffered dispatch hands the WHOLE response body as
+// one last=true frame, and for a Streamable HTTP (MCP) server that body is a
+// raw "data: {...}" SSE blob — not pre-stripped JSON. The bare json.Unmarshal
+// failed on it and silently dropped the result (response recorded with no
+// result, no invocation). OnResponseFrame must parse the embedded JSON-RPC
+// result and record an observe.
+func TestMCPParser_OnResponseFrame_RawSSEBlob(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "tools/list"}},
+	}
+	blob := []byte("event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"tools\":[{\"name\":\"get_weather\"}]}}\n\n")
+	pctx.SetCurrentPlugin("mcp-parser", pipeline.InvocationPhaseResponse)
+	p.OnResponseFrame(context.Background(), pctx, blob, true)
+	pctx.ClearCurrentPlugin()
+
+	if pctx.Extensions.MCP.Result == nil {
+		t.Fatal("Result not populated from raw SSE blob — the bug")
+	}
+	inv := pctx.Extensions.Invocations
+	if inv == nil || len(inv.Inbound)+len(inv.Outbound) == 0 {
+		t.Fatal("no mcp-parser observe recorded for the SSE response")
+	}
+}
+
 func TestMCPParser_OnResponseFrame_MalformedFrameSkipped(t *testing.T) {
 	p := NewMCPParser()
 	pctx := &pipeline.Context{