feat: Add MCP parser plugin and body access infrastructure

authbridge/authlib/pipeline/pipeline.go

@@ -3,6 +3,7 @@ package pipeline
 import (
 	"context"
 	"fmt"
+	"log/slog"
 )
 
 // Pipeline holds an ordered list of plugins and runs them sequentially.
@@ -59,12 +60,16 @@ func New(plugins []Plugin, opts ...Option) (*Pipeline, error) {
 func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action {
 	for _, plugin := range p.plugins {
 		if ctx.Err() != nil {
+			slog.Info("pipeline: request cancelled", "plugin", plugin.Name())
 			return Action{Type: Reject, Status: 499, Reason: "request cancelled"}
 		}
 		action := plugin.OnRequest(ctx, pctx)
 		if action.Type == Reject {
+			slog.Info("pipeline: plugin rejected request",
+				"plugin", plugin.Name(), "status", action.Status, "reason", action.Reason)
 			return action
 		}
+		slog.Debug("pipeline: plugin completed", "plugin", plugin.Name())
 	}
 	return Action{Type: Continue}
 }
@@ -74,16 +79,29 @@ func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action {
 func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action {
 	for i := len(p.plugins) - 1; i >= 0; i-- {
 		if ctx.Err() != nil {
+			slog.Info("pipeline: response cancelled", "plugin", p.plugins[i].Name())
 			return Action{Type: Reject, Status: 499, Reason: "request cancelled"}
 		}
 		action := p.plugins[i].OnResponse(ctx, pctx)
 		if action.Type == Reject {
+			slog.Info("pipeline: plugin rejected response",
+				"plugin", p.plugins[i].Name(), "status", action.Status, "reason", action.Reason)
 			return action
 		}
 	}
 	return Action{Type: Continue}
 }
 
+// NeedsBody returns true if any plugin in the pipeline declares BodyAccess.
+func (p *Pipeline) NeedsBody() bool {
+	for _, plugin := range p.plugins {
+		if plugin.Capabilities().BodyAccess {
+			return true
+		}
+	}
+	return false
+}
+
 // validateCapabilities checks that every slot a plugin reads has been written
 // by an earlier plugin in the chain.
 func validateCapabilities(plugins []Plugin, validSlots map[string]bool) error {

authbridge/authlib/plugins/mcpparser.go

@@ -0,0 +1,106 @@
+package plugins
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+)
+
+// MCPParser parses MCP JSON-RPC 2.0 request bodies and populates
+// pctx.Extensions.MCP with the parsed method, tool name, resource URI,
+// or prompt name for downstream policy plugins.
+type MCPParser struct{}
+
+func NewMCPParser() *MCPParser { return &MCPParser{} }
+
+func (p *MCPParser) Name() string { return "mcp-parser" }
+
+func (p *MCPParser) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{
+		Writes:     []string{"mcp"},
+		BodyAccess: true,
+	}
+}
+
+func (p *MCPParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	if len(pctx.Body) == 0 {
+		slog.Debug("mcp-parser: no body, skipping")
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+
+	var rpc jsonRPCRequest
+	if err := json.Unmarshal(pctx.Body, &rpc); err != nil {
+		slog.Debug("mcp-parser: body is not valid JSON-RPC", "error", err, "bodyLen", len(pctx.Body))
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+
+	ext := &pipeline.MCPExtension{
+		Method: rpc.Method,
+		RPCID:  rpc.ID,
+	}
+
+	switch rpc.Method {
+	case "tools/call":
+		ext.Tool = &pipeline.MCPToolMetadata{
+			Name: rpc.stringParam("name"),
+			Args: rpc.mapParam("arguments"),
+		}
+		slog.Info("mcp-parser: parsed tools/call", "tool", ext.Tool.Name)
+	case "resources/read":
+		ext.Resource = &pipeline.MCPResourceMetadata{
+			URI: rpc.stringParam("uri"),
+		}
+		slog.Info("mcp-parser: parsed resources/read", "uri", ext.Resource.URI)
+	case "prompts/get":
+		ext.Prompt = &pipeline.MCPPromptMetadata{
+			Name: rpc.stringParam("name"),
+			Args: rpc.stringMapParam("arguments"),
+		}
+		slog.Info("mcp-parser: parsed prompts/get", "prompt", ext.Prompt.Name)
+	default:
+		slog.Debug("mcp-parser: untracked method", "method", rpc.Method)
+	}
+
+	pctx.Extensions.MCP = ext
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+func (p *MCPParser) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+type jsonRPCRequest struct {
+	Method string         `json:"method"`
+	ID     any            `json:"id"`
+	Params map[string]any `json:"params"`
+}
+
+func (r *jsonRPCRequest) stringParam(key string) string {
+	if v, ok := r.Params[key].(string); ok {
+		return v
+	}
+	return ""
+}
+
+func (r *jsonRPCRequest) mapParam(key string) map[string]any {
+	if v, ok := r.Params[key].(map[string]any); ok {
+		return v
+	}
+	return nil
+}
+
+func (r *jsonRPCRequest) stringMapParam(key string) map[string]string {
+	raw, ok := r.Params[key].(map[string]any)
+	if !ok {
+		return nil
+	}
+	out := make(map[string]string, len(raw))
+	for k, v := range raw {
+		if s, ok := v.(string); ok {
+			out[k] = s
+		}
+	}
+	return out
+}

authbridge/authlib/plugins/mcpparser_test.go

@@ -0,0 +1,194 @@
+package plugins
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+)
+
+func TestMCPParser_Capabilities(t *testing.T) {
+	p := NewMCPParser()
+
+	if p.Name() != "mcp-parser" {
+		t.Errorf("Name() = %q, want %q", p.Name(), "mcp-parser")
+	}
+
+	caps := p.Capabilities()
+	if !caps.BodyAccess {
+		t.Error("BodyAccess should be true")
+	}
+	if len(caps.Writes) != 1 || caps.Writes[0] != "mcp" {
+		t.Errorf("Writes = %v, want [mcp]", caps.Writes)
+	}
+}
+
+func TestMCPParser_ToolCall(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Body: []byte(`{"jsonrpc":"2.0","method":"tools/call","id":1,"params":{"name":"get_weather","arguments":{"city":"NYC"}}}`),
+	}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP == nil {
+		t.Fatal("Extensions.MCP is nil")
+	}
+	if pctx.Extensions.MCP.Method != "tools/call" {
+		t.Errorf("Method = %q, want %q", pctx.Extensions.MCP.Method, "tools/call")
+	}
+	if pctx.Extensions.MCP.RPCID != float64(1) {
+		t.Errorf("RPCID = %v, want 1", pctx.Extensions.MCP.RPCID)
+	}
+	if pctx.Extensions.MCP.Tool == nil {
+		t.Fatal("Tool is nil")
+	}
+	if pctx.Extensions.MCP.Tool.Name != "get_weather" {
+		t.Errorf("Tool.Name = %q, want %q", pctx.Extensions.MCP.Tool.Name, "get_weather")
+	}
+	if pctx.Extensions.MCP.Tool.Args["city"] != "NYC" {
+		t.Errorf("Tool.Args[city] = %v, want %q", pctx.Extensions.MCP.Tool.Args["city"], "NYC")
+	}
+}
+
+func TestMCPParser_ResourceRead(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Body: []byte(`{"jsonrpc":"2.0","method":"resources/read","id":2,"params":{"uri":"file:///tmp/data.csv"}}`),
+	}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP == nil {
+		t.Fatal("Extensions.MCP is nil")
+	}
+	if pctx.Extensions.MCP.Resource == nil {
+		t.Fatal("Resource is nil")
+	}
+	if pctx.Extensions.MCP.Resource.URI != "file:///tmp/data.csv" {
+		t.Errorf("Resource.URI = %q, want %q", pctx.Extensions.MCP.Resource.URI, "file:///tmp/data.csv")
+	}
+}
+
+func TestMCPParser_PromptGet(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Body: []byte(`{"jsonrpc":"2.0","method":"prompts/get","id":3,"params":{"name":"summarize","arguments":{"style":"brief"}}}`),
+	}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP == nil {
+		t.Fatal("Extensions.MCP is nil")
+	}
+	if pctx.Extensions.MCP.Prompt == nil {
+		t.Fatal("Prompt is nil")
+	}
+	if pctx.Extensions.MCP.Prompt.Name != "summarize" {
+		t.Errorf("Prompt.Name = %q, want %q", pctx.Extensions.MCP.Prompt.Name, "summarize")
+	}
+	if pctx.Extensions.MCP.Prompt.Args["style"] != "brief" {
+		t.Errorf("Prompt.Args[style] = %q, want %q", pctx.Extensions.MCP.Prompt.Args["style"], "brief")
+	}
+}
+
+func TestMCPParser_UnknownMethod(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Body: []byte(`{"jsonrpc":"2.0","method":"notifications/initialized","id":4}`),
+	}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP == nil {
+		t.Fatal("Extensions.MCP is nil")
+	}
+	if pctx.Extensions.MCP.Method != "notifications/initialized" {
+		t.Errorf("Method = %q, want %q", pctx.Extensions.MCP.Method, "notifications/initialized")
+	}
+	if pctx.Extensions.MCP.Tool != nil {
+		t.Error("Tool should be nil for unknown method")
+	}
+	if pctx.Extensions.MCP.Resource != nil {
+		t.Error("Resource should be nil for unknown method")
+	}
+	if pctx.Extensions.MCP.Prompt != nil {
+		t.Error("Prompt should be nil for unknown method")
+	}
+}
+
+func TestMCPParser_NilBody(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{Body: nil}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP != nil {
+		t.Error("Extensions.MCP should be nil when body is nil")
+	}
+}
+
+func TestMCPParser_EmptyBody(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{Body: []byte{}}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP != nil {
+		t.Error("Extensions.MCP should be nil when body is empty")
+	}
+}
+
+func TestMCPParser_InvalidJSON(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{Body: []byte("not json")}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP != nil {
+		t.Error("Extensions.MCP should be nil for invalid JSON")
+	}
+}
+
+func TestMCPParser_MissingParams(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Body: []byte(`{"jsonrpc":"2.0","method":"tools/call","id":5}`),
+	}
+
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pctx.Extensions.MCP == nil {
+		t.Fatal("Extensions.MCP is nil")
+	}
+	if pctx.Extensions.MCP.Tool == nil {
+		t.Fatal("Tool should not be nil for tools/call")
+	}
+	if pctx.Extensions.MCP.Tool.Name != "" {
+		t.Errorf("Tool.Name = %q, want empty", pctx.Extensions.MCP.Tool.Name)
+	}
+}
+
+func TestMCPParser_OnResponse(t *testing.T) {
+	p := NewMCPParser()
+	action := p.OnResponse(context.Background(), &pipeline.Context{})
+	if action.Type != pipeline.Continue {
+		t.Errorf("OnResponse should return Continue, got %v", action.Type)
+	}
+}

authbridge/authlib/plugins/registry.go

@@ -13,6 +13,7 @@ type PluginFactory func(a *auth.Auth) pipeline.Plugin
 var registry = map[string]PluginFactory{
 	"jwt-validation": func(a *auth.Auth) pipeline.Plugin { return NewJWTValidation(a) },
 	"token-exchange": func(a *auth.Auth) pipeline.Plugin { return NewTokenExchange(a) },
+	"mcp-parser":     func(_ *auth.Auth) pipeline.Plugin { return NewMCPParser() },
 }
 
 // Build constructs a pipeline from an ordered list of plugin names.