diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 89b3b3a..c181781 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -70,6 +70,7 @@ func run(hub, name, token string, insecure bool) error { return err } exec := executor.Local{MaxOutput: 1024 * 1024} + mcpRuntime := cluster.NewMCPRuntime() // 初始握手附带画像 initProf := sysinfo.Collect("APPLICATION") @@ -102,6 +103,16 @@ func run(hub, name, token string, insecure bool) error { if msg.Event == "registered" { continue } + if msg.Event == "mcp_sync" { + var services []cluster.MCPService + if err := decodePayload(msg.Payload, &services); err != nil { + log.Printf("mcp sync failed: %v", err) + } else { + mcpRuntime.Sync(services) + log.Printf("MCP services synced: %d", len(services)) + } + continue + } if msg.Event == "revoked" { log.Printf("device revoked by admin: %s", msg.Error) return errors.New("revoked by admin") @@ -117,7 +128,7 @@ func run(hub, name, token string, insecure bool) error { continue } go func(msg cluster.Message) { - if err := handle(writeJSON, exec, msg); err != nil { + if err := handle(writeJSON, exec, mcpRuntime, msg); err != nil { log.Printf("request %s failed: %v", msg.ID, err) _ = c.Close() } @@ -125,8 +136,30 @@ func run(hub, name, token string, insecure bool) error { } } -func handle(writeJSON func(cluster.Message) error, exec executor.Local, msg cluster.Message) error { +func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRuntime *cluster.MCPRuntime, msg cluster.Message) error { switch msg.Action { + case "mcp_call": + var p struct { + Server string `json:"server"` + Tool string `json:"tool"` + Args any `json:"args"` + } + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + if p.Server == "" || p.Tool == "" { + return sendError(writeJSON, msg.ID, errors.New("server and tool required")) + } + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second) + defer cancel() + start := time.Now() + result, err := mcpRuntime.Call(ctx, p.Server, p.Tool, p.Args) + duration := time.Since(start).Milliseconds() + if err != nil { + _ = writeJSON(cluster.Message{ID: msg.ID, Status: "error", Error: err.Error(), Payload: map[string]any{"server": p.Server, "tool": p.Tool, "duration_ms": duration}}) + return nil + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: map[string]any{"server": p.Server, "tool": p.Tool, "result": result, "duration_ms": duration}}) case "exec": var p cluster.ExecPayload if err := decodePayload(msg.Payload, &p); err != nil { diff --git a/internal/api/api.go b/internal/api/api.go index 78f2f2c..3c0301f 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -76,6 +76,157 @@ func (s *Server) AuditsJSON(w http.ResponseWriter, r *http.Request) { jsonOut(w, logs) } +type MCPServiceRequest struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Transport string `json:"transport"` + Command string `json:"command"` + Args []string `json:"args"` + Env map[string]string `json:"env"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` + Scope string `json:"scope"` + TargetNodes []string `json:"target_nodes"` + Enabled bool `json:"enabled"` +} +type MCPCallRequest struct { + Device string `json:"device"` + Server string `json:"server"` + Tool string `json:"tool"` + Arguments any `json:"arguments"` +} + +func (s *Server) MCPServicesJSON(w http.ResponseWriter, r *http.Request) { + if s.Store == nil { + http.Error(w, "storage unavailable", 500) + return + } + switch r.Method { + case http.MethodGet: + v, err := s.Store.ListMCPServices() + if err != nil { + http.Error(w, err.Error(), 500) + return + } + jsonOut(w, v) + case http.MethodPost: + var q MCPServiceRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&q); err != nil { + http.Error(w, "bad json", 400) + return + } + if q.ID == "" || q.Name == "" || q.Transport == "stdio" && q.Command == "" || (q.Transport != "stdio" && q.URL == "") { + http.Error(w, "invalid MCP service", 400) + return + } + args, _ := json.Marshal(q.Args) + env, _ := json.Marshal(q.Env) + headers, _ := json.Marshal(q.Headers) + nodes, _ := json.Marshal(q.TargetNodes) + if q.Scope == "" { + q.Scope = "all" + } + if q.Scope != "all" && q.Scope != "custom" { + http.Error(w, "invalid scope", 400) + return + } + v := storage.MCPService{ID: q.ID, Name: q.Name, Description: q.Description, Transport: q.Transport, Command: q.Command, ArgsJSON: string(args), EnvJSON: string(env), URL: q.URL, HeadersJSON: string(headers), Scope: q.Scope, TargetNodesJSON: string(nodes), Enabled: q.Enabled} + if err := s.Store.SaveMCPService(v); err != nil { + http.Error(w, err.Error(), 500) + return + } + if s.Cluster != nil { + s.Cluster.SyncMCPServices() + } + _ = s.Store.Audit(clientIP(r), "", "mcp_service_save", q.ID, 0, 0, "") + jsonOut(w, v) + default: + http.Error(w, "method not allowed", 405) + } +} +func (s *Server) DeleteMCPService(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + http.Error(w, "method not allowed", 405) + return + } + id := r.URL.Query().Get("id") + if id == "" { + http.Error(w, "id required", 400) + return + } + if err := s.Store.DeleteMCPService(id); err != nil { + http.Error(w, err.Error(), 500) + return + } + if s.Cluster != nil { + s.Cluster.SyncMCPServices() + } + _ = s.Store.Audit(clientIP(r), "", "mcp_service_delete", id, 0, 0, "") + jsonOut(w, map[string]any{"ok": true}) +} +func (s *Server) ListAgentMCPJSON(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", 405) + return + } + device := r.URL.Query().Get("device") + if device == "" { + http.Error(w, "device required", 400) + return + } + if s.isRemote(device) && (s.Cluster == nil || !s.Cluster.Online(device)) { + http.Error(w, "device offline", 409) + return + } + if s.isRemote(device) { + v, err := s.Cluster.ListMCP(device) + if err != nil { + http.Error(w, err.Error(), 404) + return + } + jsonOut(w, map[string]any{"device": device, "services": v}) + return + } + jsonOut(w, map[string]any{"device": s.LocalName, "services": []any{}}) +} +func (s *Server) CallAgentMCPJSON(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var q MCPCallRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&q); err != nil { + http.Error(w, "bad json", 400) + return + } + if q.Device == "" || q.Server == "" || q.Tool == "" { + http.Error(w, "device, server and tool required", 400) + return + } + if s.isRemote(q.Device) && (s.Cluster == nil || !s.Cluster.Online(q.Device)) { + if s.Store != nil { + _ = s.Store.Audit(clientIP(r), q.Device, "mcp_call", q.Server+"/"+q.Tool, 1, 0, "device offline") + } + http.Error(w, "device offline", 409) + return + } + if !s.isRemote(q.Device) { + http.Error(w, "agent MCP requires a remote device", 400) + return + } + start := time.Now() + v, err := s.Cluster.CallMCP(r.Context(), q.Device, q.Server, q.Tool, q.Arguments) + ms := time.Since(start).Milliseconds() + if err != nil { + _ = s.Store.Audit(clientIP(r), q.Device, "mcp_call", q.Server+"/"+q.Tool, 1, ms, err.Error()) + http.Error(w, err.Error(), 502) + return + } + _ = s.Store.Audit(clientIP(r), q.Device, "mcp_call", q.Server+"/"+q.Tool, 0, ms, "") + jsonOut(w, v) +} + func (s *Server) Routes() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/healthz", s.health) @@ -83,6 +234,10 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("/api/v1/exec", s.exec) mux.HandleFunc("/api/v1/file/read", s.fileRead) mux.HandleFunc("/api/v1/file/write", s.fileWrite) + mux.HandleFunc("/api/v1/mcp/services", s.MCPServicesJSON) + mux.HandleFunc("/api/v1/mcp/service", s.DeleteMCPService) + mux.HandleFunc("/api/v1/mcp/list", s.ListAgentMCPJSON) + mux.HandleFunc("/api/v1/mcp/call", s.CallAgentMCPJSON) mux.HandleFunc("/agent/ws", s.agentWS) mux.HandleFunc("/openapi.json", s.openapi) return mux diff --git a/internal/api/mcp_adapter.go b/internal/api/mcp_adapter.go index f0160fc..aa4f4eb 100644 --- a/internal/api/mcp_adapter.go +++ b/internal/api/mcp_adapter.go @@ -188,3 +188,38 @@ func (b *MCPBackend) WriteFile(ctx context.Context, clientIP, device, path, cont Bytes: n, }, nil } + +func (b *MCPBackend) ListAgentMCP(ctx context.Context, device string) (any, error) { + if device == "" { + return nil, errors.New("device required") + } + if b.server.isRemote(device) && (b.server.Cluster == nil || !b.server.Cluster.Online(device)) { + return nil, fmt.Errorf("device %s is offline", device) + } + if !b.server.isRemote(device) { + return map[string]any{"device": b.server.LocalName, "services": []any{}}, nil + } + return b.server.Cluster.ListMCP(device) +} +func (b *MCPBackend) CallAgentMCP(ctx context.Context, clientIP, device, server, tool string, args any) (any, error) { + if device == "" || server == "" || tool == "" { + return nil, errors.New("device, server and tool required") + } + if b.server.isRemote(device) && (b.server.Cluster == nil || !b.server.Cluster.Online(device)) { + return nil, fmt.Errorf("device %s is offline", device) + } + if !b.server.isRemote(device) { + return nil, errors.New("agent MCP requires a remote device") + } + start := time.Now() + v, err := b.server.Cluster.CallMCP(ctx, device, server, tool, args) + ms := time.Since(start).Milliseconds() + if b.server.Store != nil { + if err != nil { + _ = b.server.Store.Audit(clientIP, device, "mcp_call", server+"/"+tool, 1, ms, err.Error()) + } else { + _ = b.server.Store.Audit(clientIP, device, "mcp_call", server+"/"+tool, 0, ms, "") + } + } + return v, err +} diff --git a/internal/api/openapi.json b/internal/api/openapi.json index 30768c1..ea93e2b 100644 --- a/internal/api/openapi.json +++ b/internal/api/openapi.json @@ -157,6 +157,69 @@ "type": "integer" } } + }, + "MCPService": { + "type": "object", + "properties": { + "id": { + "type": "string" + }, + "name": { + "type": "string" + }, + "description": { + "type": "string" + }, + "transport": { + "type": "string" + }, + "command": { + "type": "string" + }, + "args_json": { + "type": "string" + }, + "env_json": { + "type": "string" + }, + "url": { + "type": "string" + }, + "headers_json": { + "type": "string" + }, + "scope": { + "type": "string" + }, + "target_nodes_json": { + "type": "string" + }, + "enabled": { + "type": "boolean" + } + } + }, + "MCPCallRequest": { + "type": "object", + "required": [ + "device", + "server", + "tool" + ], + "properties": { + "device": { + "type": "string" + }, + "server": { + "type": "string" + }, + "tool": { + "type": "string" + }, + "arguments": { + "type": "object" + } + } } } }, @@ -281,6 +344,135 @@ } } } + }, + "/api/v1/mcp/services": { + "get": { + "operationId": "list_mcp_services", + "security": [ + { + "bearerAuth": [] + } + ], + "responses": { + "200": { + "description": "MCP services", + "content": { + "application/json": { + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/MCPService" + } + } + } + } + } + } + }, + "post": { + "operationId": "save_mcp_service", + "security": [ + { + "bearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MCPService" + } + } + } + }, + "responses": { + "200": { + "description": "Saved MCP service" + } + } + } + }, + "/api/v1/mcp/service": { + "delete": { + "operationId": "delete_mcp_service", + "security": [ + { + "bearerAuth": [] + } + ], + "parameters": [ + { + "name": "id", + "in": "query", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Deleted" + } + } + } + }, + "/api/v1/mcp/list": { + "get": { + "operationId": "list_agent_mcp", + "security": [ + { + "bearerAuth": [] + } + ], + "parameters": [ + { + "name": "device", + "in": "query", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Mounted MCP services" + }, + "409": { + "description": "Device offline" + } + } + } + }, + "/api/v1/mcp/call": { + "post": { + "operationId": "call_agent_mcp", + "security": [ + { + "bearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MCPCallRequest" + } + } + } + }, + "responses": { + "200": { + "description": "MCP result" + }, + "409": { + "description": "Device offline" + } + } + } } } } diff --git a/internal/cluster/manager.go b/internal/cluster/manager.go index ff95251..d592bb2 100644 --- a/internal/cluster/manager.go +++ b/internal/cluster/manager.go @@ -38,6 +38,7 @@ type Node struct { Arch string OS string IsLocal bool + MCP *MCPRuntime } type Result struct { @@ -63,7 +64,7 @@ func NewManager(secret string, store *storage.Store) *Manager { if d.IsLocal { continue } - n := &Node{Name: d.Name, Arch: d.Arch, OS: d.OS} + n := &Node{Name: d.Name, Arch: d.Arch, OS: d.OS, MCP: NewMCPRuntime()} n.LastSeen.Store(d.LastSeen * int64(time.Second)) n.LastHeartbeat.Store(d.LastHeartbeat * int64(time.Second)) if d.ProfileJSON != "" { @@ -113,7 +114,7 @@ func (m *Manager) ServeWS(w http.ResponseWriter, r *http.Request) { if err != nil { return } - n := &Node{Name: name, Conn: c, ConnectionID: fmt.Sprintf("%s-%d", name, time.Now().UnixNano())} + n := &Node{Name: name, Conn: c, ConnectionID: fmt.Sprintf("%s-%d", name, time.Now().UnixNano()), MCP: NewMCPRuntime()} n.LastSeen.Store(time.Now().UnixNano()) n.LastHeartbeat.Store(time.Now().UnixNano()) n.Connected.Store(true) @@ -125,6 +126,7 @@ func (m *Manager) ServeWS(w http.ResponseWriter, r *http.Request) { _ = old.Conn.Close() } _ = n.write(Message{Event: "registered", Device: name, TS: time.Now().Unix()}) + m.syncMCP(n) go m.heartbeat(n) go m.readLoop(n) } @@ -326,6 +328,63 @@ func (m *Manager) WriteFile(ctx context.Context, name string, p FileWritePayload return v.Bytes, nil } +func (m *Manager) syncMCP(n *Node) { + if n == nil || n.MCP == nil || m.Store == nil { + return + } + services, err := m.Store.ListMCPServicesForNode(n.Name) + if err != nil { + return + } + _ = n.write(Message{Event: "mcp_sync", Payload: services, TS: time.Now().Unix()}) +} + +func (m *Manager) SyncMCPServices() { + m.mu.RLock() + nodes := make([]*Node, 0, len(m.nodes)) + for _, n := range m.nodes { + nodes = append(nodes, n) + } + m.mu.RUnlock() + for _, n := range nodes { + if n.Connected.Load() { + m.syncMCP(n) + } + } +} + +func (m *Manager) ListMCP(name string) ([]MCPService, error) { + m.mu.RLock() + n := m.nodes[name] + m.mu.RUnlock() + if n == nil { + return nil, fmt.Errorf("device %s not found", name) + } + if n.MCP == nil { + n.MCP = NewMCPRuntime() + } + return n.MCP.List(), nil +} + +func (m *Manager) CallMCP(ctx context.Context, name, server, tool string, args any) (MCPCallResult, error) { + m.mu.RLock() + n := m.nodes[name] + m.mu.RUnlock() + if n == nil || !m.Online(name) { + return MCPCallResult{}, fmt.Errorf("device %s is offline", name) + } + msg, err := m.call(ctx, name, "mcp_call", map[string]any{"server": server, "tool": tool, "args": args}) + if err != nil { + return MCPCallResult{}, err + } + var out MCPCallResult + b, _ := json.Marshal(msg.Payload) + if err := json.Unmarshal(b, &out); err != nil { + return MCPCallResult{}, err + } + return out, nil +} + func (m *Manager) heartbeat(n *Node) { t := time.NewTicker(30 * time.Second) defer t.Stop() diff --git a/internal/cluster/mcp.go b/internal/cluster/mcp.go new file mode 100644 index 0000000..ff728a9 --- /dev/null +++ b/internal/cluster/mcp.go @@ -0,0 +1,329 @@ +package cluster + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "strings" + "sync" +) + +type MCPService struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Transport string `json:"transport"` + Command string `json:"command,omitempty"` + Args []string `json:"args,omitempty"` + Env map[string]string `json:"env,omitempty"` + URL string `json:"url,omitempty"` + Headers map[string]string `json:"headers,omitempty"` +} + +type MCPCallResult struct { + Server string `json:"server"` + Tool string `json:"tool"` + Result any `json:"result,omitempty"` + Error string `json:"error,omitempty"` + DurationMS int64 `json:"duration_ms"` +} + +type mcpRPC struct { + JSONRPC string `json:"jsonrpc"` + ID any `json:"id,omitempty"` + Method string `json:"method"` + Params any `json:"params,omitempty"` +} + +type mcpRPCResponse struct { + JSONRPC string `json:"jsonrpc"` + ID any `json:"id,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + Data any `json:"data,omitempty"` + } `json:"error,omitempty"` +} + +type MCPRuntime struct { + mu sync.RWMutex + services map[string]MCPService +} + +func NewMCPRuntime() *MCPRuntime { return &MCPRuntime{services: make(map[string]MCPService)} } + +func (r *MCPRuntime) Sync(services []MCPService) { + next := make(map[string]MCPService, len(services)) + for _, s := range services { + if s.ID == "" || s.Transport == "" { + continue + } + next[s.ID] = s + } + r.mu.Lock() + r.services = next + r.mu.Unlock() +} + +func (r *MCPRuntime) List() []MCPService { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]MCPService, 0, len(r.services)) + for _, s := range r.services { + out = append(out, s) + } + return out +} + +func (r *MCPRuntime) Get(id string) (MCPService, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + s, ok := r.services[id] + return s, ok +} + +func (r *MCPRuntime) Call(ctx context.Context, server, tool string, args any) (any, error) { + s, ok := r.Get(server) + if !ok { + return nil, fmt.Errorf("MCP service %q is not mounted", server) + } + switch strings.ToLower(s.Transport) { + case "stdio": + return callStdio(ctx, s, tool, args) + case "http", "sse": + return callHTTP(ctx, s, tool, args) + default: + return nil, fmt.Errorf("unsupported MCP transport %q", s.Transport) + } +} + +func callStdio(ctx context.Context, s MCPService, tool string, args any) (any, error) { + if s.Command == "" { + return nil, errors.New("MCP stdio command is empty") + } + cmd := exec.CommandContext(ctx, s.Command, s.Args...) + if len(s.Env) > 0 { + cmd.Env = append(os.Environ(), envList(s.Env)...) + } + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, err + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + cmd.Stderr = io.Discard + if err := cmd.Start(); err != nil { + return nil, err + } + defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }() + + enc := json.NewEncoder(stdin) + dec := bufio.NewReaderSize(stdout, 64*1024) + id := int64(1) + if err := enc.Encode(mcpRPC{JSONRPC: "2.0", ID: id, Method: "initialize", Params: map[string]any{ + "protocolVersion": "2024-11-05", "capabilities": map[string]any{}, + "clientInfo": map[string]string{"name": "vps-commander-agent", "version": "1.0"}, + }}); err != nil { + return nil, err + } + if _, err := readRPC(ctx, dec, id); err != nil { + return nil, err + } + if err := enc.Encode(mcpRPC{JSONRPC: "2.0", Method: "notifications/initialized", Params: map[string]any{}}); err != nil { + return nil, err + } + id++ + if err := enc.Encode(mcpRPC{JSONRPC: "2.0", ID: id, Method: "tools/call", Params: map[string]any{"name": tool, "arguments": args}}); err != nil { + return nil, err + } + return readRPC(ctx, dec, id) +} + +func readRPC(ctx context.Context, r *bufio.Reader, wantID int64) (any, error) { + for { + lineCh := make(chan []byte, 1) + errCh := make(chan error, 1) + go func() { + b, err := r.ReadBytes('\n') + if err != nil && len(b) == 0 { + errCh <- err + return + } + lineCh <- bytes.TrimSpace(b) + }() + var line []byte + select { + case <-ctx.Done(): + return nil, ctx.Err() + case err := <-errCh: + return nil, err + case line = <-lineCh: + } + if len(line) == 0 { + continue + } + var resp mcpRPCResponse + if json.Unmarshal(line, &resp) != nil { + continue + } + if resp.ID == nil { + continue + } + if n, ok := resp.ID.(float64); !ok || int64(n) != wantID { + continue + } + if resp.Error != nil { + return nil, fmt.Errorf("MCP error %d: %s", resp.Error.Code, resp.Error.Message) + } + var out any + if len(resp.Result) > 0 { + if err := json.Unmarshal(resp.Result, &out); err != nil { + return nil, err + } + } + return out, nil + } +} + +func callHTTP(ctx context.Context, s MCPService, tool string, args any) (any, error) { + if s.URL == "" { + return nil, errors.New("MCP HTTP URL is empty") + } + client := &http.Client{Timeout: 0} + var session string + post := func(req mcpRPC) (any, http.Header, error) { + b, _ := json.Marshal(req) + h := make(http.Header) + h.Set("Content-Type", "application/json") + h.Set("Accept", "application/json, text/event-stream") + for k, v := range s.Headers { + h.Set(k, v) + } + if session != "" { + h.Set("Mcp-Session-Id", session) + } + hr, err := http.NewRequestWithContext(ctx, http.MethodPost, s.URL, bytes.NewReader(b)) + if err != nil { + return nil, nil, err + } + hr.Header = h + resp, err := client.Do(hr) + if err != nil { + return nil, nil, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, resp.Header, fmt.Errorf("MCP HTTP status %s", resp.Status) + } + if v := resp.Header.Get("Mcp-Session-Id"); v != "" { + session = v + } + data, err := io.ReadAll(io.LimitReader(resp.Body, 4*1024*1024)) + if err != nil { + return nil, nil, err + } + if len(bytes.TrimSpace(data)) == 0 { + return nil, resp.Header, nil + } + if bytes.Contains(data, []byte("event:")) { + data = parseSSEData(data) + } + var rr mcpRPCResponse + if err := json.Unmarshal(bytes.TrimSpace(data), &rr); err != nil { + return nil, resp.Header, err + } + if rr.Error != nil { + return nil, resp.Header, fmt.Errorf("MCP error %d: %s", rr.Error.Code, rr.Error.Message) + } + var out any + if len(rr.Result) > 0 { + if err := json.Unmarshal(rr.Result, &out); err != nil { + return nil, resp.Header, err + } + } + return out, resp.Header, nil + } + if strings.EqualFold(s.Transport, "sse") { + endpoint, err := discoverSSEEndpoint(ctx, s) + if err != nil { + return nil, err + } + old := s.URL + s.URL = endpoint + defer func() { s.URL = old }() + } + if _, _, err := post(mcpRPC{JSONRPC: "2.0", ID: int64(1), Method: "initialize", Params: map[string]any{"protocolVersion": "2024-11-05", "capabilities": map[string]any{}, "clientInfo": map[string]string{"name": "vps-commander-agent", "version": "1.0"}}}); err != nil { + return nil, err + } + if _, _, err := post(mcpRPC{JSONRPC: "2.0", Method: "notifications/initialized", Params: map[string]any{}}); err != nil { + return nil, err + } + return func() (any, error) { + v, _, err := post(mcpRPC{JSONRPC: "2.0", ID: int64(2), Method: "tools/call", Params: map[string]any{"name": tool, "arguments": args}}) + return v, err + }() +} + +func discoverSSEEndpoint(ctx context.Context, s MCPService) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.URL, nil) + if err != nil { + return "", err + } + for k, v := range s.Headers { + req.Header.Set(k, v) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("MCP SSE status %s", resp.Status) + } + sc := bufio.NewScanner(io.LimitReader(resp.Body, 256*1024)) + isEndpoint := false + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "event:") { + isEndpoint = strings.TrimSpace(strings.TrimPrefix(line, "event:")) == "endpoint" + continue + } + if isEndpoint && strings.HasPrefix(line, "data:") { + u := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if strings.HasPrefix(u, "/") { + base := *resp.Request.URL + base.Path = u + base.RawQuery = "" + return base.String(), nil + } + return u, nil + } + } + return "", errors.New("MCP SSE endpoint event not received") +} + +func parseSSEData(b []byte) []byte { + for _, line := range strings.Split(string(b), "\n") { + if strings.HasPrefix(line, "data:") { + return []byte(strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + } + } + return b +} +func envList(m map[string]string) []string { + out := make([]string, 0, len(m)) + for k, v := range m { + out = append(out, k+"="+v) + } + return out +} diff --git a/internal/cluster/mcp_test.go b/internal/cluster/mcp_test.go new file mode 100644 index 0000000..449f189 --- /dev/null +++ b/internal/cluster/mcp_test.go @@ -0,0 +1,80 @@ +package cluster + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "testing" + "time" +) + +func TestMCPRuntimeStdioOnDemand(t *testing.T) { + if _, err := exec.LookPath("python3"); err != nil { + t.Skip("python3 unavailable") + } + dir := t.TempDir() + script := dir + "/mcp.py" + code := `import json,sys +for line in sys.stdin: + r=json.loads(line) + if r.get("method")=="initialize": + print(json.dumps({"jsonrpc":"2.0","id":r["id"],"result":{"protocolVersion":"2024-11-05"}}),flush=True) + elif r.get("method")=="tools/call": + a=r.get("params",{}).get("arguments",{}) + print(json.dumps({"jsonrpc":"2.0","id":r["id"],"result":{"content":[{"type":"text","text":"hello "+a.get("name","")}]}}),flush=True) +` + if err := os.WriteFile(script, []byte(code), 0755); err != nil { + t.Fatal(err) + } + r := NewMCPRuntime() + r.Sync([]MCPService{{ID: "test", Transport: "stdio", Command: "python3", Args: []string{script}}}) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + v, err := r.Call(ctx, "test", "hello", map[string]any{"name": "vpc"}) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(v) + if string(b) == "" { + t.Fatal("empty result") + } + if !contains(string(b), "hello vpc") { + t.Fatalf("unexpected %s", b) + } +} + +func TestMCPRuntimeHTTP(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + var q mcpRPC + _ = json.Unmarshal(b, &q) + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":{"ok":true}}`, jsonID(q.ID)) + })) + defer ts.Close() + r := NewMCPRuntime() + r.Sync([]MCPService{{ID: "http", Transport: "http", URL: ts.URL}}) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + v, err := r.Call(ctx, "http", "ping", map[string]any{}) + if err != nil { + t.Fatal(err) + } + if !contains(fmt.Sprint(v), "true") { + t.Fatalf("unexpected %v", v) + } +} +func jsonID(v any) string { b, _ := json.Marshal(v); return string(b) } +func contains(s, x string) bool { + for i := 0; i+len(x) <= len(s); i++ { + if s[i:i+len(x)] == x { + return true + } + } + return false +} diff --git a/internal/mcp/mcp.go b/internal/mcp/mcp.go index 52f1733..f4b838f 100644 --- a/internal/mcp/mcp.go +++ b/internal/mcp/mcp.go @@ -76,6 +76,11 @@ type ActionBackend interface { WriteFile(ctx context.Context, clientIP, device, path, content string) (any, error) } +type AgentMCPBackend interface { + ListAgentMCP(ctx context.Context, device string) (any, error) + CallAgentMCP(ctx context.Context, clientIP, device, server, tool string, args any) (any, error) +} + type Server struct { Backend ActionBackend Auth *auth.Manager @@ -272,6 +277,14 @@ func (s *Server) handleRequest(ctx context.Context, r *http.Request, req JSONRPC Required: []string{"path", "content"}, }, }, + { + Name: "list_agent_mcp", Description: "列出指定 Agent 当前通过 Hub 动态挂载的 MCP 服务及状态", + InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标 Agent 节点名称"}}, Required: []string{"device"}}, + }, + { + Name: "call_agent_mcp", Description: "通过 Hub 调度指定 Agent 上已挂载的 MCP 工具", + InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标 Agent 节点名称"}, "server": {Type: "string", Description: "MCP 服务 ID"}, "tool": {Type: "string", Description: "MCP 工具名称"}, "arguments": {Type: "object", Description: "MCP 工具参数 JSON 对象"}}, Required: []string{"device", "server", "tool"}}, + }, } return JSONRPCResponse{ JSONRPC: "2.0", @@ -372,6 +385,43 @@ func (s *Server) callTool(ctx context.Context, clientIP string, params CallToolP } return res, false + case "list_agent_mcp": + var args struct { + Device string `json:"device"` + } + _ = json.Unmarshal(params.Arguments, &args) + b, ok := s.Backend.(AgentMCPBackend) + if !ok { + return map[string]string{"error": "agent MCP unavailable"}, true + } + res, err := b.ListAgentMCP(ctx, args.Device) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + + case "call_agent_mcp": + var args struct { + Device string `json:"device"` + Server string `json:"server"` + Tool string `json:"tool"` + Arguments json.RawMessage `json:"arguments"` + } + _ = json.Unmarshal(params.Arguments, &args) + b, ok := s.Backend.(AgentMCPBackend) + if !ok { + return map[string]string{"error": "agent MCP unavailable"}, true + } + var av any + if len(args.Arguments) > 0 { + _ = json.Unmarshal(args.Arguments, &av) + } + res, err := b.CallAgentMCP(ctx, clientIP, args.Device, args.Server, args.Tool, av) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + default: return map[string]string{"error": "unknown tool: " + params.Name}, true } diff --git a/internal/mcp/mcp_test.go b/internal/mcp/mcp_test.go index cf40878..7c9e9ba 100644 --- a/internal/mcp/mcp_test.go +++ b/internal/mcp/mcp_test.go @@ -75,8 +75,8 @@ func TestMCPInitializeAndListTools(t *testing.T) { t.Fatalf("expected map result") } tools, ok := resMap["tools"].([]any) - if !ok || len(tools) != 4 { - t.Fatalf("expected 4 tools, got %d", len(tools)) + if !ok || len(tools) != 6 { + t.Fatalf("expected 6 tools, got %d", len(tools)) } } diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 5c5bd11..d68dcb6 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -2,6 +2,8 @@ package storage import ( "database/sql" + "encoding/json" + "errors" _ "modernc.org/sqlite" "os" "path/filepath" @@ -42,13 +44,104 @@ func (s *Store) init() error { "CREATE TABLE IF NOT EXISTS audit_logs (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp INTEGER NOT NULL, caller_ip TEXT, device TEXT, action TEXT, command TEXT, " + "exit_code INTEGER, duration_ms INTEGER, error_msg TEXT); " + - "CREATE TABLE IF NOT EXISTS revoked_devices (name TEXT PRIMARY KEY, revoked_at INTEGER NOT NULL, reason TEXT);") + "CREATE TABLE IF NOT EXISTS revoked_devices (name TEXT PRIMARY KEY, revoked_at INTEGER NOT NULL, reason TEXT); CREATE TABLE IF NOT EXISTS mcp_services (id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, transport TEXT NOT NULL, command TEXT, args_json TEXT, env_json TEXT, url TEXT, headers_json TEXT, scope TEXT NOT NULL DEFAULT 'all', target_nodes_json TEXT, enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); CREATE INDEX IF NOT EXISTS idx_mcp_services_enabled ON mcp_services(enabled);") if err != nil { return err } return s.migrateDevices() } +type MCPService struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Transport string `json:"transport"` + Command string `json:"command,omitempty"` + ArgsJSON string `json:"args_json,omitempty"` + EnvJSON string `json:"env_json,omitempty"` + URL string `json:"url,omitempty"` + HeadersJSON string `json:"headers_json,omitempty"` + Scope string `json:"scope"` + TargetNodesJSON string `json:"target_nodes_json,omitempty"` + Enabled bool `json:"enabled"` + CreatedAt int64 `json:"created_at"` + UpdatedAt int64 `json:"updated_at"` +} + +func (s *Store) SaveMCPService(v MCPService) error { + now := time.Now().Unix() + if v.CreatedAt <= 0 { + v.CreatedAt = now + } + v.UpdatedAt = now + if v.Scope == "" { + v.Scope = "all" + } + if v.Transport == "" { + return errors.New("transport required") + } + _, err := s.DB.Exec(`INSERT INTO mcp_services(id,name,description,transport,command,args_json,env_json,url,headers_json,scope,target_nodes_json,enabled,created_at,updated_at) +VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET name=excluded.name,description=excluded.description,transport=excluded.transport,command=excluded.command,args_json=excluded.args_json,env_json=excluded.env_json,url=excluded.url,headers_json=excluded.headers_json,scope=excluded.scope,target_nodes_json=excluded.target_nodes_json,enabled=excluded.enabled,updated_at=excluded.updated_at`, + v.ID, v.Name, v.Description, v.Transport, v.Command, v.ArgsJSON, v.EnvJSON, v.URL, v.HeadersJSON, v.Scope, v.TargetNodesJSON, boolInt(v.Enabled), v.CreatedAt, v.UpdatedAt) + return err +} +func boolInt(v bool) int { + if v { + return 1 + } + return 0 +} +func (s *Store) DeleteMCPService(id string) error { + _, err := s.DB.Exec(`DELETE FROM mcp_services WHERE id=?`, id) + return err +} +func (s *Store) ListMCPServices() ([]MCPService, error) { + return s.listMCP(`SELECT id,name,description,transport,COALESCE(command,''),COALESCE(args_json,''),COALESCE(env_json,''),COALESCE(url,''),COALESCE(headers_json,''),scope,COALESCE(target_nodes_json,''),enabled,created_at,updated_at FROM mcp_services ORDER BY id`) +} +func (s *Store) ListMCPServicesForNode(node string) ([]MCPService, error) { + all, err := s.listMCP(`SELECT id,name,description,transport,COALESCE(command,''),COALESCE(args_json,''),COALESCE(env_json,''),COALESCE(url,''),COALESCE(headers_json,''),scope,COALESCE(target_nodes_json,''),enabled,created_at,updated_at FROM mcp_services WHERE enabled=1 ORDER BY id`) + if err != nil { + return nil, err + } + out := make([]MCPService, 0, len(all)) + for _, v := range all { + if v.Scope != "custom" || containsNode(v.TargetNodesJSON, node) { + out = append(out, v) + } + } + return out, nil +} +func containsNode(raw, node string) bool { + var a []string + if json.Unmarshal([]byte(raw), &a) != nil { + return false + } + for _, v := range a { + if v == node { + return true + } + } + return false +} +func (s *Store) listMCP(q string) ([]MCPService, error) { + rows, err := s.DB.Query(q) + if err != nil { + return nil, err + } + defer rows.Close() + out := []MCPService{} + for rows.Next() { + var v MCPService + var en int + if err := rows.Scan(&v.ID, &v.Name, &v.Description, &v.Transport, &v.Command, &v.ArgsJSON, &v.EnvJSON, &v.URL, &v.HeadersJSON, &v.Scope, &v.TargetNodesJSON, &en, &v.CreatedAt, &v.UpdatedAt); err != nil { + return nil, err + } + v.Enabled = en != 0 + out = append(out, v) + } + return out, rows.Err() +} + func (s *Store) migrateDevices() error { cols := map[string]bool{} rows, err := s.DB.Query("PRAGMA table_info(devices)") diff --git a/internal/web/panel.go b/internal/web/panel.go index cbb441f..23444ac 100644 --- a/internal/web/panel.go +++ b/internal/web/panel.go @@ -43,10 +43,50 @@ func (p *Panel) Handler() http.Handler { mux.HandleFunc("/panel/api/audits", p.protected(p.Runner.AuditsJSON)) mux.HandleFunc("/panel/api/exec", p.protected(p.Runner.PanelExec)) mux.HandleFunc("/panel/api/key/rotate", p.protected(p.Runner.RotateAPIKey)) + mux.HandleFunc("/panel/api/mcp/services", p.protected(p.mcpServices)) + mux.HandleFunc("/panel/api/mcp/service", p.protected(p.deleteMCPService)) + mux.HandleFunc("/panel/api/mcp/list", p.protected(p.listAgentMCP)) + mux.HandleFunc("/panel/api/mcp/call", p.protected(p.callAgentMCP)) mux.HandleFunc("/", p.index) return mux } +type mcpPanelRunner interface { + MCPServicesJSON(http.ResponseWriter, *http.Request) + DeleteMCPService(http.ResponseWriter, *http.Request) + ListAgentMCPJSON(http.ResponseWriter, *http.Request) + CallAgentMCPJSON(http.ResponseWriter, *http.Request) +} + +func (p *Panel) mcpServices(w http.ResponseWriter, r *http.Request) { + if v, ok := p.Runner.(mcpPanelRunner); ok { + v.MCPServicesJSON(w, r) + return + } + http.Error(w, "MCP API unavailable", 501) +} +func (p *Panel) deleteMCPService(w http.ResponseWriter, r *http.Request) { + if v, ok := p.Runner.(mcpPanelRunner); ok { + v.DeleteMCPService(w, r) + return + } + http.Error(w, "MCP API unavailable", 501) +} +func (p *Panel) listAgentMCP(w http.ResponseWriter, r *http.Request) { + if v, ok := p.Runner.(mcpPanelRunner); ok { + v.ListAgentMCPJSON(w, r) + return + } + http.Error(w, "MCP API unavailable", 501) +} +func (p *Panel) callAgentMCP(w http.ResponseWriter, r *http.Request) { + if v, ok := p.Runner.(mcpPanelRunner); ok { + v.CallAgentMCPJSON(w, r) + return + } + http.Error(w, "MCP API unavailable", 501) +} + func (p *Panel) handleDevices(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodDelete { p.Runner.DeleteDevice(w, r) diff --git a/internal/web/static/app.css b/internal/web/static/app.css index 2ca1636..9e42639 100644 --- a/internal/web/static/app.css +++ b/internal/web/static/app.css @@ -40,3 +40,6 @@ header strong{font-size:18px}header span{margin-left:12px;color:var(--text-muted .badge-docker-text.no{background:rgba(148,163,184,.15);color:var(--text-muted);border:1px solid var(--border)} .device-card.offline{opacity:.62;filter:saturate(.45)} .device-card.offline .btn-revoke{opacity:1} + +.mcp-toolbar{display:flex;justify-content:space-between;align-items:center;gap:12px;margin-bottom:14px}.mcp-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(280px,1fr));gap:12px}.mcp-item{border:1px solid var(--line,#30343b);border-radius:10px;padding:14px;background:var(--card,#11151a)}.mcp-item h3{margin:0 0 6px;font-size:15px}.mcp-meta{font-size:12px;opacity:.72;line-height:1.7}.mcp-actions{display:flex;gap:8px;margin-top:10px}.mcp-form{display:grid;grid-template-columns:1fr 1fr;gap:10px}.mcp-form .full{grid-column:1/-1}.mcp-form label{display:flex;flex-direction:column;gap:5px;font-size:12px;opacity:.9}.mcp-form input,.mcp-form select,.mcp-form textarea{width:100%;box-sizing:border-box}.mcp-form textarea{min-height:72px;font-family:monospace}.mcp-nodes{display:flex;flex-wrap:wrap;gap:8px}.mcp-nodes label{flex-direction:row;align-items:center}.mcp-status{font-size:12px;margin-top:8px;min-height:18px} +@media(max-width:700px){.mcp-form{grid-template-columns:1fr}.mcp-form .full{grid-column:auto}.mcp-toolbar{align-items:flex-start;flex-direction:column}} diff --git a/internal/web/static/app.js b/internal/web/static/app.js index 5bafa10..b92aa16 100644 --- a/internal/web/static/app.js +++ b/internal/web/static/app.js @@ -109,3 +109,22 @@ $("#logout").onclick=async()=>{ location.href="/panel/login" }; loadDevices().then(loadAudits).catch(console.error); + +let mcpEditingId=""; +async function loadMCPServices(){ + const services=await api("/mcp/services"); + const devices=(await api("/devices")).filter(d=>!d.is_local); + $("#mcpNodes").innerHTML=devices.map(d=>``).join(""); + $("#mcpList").innerHTML=services.map(s=>`

