Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4e016fd3ba | ||
|
|
343708eebe |
@@ -70,6 +70,7 @@ func main() {
|
|||||||
AdminToken: adminToken,
|
AdminToken: adminToken,
|
||||||
ExecLimiter: make(chan struct{}, 32),
|
ExecLimiter: make(chan struct{}, 32),
|
||||||
SearchLimiter: make(chan struct{}, 8),
|
SearchLimiter: make(chan struct{}, 8),
|
||||||
|
LocalSessions: executor.NewSessionManager(0),
|
||||||
}
|
}
|
||||||
if err := s.LoadSecurityPolicySnapshot(); err != nil {
|
if err := s.LoadSecurityPolicySnapshot(); err != nil {
|
||||||
log.Fatalf("load security policy snapshot: %v", err)
|
log.Fatalf("load security policy snapshot: %v", err)
|
||||||
@@ -121,6 +122,13 @@ func main() {
|
|||||||
IdleTimeout: 90 * time.Second,
|
IdleTimeout: 90 * time.Second,
|
||||||
}
|
}
|
||||||
log.Printf("vps-commander-hub listening on %s as %s", *addr, local)
|
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() {
|
go func() {
|
||||||
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||||
log.Fatal(err)
|
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 无关,是独立的协议兼容性问题。
|
||||||
|
- 决策前不要动生产端点。
|
||||||
+83
-10
@@ -49,6 +49,7 @@ type Server struct {
|
|||||||
AdminToken string
|
AdminToken string
|
||||||
ExecLimiter chan struct{}
|
ExecLimiter chan struct{}
|
||||||
SearchLimiter chan struct{}
|
SearchLimiter chan struct{}
|
||||||
|
LocalSessions *executor.SessionManager
|
||||||
}
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@@ -782,8 +783,12 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
var req struct {
|
var req struct {
|
||||||
Device, SessionID, Command, Workdir, Data string
|
Device string `json:"device"`
|
||||||
Force bool
|
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"`
|
RingCapacity int `json:"ring_capacity"`
|
||||||
}
|
}
|
||||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&req); err != nil {
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&req); err != nil {
|
||||||
@@ -796,15 +801,17 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
|
|||||||
if !s.requireSecurity(w, r, req.Device, "exec", req.SessionID) {
|
if !s.requireSecurity(w, r, req.Device, "exec", req.SessionID) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if s.Cluster == nil {
|
|
||||||
http.Error(w, "cluster unavailable", 503)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
action := r.URL.Query().Get("action")
|
action := r.URL.Query().Get("action")
|
||||||
|
owner := sessionOwner(r)
|
||||||
var result any
|
var result any
|
||||||
var err error
|
var err error
|
||||||
|
if s.isRemote(req.Device) {
|
||||||
|
if s.Cluster == nil || !s.Cluster.Online(req.Device) {
|
||||||
|
http.Error(w, "device offline", 409)
|
||||||
|
return
|
||||||
|
}
|
||||||
switch action {
|
switch action {
|
||||||
case "start":
|
case "start":
|
||||||
if req.SessionID == "" {
|
if req.SessionID == "" {
|
||||||
@@ -814,14 +821,14 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
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})
|
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":
|
case "read":
|
||||||
result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, sessionOwner(r))
|
result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, owner)
|
||||||
case "input":
|
case "input":
|
||||||
err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, sessionOwner(r))
|
err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, owner)
|
||||||
result = map[string]any{"ok": true}
|
result = map[string]any{"ok": true}
|
||||||
case "stop":
|
case "stop":
|
||||||
err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, sessionOwner(r))
|
err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, owner)
|
||||||
result = map[string]any{"ok": true}
|
result = map[string]any{"ok": true}
|
||||||
case "list":
|
case "list":
|
||||||
result, err = s.Cluster.ListSessions(ctx, req.Device)
|
result, err = s.Cluster.ListSessions(ctx, req.Device)
|
||||||
@@ -829,6 +836,72 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
|
|||||||
http.Error(w, "unknown action", 400)
|
http.Error(w, "unknown action", 400)
|
||||||
return
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
http.Error(w, err.Error(), 409)
|
http.Error(w, err.Error(), 409)
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user