Files
MailUI4Agents/server/internal/mcp/stream.go
JianFeeeee 22107308bb fix(mcp)★★: JSON-RPC 帧一直是坏的 —— 任何 MCP 客户端都收不到事件
## 一个生产上一直存在的 bug

`Frame` 里 `"params":` 后面**少一个左花括号**:

    {"jsonrpc":"2.0",...,"params":"event_id":"1","event_type":"connected",...}
                                 ↑ params 变成了字符串

这不是非法 JSON 能靠编译发现的那种错 —— 它**编译通过**、连接是活的、
字节在流动、服务端日志里有帧,但**任何 MCP 客户端都解不开**。
症状完全像「没有人给我发信」。

## 上一轮我的生产实测为什么没抓到

    head -c 160 | cut -c1-160     ← 截断了,我没看全

当时只确认了「有帧到达」,没检查结构。**看片段不算验证。**

## 判据为什么一直是绿的

`TestFrameRendersJSONRPCNotification` 只断言「以 `{"jsonrpc"` 开头」——
坏帧恰好也满足。前缀检查、`HasSuffix("\n\n")`、payload 检查**全部通过**。

⇒ 唯一能抓到它的办法是把整帧当 JSON 解一遍。现在补上了,
连同 SSE 三行格式(id: / event: message / data:)一起断言。
两个变异(去掉 `{`、去掉 `data:` 前缀)都验证转红。

## 顺带修:SSE 方言

之前写的是**裸 JSON**(无任何 SSE 行前缀),而 Content-Type 却是
text/event-stream。标准 MCP 客户端用 EventSource / eventsource-parser
按 `data:` 取 payload,遇到裸 JSON 会读到空 payload 然后丢掉每个事件。
现在输出标准三行。

`id:` 行一并补上:SSE 规范据此维护 Last-Event-ID,断线重连时客户端
自动带上 —— 与服务端那份事件缓冲区回放是两条互补路径。

## 新增:完整生命周期 E2E(TestMCPFullLifecycleEndToEnd)

已有的 40 格判据全是**单点**(握手对了、工具列表对了、某个鉴权对了)。
它们各自通过,却不保证「一个真实客户端从头走到尾不撞墙」。
MCP 出问题时往往不在单点:

    initialize → tools/list → 订阅 SSE → 收到信 → **断开** → 重连(Last-Event-ID)
    → 回放补发 → 发信 → 读回 → 作用域延续

走真实 HTTP + 真实 SSE + 真实 AgentAuth,不是直接调 Run。

★ 这一格一写出来就抓到了上面那个坏帧(它按 data: 解析,读到 0 帧)。
  **判据抓不到的东西,真实客户端同样抓不到。**

## E2E 过程中判据自己错了三次

| 现象 | 真因(都不是产品缺陷) |
|---|---|
| `Unauthorized` | 漏挂 `AgentAuth` —— 少一层中间件 = 测的不是生产形状 |
| SSE 405 | 只挂了 `HandleHTTP`,漏了 `HandleGET`(POST/GET 是两个 handler) |
| 「回放没生效」 | `repoInsertMail` 是裸 INSERT,**不触发 broadcast**,事件从未进 ring |

第三个尤其阴险 —— 判据报出来的是一句**产品缺陷**的措辞
(「回放没生效,客户端会静默漏信」),而真实投递一定经过
`notify.Recipients`。判据数据错了会伪装成被测代码坏了。

## 同时把工具面对齐判据从手抄清单改成真读源码

`TestToolsExposeParamsOtherBridgesHave` 原来是一份手抄
`map[string][]string`:会漂移,且只查"不缺"不查"不多"。
改为每次现读 plugins/pi-mail-bridge/src/tools.mjs 做双向比对。

★ 解析器第一版用行正则判块结束,而 properties 里每行参数长得就是
  `xxx: { … },` —— 首行就把块关掉,**解析出 0 个参数却全绿**。
  改按花括号深度计,守卫从 `len(out)==0` 改成**参数总数**
  (漏掉的正是那个检查)。两个方向的变异都验证转红。

14 包全绿。
2026-10-04 17:08:15 +08:00

