From 005bcdcf87e80383e87144a4e09c417c4bf613ae Mon Sep 17 00:00:00 2001
From: Juan Fernandez <juan-fdz-hawa@users.noreply.github.com>
Date: Fri, 26 Jun 2026 12:07:13 -0400
Subject: [PATCH] fleet-mcp: run multi-host live queries via ad-hoc campaign so
 observer_plus works

Resolves #46005

Implement flow for ad-hoc distributed query campaign streamed over the
/api/v1/fleet/results/websocket endpoint, the same way the Fleet UI and
fleetctl run live queries.
---
 .../46005-fleet-mcp-live-query-observer-plus  |   1 +
 tools/fleet-mcp/.env.example                  |   8 +
 tools/fleet-mcp/README.md                     |   1 +
 tools/fleet-mcp/fleet_integration.go          | 245 +------------
 tools/fleet-mcp/fleet_integration_test.go     | 286 +++++++++++++--
 tools/fleet-mcp/go.mod                        |   4 +-
 tools/fleet-mcp/go.sum                        |   7 +-
 tools/fleet-mcp/live_query_campaign.go        | 342 ++++++++++++++++++
 tools/fleet-mcp/main.go                       |   5 -
 tools/fleet-mcp/mcp_server.go                 |   2 +-
 tools/fleet-mcp/mcp_tools_queries.go          |  23 +-
 tools/fleet-mcp/render.yaml                   |   4 +
 12 files changed, 629 insertions(+), 299 deletions(-)
 create mode 100644 changes/46005-fleet-mcp-live-query-observer-plus
 create mode 100644 tools/fleet-mcp/live_query_campaign.go

diff --git a/changes/46005-fleet-mcp-live-query-observer-plus b/changes/46005-fleet-mcp-live-query-observer-plus
new file mode 100644
index 0000000000..435ac83ac8
--- /dev/null
+++ b/changes/46005-fleet-mcp-live-query-observer-plus
@@ -0,0 +1 @@
+- Fixed fleet-mcp `run_live_query` returning a 403 error for users with the observer+ role. Multi-host live queries now run as an ad-hoc live query campaign (raw SQL, streamed over the results websocket) instead of creating a temporary saved query, so they require only the live-query permission that observer+ already has.
diff --git a/tools/fleet-mcp/.env.example b/tools/fleet-mcp/.env.example
index d962e0f7a5..377775fd04 100644
--- a/tools/fleet-mcp/.env.example
+++ b/tools/fleet-mcp/.env.example
@@ -43,6 +43,14 @@ FLEET_API_KEY=YOUR_FLEET_API_KEY
 # Verbosity: debug | info | warn | error
 LOG_LEVEL=info
 
+# ── Live queries ──────────────────────────────────────────────────────────────
+
+# How long run_live_query waits for hosts to report before returning. Accepts
+# any Go duration string (e.g. 25s, 1m). Multi-host runs stop early once every
+# online host has responded; this is just the upper bound. Mirrors the same
+# variable on the Fleet server — keep this >= the server's value. Default 25s.
+# FLEET_LIVE_QUERY_REST_PERIOD=25s
+
 # ── TLS (only if your Fleet uses a self-signed cert; pick AT MOST one) ────────
 
 # Option A: Skip TLS verification — DEV/TEST ONLY, never use in production.
