diff --git a/internal/mcpserver/server.go b/internal/mcpserver/server.go index 6af11c7..f736af6 100644 --- a/internal/mcpserver/server.go +++ b/internal/mcpserver/server.go @@ -86,11 +86,12 @@ func New(cfg Config) (*Server, error) { } deps := tools.Deps{ - Exec: exectool.NewRunner(), - ProcFS: procfs.Default, - SysFS: sysfs.Default, - Logger: cfg.Logger, - Fleet: fleetReg, + Exec: exectool.NewRunner(), + ProcFS: procfs.Default, + SysFS: sysfs.Default, + Logger: cfg.Logger, + Fleet: fleetReg, + FleetToken: cfg.Token, } tools.RegisterAll(mcpSrv, deps) diff --git a/internal/tools/fleet_query.go b/internal/tools/fleet_query.go new file mode 100644 index 0000000..5f1ed9f --- /dev/null +++ b/internal/tools/fleet_query.go @@ -0,0 +1,145 @@ +package tools + +import ( + "context" + "fmt" + "net/http" + "strings" + "sync" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +type fleetQueryIn struct { + Tool string `json:"tool"` + Arguments map[string]any `json:"arguments,omitempty"` +} + +type fleetNodeResult struct { + NodeID string `json:"node_id"` + AdvertiseURL string `json:"advertise_url"` + Self bool `json:"self"` + Text string `json:"text,omitempty"` + Result map[string]any `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +type fleetQueryOut struct { + Tool string `json:"tool"` + Results []fleetNodeResult `json:"results"` +} + +func registerFleetQuery(s *mcp.Server, d Deps) { + mcp.AddTool(s, &mcp.Tool{ + Name: "fleet_query", + Description: "Fan-out a tool call to every node in the fleet (including self) and return " + + "per-node results. Requires fleet mode to be enabled. " + + "'tool' is the name of any FleetMind tool. " + + "Optional 'arguments' are passed verbatim to each node.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in fleetQueryIn) (*mcp.CallToolResult, fleetQueryOut, error) { + out := fleetQueryOut{Tool: in.Tool} + if d.Fleet == nil { + return textResult("fleet mode disabled"), out, nil + } + if in.Tool == "" { + return textResult("tool name is required"), out, nil + } + + roster := d.Fleet.Roster() + self := d.Fleet.Self() + + results := make([]fleetNodeResult, len(roster)) + var wg sync.WaitGroup + + for i, p := range roster { + wg.Add(1) + go func(idx int, advertiseURL, nodeID string) { + defer wg.Done() + nr := fleetNodeResult{ + NodeID: nodeID, + AdvertiseURL: advertiseURL, + Self: nodeID == self.NodeID, + } + res, err := peerCallTool(ctx, advertiseURL, d.FleetToken, in.Tool, in.Arguments) + if err != nil { + nr.Error = err.Error() + } else if res.IsError { + for _, c := range res.Content { + if tc, ok := c.(*mcp.TextContent); ok { + nr.Error = tc.Text + break + } + } + } else { + var parts []string + for _, c := range res.Content { + if tc, ok := c.(*mcp.TextContent); ok { + parts = append(parts, tc.Text) + } + } + nr.Text = strings.Join(parts, "\n") + if sc, ok := res.StructuredContent.(map[string]any); ok { + nr.Result = sc + } + } + results[idx] = nr + }(i, p.AdvertiseURL, p.NodeID) + } + wg.Wait() + + out.Results = results + var b strings.Builder + fmt.Fprintf(&b, "fleet_query %q — %d node(s):\n", in.Tool, len(results)) + for _, r := range results { + label := r.NodeID + if r.Self { + label += " (self)" + } + if r.Error != "" { + fmt.Fprintf(&b, " %-40s error: %s\n", label, r.Error) + } else if r.Text != "" { + fmt.Fprintf(&b, " %-40s %s\n", label, r.Text) + } else { + fmt.Fprintf(&b, " %-40s (no output)\n", label) + } + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: strings.TrimSpace(b.String())}}, + }, out, nil + }) +} + +// peerCallTool opens a short-lived MCP session to advertiseURL and calls the +// named tool, returning the raw result. A 30-second per-peer timeout is applied. +func peerCallTool(ctx context.Context, advertiseURL, token, toolName string, args map[string]any) (*mcp.CallToolResult, error) { + transport := &mcp.StreamableClientTransport{ + Endpoint: advertiseURL + "/mcp", + HTTPClient: &http.Client{ + Transport: &fleetTokenTransport{base: http.DefaultTransport, token: token}, + Timeout: 30 * time.Second, + }, + DisableStandaloneSSE: true, + } + c := mcp.NewClient(&mcp.Implementation{Name: "fleetmind", Version: "internal"}, nil) + session, err := c.Connect(ctx, transport, nil) + if err != nil { + return nil, fmt.Errorf("connect to %s: %w", advertiseURL, err) + } + defer session.Close() + if args == nil { + args = map[string]any{} + } + return session.CallTool(ctx, &mcp.CallToolParams{Name: toolName, Arguments: args}) +} + +type fleetTokenTransport struct { + base http.RoundTripper + token string +} + +func (t *fleetTokenTransport) RoundTrip(req *http.Request) (*http.Response, error) { + r := req.Clone(req.Context()) + r.Header.Set("Authorization", "Bearer "+t.token) + return t.base.RoundTrip(r) +} diff --git a/internal/tools/tools.go b/internal/tools/tools.go index eaef028..cd7e60b 100644 --- a/internal/tools/tools.go +++ b/internal/tools/tools.go @@ -24,8 +24,11 @@ type Deps struct { SysFS sysfs.Root Logger *slog.Logger // Fleet is the local fleet registry. Nil when fleet mode is disabled — - // list_fleet is still registered but reports an empty roster. + // list_fleet and fleet_query are still registered but report disabled. Fleet *fleet.Registry + // FleetToken is the shared bearer secret used to authenticate outbound + // MCP calls to fleet peers. Empty when fleet mode is disabled. + FleetToken string } // AllToolNames lists every MCP tool name RegisterAll attaches. Kept in sync @@ -51,6 +54,7 @@ var AllToolNames = []string{ "read_journal", "read_dmesg", "list_fleet", + "fleet_query", } // RegisterAll attaches every tool to s. @@ -69,6 +73,7 @@ func RegisterAll(s *mcp.Server, d Deps) { registerHardware(s, d) registerLogs(s, d) registerFleet(s, d) + registerFleetQuery(s, d) } // textResult builds an MCP CallToolResult with a single textual summary line.