AutoZero/mcp/http.go

package mcp

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
)

// httpTransport 实现 Streamable HTTP 传输（2026-07-28：无会话）。
//
// 每个 JSON-RPC 请求一次 POST，响应自带结果——与 stdio 共用 Client 的年代探测与 MRTR 逻辑。
//
// 规范要点：
//   - POST 单条请求（不支持批处理），Content-Type: application/json；
//   - Accept 必须同时列 application/json 与 text/event-stream：服务端可能回 JSON 对象，
//     也可能回 SSE 流，客户端两种都要支持；
//   - 头：MCP-Protocol-Version（必须与 params._meta 里的版本一致）、Mcp-Method（所有请求）、
//     Mcp-Name（tools/call / resources/read / prompts/get）；
//   - 无 Mcp-Session-Id、无 Last-Event-ID 续传、无 GET 流端点（那是已废弃的 HTTP+SSE）。
//   - 取消 = 关闭响应流（交给 ctx）。
//
// 年代回退：旧版 Streamable HTTP（2025-11-25 及更早）需要先 POST initialize 并在响应里拿
// Mcp-Session-Id，之后每次请求带上它。sessionID 只在这条分支使用。

type httpTransport struct {
	url       string
	headers   map[string]string
	client    *http.Client
	sessionID string
}

func newHTTPTransport(url string, headers map[string]string) *httpTransport {
	return &httpTransport{url: url, headers: headers, client: &http.Client{}}
}

func (t *httpTransport) Kind() string { return "http" }

// SetSession 记录旧版传输的会话 id（initialize 之后调用）。
func (t *httpTransport) SetSession(id string) { t.sessionID = id }

func (t *httpTransport) Call(ctx context.Context, body []byte, meta reqMeta) (rpcMsg, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.url, bytes.NewReader(body))
	if err != nil {
		return rpcMsg{}, err
	}
	t.applyHeaders(req, meta)
	resp, err := t.client.Do(req)
	if err != nil {
		return rpcMsg{}, err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return t.classifyError(resp)
	}
	// 旧版语义：initialize 的响应里下发 Mcp-Session-Id，之后每次请求都要带上。现代服务端
	// 不发这个头，所以这里只是顺手接住，不产生副作用。
	if id := resp.Header.Get("Mcp-Session-Id"); id != "" {
		t.sessionID = id
	}
	return DecodeHTTPResponse(resp)
}

// applyHeaders 生成规范要求的请求头。
func (t *httpTransport) applyHeaders(req *http.Request, meta reqMeta) {
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/json, text/event-stream")
	req.Header.Set("MCP-Protocol-Version", meta.version)
	if meta.method != "" {
		req.Header.Set("Mcp-Method", meta.method)
	}
	if meta.name != "" {
		req.Header.Set("Mcp-Name", meta.name)
	}
	for k, v := range t.headers {
		req.Header.Set(k, v)
	}
	if t.sessionID != "" {
		req.Header.Set("Mcp-Session-Id", t.sessionID)
	}
}

// classifyError 把非 200 分成三类：
//  1. body 是"可识别的现代错误"（错误码落在规范保留的 -32020..-32099 段）→ 原样返回响应，
//     上层据此重试/换版本，**不回退**；
//  2. 400/404/405 且 body 不是现代错误 → 判为 legacy，上层回退 initialize；
//  3. 其它状态码 → 直接报错。
func (t *httpTransport) classifyError(resp *http.Response) (rpcMsg, error) {
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	snippet := strings.TrimSpace(string(body))
	var m rpcMsg
	if json.Unmarshal(bytes.TrimSpace(body), &m) == nil && m.Error != nil && m.Error.Code <= -32020 && m.Error.Code >= -32099 {
		return m, nil
	}
	switch resp.StatusCode {
	case http.StatusBadRequest, http.StatusNotFound, http.StatusMethodNotAllowed:
		return rpcMsg{}, &legacyServerError{
			reason: fmt.Sprintf("服务端返回 HTTP %d 且响应体不是可识别的现代错误，按旧版处理（%s）", resp.StatusCode, snippet)}
	case http.StatusUnauthorized, http.StatusForbidden:
		// 常见场景：远程 MCP 要求 OAuth。本客户端还没实现 MCP 的授权流程——给一条可操作的
		// 提示：先用 headers 直接塞 Bearer token（用户的 westock-mcp 正是这种情况）。
		return rpcMsg{}, &authRequiredError{msg: fmt.Sprintf(
			"服务端要求鉴权（HTTP %d：%s）。本客户端尚未实现 MCP 的 OAuth 授权流程；如已有 token，可暂用配置的 headers 传 Authorization。",
			resp.StatusCode, snippet)}
	}
	return rpcMsg{}, fmt.Errorf("mcp http %d: %s", resp.StatusCode, snippet)
}

// DecodeHTTPResponse 解析 200 响应：单个 JSON 对象，或 SSE 流（流上先来的是与本次请求相关
// 的通知，最后一条带 id 的就是响应；`:` 开头的注释行忽略）。
func DecodeHTTPResponse(resp *http.Response) (rpcMsg, error) {
	ct := resp.Header.Get("Content-Type")
	if strings.Contains(ct, "text/event-stream") {
		return decodeSSE(resp.Body)
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<24))
	if err != nil {
		return rpcMsg{}, err
	}
	var m rpcMsg
	if err := json.Unmarshal(bytes.TrimSpace(body), &m); err != nil {
		return rpcMsg{}, fmt.Errorf("解析 MCP 响应失败: %w", err)
	}
	return m, nil
}

// decodeSSE 读 SSE 流，返回第一条带 id 的 JSON-RPC 消息。
func decodeSSE(r io.Reader) (rpcMsg, error) {
	sc := bufio.NewScanner(r)
	var data []string
	flush := func() (rpcMsg, bool) {
		if len(data) == 0 {
			return rpcMsg{}, false
		}
		payload := strings.Join(data, "\n")
		data = nil
		var m rpcMsg
		if json.Unmarshal([]byte(payload), &m) != nil {
			return rpcMsg{}, false
		}
		return m, true
	}
	for sc.Scan() {
		line := sc.Text()
		switch {
		case strings.HasPrefix(line, ":"):
			continue // 注释 / keep-alive
		case line == "":
			if m, ok := flush(); ok && len(m.ID) > 0 {
				return m, nil
			}
		case strings.HasPrefix(line, "data:"):
			data = append(data, strings.TrimPrefix(line, "data:"))
		}
	}
	if err := sc.Err(); err != nil {
		return rpcMsg{}, err
	}
	if m, ok := flush(); ok && len(m.ID) > 0 {
		return m, nil
	}
	return rpcMsg{}, fmt.Errorf("SSE 流结束但没拿到 JSON-RPC 响应")
}

// Notify 发送无 id 通知（仅 legacy 握手的 notifications/initialized 用；现代协议没有
// 客户端通知）。
func (t *httpTransport) Notify(body []byte, meta reqMeta) error {
	req, err := http.NewRequest(http.MethodPost, t.url, bytes.NewReader(body))
	if err != nil {
		return err
	}
	t.applyHeaders(req, meta)
	resp, err := t.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
		return fmt.Errorf("通知被拒绝: HTTP %d %s", resp.StatusCode, strings.TrimSpace(string(body)))
	}
	return nil
}

func (t *httpTransport) Close() error { return nil }