Add stdio transport for subprocess MCP communication

cmd/proxy/main.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"io"
 	"os"
 	"os/signal"
 	"strings"
@@ -17,7 +18,9 @@ import (
 	"github.com/agentfacts/mcp-proxy/internal/policy"
 	"github.com/agentfacts/mcp-proxy/internal/router"
 	"github.com/agentfacts/mcp-proxy/internal/session"
+	"github.com/agentfacts/mcp-proxy/internal/transport"
 	"github.com/agentfacts/mcp-proxy/internal/transport/sse"
+	"github.com/agentfacts/mcp-proxy/internal/transport/stdio"
 	"github.com/agentfacts/mcp-proxy/internal/upstream"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
@@ -34,7 +37,7 @@ type Application struct {
 	cfg            *config.Config
 	sessionManager *session.Manager
 	router         *router.Router
-	sseServer      *sse.Server
+	transport      transport.Transport
 	upstreamClient *upstream.Client
 	policyEngine   *policy.Engine
 	auditStore     *audit.Store
@@ -270,11 +273,19 @@ func newApplication(cfg *config.Config) (*Application, error) {
 		}, nil
 	})
 
-	// Initialize SSE server
-	app.sseServer = sse.NewServer(cfg.Server, cfg.Agent, app.sessionManager)
+	// Initialize transport based on config
+	switch cfg.Server.Transport {
+	case "sse":
+		app.transport = sse.NewServer(cfg.Server, cfg.Agent, app.sessionManager)
+	case "stdio":
+		stdioServer := stdio.NewServer(cfg.Agent, app.sessionManager)
+		app.transport = stdioServer
+	default:
+		return nil, fmt.Errorf("unknown transport: %s", cfg.Server.Transport)
+	}
 
 	// Set up message handler to use router
-	app.sseServer.SetMessageHandler(app.handleMessage)
+	app.transport.SetMessageHandler(app.handleMessage)
 
 	// Initialize observability
 	app.metrics = observability.NewMetrics("mcp_proxy")
@@ -347,9 +358,9 @@ func (app *Application) Start(ctx context.Context) error {
 		}
 	}
 
-	// Start SSE server
-	if err := app.sseServer.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start SSE server: %w", err)
+	// Start transport server
+	if err := app.transport.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start %s server: %w", app.transport.Name(), err)
 	}
 
 	// Start observability server
@@ -375,9 +386,9 @@ func (app *Application) Stop(ctx context.Context) error {
 		log.Error().Err(err).Msg("Error stopping observability server")
 	}
 
-	// Stop SSE server first (stop accepting new connections)
-	if err := app.sseServer.Stop(ctx); err != nil {
-		log.Error().Err(err).Msg("Error stopping SSE server")
+	// Stop transport server first (stop accepting new connections)
+	if err := app.transport.Stop(ctx); err != nil {
+		log.Error().Err(err).Msg("Error stopping transport server")
 	}
 
 	// Disconnect from upstream
@@ -417,16 +428,27 @@ func initLogger(cfg config.LoggingConfig) {
 	}
 	zerolog.SetGlobalLevel(level)
 
+	// Determine output destination
+	var output io.Writer = os.Stdout
+	switch cfg.Output {
+	case "stderr":
+		output = os.Stderr
+	case "stdout", "":
+		output = os.Stdout
+		// File output could be added here if needed
+	}
+
 	// Configure output format
 	if cfg.Format == "text" {
 		log.Logger = log.Output(zerolog.ConsoleWriter{
-			Out:        os.Stdout,
+			Out:        output,
 			TimeFormat: time.RFC3339,
 		})
 	} else {
 		// JSON format (default)
 		zerolog.TimeFieldFormat = time.RFC3339Nano
+		log.Logger = log.Output(output)
 	}
 
-	log.Debug().Str("level", cfg.Level).Str("format", cfg.Format).Msg("Logger initialized")
+	log.Debug().Str("level", cfg.Level).Str("format", cfg.Format).Str("output", cfg.Output).Msg("Logger initialized")
 }

internal/policy/compiler/capability.go