119 lines
5.5 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package mcp
// GET /mcp —— 服务端→客户端的事件流(Streamable HTTP 的 SSE 通道)。
//
// # 这半边解决什么
//
// 工具面(POST /mcp)只解决「接入方**问**」。这一条解决「服务端**说**」:
// 邮件投递时把 new_mail / session_update 推给接入方,让它**拉起对话** ——
// 与各桥靠 /api/v1/events/stream 收信是同一件事,只是方言不同:
//
// 桥: id: 7\nevent: new_mail\ndata: {…}\n\n
// MCP: {"jsonrpc":"2.0","method":"notifications/message","params":{…}}
//
// # 为什么复用 sse.Manager 而不是另起一套
//
// Manager 里那些东西**都是踩过坑才对的**:writeMu 串行化(2026-09-28 -race
// 实测 http.ResponseWriter 并发写会把 JSON 劈成半截,800 帧只切出 459 个完整)、
// Last-Event-ID 回放(宁可重复也不丢失)、心跳(反代按空闲 30-58s 掐连接)、
// 环形缓冲上限、断线清理。复制一份等于把那些坑再踩一遍,
// 而两边的修复从此各走各的。
//
// 代价是 sse.Client 多了一个 Frame 钩子(见 sse/manager.go)。
// 默认 nil = AgentMail 原格式,各桥与 WebUI 行为一字未变。
import (
"fmt"
"net/http"
"strings"
"github.com/agentmail/gateway/internal/middleware"
"github.com/agentmail/gateway/internal/sse"
)
// Frame 把一条 AgentMail 事件渲染成 MCP 方言(JSON-RPC 通知)。
//
// 独立成函数是为了能被单测直接调用、逐字节断言。
//
// data 已是 JSON 字节(SendWithID marshal 过;回放时本来就是原始字节),
// 嵌进 params 时**原样拼接**而不是再 marshal 一次 —— 再 marshal 会把已序列化的
// JSON 转义成字符串,客户端得解析两次,且长度翻倍。
func Frame(id, eventType string, data []byte) string {
// ★ 必须是**标准 SSE 行**(2026-10-04 修)。
//
// 原实现直接写裸 JSON(无 `data:` 前缀),而 Content-Type 却是
// text/event-stream —— 那个组合会让**标准 MCP 客户端静默收不到事件**:
// 它们用 EventSource / eventsource-parser 按 `data:` 行取 payload,
// 遇到裸 JSON 会解析出空 payload,然后丢掉每一个事件。
//
// ★ 症状具有欺骗性:连接是活的、字节在流动、服务端日志里有帧,
// 看起来完全像「没有人给我发信」。本项目的 E2E 判据
// (TestMCPFullLifecycleEndToEnd)正是因此红了 —— 它按 data: 解析,
// 读到 0 帧。**判据抓不到的东西,真实客户端同样抓不到。**
//
// `id:` 行一并补上:SSE 规范据此维护 Last-Event-ID,断线重连时
// 客户端会自动带上 —— 与我们服务端那份事件缓冲区回放是**两条互补
// 的路径**(回放不依赖它,但规范客户端靠它)。
payload := fmt.Sprintf(`{"jsonrpc":"2.0","method":"notifications/message","params":{`+
`"event_id":%q,"event_type":%q,"payload":%s}}`, id, eventType, string(data))
return fmt.Sprintf("id: %s\nevent: message\ndata: %s\n\n", id, payload)
}
// HandleGET 处理 GET /mcp:开一条 SSE 长连,把投递事件以 MCP 方言推给接入方。
//
// 认证在**外层**(与 POST 同一条 AgentAuth 路由):身份已在 context 里,
// 未带凭证的请求根本到不了这里。
func (s *Server) HandleGET(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
w.Header().Set("Allow", "GET, POST")
http.Error(w, "MCP GET 通道只接受 GET(工具面走 POST)", http.StatusMethodNotAllowed)
return
}
agent := middleware.GetAgentName(r)
if agent == "" {
// AgentAuth 已拦过;能到这里说明路由挂错了位置。
// 报出来而不是默默接受 —— 匿名订阅会变成一处静默的越权读。
http.Error(w, "MCP 事件流需要 Agent 凭证", http.StatusUnauthorized)
return
}
// 必须声明接受 SSE:否则客户端拿到一堆自己解析不了的文本流。
// 这条不是形式主义 —— 规范里 POST 与 GET 的语义不同,接错的客户端
// 会静默地什么都不收到(连接开着,但内容它不是当 SSE 读的)。
if !acceptsEventStream(r) {
http.Error(w, "MCP 事件流要求 Accept: text/event-stream", http.StatusNotAcceptable)
return
}
// ★ 同一把 Manager、同一个缓冲区、同一套回放 —— 与各桥唯一的差别是方言。
//
// Frame 在**注册时**传入(而不是注册后设):回放发生在"先写响应、再注册"
// 的前半段,事后设会让重连那一批走默认格式 —— 同一个连接两种方言,
// 且只在真实断线时现形。见 sse.AddClientWithFrame 的注释。
client := sse.Default.AddClientWithFrame(w, r, agent, "", Frame)
if client == nil {
http.Error(w, "SSE not supported", http.StatusInternalServerError)
return
}
// 连接还活着就阻塞在这里;断开(客户端关、反代掐、服务关停)即返回。
<-r.Context().Done()
sse.Default.RemoveClient(client.ID)
}
// acceptsEventStream 判 Accept 头里有没有 text/event-stream。
//
// 手写而不是用 mime.ParseMediaType + 循环:Accept 是**列表**且带 q 值,
// 完整解析要处理 `text/event-stream;q=0.9, application/json;q=0.5`。
// 这里只关心「有没有声明」,不关心优先级 —— 多解析的那部分没有消费者。
func acceptsEventStream(r *http.Request) bool {
for _, part := range strings.Split(r.Header.Get("Accept"), ",") {
media := strings.TrimSpace(strings.SplitN(part, ";", 2)[0])
if strings.EqualFold(media, "text/event-stream") {
return true
}
}
return false
}