mcp: add client-side OAuth flow (preliminary)

This is a preliminary implementation of OAuth 2.1 for the client.

When a StreamableClientTransport encounters a 401 Unauthorized response
from the server, it initiates the OAuth flow described in thec
authorization section of the MCP spec
(https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization).
On success, the transport obtains an access token which it passes
to all subsequent requests.

Much remains to be done here:

- Dynamic client registration is not implemented. Since it is optional,
  we also need another way of supplying the client ID and secret to
  this code.

- Resource Indicators, as described in section 2.5.1 of the MCP spec.

- There is no way for the user to provide a redirect URL.

- All of this is unexported, so it is available only to our own
  StreamingClientTransport. We should add API so people can use it
  with their own transports.

- And, of course, tests. We should test against fake implementations
  but also, if we can find any, real reference implementations.

Author: jba

go.mod

@@ -5,5 +5,6 @@ go 1.23.0
 require (
 	github.com/google/go-cmp v0.7.0
 	github.com/yosida95/uritemplate/v3 v3.0.2
+	golang.org/x/oauth2 v0.30.0
 	golang.org/x/tools v0.34.0
 )

go.sum

@@ -2,5 +2,7 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
+golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI=
+golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU=
 golang.org/x/tools v0.34.0 h1:qIpSLOxeCYGg9TrcJokLBG4KFA6d795g0xkBkiESGlo=
 golang.org/x/tools v0.34.0/go.mod h1:pAP9OwEaY1CAW3HOmg3hLZC5Z0CCmzjAF2UQMSqNARg=

internal/oauthex/resource_meta.go

