feat(serveapi): #498 Lane B M2 — request-scoped MCP adapter and frozen wire envelopes

internal/apiserver/authorized_surface.go

@@ -4,6 +4,7 @@
 	"encoding/json"
 	"fmt"
 	"sort"
+	"strings"
 )
 
 // ToolDescriptor is the internal, protocol-neutral form of a host tool.
@@ -60,6 +61,9 @@
 	if input == nil || input["type"] != "object" {
 		return fmt.Errorf("tool %q: input schema must have type object", tool.Name)
 	}
+	if err := validateHeaderAnnotations(input); err != nil {
+		return fmt.Errorf("tool %q: invalid parameter header annotations: %w", tool.Name, err)
+	}
 	if tool.OutputSchema != nil {
 		var output any
 		if err := json.Unmarshal(tool.OutputSchema, &output); err != nil {
@@ -69,6 +73,51 @@
 	return nil
 }
 
+func validateHeaderAnnotations(schema map[string]any) error {
+	seen := make(map[string]bool)
+	var walk func(map[string]any, string) error
+	walk = func(node map[string]any, prefix string) error {
+		properties, _ := node["properties"].(map[string]any)
+		for name, value := range properties {
+			property, _ := value.(map[string]any)
+			path := name
+			if prefix != "" {
+				path = prefix + "." + name
+			}
+			if annotation, ok := property["x-mcp-header"]; ok {
+				typeName, _ := property["type"].(string)
+				if typeName != "string" && typeName != "integer" && typeName != "boolean" {
+					return fmt.Errorf("property %q: x-mcp-header requires a primitive type", path)
+				}
+				header, ok := annotation.(string)
+				if !ok || header == "" || !validHTTPFieldName(header) {
+					return fmt.Errorf("property %q: invalid x-mcp-header value", path)
+				}
+				key := strings.ToLower(header)
+				if seen[key] {
+					return fmt.Errorf("property %q: duplicate x-mcp-header value %q", path, header)
+				}
+				seen[key] = true
+			}
+			if err := walk(property, path); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+	return walk(schema, "")
+}
+
+func validHTTPFieldName(name string) bool {
+	for _, c := range name {
+		if !((c >= '0' && c <= '9') || (c >= 'A' && c <= 'Z') ||
+			(c >= 'a' && c <= 'z') || strings.ContainsRune("!#$%&'*+-.^_`|~", c)) {
+			return false
+		}
+	}
+	return name != ""
+}
+
 func cloneToolDescriptor(tool ToolDescriptor) ToolDescriptor {
 	tool.InputSchema = append(json.RawMessage(nil), tool.InputSchema...)
 	tool.OutputSchema = append(json.RawMessage(nil), tool.OutputSchema...)

internal/apiserver/embedded_mcp.go

@@ -0,0 +1,232 @@
+package apiserver
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"sync"
+
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+)
+
+// EmbeddedMCPConfig supplies the request-scoped host operations used by the
+// public serveapi facade without introducing an internal-to-public import.
+type EmbeddedMCPConfig struct {
+	AgentName    string
+	AgentVersion string
+	Runner       *CallbackRunner
+	Resolve      func(context.Context, *http.Request) (any, error)
+	Tools        func(context.Context, any) ([]ToolDescriptor, error)
+	Invoke       func(context.Context, any, string, json.RawMessage) (json.RawMessage, error)
+}
+
+type embeddedMCPHandler struct {
+	config    EmbeddedMCPConfig
+	transport http.Handler
+}
+
+type embeddedMCPContext struct {
+	surface *AuthorizedSurface
+	session any
+	failure *embeddedCallbackFailure
+}
+
+type embeddedCallbackFailure struct {
+	mu      sync.Mutex
+	message string
+}
+
+type embeddedMCPContextKey struct{}
+
+// NewEmbeddedMCPHandler builds the stateless SDK transport once. The server
+// returned to it is still new for every authorized POST.
+func NewEmbeddedMCPHandler(config EmbeddedMCPConfig) http.Handler {
+	h := &embeddedMCPHandler{config: config}
+	h.transport = mcp.NewStreamableHTTPHandler(h.serverForRequest,
+		&mcp.StreamableHTTPOptions{Stateless: true})
+	return h
+}
+
+func (h *embeddedMCPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		h.transport.ServeHTTP(w, r)
+		return
+	}
+
+	body, err := io.ReadAll(io.LimitReader(r.Body, mcp.DefaultMaxRequestBodyBytes+1))
+	if err != nil || len(body) > mcp.DefaultMaxRequestBodyBytes {
+		writeMCPEnvelope(w, requestID(body), "invalid MCP request body")
+		return
+	}
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	id := requestID(body)
+
+	session, err := RunCallback(r.Context(), h.config.Runner, func(ctx context.Context) (any, error) {
+		return h.config.Resolve(ctx, r)
+	})
+	if err != nil {
+		if status := authorizationStatus(err); status != 0 {
+			http.Error(w, err.Error(), status)
+			return
+		}
+		writeMCPCallbackError(w, id, err)
+		return
+	}
+
+	descriptors, err := RunCallback(r.Context(), h.config.Runner, func(ctx context.Context) ([]ToolDescriptor, error) {
+		return h.config.Tools(ctx, session)
+	})
+	if err != nil {
+		writeMCPCallbackError(w, id, err)
+		return
+	}
+	surface, err := callerSurface(descriptors)
+	if err != nil {
+		writeMCPEnvelope(w, id, err.Error())
+		return
+	}
+
+	failure := &embeddedCallbackFailure{}
+	ctx := context.WithValue(r.Context(), embeddedMCPContextKey{}, embeddedMCPContext{surface, session, failure})
+	r = r.WithContext(ctx)
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	h.serveTransport(w, r, id)
+}
+
+func (h *embeddedMCPHandler) serveTransport(w http.ResponseWriter, r *http.Request, id json.RawMessage) {
+	buffer := newBufferedResponseWriter()
+	defer func() {
+		if recover() != nil {
+			writeMCPEnvelope(w, id, "host tool registration failed")
+		}
+	}()
+	h.transport.ServeHTTP(buffer, r)
+	requestContext := r.Context().Value(embeddedMCPContextKey{}).(embeddedMCPContext)
+	requestContext.failure.mu.Lock()
+	message := requestContext.failure.message
+	requestContext.failure.mu.Unlock()
+	if message != "" {
+		writeMCPEnvelope(w, id, message)
+		return
+	}
+	for name, values := range buffer.header {
+		w.Header()[name] = append([]string(nil), values...)
+	}
+	w.WriteHeader(buffer.status)
+	_, _ = w.Write(buffer.body.Bytes())
+}
+
+type bufferedResponseWriter struct {
+	header http.Header
+	body   bytes.Buffer
+	status int
+}
+
+func newBufferedResponseWriter() *bufferedResponseWriter {
+	return &bufferedResponseWriter{header: make(http.Header), status: http.StatusOK}
+}
+
+func (w *bufferedResponseWriter) Header() http.Header            { return w.header }
+func (w *bufferedResponseWriter) WriteHeader(status int)         { w.status = status }
+func (w *bufferedResponseWriter) Write(data []byte) (int, error) { return w.body.Write(data) }
+func (w *bufferedResponseWriter) Flush()                         {}
+
+func (h *embeddedMCPHandler) serverForRequest(r *http.Request) *mcp.Server {
+	requestContext, ok := r.Context().Value(embeddedMCPContextKey{}).(embeddedMCPContext)
+	if !ok {
+		return nil
+	}
+	server := mcp.NewServer(&mcp.Implementation{
+		Name: h.config.AgentName, Version: h.config.AgentVersion,
+	}, nil)
+	for _, descriptor := range requestContext.surface.All() {
+		descriptor := descriptor
+		server.AddTool(&mcp.Tool{
+			Name: descriptor.Name, Description: descriptor.Description,
+			InputSchema: descriptor.InputSchema, OutputSchema: descriptor.OutputSchema,
+		}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+			result, err := RunCallback(ctx, h.config.Runner, func(callCtx context.Context) (json.RawMessage, error) {
+				return h.config.Invoke(callCtx, requestContext.session, descriptor.Name, req.Params.Arguments)
+			})
+			if err != nil {
+				requestContext.failure.mu.Lock()
+				requestContext.failure.message = callbackMessage(err)
+				requestContext.failure.mu.Unlock()
+				return mcpError(callbackMessage(err)), nil
+			}
+			return &mcp.CallToolResult{
+				Content:           []mcp.Content{&mcp.TextContent{Text: string(result)}},
+				StructuredContent: result,
+			}, nil
+		})
+	}
+	return server
+}
+
+func requestID(body []byte) json.RawMessage {
+	var request struct {
+		ID json.RawMessage `json:"id"`
+	}
+	if json.Unmarshal(body, &request) != nil || len(request.ID) == 0 || !json.Valid(request.ID) {
+		return json.RawMessage("null")
+	}
+	return append(json.RawMessage(nil), request.ID...)
+}
+
+func writeMCPCallbackError(w http.ResponseWriter, id json.RawMessage, err error) {
+	writeMCPEnvelope(w, id, callbackMessage(err))
+}
+
+func callbackMessage(err error) string {
+	switch {
+	case errors.Is(err, ErrCallbackCapacity):
+		return "host callback capacity exceeded"
+	case errors.Is(err, context.DeadlineExceeded):
+		return "host callback timed out"
+	case errors.Is(err, context.Canceled):
+		return "host callback canceled"
+	default:
+		return "host callback failed"
+	}
+}
+
+func writeMCPEnvelope(w http.ResponseWriter, id json.RawMessage, message string) {
+	if len(id) == 0 || !json.Valid(id) {
+		id = json.RawMessage("null")
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(w).Encode(struct {
+		JSONRPC string          `json:"jsonrpc"`
+		ID      json.RawMessage `json:"id"`
+		Error   struct {
+			Code    int    `json:"code"`
+			Message string `json:"message"`
+		} `json:"error"`
+	}{JSONRPC: "2.0", ID: id, Error: struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	}{Code: -32603, Message: message}})
+}
+
+func authorizationStatus(err error) int {
+	var statusError interface{ HTTPStatus() int }
+	if errors.As(err, &statusError) {
+		status := statusError.HTTPStatus()
+		if status == http.StatusUnauthorized || status == http.StatusForbidden {
+			return status
+		}
+	}
+	return 0
+}
+
+// mcpError creates an MCP tool error result shared by standalone and embedded servers.
+func mcpError(msg string) *mcp.CallToolResult {
+	return &mcp.CallToolResult{
+		Content: []mcp.Content{&mcp.TextContent{Text: msg}},
+		IsError: true,
+	}
+}