Compare commits

..
2 Commits
Author SHA1 Message Date
kory 4e016fd3ba docs: record streamable http compatibility gap
Release / Build & Release (push) Canceled after 0s
2026-10-02 14:14:26 +08:00
kory 343708eebe fix: repair session API and local session fallback
Release / Build & Release (push) Canceled after 0s
2026-10-02 14:10:56 +08:00
3 changed files with 158 additions and 28 deletions
+8
View File
@@ -70,6 +70,7 @@ func main() {
AdminToken: adminToken,
ExecLimiter: make(chan struct{}, 32),
SearchLimiter: make(chan struct{}, 8),
LocalSessions: executor.NewSessionManager(0),
}
if err := s.LoadSecurityPolicySnapshot(); err != nil {
log.Fatalf("load security policy snapshot: %v", err)
@@ -121,6 +122,13 @@ func main() {
IdleTimeout: 90 * time.Second,
}
log.Printf("vps-commander-hub listening on %s as %s", *addr, local)
go func() {
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for range ticker.C {
s.LocalSessions.Cleanup()
}
}()
go func() {
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal(err)
@@ -0,0 +1,49 @@
# 已知问题:MCP Streamable HTTP 传输不受支持(待排期)
- **发现日期**:2026-10-02
- **发现人**:审核专家 B(生产问题分析)
- **状态**:待排期,未修复
---
## 现象
Android 客户端使用 MCP Kotlin SDK 的 `StreamableHttpClientTransport` 连接 vps-commander,
发送第一条消息(`initialize`)时失败:
```
io.modelcontextprotocol.kotlin.sdk.types.McpException: Error while sending message: Streamable HTTP error:
Caused by: io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpError: Streamable HTTP error:
at ...StreamableHttpClientTransport.performSend(...)
```
## 根因
传输协议代际不匹配:
| | 服务端 (vps-commander) | 客户端 (Kotlin SDK) |
|---|---|---|
| 协议 | SSE 传输(MCP 2024-11-05) | Streamable HTTP(MCP 2025-03-26) |
| 端点 | `GET /mcp/sse` + `POST /mcp/message` | `POST /mcp`(单端点) |
| 行为 | `HandleSSE` 非 GET 直接 405(`internal/mcp/mcp.go:120`) | POST 收到非 2xx 即抛 `StreamableHttpError` |
服务端没有 `POST /mcp` 路由。客户端 POST 打到 `/mcp/sse` 得 405,打到 `/mcp` 得 404,
两种情况都会触发该异常。SDK 的错误信息为空(未携带 HTTP 状态码),需抓包/curl 确认。
## 验证
```bash
curl -i -X POST https://vpstool.kory.kdns.fr/mcp/sse \
-H "Content-Type: application/json" -d '{}'
# 预期:405 method not allowed(坐实协议不匹配)
```
## 修复选项(待决策)
1. **客户端改(快)**:Android 端改用 `SseClientTransport`,URL 指向 `/mcp/sse`,与服务端现状匹配。
2. **服务端加(彻底)**:实现 Streamable HTTP(`POST /mcp`,MCP 2025-03-26),属新功能开发。
## 备注
- 与 2026-10-02 bug hunt 的 4 个 bug 无关,是独立的协议兼容性问题。
- 决策前不要动生产端点。
+101 -28
View File
@@ -49,6 +49,7 @@ type Server struct {
AdminToken string
ExecLimiter chan struct{}
SearchLimiter chan struct{}
LocalSessions *executor.SessionManager
}
const (
@@ -782,9 +783,13 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
return
}
var req struct {
Device, SessionID, Command, Workdir, Data string
Force bool
RingCapacity int `json:"ring_capacity"`
Device string `json:"device"`
SessionID string `json:"session_id"`
Command string `json:"command"`
Workdir string `json:"workdir"`
Data string `json:"data"`
Force bool `json:"force"`
RingCapacity int `json:"ring_capacity"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&req); err != nil {
http.Error(w, "bad json", 400)
@@ -796,38 +801,106 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
if !s.requireSecurity(w, r, req.Device, "exec", req.SessionID) {
return
}
if s.Cluster == nil {
http.Error(w, "cluster unavailable", 503)
return
}
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
defer cancel()
action := r.URL.Query().Get("action")
owner := sessionOwner(r)
var result any
var err error
switch action {
case "start":
if req.SessionID == "" {
req.SessionID, err = newSessionID()
if err != nil {
http.Error(w, "session id generation failed", 500)
return
if s.isRemote(req.Device) {
if s.Cluster == nil || !s.Cluster.Online(req.Device) {
http.Error(w, "device offline", 409)
return
}
switch action {
case "start":
if req.SessionID == "" {
req.SessionID, err = newSessionID()
if err != nil {
http.Error(w, "session id generation failed", 500)
return
}
}
result, err = s.Cluster.StartProcess(ctx, req.Device, cluster.StartProcessPayload{Command: req.Command, Workdir: req.Workdir, SessionID: req.SessionID, Owner: owner, RingCapacity: req.RingCapacity})
case "read":
result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, owner)
case "input":
err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, owner)
result = map[string]any{"ok": true}
case "stop":
err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, owner)
result = map[string]any{"ok": true}
case "list":
result, err = s.Cluster.ListSessions(ctx, req.Device)
default:
http.Error(w, "unknown action", 400)
return
}
} else {
if s.LocalSessions == nil {
http.Error(w, "local session manager unavailable", 503)
return
}
switch action {
case "start":
if req.SessionID == "" {
req.SessionID, err = newSessionID()
if err != nil {
http.Error(w, "session id generation failed", 500)
return
}
}
var sess *executor.Session
sess, err = s.LocalSessions.Start(ctx, req.SessionID, req.Command, req.Workdir, owner, req.RingCapacity)
if err == nil {
result = map[string]any{"session_id": sess.ID, "started_at": sess.StartedAt.Unix()}
}
case "read":
var sess *executor.Session
sess, ok := s.LocalSessions.Get(req.SessionID)
if !ok {
err = fmt.Errorf("session not found")
break
}
if !sess.Authorized(owner) {
err = fmt.Errorf("session ownership denied")
break
}
code, done, _ := sess.Status()
stdout, stderr := sess.Output()
result = cluster.ProcessOutputPayload{SessionID: req.SessionID, Stdout: stdout, Stderr: stderr, ExitCode: code, Running: !done}
case "input":
var sess *executor.Session
sess, ok := s.LocalSessions.Get(req.SessionID)
if !ok {
err = fmt.Errorf("session not found")
break
}
if !sess.Authorized(owner) {
err = fmt.Errorf("session ownership denied")
break
}
err = sess.Stdin([]byte(req.Data))
result = map[string]any{"ok": true}
case "stop":
var sess *executor.Session
sess, ok := s.LocalSessions.Get(req.SessionID)
if !ok {
err = fmt.Errorf("session not found")
break
}
if !sess.Authorized(owner) {
err = fmt.Errorf("session ownership denied")
break
}
err = s.LocalSessions.Stop(req.SessionID, true)
result = map[string]any{"ok": true}
case "list":
result = s.LocalSessions.List()
default:
http.Error(w, "unknown action", 400)
return
}
result, err = s.Cluster.StartProcess(ctx, req.Device, cluster.StartProcessPayload{Command: req.Command, Workdir: req.Workdir, SessionID: req.SessionID, Owner: sessionOwner(r), RingCapacity: req.RingCapacity})
case "read":
result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, sessionOwner(r))
case "input":
err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, sessionOwner(r))
result = map[string]any{"ok": true}
case "stop":
err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, sessionOwner(r))
result = map[string]any{"ok": true}
case "list":
result, err = s.Cluster.ListSessions(ctx, req.Device)
default:
http.Error(w, "unknown action", 400)
return
}
if err != nil {
http.Error(w, err.Error(), 409)