mcp: expose streamable HTTP request summaries

mcp/streamable.go

@@ -69,6 +69,25 @@ type sessionInfo struct {
 	timer   *time.Timer
 }
 
+// StreamableHTTPRequestSummary contains redacted metadata about a single
+// JSON-RPC message decoded from a streamable HTTP POST body.
+type StreamableHTTPRequestSummary struct {
+	// Method is the method of a decoded JSON-RPC request. It is empty for a
+	// response. The method has not yet been validated and may contain an
+	// arbitrary attacker-controlled string.
+	Method string
+
+	// RequestID is valid only for a JSON-RPC call. IDs use the same coercion
+	// rules as [jsonrpc.DecodeMessage].
+	RequestID jsonrpc.ID
+
+	// IsNotification reports whether the message is a JSON-RPC notification.
+	IsNotification bool
+
+	// IsResponse reports whether the message is a JSON-RPC response.
+	IsResponse bool
+}
+
 // startPOST signals that a POST request for this session is starting (which
 // carries a client->server message), pausing the session timeout if it was
 // running.
@@ -218,6 +237,23 @@ type StreamableHTTPOptions struct {
 	// the allowsessionsinstateless compatibility path) are unaffected.
 	PropagateRequestCancellation bool
 
+	// OnRequestSummary, when non-nil, observes redacted metadata for a single
+	// JSON-RPC message decoded from a streamable HTTP POST body. It is not called
+	// for JSON-RPC batches. The callback receives the HTTP request's context and
+	// runs synchronously before validation and dispatch of the decoded message.
+	// It may be called concurrently for different requests and should return
+	// promptly; in particular, it must not wait for processing of the same
+	// request. Panics are not recovered. Only the summary is redacted; the context
+	// may contain values added by authentication or other middleware.
+	//
+	// The callback is not invoked for requests rejected before this point,
+	// including HTTP, authorization, session-routing, and connection failures,
+	// even if connection setup previously inspected the body. Use HTTP middleware
+	// to observe those failures. Use [Server.AddReceivingMiddleware] to observe
+	// dispatched messages; this callback additionally observes decoded messages
+	// that are rejected before dispatch without exposing their parameters.
+	OnRequestSummary func(context.Context, StreamableHTTPRequestSummary)
+
 	// StreamKeepAlive is how long the response stream of a subscriptions/listen
 	// request may carry no bytes before an SSE comment line is written, so
 	// that idle-timeout intermediaries do not sever the long-lived stream, as
@@ -444,15 +480,8 @@ func (h *StreamableHTTPHandler) serveStateless(w http.ResponseWriter, req *http.
 		}
 	}
 
