From 2a9d01d7f7c0d83669260be5d83291eb4028221b Mon Sep 17 00:00:00 2001 From: kory Date: Mon, 28 Sep 2026 13:20:42 +0800 Subject: [PATCH] fix: sync agent MCP payloads and query runtime state --- cmd/agent/main.go | 2 ++ cmd/mcp-stdio/main.go | 29 +++++++++++++++++++++++++++++ internal/cluster/manager.go | 30 ++++++++++++++++++++++++++---- 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/cmd/agent/main.go b/cmd/agent/main.go index c181781..7f27fb3 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -138,6 +138,8 @@ func run(hub, name, token string, insecure bool) error { func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRuntime *cluster.MCPRuntime, msg cluster.Message) error { switch msg.Action { + case "mcp_list": + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: mcpRuntime.List()}) case "mcp_call": var p struct { Server string `json:"server"` diff --git a/cmd/mcp-stdio/main.go b/cmd/mcp-stdio/main.go index 746153f..44610ec 100644 --- a/cmd/mcp-stdio/main.go +++ b/cmd/mcp-stdio/main.go @@ -9,6 +9,7 @@ import ( "fmt" "io" "net/http" + "net/url" "os" "strings" "time" @@ -147,3 +148,31 @@ func main() { os.Stdout.WriteString("\n") } } + +func (h *HubRemoteBackend) ListAgentMCP(ctx context.Context, device string) (any, error) { + var res any + err := h.get(ctx, "/api/v1/mcp/list?device="+url.QueryEscape(device), &res) + return res, err +} +func (h *HubRemoteBackend) CallAgentMCP(ctx context.Context, clientIP, device, server, tool string, args any) (any, error) { + payload := map[string]any{"device": device, "server": server, "tool": tool, "arguments": args} + var res any + return res, h.post("/api/v1/mcp/call", payload, &res) +} +func (h *HubRemoteBackend) get(ctx context.Context, path string, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(h.BaseURL, "/")+path, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+h.APIKey) + resp, err := h.HTTPClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + b, _ := io.ReadAll(resp.Body) + return fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(b)) + } + return json.NewDecoder(resp.Body).Decode(out) +} diff --git a/internal/cluster/manager.go b/internal/cluster/manager.go index d592bb2..0cd94a7 100644 --- a/internal/cluster/manager.go +++ b/internal/cluster/manager.go @@ -332,13 +332,25 @@ func (m *Manager) syncMCP(n *Node) { if n == nil || n.MCP == nil || m.Store == nil { return } - services, err := m.Store.ListMCPServicesForNode(n.Name) + stored, err := m.Store.ListMCPServicesForNode(n.Name) if err != nil { return } + services := make([]MCPService, 0, len(stored)) + for _, v := range stored { + services = append(services, decodeStoredMCPService(v)) + } _ = n.write(Message{Event: "mcp_sync", Payload: services, TS: time.Now().Unix()}) } +func decodeStoredMCPService(v storage.MCPService) MCPService { + out := MCPService{ID: v.ID, Name: v.Name, Description: v.Description, Transport: v.Transport, Command: v.Command, URL: v.URL} + _ = json.Unmarshal([]byte(v.ArgsJSON), &out.Args) + _ = json.Unmarshal([]byte(v.EnvJSON), &out.Env) + _ = json.Unmarshal([]byte(v.HeadersJSON), &out.Headers) + return out +} + func (m *Manager) SyncMCPServices() { m.mu.RLock() nodes := make([]*Node, 0, len(m.nodes)) @@ -357,13 +369,23 @@ 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 == nil || !m.Online(name) { + return nil, fmt.Errorf("device %s is offline", name) + } + msg, err := m.call(context.Background(), name, "mcp_list", nil) + if err != nil { + return nil, err + } + var out []MCPService + b, _ := json.Marshal(msg.Payload) + if err := json.Unmarshal(b, &out); err != nil { + return nil, err } if n.MCP == nil { n.MCP = NewMCPRuntime() } - return n.MCP.List(), nil + n.MCP.Sync(out) + return out, nil } func (m *Manager) CallMCP(ctx context.Context, name, server, tool string, args any) (MCPCallResult, error) {