${esc(s.name)} ${esc(s.id)}

${esc(s.transport)} · ${s.scope==='custom'?'指定节点':'全部节点'} · ${s.enabled?'启用':'停用'}${s.url?' · '+esc(s.url):s.command?' · '+esc(s.command):''}
`).join("")||'
暂无 MCP 服务配置。
'; +} +function resetMCPForm(){ + mcpEditingId=""; $("#mcpEditor").hidden=false; $("#mcpId").disabled=false; ["mcpId","mcpName","mcpCommand","mcpUrl"].forEach(id=>$("#"+id).value=""); $("#mcpTransport").value="stdio"; $("#mcpEnabled").value="1"; $("#mcpArgs").value="[]"; $("#mcpHeaders").value="{}"; $("#mcpEnv").value="{}"; $("#mcpScope").value="all"; $("#mcpStatus").textContent=""; $("#mcpNodes").querySelectorAll("input").forEach(x=>x.checked=false); +} +async function editMCP(id){ + const services=await api("/mcp/services"); const s=services.find(x=>x.id===id); if(!s)return; resetMCPForm(); mcpEditingId=id; $("#mcpId").value=s.id; $("#mcpId").disabled=true; $("#mcpName").value=s.name||""; $("#mcpTransport").value=s.transport; $("#mcpEnabled").value=s.enabled?"1":"0"; $("#mcpCommand").value=s.command||""; $("#mcpUrl").value=s.url||""; $("#mcpArgs").value=s.args_json||"[]"; $("#mcpHeaders").value=s.headers_json||"{}"; $("#mcpEnv").value=s.env_json||"{}"; $("#mcpScope").value=s.scope||"all"; let nodes=[];try{nodes=JSON.parse(s.target_nodes_json||"[]")}catch(e){};$("#mcpNodes").querySelectorAll("input").forEach(x=>x.checked=nodes.includes(x.value)); +} +async function deleteMCP(id){if(!confirm(`确认删除 MCP 服务 [${id}]？`))return;try{await api("/mcp/service?id="+encodeURIComponent(id),{method:"DELETE"});await loadMCPServices()}catch(e){alert("删除失败: "+e.message)}} +$("#mcpNew").onclick=resetMCPForm; $("#mcpCancel").onclick=()=>$("#mcpEditor").hidden=true; +$("#mcpSave").onclick=async()=>{try{const scope=$("#mcpScope").value;const body={id:$("#mcpId").value.trim(),name:$("#mcpName").value.trim(),transport:$("#mcpTransport").value,command:$("#mcpCommand").value.trim(),url:$("#mcpUrl").value.trim(),scope,enabled:$("#mcpEnabled").value==="1",args:JSON.parse($("#mcpArgs").value||"[]"),headers:JSON.parse($("#mcpHeaders").value||"{}"),env:JSON.parse($("#mcpEnv").value||"{}"),target_nodes:[...$("#mcpNodes").querySelectorAll("input:checked")].map(x=>x.value)};if(!body.id||!body.name)throw Error("服务标识和显示名称不能为空");await api("/mcp/services",{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify(body)});$("#mcpStatus").textContent="已保存并向在线节点同步。";await loadMCPServices()}catch(e){$("#mcpStatus").textContent="保存失败: "+e.message}}; +$("#mcpTransport").onchange=()=>{const stdio=$("#mcpTransport").value==="stdio";$("#mcpCommand").parentElement.style.display=stdio?"":"none";$("#mcpArgs").parentElement.style.display=stdio?"":"none";$("#mcpUrl").parentElement.style.display=stdio?"none":""}; +loadMCPServices().catch(console.error); diff --git a/internal/web/static/index.html b/internal/web/static/index.html index 632123e..3715c04 100644 --- a/internal/web/static/index.html +++ b/internal/web/static/index.html @@ -1 +1,2 @@ -VPS-Commander
VPS-CommanderPrivate Infrastructure Console

设备看板

Actions 向导

当前 Hub 的 OpenAPI Schema：

/openapi.json
快速控制台
审计日志
时间设备动作命令/路径退出码耗时来源

+VPS-Commander
VPS-CommanderPrivate Infrastructure Console
设备看板
Actions 向导
当前 Hub 的 OpenAPI Schema：
/openapi.json
MCP 挂载

+
快速控制台
审计日志
时间设备动作命令/路径退出码耗时来源