-	transport := &StreamableServerTransport{
-		SessionID:                   sessionID,
-		Stateless:                   true,
-		EventStore:                  h.opts.EventStore,
-		jsonResponse:                h.opts.JSONResponse,
-		streamKeepAlive:             h.opts.StreamKeepAlive,
-		logger:                      h.opts.Logger,
-		shouldPropagateCancellation: info.usesNewProtocol && (info.isSubscriptionsListen || h.opts.PropagateRequestCancellation),
-	}
+	transport := h.newStreamableServerTransport(sessionID, true)
+	transport.shouldPropagateCancellation = info.usesNewProtocol && (info.isSubscriptionsListen || h.opts.PropagateRequestCancellation)
 
 	session, err := connectStreamable(req.Context(), server, transport, info.opts)
 	if err != nil {
@@ -561,6 +590,18 @@ func connectStreamable(ctx context.Context, server *Server, transport *Streamabl
 	return s, nil
 }
 
+func (h *StreamableHTTPHandler) newStreamableServerTransport(sessionID string, stateless bool) *StreamableServerTransport {
+	return &StreamableServerTransport{
+		SessionID:        sessionID,
+		Stateless:        stateless,
+		EventStore:       h.opts.EventStore,
+		jsonResponse:     h.opts.JSONResponse,
+		streamKeepAlive:  h.opts.StreamKeepAlive,
+		logger:           h.opts.Logger,
+		onRequestSummary: h.opts.OnRequestSummary,
+	}
+}
+
 // serveStateful handles requests for stateful servers.
 // Stateful servers support GET, POST, and DELETE, and maintain persistent
 // sessions keyed by session ID.
@@ -679,14 +720,7 @@ func (h *StreamableHTTPHandler) serveStatefulPOST(w http.ResponseWriter, req *ht
 	}
 	sessionID = server.opts.GetSessionID()
 
-	transport := &StreamableServerTransport{
-		SessionID:       sessionID,
-		Stateless:       false,
-		EventStore:      h.opts.EventStore,
-		jsonResponse:    h.opts.JSONResponse,
-		streamKeepAlive: h.opts.StreamKeepAlive,
-		logger:          h.opts.Logger,
-	}
+	transport := h.newStreamableServerTransport(sessionID, false)
 
 	// Sessions without a session ID (GetSessionID returned "") are ephemeral:
 	// there's no way to address them, so they are closed after the request.
@@ -867,6 +901,10 @@ type StreamableServerTransport struct {
 	// [streamableServerConn]. See its docstring.
 	shouldPropagateCancellation bool
 
+	// onRequestSummary is forwarded from StreamableHTTPOptions for transports
+	// created by StreamableHTTPHandler.
+	onRequestSummary func(context.Context, StreamableHTTPRequestSummary)
+
 	// connection is non-nil if and only if the transport has been connected.
 	connection *streamableServerConn
 }
@@ -884,6 +922,7 @@ func (t *StreamableServerTransport) Connect(ctx context.Context) (Connection, er
 		streamKeepAlive:             t.streamKeepAlive,
 		logger:                      ensureLogger(t.logger), // see #556: must be non-nil
 		shouldPropagateCancellation: t.shouldPropagateCancellation,
+		onRequestSummary:            t.onRequestSummary,
 		incoming:                    make(chan jsonrpc.Message, 10),
 		done:                        make(chan struct{}),
 		streams:                     make(map[string]*stream),
@@ -912,10 +951,11 @@ func (t *StreamableServerTransport) SupportsProtocolVersion(version string) bool
 }
 
 type streamableServerConn struct {
-	sessionID    string
-	stateless    bool
-	jsonResponse bool
-	eventStore   EventStore
+	sessionID        string
+	stateless        bool
+	jsonResponse     bool
+	eventStore       EventStore
+	onRequestSummary func(context.Context, StreamableHTTPRequestSummary)
 
 	// streamKeepAlive is the idle interval for SSE keep-alive comments, or
 	// non-positive to disable them; see [StreamableHTTPOptions.StreamKeepAlive].
@@ -1572,6 +1612,9 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
 		http.Error(w, fmt.Sprintf("malformed payload: %v", err), http.StatusBadRequest)
 		return
 	}
+	if c.onRequestSummary != nil && !isBatch && len(incoming) == 1 {
+		c.onRequestSummary(req.Context(), summarizeStreamableHTTPRequest(incoming[0]))
+	}
 
 	protocolVersion := protocolVersionFromContext(req.Context())
 	if protocolVersion == "" {
@@ -1908,6 +1951,22 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
 	c.hangResponse(req.Context(), done)
 }
 
+func summarizeStreamableHTTPRequest(msg jsonrpc.Message) StreamableHTTPRequestSummary {
+	var summary StreamableHTTPRequestSummary
+	switch msg := msg.(type) {
+	case *jsonrpc.Request:
+		summary.Method = msg.Method
+		if msg.IsCall() {
+			summary.RequestID = msg.ID
+		} else {
+			summary.IsNotification = true
+		}
+	case *jsonrpc.Response:
+		summary.IsResponse = true
+	}
+	return summary
+}
+
 // Event IDs: encode both the logical connection ID and the index, as
 // <streamID>_<idx>, to be consistent with the typescript implementation.