1package mcp
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "mime"
12 "net/http"
13 "os"
14 "strings"
15 "sync"
16 "sync/atomic"
17 "time"
18 "unicode/utf8"
19)
20
21const (
22 maxHTTPBody = 64 << 20
23 maxErrorBody = 4 << 10
24 headerSessionID = "Mcp-Session-Id"
25 headerProtocol = "MCP-Protocol-Version"
26)
27
28// httpTransport implements the MCP streamable HTTP transport: every client
29// message is a POST; responses arrive as JSON or as an SSE stream. A GET
30// stream carries server-initiated messages when the server supports it.
31type httpTransport struct {
32 name string
33 url string
34 headers map[string]string
35 client *http.Client
36 red *redactor
37
38 mu sync.Mutex
39 sessionID string
40 protocol string
41 handler incomingHandler
42 err error
43 // sessionGone is set when the server dropped our session (HTTP 404); the
44 // DELETE on close is pointless then.
45 sessionGone bool
46 deleted bool
47
48 ctx context.Context
49 cancel context.CancelFunc
50 doneCh chan struct{}
51 closeOnce sync.Once
52 listening bool
53
54 // cancelsSent counts notifications/cancelled sends started.
55 cancelsSent atomic.Int64
56}
57
58func newHTTPTransport(name string, cfg ServerConfig, client *http.Client) (*httpTransport, error) {
59 headers := map[string]string{}
60 for k, v := range cfg.Headers {
61 headers[k] = v
62 }
63 for header, envName := range cfg.EnvHeaders {
64 if v, ok := os.LookupEnv(envName); ok && v != "" {
65 headers[header] = v
66 }
67 }
68 if cfg.BearerTokenEnvVar != "" {
69 token, ok := os.LookupEnv(cfg.BearerTokenEnvVar)
70 if !ok || strings.TrimSpace(token) == "" {
71 return nil, fmt.Errorf("bearer token variable %s is not set", cfg.BearerTokenEnvVar)
72 }
73 headers["Authorization"] = "Bearer " + strings.TrimSpace(token)
74 }
75 if client == nil {
76 client = &http.Client{}
77 }
78 ctx, cancel := context.WithCancel(context.Background())
79 return &httpTransport{
80 name: name,
81 url: cfg.URL,
82 headers: headers,
83 client: client,
84 red: newRedactor(cfg),
85 ctx: ctx,
86 cancel: cancel,
87 doneCh: make(chan struct{}),
88 }, nil
89}
90
91func (t *httpTransport) newRequest(ctx context.Context, method string, body []byte) (*http.Request, error) {
92 var reader io.Reader
93 if body != nil {
94 reader = bytes.NewReader(body)
95 }
96 req, err := http.NewRequestWithContext(ctx, method, t.url, reader)
97 if err != nil {
98 // url parse errors quote the whole URL, query token included.
99 return nil, t.red.scrubError(err)
100 }
101 for k, v := range t.headers {
102 req.Header.Set(k, v)
103 }
104 req.Header.Set("User-Agent", "zut-mcp-client")
105 if body != nil {
106 req.Header.Set("Content-Type", "application/json")
107 }
108 req.Header.Set("Accept", "application/json, text/event-stream")
109 t.mu.Lock()
110 if t.sessionID != "" {
111 req.Header.Set(headerSessionID, t.sessionID)
112 }
113 if t.protocol != "" {
114 req.Header.Set(headerProtocol, t.protocol)
115 }
116 t.mu.Unlock()
117 return req, nil
118}
119
120// bind returns a context that ends when ctx ends or the transport fails or
121// closes, so no request outlives the connection.
122func (t *httpTransport) bind(ctx context.Context) (context.Context, context.CancelFunc) {
123 bound, cancel := context.WithCancel(ctx)
124 stop := context.AfterFunc(t.ctx, cancel)
125 return bound, func() {
126 stop()
127 cancel()
128 }
129}
130
131func (t *httpTransport) post(ctx context.Context, frame []byte) (*http.Response, error) {
132 req, err := t.newRequest(ctx, http.MethodPost, frame)
133 if err != nil {
134 return nil, err
135 }
136 resp, err := t.client.Do(req)
137 if err != nil {
138 return nil, t.red.scrubError(err)
139 }
140 if sid := resp.Header.Get(headerSessionID); sid != "" {
141 t.mu.Lock()
142 t.sessionID = sid
143 t.mu.Unlock()
144 }
145 if resp.StatusCode >= 300 {
146 defer resp.Body.Close()
147 // Read past the display limit by the longest secret's length, so a
148 // secret straddling the cut is seen whole and can be redacted.
149 limit := maxErrorBody + t.red.overlap()
150 snippet, _ := io.ReadAll(io.LimitReader(resp.Body, int64(limit)))
151 return nil, t.statusFailure(req, resp.StatusCode, snippet, len(snippet) >= limit)
152 }
153 return resp, nil
154}
155
156// statusFailure builds the error for a non-success status. A 404 for a
157// request that carried a session id means the server forgot the session
158// (expired or restarted): the transport becomes unusable so the manager
159// reconnects with a fresh initialize. The failed call is not replayed.
160func (t *httpTransport) statusFailure(req *http.Request, status int, body []byte, truncated bool) error {
161 err := t.red.statusErrorBody(status, body, truncated)
162 if status == http.StatusNotFound && req.Header.Get(headerSessionID) != "" {
163 err = fmt.Errorf("mcp session expired: %w", err)
164 t.mu.Lock()
165 t.sessionGone = true
166 t.mu.Unlock()
167 t.fail(err)
168 }
169 return err
170}
171
172// HTTPStatusError reports a non-success HTTP status from a server.
173type HTTPStatusError struct {
174 Status int
175 Body string
176}
177
178func (e *HTTPStatusError) Error() string {
179 msg := fmt.Sprintf("http %d %s", e.Status, http.StatusText(e.Status))
180 if e.Status == http.StatusUnauthorized || e.Status == http.StatusForbidden {
181 msg += " (check headers, envHeaders, or bearerTokenEnvVar)"
182 }
183 if e.Body != "" {
184 msg += ": " + e.Body
185 }
186 return msg
187}
188
189func httpStatusError(status int, body []byte) error {
190 return (*redactor)(nil).statusError(status, body)
191}
192
193// statusError builds an HTTPStatusError from a complete body.
194func (r *redactor) statusError(status int, body []byte) error {
195 return r.statusErrorBody(status, body, false)
196}
197
198// maxStatusBodyDisplay is how much of an error body is shown.
199const maxStatusBodyDisplay = 300
200
201// statusErrorBody builds an HTTPStatusError whose body is redacted, stripped of
202// terminal control sequences, and shortened for display. truncated says the
203// body was cut by the reader at an arbitrary byte, so it may end inside a
204// secret; whatever fragment of one remains at the very end is removed. The
205// final display cut gets the same protection.
206func (r *redactor) statusErrorBody(status int, body []byte, truncated bool) error {
207 text := r.scrubOneLine(string(body))
208 if truncated {
209 text = r.trimPartialSecret(text)
210 }
211 text = strings.TrimSpace(text)
212 if len(text) > maxStatusBodyDisplay {
213 cut := maxStatusBodyDisplay
214 for cut > 0 && !utf8.RuneStart(text[cut]) {
215 cut--
216 }
217 text = r.trimPartialSecret(text[:cut]) + "…"
218 }
219 return &HTTPStatusError{Status: status, Body: text}
220}
221
222func (t *httpTransport) request(ctx context.Context, id int64, method string, params any) (json.RawMessage, error) {
223 if err := t.failure(); err != nil {
224 return nil, err
225 }
226 frame, err := encodeRequest(id, method, params)
227 if err != nil {
228 return nil, err
229 }
230 // Bind the request to both the caller and the transport lifetime.
231 reqCtx, cancel := t.bind(ctx)
232 defer cancel()
233 resp, err := t.post(reqCtx, frame)
234 if err != nil {
235 if ctx.Err() != nil {
236 t.cancelRequest(id, method, ctx.Err().Error())
237 return nil, ctx.Err()
238 }
239 if t.ctx.Err() != nil {
240 return nil, t.failure()
241 }
242 return nil, err
243 }
244 defer resp.Body.Close()
245 want := fmt.Sprint(id)
246 var result *message
247 handle := func(msg message) bool {
248 if msg.isResponse() && idKey(msg.ID) == want {
249 m := msg
250 result = &m
251 return true
252 }
253 t.dispatchIncoming(msg)
254 return false
255 }
256 mediaType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type"))
257 switch mediaType {
258 case "text/event-stream":
259 err = readSSEFrames(resp.Body, handle)
260 default:
261 var raw []byte
262 raw, err = io.ReadAll(io.LimitReader(resp.Body, maxHTTPBody+1))
263 if err == nil && len(raw) > maxHTTPBody {
264 err = fmt.Errorf("response exceeds %d bytes", maxHTTPBody)
265 }
266 if err == nil {
267 var frames []message
268 frames, err = parseFrames(raw)
269 for _, msg := range frames {
270 if handle(msg) {
271 break
272 }
273 }
274 }
275 }
276 if result == nil {
277 if ctx.Err() != nil {
278 t.cancelRequest(id, method, ctx.Err().Error())
279 return nil, ctx.Err()
280 }
281 if t.ctx.Err() != nil {
282 return nil, t.failure()
283 }
284 if err != nil {
285 return nil, fmt.Errorf("read response: %w", t.red.scrubError(err))
286 }
287 return nil, errors.New("server closed the response without a result")
288 }
289 if result.Error != nil {
290 return nil, t.red.scrubRPCError(result.Error)
291 }
292 return result.Result, nil
293}
294
295// cancelRequest announces an abandoned request on a best-effort basis. It
296// never delays the caller, is bounded by cancelNotifyTimeout, and ends with
297// the transport. initialize is never cancelled (MCP lifecycle).
298func (t *httpTransport) cancelRequest(id int64, method, reason string) {
299 if !shouldSendCancel(method) || t.ctx.Err() != nil {
300 return
301 }
302 t.cancelsSent.Add(1)
303 go func() {
304 ctx, cancel := context.WithTimeout(t.ctx, cancelNotifyTimeout)
305 defer cancel()
306 _ = t.notify(ctx, "notifications/cancelled", map[string]any{"requestId": id, "reason": reason})
307 }()
308}
309
310// errSSEEventTooLarge reports an SSE event whose aggregated data exceeds the
311// body limit.
312type errSSEEventTooLarge struct{ limit int }
313
314func (e errSSEEventTooLarge) Error() string {
315 return fmt.Sprintf("sse event data exceeds %d bytes", e.limit)
316}
317
318// readSSEFrames delivers each SSE data payload as JSON-RPC frames until
319// handle returns true or the stream ends.
320func readSSEFrames(r io.Reader, handle func(message) bool) error {
321 return readSSEFramesLimit(r, maxHTTPBody, handle)
322}
323
324// readSSEFramesLimit is readSSEFrames with an explicit bound on the data of
325// one event, counted across all of its data: lines.
326func readSSEFramesLimit(r io.Reader, limit int, handle func(message) bool) error {
327 sc := bufio.NewScanner(r)
328 sc.Buffer(make([]byte, 0, 64*1024), limit+1)
329 var data strings.Builder
330 flush := func() bool {
331 if data.Len() == 0 {
332 return false
333 }
334 frames, err := parseFrames([]byte(data.String()))
335 data.Reset()
336 if err != nil {
337 return false
338 }
339 for _, msg := range frames {
340 if handle(msg) {
341 return true
342 }
343 }
344 return false
345 }
346 for sc.Scan() {
347 line := sc.Text()
348 if line == "" {
349 if flush() {
350 return nil
351 }
352 continue
353 }
354 if strings.HasPrefix(line, ":") {
355 continue
356 }
357 field, value, _ := strings.Cut(line, ":")
358 value = strings.TrimPrefix(value, " ")
359 if field == "data" {
360 extra := len(value)
361 if data.Len() > 0 {
362 extra++
363 }
364 if data.Len()+extra > limit {
365 return errSSEEventTooLarge{limit: limit}
366 }
367 if data.Len() > 0 {
368 data.WriteByte('\n')
369 }
370 data.WriteString(value)
371 }
372 }
373 if err := sc.Err(); err != nil {
374 if errors.Is(err, bufio.ErrTooLong) {
375 return errSSEEventTooLarge{limit: limit}
376 }
377 return err
378 }
379 flush()
380 return nil
381}
382
383func (t *httpTransport) dispatchIncoming(msg message) {
384 t.mu.Lock()
385 handler := t.handler
386 t.mu.Unlock()
387 if msg.isRequest() {
388 go func() {
389 var result any
390 rpcErr := &RPCError{Code: codeMethodNotFound, Message: "method not found: " + msg.Method}
391 if handler != nil {
392 result, rpcErr = handler(msg)
393 }
394 frame, err := encodeResponse(msg.ID, result, rpcErr)
395 if err != nil {
396 return
397 }
398 ctx, cancel := context.WithTimeout(t.ctx, 30*time.Second)
399 defer cancel()
400 if resp, err := t.post(ctx, frame); err == nil {
401 _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBody))
402 resp.Body.Close()
403 }
404 }()
405 return
406 }
407 if msg.Method != "" && handler != nil {
408 go handler(msg)
409 }
410}
411
412func (t *httpTransport) notify(ctx context.Context, method string, params any) error {
413 frame, err := encodeNotification(method, params)
414 if err != nil {
415 return err
416 }
417 ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
418 defer cancel()
419 ctx, stop := t.bind(ctx)
420 defer stop()
421 resp, err := t.post(ctx, frame)
422 if err != nil {
423 return err
424 }
425 _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBody))
426 return resp.Body.Close()
427}
428
429// listen opens the optional GET stream for server-initiated notifications
430// such as tools/list_changed. Servers that answer 405 do not offer it.
431func (t *httpTransport) listen() {
432 t.mu.Lock()
433 if t.listening {
434 t.mu.Unlock()
435 return
436 }
437 t.listening = true
438 t.mu.Unlock()
439 go func() {
440 backoff := time.Second
441 for t.ctx.Err() == nil {
442 req, err := t.newRequest(t.ctx, http.MethodGet, nil)
443 if err != nil {
444 return
445 }
446 req.Header.Set("Accept", "text/event-stream")
447 resp, err := t.client.Do(req)
448 if err != nil {
449 if !sleepCtx(t.ctx, backoff) {
450 return
451 }
452 backoff = min(backoff*2, 30*time.Second)
453 continue
454 }
455 if resp.StatusCode != http.StatusOK {
456 resp.Body.Close()
457 if resp.StatusCode == http.StatusNotFound && req.Header.Get(headerSessionID) != "" {
458 t.mu.Lock()
459 t.sessionGone = true
460 t.mu.Unlock()
461 t.fail(fmt.Errorf("mcp session expired: %w", t.red.statusError(resp.StatusCode, nil)))
462 }
463 // 405: no server stream. Other statuses: give up quietly; the
464 // POST path still works.
465 return
466 }
467 backoff = time.Second
468 _ = readSSEFrames(resp.Body, func(msg message) bool {
469 t.dispatchIncoming(msg)
470 return false
471 })
472 resp.Body.Close()
473 if !sleepCtx(t.ctx, backoff) {
474 return
475 }
476 }
477 }()
478}
479
480func sleepCtx(ctx context.Context, d time.Duration) bool {
481 timer := time.NewTimer(d)
482 defer timer.Stop()
483 select {
484 case <-ctx.Done():
485 return false
486 case <-timer.C:
487 return true
488 }
489}
490
491func (t *httpTransport) setHandler(h incomingHandler) {
492 t.mu.Lock()
493 t.handler = h
494 t.mu.Unlock()
495}
496
497func (t *httpTransport) setProtocolVersion(v string) {
498 t.mu.Lock()
499 t.protocol = v
500 t.mu.Unlock()
501}
502
503func (t *httpTransport) done() <-chan struct{} { return t.doneCh }
504
505func (t *httpTransport) failure() error {
506 t.mu.Lock()
507 defer t.mu.Unlock()
508 return t.err
509}
510
511// fail marks the transport unusable: pending and future requests error out,
512// background streams stop, and done() is closed so the manager reconnects.
513func (t *httpTransport) fail(err error) {
514 t.mu.Lock()
515 if t.err == nil {
516 t.err = err
517 }
518 t.mu.Unlock()
519 t.cancel()
520 t.closeOnce.Do(func() { close(t.doneCh) })
521}
522
523// close ends the session with DELETE (best effort) and stops all streams.
524func (t *httpTransport) close() error {
525 t.fail(errors.New("connection closed"))
526 t.mu.Lock()
527 sid, gone := t.sessionID, t.sessionGone
528 send := !t.deleted && sid != "" && !gone
529 t.deleted = true
530 t.mu.Unlock()
531 if send {
532 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
533 defer cancel()
534 // Built like every other request so it carries the protocol version
535 // and configured headers as well as the session id.
536 if req, err := t.newRequest(ctx, http.MethodDelete, nil); err == nil {
537 if resp, err := t.client.Do(req); err == nil {
538 resp.Body.Close()
539 }
540 }
541 }
542 return nil
543}