@@ -145,11 +145,11 @@ func GetProtectedResourceMetadataFromID(ctx context.Context, resourceID string,
 // If there is no URL in the request, it returns nil, nil.
 func GetProtectedResourceMetadataFromHeader(ctx context.Context, header http.Header, c *http.Client) (_ *ProtectedResourceMetadata, err error) {
 	defer util.Wrapf(&err, "GetProtectedResourceMetadataFromHeader")
-	headers := header[http.CanonicalHeaderKey("WWW-Authenticate")]
-	if len(headers) == 0 {
+	authHeaders := header[http.CanonicalHeaderKey("WWW-Authenticate")]
+	if len(authHeaders) == 0 {
 		return nil, nil
 	}
-	cs, err := parseWWWAuthenticate(headers)
+	cs, err := parseWWWAuthenticate(authHeaders)
 	if err != nil {
 		return nil, err
 	}

mcp/auth.go

@@ -0,0 +1,71 @@
+// Copyright 2025 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package mcp
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/modelcontextprotocol/go-sdk/internal/oauthex"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/authhandler"
+)
+
+// newAuthClient returns a shallow copy of c with its tranport replaced by one that
+// authorizes with the token source.
+func newAuthClient(c *http.Client, ts oauth2.TokenSource) *http.Client {
+	c2 := *c
+	c2.Transport = &oauth2.Transport{
+		Base:   c.Transport,
+		Source: ts,
+	}
+	return &c2
+}
+
+// doOauth runs the OAuth 2.1 flow for MCP as described in
+// https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization.
+// It returns the resulting TokenSource.
+func doOauth(ctx context.Context, header http.Header, c *http.Client, oauthHandler authhandler.AuthorizationHandler) (oauth2.TokenSource, error) {
+	prm, err := oauthex.GetProtectedResourceMetadataFromHeader(ctx, header, c)
+	if err != nil {
+		return nil, err
+	}
+	if len(prm.AuthorizationServers) == 0 {
+		return nil, fmt.Errorf("resource %s provided no authorization servers", prm.Resource)
+	}
+	// TODO: try more than one?
+	authServer := prm.AuthorizationServers[0]
+	// TODO: which scopes to ask for? All of them?
+	scopes := prm.ScopesSupported
+	asm, err := oauthex.GetAuthServerMeta(ctx, authServer, c)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: register the client with the auth server if not registered yet,
+	// or find another way to get the client ID and secret.
+
+	// Get an access token from the auth server.
+	config := &oauth2.Config{
+		ClientID:     "TODO: from registration",
+		ClientSecret: "TODO: from registration",
+		Endpoint: oauth2.Endpoint{
+			AuthURL:  asm.AuthorizationEndpoint,
+			TokenURL: asm.TokenEndpoint,
+			// DeviceAuthURL: "",
+			// AuthStyle: "from auth meta?",
+		},
+		RedirectURL: "", // ???
+		Scopes:      scopes,
+	}
+	v := oauth2.GenerateVerifier()
+	pkceParams := authhandler.PKCEParams{
+		ChallengeMethod: "S256",
+		Challenge:       oauth2.S256ChallengeFromVerifier(v),
+		Verifier:        v,
+	}
+	state := randText()
+	return authhandler.TokenSourceWithPKCE(ctx, config, state, oauthHandler, &pkceParams), nil
+}

mcp/streamable.go

@@ -20,7 +20,9 @@ import (
 	"time"
 
 	"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
+	"github.com/modelcontextprotocol/go-sdk/internal/util"
 	"github.com/modelcontextprotocol/go-sdk/jsonrpc"
+	"golang.org/x/oauth2/authhandler"
 )
 
 const (
@@ -656,7 +658,7 @@ type StreamableReconnectOptions struct {
 }
 
 // DefaultReconnectOptions provides sensible defaults for reconnect logic.
-var DefaultReconnectOptions = &StreamableReconnectOptions{
+var DefaultReconnectOptions = StreamableReconnectOptions{
 	MaxRetries:   5,
 	growFactor:   1.5,
 	initialDelay: 1 * time.Second,
@@ -666,10 +668,18 @@ var DefaultReconnectOptions = &StreamableReconnectOptions{
 // StreamableClientTransportOptions provides options for the
 // [NewStreamableClientTransport] constructor.
 type StreamableClientTransportOptions struct {
-	// HTTPClient is the client to use for making HTTP requests. If nil,
-	// http.DefaultClient is used.
-	HTTPClient       *http.Client
-	ReconnectOptions *StreamableReconnectOptions
+	// ReconnectOptions control the transport's behavior when it is disconnected
+	// from the server.
+	ReconnectOptions StreamableReconnectOptions
+	// HTTPClient is the client to use for making unauthenticaed HTTP requests.
+	// If nil, http.DefaultClient is used.
+	// For authenticated requests, a shallow clone of the client will be used,
+	// with a different transport. The cookie jar will not be copied.
+	HTTPClient *http.Client
+	// AuthHandler is a function that handles the user interaction part of the OAuth 2.1 flow.
+	// It should prompt the user at the given URL and return the expected OAuth values.
+	// See [authhandler.AuthorizationHandler] for more.
+	AuthHandler authhandler.AuthorizationHandler
 }
 
 // NewStreamableClientTransport returns a new client transport that connects to
@@ -679,6 +689,12 @@ func NewStreamableClientTransport(url string, opts *StreamableClientTransportOpt
 	if opts != nil {
 		t.opts = *opts
 	}
+	if t.opts.HTTPClient == nil {
+		t.opts.HTTPClient = http.DefaultClient
+	}
+	if t.opts.ReconnectOptions == (StreamableReconnectOptions{}) {
+		t.opts.ReconnectOptions = DefaultReconnectOptions
+	}
 	return t
 }
 
@@ -691,36 +707,26 @@ func NewStreamableClientTransport(url string, opts *StreamableClientTransportOpt
 // When closed, the connection issues a DELETE request to terminate the logical
 // session.
 func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, error) {
-	client := t.opts.HTTPClient
-	if client == nil {
-		client = http.DefaultClient
-	}
-	reconnOpts := t.opts.ReconnectOptions
-	if reconnOpts == nil {
-		reconnOpts = DefaultReconnectOptions
-	}
 	// Create a new cancellable context that will manage the connection's lifecycle.
 	// This is crucial for cleanly shutting down the background SSE listener by
 	// cancelling its blocking network operations, which prevents hangs on exit.
 	connCtx, cancel := context.WithCancel(context.Background())
-	conn := &streamableClientConn{
-		url:              t.url,
-		client:           client,
-		incoming:         make(chan []byte, 100),
-		done:             make(chan struct{}),
-		ReconnectOptions: reconnOpts,
-		ctx:              connCtx,
-		cancel:           cancel,
-	}
-	return conn, nil
+	return &streamableClientConn{
+		url:      t.url,
+		opts:     t.opts,
+		incoming: make(chan []byte, 100),
+		done:     make(chan struct{}),
+		ctx:      connCtx,
+		cancel:   cancel,
+	}, nil
 }
 
 type streamableClientConn struct {
-	url              string
-	client           *http.Client
-	incoming         chan []byte
-	done             chan struct{}
-	ReconnectOptions *StreamableReconnectOptions
+	url        string
+	opts       StreamableClientTransportOptions
+	authClient *http.Client
+	incoming   chan []byte
+	done       chan struct{}
 
 	closeOnce sync.Once
 	closeErr  error
@@ -800,7 +806,11 @@ func (s *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e
 	return nil
 }
 
-func (s *streamableClientConn) postMessage(ctx context.Context, sessionID string, msg jsonrpc.Message) (string, error) {
+// postMessage makes a POST request to the server with msg as the body.
+// It returns the session ID.
+func (s *streamableClientConn) postMessage(ctx context.Context, sessionID string, msg jsonrpc.Message) (_ string, err error) {
+	defer util.Wrapf(&err, "MCP client posting message, session ID %q", sessionID)
+
 	data, err := jsonrpc2.EncodeMessage(msg)
 	if err != nil {
 		return "", err
@@ -819,28 +829,59 @@ func (s *streamableClientConn) postMessage(ctx context.Context, sessionID string
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json, text/event-stream")
 
-	resp, err := s.client.Do(req)
+	// Use an HTTP client that does authentication, if there is one.
+	// Otherwise, use the one provided by the user.
+	client := s.authClient
+	if client == nil {
+		client = s.opts.HTTPClient
+	}
+	// TODO: Resource Indicators, as in
+	// https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization#resource-parameter-implementation
+	resp, err := client.Do(req)
 	if err != nil {
 		return "", err
 	}
+	bodyClosed := false // avoid a second call to Close: undefined behavior (see [io.Closer])
+	defer func() {
+		if resp != nil && !bodyClosed {
+			resp.Body.Close()
+		}
+	}()
+
+	if resp.StatusCode == http.StatusUnauthorized {
+		if client == s.authClient {
+			return "", errors.New("got StatusUnauthorized when already authorized")
+		}
+		tokenSource, err := doOauth(ctx, resp.Header, s.opts.HTTPClient, s.opts.AuthHandler)
+		if err != nil {
+			return "", err
+		}
+		s.authClient = newAuthClient(s.opts.HTTPClient, tokenSource)
+		resp.Body.Close() // because we're about to replace resp
+		resp, err = s.authClient.Do(req)
+		if err != nil {
+			return "", err
+		}
+		if resp.StatusCode == http.StatusUnauthorized {
+			return "", errors.New("got StatusUnauthorized just after authorization")
+		}
+	}
 
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		// TODO: do a best effort read of the body here, and format it in the error.
-		resp.Body.Close()
 		return "", fmt.Errorf("broken session: %v", resp.Status)
 	}
 
 	sessionID = resp.Header.Get(sessionIDHeader)
 	switch ct := resp.Header.Get("Content-Type"); ct {
 	case "text/event-stream":
 		// Section 2.1: The SSE stream is initiated after a POST.
+		bodyClosed = true // handleSSE will close.
 		go s.handleSSE(resp)
 	case "application/json":
 		// TODO: read the body and send to s.incoming (in a select that also receives from s.done).
-		resp.Body.Close()
 		return "", fmt.Errorf("streamable HTTP client does not yet support raw JSON responses")
 	default:
-		resp.Body.Close()
 		return "", fmt.Errorf("unsupported content type %q", ct)
 	}
 	return sessionID, nil
@@ -912,12 +953,13 @@ func (s *streamableClientConn) processStream(resp *http.Response) (lastEventID s
 // an error if all retries are exhausted.
 func (s *streamableClientConn) reconnect(lastEventID string) (*http.Response, error) {
 	var finalErr error
+	maxRetries := s.opts.ReconnectOptions.MaxRetries
 
-	for attempt := 0; attempt < s.ReconnectOptions.MaxRetries; attempt++ {
+	for attempt := 0; attempt < maxRetries; attempt++ {
 		select {
 		case <-s.done:
 			return nil, fmt.Errorf("connection closed by client during reconnect")
-		case <-time.After(calculateReconnectDelay(s.ReconnectOptions, attempt)):
+		case <-time.After(calculateReconnectDelay(&s.opts.ReconnectOptions, attempt)):
 			resp, err := s.establishSSE(lastEventID)
 			if err != nil {
 				finalErr = err // Store the error and try again.
@@ -935,9 +977,9 @@ func (s *streamableClientConn) reconnect(lastEventID string) (*http.Response, er
 	}
 	// If the loop completes, all retries have failed.
 	if finalErr != nil {
-		return nil, fmt.Errorf("connection failed after %d attempts: %w", s.ReconnectOptions.MaxRetries, finalErr)
+		return nil, fmt.Errorf("connection failed after %d attempts: %w", maxRetries, finalErr)
 	}
-	return nil, fmt.Errorf("connection failed after %d attempts", s.ReconnectOptions.MaxRetries)
+	return nil, fmt.Errorf("connection failed after %d attempts", maxRetries)
 }
 
 // isResumable checks if an HTTP response indicates a valid SSE stream that can be processed.
@@ -966,7 +1008,7 @@ func (s *streamableClientConn) Close() error {
 				req.Header.Set(protocolVersionHeader, s.protocolVersion)
 			}
 			req.Header.Set(sessionIDHeader, s._sessionID)
-			if _, err := s.client.Do(req); err != nil {
+			if _, err := s.opts.HTTPClient.Do(req); err != nil {
 				s.closeErr = err
 			}
 		}
@@ -992,7 +1034,7 @@ func (s *streamableClientConn) establishSSE(lastEventID string) (*http.Response,
 	}
 	req.Header.Set("Accept", "text/event-stream")
 
-	return s.client.Do(req)
+	return s.opts.HTTPClient.Do(req)
 }
 
 // calculateReconnectDelay calculates a delay using exponential backoff with full jitter.