diff --git a/tools/fleet-mcp/README.md b/tools/fleet-mcp/README.md
index 7f697657d7..4645002b38 100644
--- a/tools/fleet-mcp/README.md
+++ b/tools/fleet-mcp/README.md
@@ -95,6 +95,7 @@ Configure the server using environment variables or a `.env` file (in the same d
 | `LOG_LEVEL` | `info` | Log verbosity: `debug` / `info` / `warn` / `error`. Note: `debug` logs the route shape of every Fleet API call (path before query string only — no PII identifiers). Avoid `debug` in production deployments where logs are shipped to a centralized aggregator. |
 | `FLEET_TLS_SKIP_VERIFY` | `false` | Skip TLS certificate verification. **Hard-gated to localhost — the server refuses to start with this set and a non-loopback `FLEET_BASE_URL`.** Conflicts with `FLEET_CA_FILE`. |
 | `FLEET_CA_FILE` | *(optional)* | Path to a PEM CA certificate for self-signed Fleet instances |
+| `FLEET_LIVE_QUERY_REST_PERIOD` | `25s` | How long `run_live_query` waits for hosts to report before returning. Accepts any Go duration string (e.g. `25s`, `1m`). Multi-host runs stop early once every online host has responded; this is the upper bound for the wait. Mirrors the same variable on the Fleet server — keep this ≥ the server's value so the MCP doesn't give up before the server finishes the campaign. |
 
 Copy the provided template:
 
diff --git a/tools/fleet-mcp/fleet_integration.go b/tools/fleet-mcp/fleet_integration.go
index 3c6100e17f..b5e2db8fe6 100644
--- a/tools/fleet-mcp/fleet_integration.go
+++ b/tools/fleet-mcp/fleet_integration.go
@@ -3,10 +3,8 @@ package main
 import (
 	"bytes"
 	"context"
-	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -29,22 +27,6 @@ import (
 // of simultaneous requests.
 const teamFanOutConcurrency = 8
 
-// tempQueryNamePrefix is the prefix used by all transient saved queries created
-// by runMultiHostQuery. Sweeping leftover queries at startup uses this prefix
-// to find them.
-const tempQueryNamePrefix = "fleet-mcp-temp-"
-
-// randomHexSuffix returns a hex-encoded random string for unique temp-query
-// names. Falls back to time.Now().UnixNano() if crypto/rand is unavailable
-// (extremely unlikely, but the fallback keeps runMultiHostQuery functional).
-func randomHexSuffix(nBytes int) string {
-	b := make([]byte, nBytes)
-	if _, err := rand.Read(b); err != nil {
-		return strconv.FormatInt(time.Now().UnixNano(), 16)
-	}
-	return hex.EncodeToString(b)
-}
-
 // FleetClient represents a client for interacting with Fleet API
 type FleetClient struct {
 	baseURL    string
@@ -245,27 +227,7 @@ type AdHocQueryResponse struct {
 	Rows   []map[string]interface{} `json:"rows"`
 }
 
-// MultiQueryRunRequest is the body for running a saved query against multiple hosts
-type MultiQueryRunRequest struct {
-	HostIDs []uint `json:"host_ids,omitempty"`
-}
-
-// LiveQueryHostResult is a single host's result from a multi-host query run
-type LiveQueryHostResult struct {
-	HostID uint                     `json:"host_id"`
-	Rows   []map[string]interface{} `json:"rows"`
-	Error  *string                  `json:"error"`
-}
-
-// MultiQueryRunResponse is the response from POST /api/v1/fleet/queries/:id/run
-type MultiQueryRunResponse struct {
-	QueryID            uint                  `json:"query_id"`
-	TargetedHostCount  int                   `json:"targeted_host_count"`
-	RespondedHostCount int                   `json:"responded_host_count"`
-	Results            []LiveQueryHostResult `json:"results"`
-}
-
-// LiveQueryResult is a unified result returned from RunLiveQuery
+// LiveQueryResult is a unified result returned from a live query run.
 type LiveQueryResult struct {
 	TargetedHostCount  int                      `json:"targeted_host_count"`
 	RespondedHostCount int                      `json:"responded_host_count"`
@@ -1539,10 +1501,6 @@ func (fc *FleetClient) CreateSavedQuery(ctx context.Context, name, description,
 	return &result.Query, nil
 }
 
-// RunLiveQuery executes a live query against the specified targets using Fleet's modern REST API.
-// Uses targeted API calls per dimension to avoid fetching all hosts.
-// For single hosts: uses per-host ad hoc endpoint (POST /api/v1/fleet/hosts/:id/query).
-// For multiple hosts: creates a temp saved query → runs by ID → deletes it.
 // LiveQueryTargetSpec captures every dimension that scopes a live query.
 //
 // Filter dimensions (Fleet / Platform / Label / Status / Query / PolicyID /
@@ -1730,78 +1688,6 @@ func intersectHostsByID(a, b []Endpoint) []Endpoint {
 	return out
 }
 
-func (fc *FleetClient) RunLiveQuery(ctx context.Context, sql string, hostnames, labels, platforms, teams []string) (*LiveQueryResult, error) {
-	// Legacy entry point — preserved so existing callers keep working.
-	// New code should use RunLiveQueryWithSpec for full filter dimensions.
-	spec := LiveQueryTargetSpec{
-		Hostnames:       hostnames,
-		LegacyLabels:    labels,
-		LegacyPlatforms: platforms,
-		LegacyFleets:    teams,
-	}
-	return fc.RunLiveQueryWithSpec(ctx, sql, spec)
-}
-
-// RunLiveQueryWithSpec resolves the spec to an exact target host list using
-// the same intersection semantics as ResolveLiveQueryTargets, then dispatches
-// to single-host or multi-host osquery distribution.
-//
-// When spec.Fleet (or the legacy spec.LegacyFleets[0]) is set, the team is
-// resolved here and the team_id is threaded through runMultiHostQuery so the
-// transient saved query is created under that team instead of Global. The
-// host targeting itself is already team-scoped via ResolveLiveQueryTargets;
-// this additionally aligns the saved-query ownership / RBAC with the team.
-func (fc *FleetClient) RunLiveQueryWithSpec(ctx context.Context, sql string, spec LiveQueryTargetSpec) (*LiveQueryResult, error) {
-	targets, err := fc.ResolveLiveQueryTargets(ctx, spec)
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve target hosts: %w", err)
-	}
-	if len(targets) == 0 {
-		return nil, fmt.Errorf("no matching hosts found for the provided targets")
-	}
-
-	teamID, err := fc.resolveLiveQueryTeamID(ctx, spec)
-	if err != nil {
-		return nil, err
-	}
-
-	hostIDs := make([]uint, 0, len(targets))
-	nameByID := make(map[uint]Endpoint, len(targets))
-	for _, t := range targets {
-		hostIDs = append(hostIDs, t.ID)
-		nameByID[t.ID] = t
-	}
-
-	if len(hostIDs) == 1 {
-		// Ad-hoc single-host path uses POST /hosts/:id/query directly — no
-		// saved query is created so team scoping does not apply.
-		return fc.runAdHocSingleHost(ctx, hostIDs[0], sql, nameByID)
-	}
-	return fc.runMultiHostQuery(ctx, hostIDs, sql, nameByID, teamID)
-}
-
-// resolveLiveQueryTeamID translates spec.Fleet (with LegacyFleets fallback)
-// into a *uint team_id suitable for CreateSavedQuery / CreateQueryRequest.
-// Returns (nil, nil) when no team is requested — that's the Global scope.
-func (fc *FleetClient) resolveLiveQueryTeamID(ctx context.Context, spec LiveQueryTargetSpec) (*uint, error) {
-	teamName := strings.TrimSpace(spec.Fleet)
-	if teamName == "" && len(spec.LegacyFleets) > 0 {
-		teamName = strings.TrimSpace(spec.LegacyFleets[0])
-	}
-	if teamName == "" {
-		return nil, nil
-	}
-	ids, err := fc.resolveTeamNames(ctx, []string{teamName})
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve fleet %q for query scoping: %w", teamName, err)
-	}
-	if len(ids) == 0 {
-		return nil, fmt.Errorf("fleet %q resolved to no team IDs", teamName)
-	}
-	id := ids[0]
-	return &id, nil
-}
-
 // runAdHocSingleHost uses POST /api/v1/fleet/hosts/:id/query (Fleet 4.43+ synchronous REST).
 func (fc *FleetClient) runAdHocSingleHost(ctx context.Context, hostID uint, sql string, endpointByID map[uint]Endpoint) (*LiveQueryResult, error) {
 	endpointPath := fmt.Sprintf("/api/v1/fleet/hosts/%d/query", hostID)
@@ -1850,103 +1736,6 @@ func (fc *FleetClient) runAdHocSingleHost(ctx context.Context, hostID uint, sql
 	}, nil
 }
 
-// runMultiHostQuery creates a temporary saved query, runs it by ID, then deletes it.
-// Uses POST /api/v1/fleet/queries/:id/run (Fleet 4.43+ synchronous REST).
-//
-// The temp query name pairs a millisecond timestamp with 8 random bytes — the
-// timestamp keeps lexical order useful for log scans, the random suffix makes
-// concurrent invocations from the same MCP process collision-proof. If the
-// DELETE in the deferred cleanup fails (network blip, Fleet 5xx, MCP killed),
-// the leftover is logged at error level so an operator can run the startup
-// sweeper or clean it up by hand. SweepLeftoverTempQueries() also removes any
-// such residue at next MCP boot.
-func (fc *FleetClient) runMultiHostQuery(ctx context.Context, hostIDs []uint, sql string, endpointByID map[uint]Endpoint, teamID *uint) (*LiveQueryResult, error) {
-	tempName := fmt.Sprintf("%s%d-%s", tempQueryNamePrefix, time.Now().UnixMilli(), randomHexSuffix(8))
-	// teamID propagates from the caller's spec.Fleet — when set, the temp
-	// saved query lives under that team (Fleet) instead of Global, so RBAC,
-	// listings, and audit trail all reflect the intended scope.
-	savedQuery, err := fc.CreateSavedQuery(ctx, tempName, "Temporary MCP live query", sql, "", teamID)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create temporary query: %w", err)
-	}
-	defer func() {
-		// Detach from the request ctx — if the caller cancelled (MCP client
-		// hung up, request timeout), we still want to clean up the temp
-		// query rather than wait for the next startup sweep. Bound the
-		// detached call with a short timeout so a wedged Fleet doesn't pin
-		// the goroutine forever.
-		cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer cancel()
-		delEndpoint := fmt.Sprintf("/api/v1/fleet/reports/id/%d", savedQuery.ID)
-		r, delErr := fc.makeFleetRequest(cleanupCtx, "DELETE", delEndpoint, nil)
-		if r != nil {
-			r.Body.Close()
-		}
-		if delErr != nil {
-			logrus.Errorf("failed to delete temp query %s (id=%d): %v — will be swept on next startup", tempName, savedQuery.ID, delErr)
-		} else if r != nil && r.StatusCode != http.StatusOK && r.StatusCode != http.StatusNoContent {
-			logrus.Errorf("temp query DELETE returned status %d for %s (id=%d) — will be swept on next startup", r.StatusCode, tempName, savedQuery.ID)
-		}
-	}()
-
-	logrus.Infof("Created temp query ID=%d, running against %d hosts", savedQuery.ID, len(hostIDs))
-
-	runEndpoint := fmt.Sprintf("/api/v1/fleet/reports/%d/run", savedQuery.ID)
-	resp, err := fc.makeFleetRequest(ctx, "POST", runEndpoint, MultiQueryRunRequest{HostIDs: hostIDs})
-	if err != nil {
-		return nil, fmt.Errorf("failed to run live query: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(resp.Body)
-		return nil, fmt.Errorf("live query run failed with status %d: %s", resp.StatusCode, string(body))
-	}
-
-	var runResp MultiQueryRunResponse
-	if err := json.NewDecoder(resp.Body).Decode(&runResp); err != nil {
-		return nil, fmt.Errorf("failed to decode live query run response: %w", err)
-	}
-
-	var enriched []map[string]interface{}
-	for _, r := range runResp.Results {
-		row := map[string]interface{}{
-			"host_id": r.HostID,
-			"rows":    r.Rows,
-		}
-		if ep, ok := endpointByID[r.HostID]; ok {
-			name := ep.DisplayName
-			if name == "" {
-				name = ep.Name
-			}
-			row["host_name"] = name
-		}
-		if r.Error != nil {
-			row["error"] = *r.Error
-		}
-		enriched = append(enriched, row)
-	}
-
-	return &LiveQueryResult{
-		TargetedHostCount:  runResp.TargetedHostCount,
-		RespondedHostCount: runResp.RespondedHostCount,
-		Results:            enriched,
-	}, nil
-}
-
-// isTempQueryName reports whether name marks a transient saved query
-// created by runMultiHostQuery. Tolerates the "[<team>] " prefix that
-// GetQueries prepends to team-scoped queries so team-scoped temp queries
-// are detected alongside global ones.
-func isTempQueryName(name string) bool {
-	if strings.HasPrefix(name, "[") {
-		if idx := strings.Index(name, "] "); idx > 0 {
-			name = name[idx+2:]
-		}
-	}
-	return strings.HasPrefix(name, tempQueryNamePrefix)
-}
-
 // endpointMatchesHostname reports whether ep's hostname-like fields (Name,
 // ComputerName, DisplayName) equal name case-insensitively. Used to verify
 // a singleton substring hit from Fleet's /hosts?query= actually matched on
@@ -1957,38 +1746,6 @@ func endpointMatchesHostname(ep Endpoint, name string) bool {
 		strings.EqualFold(ep.DisplayName, name)
 }
 
-// SweepLeftoverTempQueries deletes any saved queries whose name begins with
-// tempQueryNamePrefix. Called once at MCP startup to clean up residue from
-// previous runMultiHostQuery invocations whose deferred DELETE failed (process
-// killed mid-run, Fleet 5xx, network partition). Best-effort: errors are
-// logged but do not block startup.
-func (fc *FleetClient) SweepLeftoverTempQueries(ctx context.Context) {
-	queries, err := fc.GetQueries(ctx)
-	if err != nil {
-		logrus.Warnf("temp-query sweep: failed to list queries: %v", err)
-		return
-	}
-	swept := 0
-	for _, q := range queries {
-		if !isTempQueryName(q.Name) {
-			continue
-		}
-		delEndpoint := fmt.Sprintf("/api/v1/fleet/reports/id/%d", q.ID)
-		r, err := fc.makeFleetRequest(ctx, "DELETE", delEndpoint, nil)
-		if r != nil {
-			r.Body.Close()
-		}
-		if err != nil {
-			logrus.Warnf("temp-query sweep: failed to delete %s (id=%d): %v", q.Name, q.ID, err)
-			continue
-		}
-		swept++
-	}
-	if swept > 0 {
-		logrus.Infof("temp-query sweep: deleted %d leftover %s* queries", swept, tempQueryNamePrefix)
-	}
-}
-
 // makeFleetRequest builds and executes a Fleet API request bound to ctx.
 //
 // ctx propagation: when the MCP caller cancels the request (client disconnect,
diff --git a/tools/fleet-mcp/fleet_integration_test.go b/tools/fleet-mcp/fleet_integration_test.go
index 6d79160b52..9538a87a6d 100644
--- a/tools/fleet-mcp/fleet_integration_test.go
+++ b/tools/fleet-mcp/fleet_integration_test.go
@@ -9,6 +9,9 @@ import (
 	"strings"
 	"sync/atomic"
 	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
 )
 
 func newTestClient(serverURL string) *FleetClient {
@@ -19,30 +22,6 @@ func newTestClient(serverURL string) *FleetClient {
 	}
 }
 
-func TestIsTempQueryName(t *testing.T) {
-	cases := []struct {
-		name string
-		in   string
-		want bool
-	}{
-		{"global temp", tempQueryNamePrefix + "1234-abc", true},
-		{"team-scoped temp", "[Workstations] " + tempQueryNamePrefix + "1234-abc", true},
-		{"team-scoped temp with emoji", "[💻 Workstations] " + tempQueryNamePrefix + "1234-abc", true},
-		{"unrelated global query", "Top-level CPU usage", false},
-		{"unrelated team query", "[Servers] Disk space check", false},
-		{"prefix substring not at start", "prefixed-" + tempQueryNamePrefix + "abc", false},
-		{"empty", "", false},
-		{"just brackets", "[abc]", false},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := isTempQueryName(tc.in); got != tc.want {
-				t.Errorf("isTempQueryName(%q) = %v, want %v", tc.in, got, tc.want)
-			}
-		})
-	}
-}
-
 func TestEndpointMatchesHostname(t *testing.T) {
 	cases := []struct {
 		name string
@@ -297,16 +276,16 @@ func TestValidateCVEID(t *testing.T) {
 		wantErr bool
 	}{
 		{"CVE-2026-12345", false},
-		{"CVE-1999-0001", false},     // 4-digit minimum
+		{"CVE-1999-0001", false},      // 4-digit minimum
 		{"  CVE-2026-12345  ", false}, // trims
 		{"", true},
 		{"   ", true},
-		{"cve-2026-12345", true},   // case-sensitive
-		{"CVE-26-12345", true},     // year too short
-		{"CVE-2026-123", true},     // suffix too short
-		{"CVE-2026-12345x", true},  // trailing junk
-		{"CVE-2026", true},         // missing suffix
-		{"<script>", true},         // injection-shaped junk
+		{"cve-2026-12345", true},  // case-sensitive
+		{"CVE-26-12345", true},    // year too short
+		{"CVE-2026-123", true},    // suffix too short
+		{"CVE-2026-12345x", true}, // trailing junk
+		{"CVE-2026", true},        // missing suffix
+		{"<script>", true},        // injection-shaped junk
 	}
 	for _, tc := range cases {
 		t.Run(tc.in, func(t *testing.T) {
@@ -403,3 +382,248 @@ func TestGetHostsForCVE_PaginatesTitles(t *testing.T) {
 		t.Errorf("expected 2 titles pages (100 + 30 short page), got %d", got)
 	}
 }
+
+// campaignTestServer stands up an httptest server that answers the campaign
+// create POST and upgrades the results websocket, then hands the connection to
+// drive() (after consuming the auth + select_campaign handshake) so each test
+// can script the frames the server sends back.
+func campaignTestServer(t *testing.T, campaignID uint, drive func(conn *websocket.Conn)) *httptest.Server {
+	up := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodPost && r.URL.Path == "/api/v1/fleet/reports/run":
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"campaign": map[string]interface{}{"id": campaignID},
+			})
+		case r.URL.Path == "/api/v1/fleet/results/websocket":
+			conn, err := up.Upgrade(w, r, nil)
+			if err != nil {
+				t.Errorf("websocket upgrade: %v", err)
+				return
+			}
+			defer conn.Close()
+			// Validate the client speaks the handshake protocol the real server
+			// enforces: an auth frame carrying a token, then a select_campaign
+			// frame naming this campaign.
+			var msg map[string]interface{}
+			if err := conn.ReadJSON(&msg); err != nil {
+				t.Errorf("read auth frame: %v", err)
+				return
+			}
+			if msg["type"] != "auth" {
+				t.Errorf("first frame type = %v, want auth", msg["type"])
+			}
+			if data, _ := msg["data"].(map[string]interface{}); data["token"] == "" || data["token"] == nil {
+				t.Errorf("auth frame missing token, got %v", msg["data"])
+			}
+			if err := conn.ReadJSON(&msg); err != nil {
+				t.Errorf("read select_campaign frame: %v", err)
+				return
+			}
+			if msg["type"] != "select_campaign" {
+				t.Errorf("second frame type = %v, want select_campaign", msg["type"])
+			}
+			// JSON numbers decode to float64 in an interface{} map.
+			if data, _ := msg["data"].(map[string]interface{}); data["campaign_id"] != float64(campaignID) {
+				t.Errorf("select_campaign campaign_id = %v, want %d", data["campaign_id"], campaignID)
+			}
+			drive(conn)
+		default:
+			t.Errorf("unexpected request %s %s", r.Method, r.URL.Path)
+			http.NotFound(w, r)
+		}
+	}))
+}
+
+func writeWSFrame(t *testing.T, conn *websocket.Conn, typ string, data interface{}) {
+	if err := conn.WriteJSON(map[string]interface{}{"type": typ, "data": data}); err != nil {
+		t.Errorf("server write %s frame: %v", typ, err)
+	}
+}
+
+// runMultiHostCampaign creates an ad-hoc campaign and streams results over the
+// websocket, aggregating each host's rows into one result.
+func TestRunMultiHostCampaign_AggregatesResults(t *testing.T) {
+	t.Setenv("FLEET_LIVE_QUERY_REST_PERIOD", "5s")
+	srv := campaignTestServer(t, 42, func(conn *websocket.Conn) {
+		writeWSFrame(t, conn, "totals", map[string]interface{}{"count": 2, "online": 2})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host": map[string]interface{}{"id": 10, "hostname": "h10", "display_name": "Host 10"},
+			"rows": []map[string]string{{"answer": "42"}},
+		})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host": map[string]interface{}{"id": 20, "hostname": "h20", "display_name": "Host 20"},
+			"rows": []map[string]string{{"answer": "43"}},
+		})
+		writeWSFrame(t, conn, "status", map[string]interface{}{"expected_results": 2, "actual_results": 2, "status": "finished"})
+	})
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	nameByID := map[uint]Endpoint{10: {ID: 10, Name: "host-10"}, 20: {ID: 20, Name: "host-20"}}
+	res, err := fc.runMultiHostCampaign(t.Context(), []uint{10, 20}, "SELECT 1;", nameByID)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.TargetedHostCount != 2 {
+		t.Errorf("TargetedHostCount = %d, want 2 (from totals)", res.TargetedHostCount)
+	}
+	if res.RespondedHostCount != 2 {
+		t.Errorf("RespondedHostCount = %d, want 2 (from status.actual_results)", res.RespondedHostCount)
+	}
+	if len(res.Results) != 2 {
+		t.Fatalf("len(Results) = %d, want 2", len(res.Results))
+	}
+	// The locally-resolved host name wins over the server-reported one.
+	for _, row := range res.Results {
+		if row["host_id"] == uint(10) && row["host_name"] != "host-10" {
+			t.Errorf("host 10 name = %v, want host-10", row["host_name"])
+		}
+	}
+}
+
+// Offline hosts never report, so the stream stops once every online host has
+// responded rather than waiting out the deadline.
+func TestRunMultiHostCampaign_StopsWhenOnlineHostsRespond(t *testing.T) {
+	t.Setenv("FLEET_LIVE_QUERY_REST_PERIOD", "5s")
+	srv := campaignTestServer(t, 7, func(conn *websocket.Conn) {
+		// 3 targeted, only 2 online; host 30 is offline and silent.
+		writeWSFrame(t, conn, "totals", map[string]interface{}{"count": 3, "online": 2})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host": map[string]interface{}{"id": 10}, "rows": []map[string]string{{"k": "v"}},
+		})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host": map[string]interface{}{"id": 20}, "rows": []map[string]string{{"k": "v"}},
+		})
+		writeWSFrame(t, conn, "status", map[string]interface{}{"expected_results": 2, "actual_results": 2, "status": "finished"})
+	})
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	nameByID := map[uint]Endpoint{10: {ID: 10}, 20: {ID: 20}, 30: {ID: 30}}
+	start := time.Now()
+	res, err := fc.runMultiHostCampaign(t.Context(), []uint{10, 20, 30}, "SELECT 1;", nameByID)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 4*time.Second {
+		t.Errorf("expected prompt return once online hosts responded, took %s", elapsed)
+	}
+	if res.TargetedHostCount != 3 {
+		t.Errorf("TargetedHostCount = %d, want 3", res.TargetedHostCount)
+	}
+	if res.RespondedHostCount != 2 {
+		t.Errorf("RespondedHostCount = %d, want 2", res.RespondedHostCount)
+	}
+	if len(res.Results) != 2 {
+		t.Errorf("len(Results) = %d, want 2 (offline host produced no row)", len(res.Results))
+	}
+}
+
+// A per-host osquery error in a result frame surfaces as an error on that host's
+// row without failing the whole query.
+func TestRunMultiHostCampaign_HostErrorRow(t *testing.T) {
+	t.Setenv("FLEET_LIVE_QUERY_REST_PERIOD", "5s")
+	srv := campaignTestServer(t, 1, func(conn *websocket.Conn) {
+		writeWSFrame(t, conn, "totals", map[string]interface{}{"count": 1, "online": 1})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host":  map[string]interface{}{"id": 10},
+			"rows":  []map[string]string{},
+			"error": "no such table: bogus",
+		})
+		writeWSFrame(t, conn, "status", map[string]interface{}{"expected_results": 1, "actual_results": 1, "status": "finished"})
+	})
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	res, err := fc.runMultiHostCampaign(t.Context(), []uint{10}, "SELECT * FROM bogus;", map[uint]Endpoint{10: {ID: 10}})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(res.Results) != 1 {
+		t.Fatalf("len(Results) = %d, want 1", len(res.Results))
+	}
+	if res.Results[0]["error"] != "no such table: bogus" {
+		t.Errorf("expected host error row, got %+v", res.Results[0])
+	}
+}
+
+// A failed campaign creation surfaces as an error (no websocket is opened).
+func TestRunMultiHostCampaign_CreateFails(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/api/v1/fleet/results/websocket" {
+			t.Errorf("websocket must not be dialed when campaign creation fails")
+		}
+		http.Error(w, `{"message":"boom"}`, http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	_, err := fc.runMultiHostCampaign(t.Context(), []uint{10, 20}, "SELECT 1;", map[uint]Endpoint{})
+	if err == nil {
+		t.Fatal("expected error on failed campaign creation, got nil")
+	}
+}
+
+// An "error" frame from the server (campaign not found, unauthorized, pubsub
+// failure) must surface as an error, not a silent empty result.
+func TestRunMultiHostCampaign_ServerErrorFrame(t *testing.T) {
+	t.Setenv("FLEET_LIVE_QUERY_REST_PERIOD", "5s")
+	srv := campaignTestServer(t, 99, func(conn *websocket.Conn) {
+		writeWSFrame(t, conn, "error", "cannot find campaign for ID 99")
+	})
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	_, err := fc.runMultiHostCampaign(t.Context(), []uint{10, 20}, "SELECT 1;", map[uint]Endpoint{})
+	if err == nil {
+		t.Fatal("expected error from server error frame, got nil")
+	}
+	if !strings.Contains(err.Error(), "cannot find campaign") {
+		t.Errorf("error %q should include the server's message", err)
+	}
+}
+
+// A mid-stream read failure (connection dropped before a terminal status) must
+// surface as an error rather than masquerading as an empty successful result.
+func TestRunMultiHostCampaign_StreamReadError(t *testing.T) {
+	t.Setenv("FLEET_LIVE_QUERY_REST_PERIOD", "5s")
+	srv := campaignTestServer(t, 5, func(conn *websocket.Conn) {
+		// Two hosts online but only one responds, then the connection drops
+		// abruptly (no "finished" status, no clean close handshake).
+		writeWSFrame(t, conn, "totals", map[string]interface{}{"count": 2, "online": 2})
+		writeWSFrame(t, conn, "result", map[string]interface{}{
+			"host": map[string]interface{}{"id": 10}, "rows": []map[string]string{{"k": "v"}},
+		})
+		conn.Close()
+	})
+	defer srv.Close()
+
+	fc := newTestClient(srv.URL)
+	_, err := fc.runMultiHostCampaign(t.Context(), []uint{10, 20}, "SELECT 1;", map[uint]Endpoint{10: {ID: 10}, 20: {ID: 20}})
+	if err == nil {
+		t.Fatal("expected error on abrupt stream drop, got nil")
+	}
+}
+
+func TestCampaignWebsocketURL(t *testing.T) {
+	cases := []struct {
+		base string
+		want string
+	}{
+		{"http://localhost:8080", "ws://localhost:8080/api/v1/fleet/results/websocket"},
+		{"https://fleet.example.com", "wss://fleet.example.com/api/v1/fleet/results/websocket"},
+		{"https://fleet.example.com/", "wss://fleet.example.com/api/v1/fleet/results/websocket"},
+	}
+	for _, tc := range cases {
+		fc := newTestClient(tc.base)
+		got, err := fc.campaignWebsocketURL()
+		if err != nil {
+			t.Errorf("campaignWebsocketURL(%q) error: %v", tc.base, err)
+			continue
+		}
+		if got != tc.want {
+			t.Errorf("campaignWebsocketURL(%q) = %q, want %q", tc.base, got, tc.want)
+		}
+	}
+}
diff --git a/tools/fleet-mcp/go.mod b/tools/fleet-mcp/go.mod
index 2d40b5d64a..0261fcfe3f 100644
--- a/tools/fleet-mcp/go.mod
+++ b/tools/fleet-mcp/go.mod
@@ -3,6 +3,7 @@ module fleet-mcp
 go 1.26.4
 
 require (
+	github.com/gorilla/websocket v1.5.1
 	github.com/joho/godotenv v1.5.1
 	github.com/mark3labs/mcp-go v0.44.0
 	github.com/sirupsen/logrus v1.9.3
@@ -18,6 +19,7 @@ require (
 	github.com/spf13/cast v1.7.1 // indirect
 	github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
-	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
+	golang.org/x/net v0.56.0 // indirect
+	golang.org/x/sys v0.46.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/tools/fleet-mcp/go.sum b/tools/fleet-mcp/go.sum
index 05c07b30f7..950b495412 100644
--- a/tools/fleet-mcp/go.sum
+++ b/tools/fleet-mcp/go.sum
@@ -11,6 +11,8 @@ github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
+github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
 github.com/invopop/jsonschema v0.13.0 h1:KvpoAJWEjR3uD9Kbm2HWJmqsEaHt8lBUpd0qHcIi21E=
 github.com/invopop/jsonschema v0.13.0/go.mod h1:ffZ5Km5SWWRAIN6wbDXItl95euhFz2uON45H2qjYt+0=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
@@ -40,8 +42,11 @@ github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/
 github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
-golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
+golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
+golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
diff --git a/tools/fleet-mcp/live_query_campaign.go b/tools/fleet-mcp/live_query_campaign.go
new file mode 100644
index 0000000000..413809364d
--- /dev/null
+++ b/tools/fleet-mcp/live_query_campaign.go
@@ -0,0 +1,342 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/sirupsen/logrus"
+)
+
+// createCampaignRequest is the body for POST /api/v1/fleet/reports/run. Selected
+// targets the campaign by already-resolved host IDs.
+type createCampaignRequest struct {
+	Query    string          `json:"query"`
+	Selected campaignTargets `json:"selected"`
+}
+
+type campaignTargets struct {
+	Hosts []uint `json:"hosts"`
+}
+
+// createCampaignResponse captures the campaign ID returned when the campaign is
+// created. Only the ID is needed to subscribe to the result stream.
+type createCampaignResponse struct {
+	Campaign struct {
+		ID uint `json:"id"`
+	} `json:"campaign"`
+}
+
+// wsJSONMessage mirrors server/websocket.JSONMessage on the read side. Data is
+// left raw and decoded per Type.
+type wsJSONMessage struct {
+	Type string          `json:"type"`
+	Data json.RawMessage `json:"data"`
+}
+
+// wsDistributedResult mirrors fleet.DistributedQueryResult (the "result" frame).
+type wsDistributedResult struct {
+	Host struct {
+		ID          uint   `json:"id"`
+		Hostname    string `json:"hostname"`
+		DisplayName string `json:"display_name"`
+	} `json:"host"`
+	Rows  []map[string]string `json:"rows"`
+	Error *string             `json:"error,omitempty"`
+}
+
+// wsTotals mirrors the service targetTotals struct (the "totals" frame).
+type wsTotals struct {
+	Total           uint `json:"count"`
+	Online          uint `json:"online"`
+	Offline         uint `json:"offline"`
+	MissingInAction uint `json:"missing_in_action"`
+}
+
+// wsStatus mirrors the service campaignStatus struct (the "status" frame).
+// ActualResults is the authoritative count of hosts that have reported
+// (with or without rows).
+type wsStatus struct {
+	ExpectedResults uint   `json:"expected_results"`
+	ActualResults   uint   `json:"actual_results"`
+	Status          string `json:"status"`
+}
+
+// runMultiHostCampaign runs raw SQL against the given hosts via an ad-hoc live
+// query campaign and returns the aggregated results. It is the multi-host live
+// query path for every role.
+func (fc *FleetClient) runMultiHostCampaign(ctx context.Context, hostIDs []uint, sql string, endpointByID map[uint]Endpoint) (*LiveQueryResult, error) {
+	resp, err := fc.makeFleetRequest(ctx, "POST", "/api/v1/fleet/reports/run", createCampaignRequest{
+		Query:    sql,
+		Selected: campaignTargets{Hosts: hostIDs},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create live query campaign: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("live query campaign creation failed with status %d: %s", resp.StatusCode, string(body))
+	}
+
+	var camp createCampaignResponse
+	if err := json.NewDecoder(resp.Body).Decode(&camp); err != nil {
+		return nil, fmt.Errorf("failed to decode campaign response: %w", err)
+	}
+	if camp.Campaign.ID == 0 {
+		return nil, fmt.Errorf("live query campaign creation returned no campaign id")
+	}
+
+	logrus.Infof("Created live query campaign ID=%d, streaming results for %d hosts", camp.Campaign.ID, len(hostIDs))
+	return fc.streamCampaignResults(ctx, camp.Campaign.ID, hostIDs, endpointByID)
+}
+
+// streamCampaignResults opens the results websocket, subscribes to the campaign,
+// and reads frames until all online hosts have reported or the live-query
+// deadline elapses. Partial results gathered before the deadline are returned.
+func (fc *FleetClient) streamCampaignResults(ctx context.Context, campaignID uint, hostIDs []uint, endpointByID map[uint]Endpoint) (*LiveQueryResult, error) {
+	wsURL, err := fc.campaignWebsocketURL()
+	if err != nil {
+		return nil, err
+	}
+
+	dialer := &websocket.Dialer{
+		Proxy:            http.ProxyFromEnvironment,
+		HandshakeTimeout: 45 * time.Second,
+		TLSClientConfig:  fc.tlsClientConfig(),
+	}
+	conn, _, err := dialer.DialContext(ctx, wsURL, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open live query results websocket: %w", err)
+	}
+	defer conn.Close()
+
+	// Authenticate with the same token used for REST calls, then subscribe to
+	// the campaign. WriteJSON marshals an arbitrary value, matching the server's
+	// JSONMessage{Type, Data} envelope.
+	if err := conn.WriteJSON(map[string]interface{}{"type": "auth", "data": map[string]string{"token": fc.apiKey}}); err != nil {
+		return nil, fmt.Errorf("failed to authenticate results websocket: %w", err)
+	}
+	if err := conn.WriteJSON(map[string]interface{}{"type": "select_campaign", "data": map[string]interface{}{"campaign_id": campaignID}}); err != nil {
+		return nil, fmt.Errorf("failed to subscribe to campaign: %w", err)
+	}
+
+	// Bound the read loop by the same deadline the synchronous REST endpoints
+	// use, so a wedged or offline-heavy fleet can't pin us indefinitely.
+	deadline := time.Now().Add(liveQueryDeadline())
+	_ = conn.SetReadDeadline(deadline)
+
+	// Closing the connection unblocks the blocking ReadJSON below, so a caller
+	// cancellation (MCP client hangs up) interrupts the stream immediately
+	// rather than waiting out the read deadline. The done channel stops this
+	// watcher when the function returns normally.
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			_ = conn.Close()
+		case <-done:
+		}
+	}()
+
+	results := make([]map[string]interface{}, 0, len(hostIDs))
+	var totals wsTotals
+	var status wsStatus
+	var gotTotals, gotStatus bool
+
+	for {
+		var msg wsJSONMessage
+		if err := conn.ReadJSON(&msg); err != nil {
+			// A cancelled context means the read was interrupted on purpose.
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				return nil, ctxErr
+			}
+			// The read deadline elapsing is the expected upper bound — return
+			// the partial results gathered so far rather than failing.
+			var ne net.Error
+			if errors.As(err, &ne) && ne.Timeout() {
+				break
+			}
+			// A clean server-side close after it has streamed its frames is a
+			// normal end of stream, not a failure.
+			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+				break
+			}
+			// Anything else — auth rejection, protocol violation, dropped
+			// connection — is a real error that must not masquerade as an empty
+			// successful result.
+			return nil, fmt.Errorf("live query campaign %d: results stream read failed: %w", campaignID, err)
+		}
+
+		switch msg.Type {
+		case "result":
+			var r wsDistributedResult
+			if err := json.Unmarshal(msg.Data, &r); err != nil {
+				logrus.Warnf("live query campaign %d: failed to decode result frame: %v", campaignID, err)
+				continue
+			}
+			results = append(results, fc.campaignResultRow(r, endpointByID))
+		case "totals":
+			if err := json.Unmarshal(msg.Data, &totals); err != nil {
+				logrus.Warnf("live query campaign %d: failed to decode totals frame: %v", campaignID, err)
+				continue
+			}
+			gotTotals = true
+		case "status":
+			if err := json.Unmarshal(msg.Data, &status); err != nil {
+				logrus.Warnf("live query campaign %d: failed to decode status frame: %v", campaignID, err)
+				continue
+			}
+			gotStatus = true
+		case "error":
+			// The server reports post-subscription failures (campaign not found,
+			// unauthorized, pubsub error) as an error frame, then closes the
+			// stream. Surface it instead of returning an empty success.
+			var em string
+			if err := json.Unmarshal(msg.Data, &em); err != nil {
+				em = string(msg.Data)
+			}
+			return nil, fmt.Errorf("live query campaign %d failed: %s", campaignID, em)
+		}
+
+		// The server reports a terminal "finished" status once the campaign
+		// completes. Treat that as authoritative so termination never depends
+		// solely on the count arithmetic below.
+		if gotStatus && status.Status == "finished" {
+			break
+		}
+
+		// Stop as soon as every online host has reported. Offline hosts never
+		// report, so waiting for them would just burn the deadline. The server
+		// only refreshes status on a 5s ticker, so also count result frames
+		// directly — that lets the common all-hosts-return-rows case finish
+		// promptly instead of idling until the next status tick.
+		if gotTotals && totals.Online == 0 {
+			break
+		}
+		if gotTotals && uint(len(results)) >= totals.Online {
+			break
+		}
+		if gotTotals && gotStatus && status.ActualResults >= totals.Online {
+			break
+		}
+	}
+
+	targeted := len(hostIDs)
+	if gotTotals {
+		targeted = int(totals.Total)
+	}
+	// Count both result frames (hosts that returned rows) and the server's
+	// status tally (which also counts hosts that responded with no rows), and
+	// take the larger — status can lag behind the result frames we just read.
+	responded := len(results)
+	if gotStatus && int(status.ActualResults) > responded {
+		responded = int(status.ActualResults)
+	}
+
+	return &LiveQueryResult{
+		TargetedHostCount:  targeted,
+		RespondedHostCount: responded,
+		Results:            results,
+	}, nil
+}
+
+// campaignResultRow converts a websocket result frame into the same row shape
+// produced by the single-host ad-hoc path, preferring the locally-resolved host
+// name and falling back to the name the server reports.
+func (fc *FleetClient) campaignResultRow(r wsDistributedResult, endpointByID map[uint]Endpoint) map[string]interface{} {
+	// A result frame means the host reported, so it is online — mirror the
+	// single-host path, which always carries a "status" field. The server
+	// injects host_hostname/host_display_name into every result row; strip them
+	// so the campaign rows match the single-host shape (host identity lives in
+	// the row's own host_id/host_name keys, not inside the osquery columns).
+	for _, qr := range r.Rows {
+		delete(qr, "host_hostname")
+		delete(qr, "host_display_name")
+	}
+
+	row := map[string]interface{}{
+		"host_id": r.Host.ID,
+		"status":  "online",
+		"rows":    r.Rows,
+	}
+
+	name := ""
+	if ep, ok := endpointByID[r.Host.ID]; ok {
+		name = ep.DisplayName
+		if name == "" {
+			name = ep.Name
+		}
+	}
+	if name == "" {
+		name = r.Host.DisplayName
+	}
+	if name == "" {
+		name = r.Host.Hostname
+	}
+	if name != "" {
+		row["host_name"] = name
+	}
+
+	if r.Error != nil {
+		row["error"] = *r.Error
+	}
+	return row
+}
+
+// campaignWebsocketURL derives the results websocket URL from the configured
+// base URL, mapping http→ws and https→wss.
+func (fc *FleetClient) campaignWebsocketURL() (string, error) {
+	u, err := url.Parse(fc.baseURL)
+	if err != nil {
+		return "", fmt.Errorf("invalid Fleet base URL %q: %w", fc.baseURL, err)
+	}
+	switch strings.ToLower(u.Scheme) {
+	case "https":
+		u.Scheme = "wss"
+	case "http":
+		u.Scheme = "ws"
+	default:
+		return "", fmt.Errorf("unsupported Fleet base URL scheme %q", u.Scheme)
+	}
+	u.Path = strings.TrimRight(u.Path, "/") + "/api/v1/fleet/results/websocket"
+	return u.String(), nil
+}
+
+// tlsClientConfig returns the TLS settings configured on the REST HTTP client
+// (skip-verify / custom CA) so the websocket dialer trusts the same way. Returns
+// nil when the transport isn't a *http.Transport (e.g. test clients), in which
+// case the dialer uses its defaults.
+func (fc *FleetClient) tlsClientConfig() *tls.Config {
+	if tr, ok := fc.httpClient.Transport.(*http.Transport); ok {
+		return tr.TLSClientConfig
+	}
+	return nil
+}
+
+// liveQueryDeadline is how long to wait for hosts to report. It mirrors the
+// synchronous REST endpoints' FLEET_LIVE_QUERY_REST_PERIOD (default 25s).
+func liveQueryDeadline() time.Duration {
+	period := os.Getenv("FLEET_LIVE_QUERY_REST_PERIOD")
+	if period == "" {
+		return 25 * time.Second
+	}
+	d, err := time.ParseDuration(period)
+	if err != nil {
+		logrus.Warnf("invalid FLEET_LIVE_QUERY_REST_PERIOD %q, defaulting to 25s: %v", period, err)
+		return 25 * time.Second
+	}
+	return d
+}
diff --git a/tools/fleet-mcp/main.go b/tools/fleet-mcp/main.go
index 2e11ee3bf8..8d7871f3c3 100644
--- a/tools/fleet-mcp/main.go
+++ b/tools/fleet-mcp/main.go
@@ -53,11 +53,6 @@ func main() {
 
 	fleetClient := NewFleetClient(config.FleetBaseURL, config.FleetAPIKey, config.TLSSkipVerify, config.TLSCAFile)
 
-	// Best-effort cleanup of any fleet-mcp-temp-* saved queries left over from
-	// previous runs whose DELETE failed. Synchronous so any temporary cleanup
-	// failures show up immediately in startup logs; errors are logged not fatal.
-	fleetClient.SweepLeftoverTempQueries(context.Background())
-
 	if *seed {
 		SeedFleet(config, fleetClient)
 		return
diff --git a/tools/fleet-mcp/mcp_server.go b/tools/fleet-mcp/mcp_server.go
index e96e36a0f3..b85269895c 100644
--- a/tools/fleet-mcp/mcp_server.go
+++ b/tools/fleet-mcp/mcp_server.go
@@ -21,7 +21,7 @@ CRITICAL WORKFLOW for any tool that takes a 'sql' argument (run_live_query, crea
 
 Schema freshness: the in-memory schema is refreshed periodically from https://raw.githubusercontent.com/fleetdm/fleet/main/schema/osquery_fleet_schema.json (the JSON behind https://fleetdm.com/tables). If you suspect a schema mismatch — e.g. fleet docs show a column the response is missing — call refresh_osquery_schema and try again.
 
-Team (Fleet) scoping: when the user names a team in the conversation (e.g. "Workstations", "Servers"), pass it as the 'fleet' argument to run_live_query and create_saved_query. Both tools then create the underlying saved query under that team — not Global — so it inherits the team's RBAC and shows up in the right place in the Fleet UI. Only omit 'fleet' when the user explicitly wants a Global-scope query.
+Team (Fleet) scoping: when the user names a team in the conversation (e.g. "Workstations", "Servers"), pass it as the 'fleet' argument. For run_live_query this restricts the targeted hosts to that team; for create_saved_query it creates the query under that team — not Global — so it inherits the team's RBAC and shows up in the right place in the Fleet UI. Only omit 'fleet' when the user explicitly wants Global scope.
 
 Skipping step 1 produces queries that parse and run but return wrong or empty results. Always verify before emitting SQL.`
 
diff --git a/tools/fleet-mcp/mcp_tools_queries.go b/tools/fleet-mcp/mcp_tools_queries.go
index 27de516e64..a05a6e8c44 100644
--- a/tools/fleet-mcp/mcp_tools_queries.go
+++ b/tools/fleet-mcp/mcp_tools_queries.go
@@ -259,7 +259,7 @@ func inferPlatformFromTargets(targets []Endpoint) string {
 
 func registerRunLiveQuery(s *server.MCPServer, fleetClient *FleetClient) {
 	tool := mcp.NewTool("run_live_query",
-		mcp.WithDescription("Step 2 of 2. MUST call get_osquery_schema(platform=<target>) (or prepare_live_query, which embeds the schema response) BEFORE writing the sql argument. This verifies column NAMES and TYPES against the canonical schema — many osquery columns are TEXT despite numeric-looking values (e.g. windows_update_history.result_code is TEXT 'Succeeded'/'Failed', not an integer). Skipping the schema check produces queries that run but silently return zero rows.\n\nResolve targets and run an OSQuery SQL statement against Fleet devices. Accepts the SAME filter dimensions as prepare_live_query (intersection across fleet, platform, label, status, query, policy, CVE, hostnames, host_ids). Resolved target set is included in the response so the caller sees exactly which hosts were queried.\n\nTeam scoping: when `fleet` is set, the transient saved query that this tool creates internally is also scoped to that team — visible only under that team in the Fleet UI / audit log, with that team's RBAC. When the user mentions a team (e.g. 'Workstations'), pass it as `fleet`; do not run queries Globally and rely on host filters alone.\n\nUse the smallest target set that answers the question. Example: a CVE remediation check should target only hosts impacted by that CVE — pass cve_id + fleet, not platform=all."),
+		mcp.WithDescription("Step 2 of 2. MUST call get_osquery_schema(platform=<target>) (or prepare_live_query, which embeds the schema response) BEFORE writing the sql argument. This verifies column NAMES and TYPES against the canonical schema — many osquery columns are TEXT despite numeric-looking values (e.g. windows_update_history.result_code is TEXT 'Succeeded'/'Failed', not an integer). Skipping the schema check produces queries that run but silently return zero rows.\n\nResolve targets and run an OSQuery SQL statement against Fleet devices. Accepts the SAME filter dimensions as prepare_live_query (intersection across fleet, platform, label, status, query, policy, CVE, hostnames, host_ids). Resolved target set is included in the response so the caller sees exactly which hosts were queried.\n\nTeam scoping: when `fleet` is set, only hosts in that team are targeted. When the user mentions a team (e.g. 'Workstations'), pass it as `fleet`; do not run queries Globally and rely on host filters alone.\n\nUse the smallest target set that answers the question. Example: a CVE remediation check should target only hosts impacted by that CVE — pass cve_id + fleet, not platform=all."),
 		mcp.WithString("sql", mcp.Required(), mcp.Description("The OSQuery SQL statement to run (e.g. 'SELECT * FROM os_version;')")),
 		mcp.WithString("fleet", mcp.Description("Fleet (team) name.")),
 		mcp.WithString("platform", mcp.Description("Platform: 'macos' / 'windows' / 'linux' / 'chromeos'.")),
@@ -274,11 +274,10 @@ func registerRunLiveQuery(s *server.MCPServer, fleetClient *FleetClient) {
 		mcp.WithString("labels", mcp.Description("LEGACY — first item only; prefer 'label'.")),
 		mcp.WithString("platforms", mcp.Description("LEGACY — first item only; prefer 'platform'.")),
 		mcp.WithString("fleets", mcp.Description("LEGACY — first item only; prefer 'fleet'.")),
-		// run_live_query fires osquery against every targeted device and creates
-		// (then deletes) a transient saved query on the Fleet server. Even when the
-		// SQL itself is a SELECT, it consumes device CPU, surfaces in EDR telemetry,
-		// and writes/deletes Fleet state. NOT read-only, IS destructive: MCP clients
-		// must prompt for explicit user approval.
+		// run_live_query fires osquery against every targeted device via an ad-hoc
+		// live query campaign. Even when the SQL itself is a SELECT, it consumes
+		// device CPU and surfaces in EDR telemetry. NOT read-only, IS destructive:
+		// MCP clients must prompt for explicit user approval.
 		mcp.WithReadOnlyHintAnnotation(false),
 		mcp.WithDestructiveHintAnnotation(true),
 		mcp.WithIdempotentHintAnnotation(false),
@@ -286,7 +285,7 @@ func registerRunLiveQuery(s *server.MCPServer, fleetClient *FleetClient) {
 	s.AddTool(tool, func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
 		logrus.Info("Tool invoked: run_live_query")
 		sql, err := request.RequireString("sql")
-		if err != nil || sql == "" {
+		if err != nil || strings.TrimSpace(sql) == "" {
 			return mcp.NewToolResultError("sql is required"), nil
 		}
 
@@ -345,14 +344,6 @@ func registerRunLiveQuery(s *server.MCPServer, fleetClient *FleetClient) {
 			}
 		}
 
-		// Resolve team scoping for the transient saved query that
-		// runMultiHostQuery creates. When `fleet` is set, the query lives
-		// under that team in Fleet's UI / RBAC instead of Global.
-		teamID, tErr := fleetClient.resolveLiveQueryTeamID(ctx, spec)
-		if tErr != nil {
-			return mcp.NewToolResultError(fmt.Sprintf("Team scoping failed: %v", tErr)), nil
-		}
-
 		hostIDs := make([]uint, 0, len(targets))
 		nameByID := make(map[uint]Endpoint, len(targets))
 		for _, t := range targets {
@@ -364,7 +355,7 @@ func registerRunLiveQuery(s *server.MCPServer, fleetClient *FleetClient) {
 		if len(hostIDs) == 1 {
 			results, err = fleetClient.runAdHocSingleHost(ctx, hostIDs[0], sql, nameByID)
 		} else {
-			results, err = fleetClient.runMultiHostQuery(ctx, hostIDs, sql, nameByID, teamID)
+			results, err = fleetClient.runMultiHostCampaign(ctx, hostIDs, sql, nameByID)
 		}
 		if err != nil {
 			return mcp.NewToolResultError(fmt.Sprintf("Failed to run live query: %v", err)), nil
diff --git a/tools/fleet-mcp/render.yaml b/tools/fleet-mcp/render.yaml
index 6448afe833..4853b9677f 100644
--- a/tools/fleet-mcp/render.yaml
+++ b/tools/fleet-mcp/render.yaml
@@ -15,3 +15,7 @@ services:
         sync: false
       - key: LOG_LEVEL
         value: info
+      # Upper bound run_live_query waits for hosts to report (Go duration).
+      # Keep >= the Fleet server's FLEET_LIVE_QUERY_REST_PERIOD. Default 25s.
+      - key: FLEET_LIVE_QUERY_REST_PERIOD
+        value: 25s