@@ -33,9 +33,9 @@ func CompileCapabilityRules(rules []RuleDefinition, policyName string) (string,
 
 		// Replace placeholders in message
 		message = replacePlaceholders(message, map[string]string{
-			"agent.id":  "' + input.agent.id + '",
-			"tool":      tool,
-			"required":  capability,
+			"agent.id": "' + input.agent.id + '",
+			"tool":     tool,
+			"required": capability,
 		})
 
 		data := CapabilityData{

internal/policy/compiler/templates.go

@@ -9,6 +9,7 @@ import (
 // Templates for Rego code generation.
 var templates *template.Template
 
+//nolint:gochecknoinits // template initialization is idiomatic with init
 func init() {
 	templates = template.New("rego").Funcs(template.FuncMap{
 		"quote":      quoteString,

internal/policy/compiler/validation.go

@@ -157,13 +157,11 @@ func (v *Validator) validateRateLimitRule(rule *RuleDefinition) error {
 		return fmt.Errorf("'limit' must be a number")
 	}
 
-	// Either agent_id or agent_pattern should be specified
-	_, hasID := rule.Conditions["agent_id"]
-	_, hasPattern := rule.Conditions["agent_pattern"]
-
-	if !hasID && !hasPattern {
-		// Allow default rate limit without agent specification
-	}
+	// Either agent_id or agent_pattern can be specified (both optional for default rate limits)
+	// No validation needed - all combinations are valid:
+	// - No agent specification = default rate limit for all agents
+	// - agent_id = rate limit for specific agent
+	// - agent_pattern = rate limit for agents matching pattern
 
 	if window, ok := rule.Conditions["window"]; ok {
 		w, ok := window.(string)

internal/transport/sse/handler.go

@@ -1,7 +1,6 @@
 package sse
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -11,11 +10,12 @@ import (
 
 	"github.com/agentfacts/mcp-proxy/internal/config"
 	"github.com/agentfacts/mcp-proxy/internal/session"
+	"github.com/agentfacts/mcp-proxy/internal/transport"
 	"github.com/rs/zerolog/log"
 )
 
-// MessageHandler is the callback for processing incoming MCP messages.
-type MessageHandler func(ctx context.Context, sess *session.Session, message []byte) ([]byte, error)
+// MessageHandler is an alias for the transport.MessageHandler type.
+type MessageHandler = transport.MessageHandler
 
 // Handler handles SSE connections and messages.
 type Handler struct {

internal/transport/stdio/reader.go

@@ -0,0 +1,61 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"io"
+)
+
+// DefaultMaxMessageSize is the default maximum size of a single JSON message (1MB).
+const DefaultMaxMessageSize = 1024 * 1024
+
+// Reader handles reading newline-delimited JSON messages from stdin.
+type Reader struct {
+	scanner        *bufio.Scanner
+	maxMessageSize int
+}
+
+// NewReader creates a new Reader for the given input stream.
+func NewReader(in io.Reader) *Reader {
+	return NewReaderWithMaxSize(in, DefaultMaxMessageSize)
+}
+
+// NewReaderWithMaxSize creates a new Reader with a custom max message size.
+func NewReaderWithMaxSize(in io.Reader, maxSize int) *Reader {
+	scanner := bufio.NewScanner(in)
+	scanner.Buffer(make([]byte, 0, 64*1024), maxSize)
+
+	return &Reader{
+		scanner:        scanner,
+		maxMessageSize: maxSize,
+	}
+}
+
+// ReadMessage reads the next JSON message from the input.
+// Returns io.EOF when there are no more messages.
+func (r *Reader) ReadMessage() ([]byte, error) {
+	if !r.scanner.Scan() {
+		if err := r.scanner.Err(); err != nil {
+			return nil, fmt.Errorf("reading input: %w", err)
+		}
+		return nil, io.EOF
+	}
+
+	line := r.scanner.Bytes()
+	if len(line) == 0 {
+		// Skip empty lines
+		return r.ReadMessage()
+	}
+
+	// Make a copy since scanner reuses the buffer
+	msg := make([]byte, len(line))
+	copy(msg, line)
+
+	// Validate JSON
+	if !json.Valid(msg) {
+		return nil, fmt.Errorf("invalid JSON message")
+	}
+
+	return msg, nil
+}