Add mcp_listening_servers table (#34286)

**Related issue:** Resolves #34330 # Checklist for submitter If some of the following don't apply, delete the relevant line. - [x] Changes file added for user-visible changes in `changes/`, `orbit/changes/` or `ee/fleetd-chrome/changes`. See [Changes files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files) for more information. - [x] Input data is properly validated, `SELECT *` is avoided, SQL injection is prevented (using placeholders for values in statements) ## Testing - [x] Added/updated automated tests - [x] QA'd all new/changed functionality manually (so far just macOS) ## fleetd/orbit/Fleet Desktop - [x] Verified compatibility with the latest released version of Fleet (see [Must rule](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/workflows/fleetd-development-and-release-strategy.md)) - [x] Verified that fleetd runs on macOS, Linux and Windows  ## Summary by CodeRabbit * **New Features** * Added a built-in mcp_listening_servers table to discover MCP servers by inspecting listening ports and probing endpoints; returns process info, server metadata, capabilities, tools, prompts, and resources (supports macOS, Windows, Linux). * **Tests** * Added comprehensive unit tests covering detection, IPv6 handling, SSE responses, and session lifecycle.
2025-10-21 11:25:11 -07:00
parent caa57019f0
commit 6726bb196f
10 changed files with 1073 additions and 6 deletions
@@ -0,0 +1 @@
+* Add `mcp_listening_servers` table to find MCP servers listening over HTTP.
@@ -50,10 +50,10 @@ func main() {
 		log.Fatalln(err)
 	}

-	plugins := orbittable.OrbitDefaultTables()
 	opts := orbittable.PluginOpts{
 		Socket: *socket,
 	}
+	plugins := orbittable.OrbitDefaultTables(opts)
 	platformTables, err := orbittable.PlatformTables(opts)
 	if err != nil {
 		log.Fatalln(err)
@@ -330,7 +330,6 @@ func (r *Runner) installWithRetry(ctx context.Context, installer *fleet.Software

 // attemptInstall performs a single installation attempt (download + install)
 func (r *Runner) attemptInstall(ctx context.Context, installer *fleet.SoftwareInstallDetails, payload *fleet.HostSoftwareInstallResultPayload, logger zerolog.Logger) (*fleet.HostSoftwareInstallResultPayload, error) {
-
 	tmpDirFn := r.tempDirFn
 	if tmpDirFn == nil {
 		tmpDirFn = os.MkdirTemp
@@ -12,6 +12,7 @@ import (
 	"github.com/fleetdm/fleet/v4/orbit/pkg/table/dataflattentable"
 	"github.com/fleetdm/fleet/v4/orbit/pkg/table/firefox_preferences"
 	"github.com/fleetdm/fleet/v4/orbit/pkg/table/fleetd_logs"
+	"github.com/fleetdm/fleet/v4/orbit/pkg/table/mcp_listening_servers"
 	"github.com/fleetdm/fleet/v4/orbit/pkg/table/sntp_request"
 	"github.com/macadmins/osquery-extension/tables/chromeuserprofiles"
 	"github.com/macadmins/osquery-extension/tables/fileline"
@@ -107,9 +108,8 @@ func (r *Runner) Execute() error {
 		}
 	}

-	plugins := OrbitDefaultTables()
-
 	opts := PluginOpts{Socket: r.socket}
+	plugins := OrbitDefaultTables(opts)
 	platformTables, err := PlatformTables(opts)
 	if err != nil {
 		return fmt.Errorf("populating platform tables: %w", err)
@@ -132,7 +132,7 @@ func (r *Runner) Execute() error {
 	return nil
 }

-func OrbitDefaultTables() []osquery.OsqueryPlugin {
+func OrbitDefaultTables(opts PluginOpts) []osquery.OsqueryPlugin {
 	plugins := []osquery.OsqueryPlugin{
 		// MacAdmins extensions.
 		table.NewPlugin("puppet_info", puppet.PuppetInfoColumns(), puppet.PuppetInfoGenerate),
@@ -157,6 +157,14 @@ func OrbitDefaultTables() []osquery.OsqueryPlugin {
 		dataflattentable.TablePlugin(log.Logger, dataflattentable.XmlType),   // table name is "parse_xml"
 		dataflattentable.TablePlugin(log.Logger, dataflattentable.IniType),   // table name is "parse_ini"

+		// mcp_listening_servers: lists running processes from core processes table via osquery client
+		table.NewPlugin(
+			"mcp_listening_servers",
+			mcp_listening_servers.Columns(),
+			func(ctx context.Context, queryContext table.QueryContext) ([]map[string]string, error) {
+				return mcp_listening_servers.Generate(ctx, queryContext, opts.Socket)
+			},
+		),
 	}
 	return plugins
 }
@@ -7,4 +7,4 @@ package table

 import "github.com/osquery/osquery-go"

-func PlatformTables(_ PluginOpts) []osquery.OsqueryPlugin { return nil }
+func PlatformTables(_ PluginOpts) ([]osquery.OsqueryPlugin, error) { return nil, nil }
@@ -4,6 +4,7 @@ package table

 import (
 	"fmt"
+
 	"github.com/fleetdm/fleet/v4/orbit/pkg/table/bitlocker_key_protectors"
 	cisaudit "github.com/fleetdm/fleet/v4/orbit/pkg/table/cis_audit"
 	mdmbridge "github.com/fleetdm/fleet/v4/orbit/pkg/table/mdm"
@@ -0,0 +1,449 @@
+package mcp_listening_servers
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/fleetdm/fleet/v4/orbit/pkg/build"
+	"github.com/fleetdm/fleet/v4/pkg/fleethttp"
+	osqclient "github.com/osquery/osquery-go"
+	"github.com/osquery/osquery-go/plugin/table"
+	"github.com/rs/zerolog/log"
+)
+
+// Columns defines the schema for the mcp_listening_servers table.
+func Columns() []table.ColumnDefinition {
+	return []table.ColumnDefinition{
+		table.BigIntColumn("pid"),
+		table.TextColumn("name"),
+		table.TextColumn("cmdline"),
+		table.IntegerColumn("port"),
+		table.TextColumn("address"),
+		table.TextColumn("protocol_version"),
+		table.TextColumn("server_name"),
+		table.TextColumn("server_title"),
+		table.TextColumn("server_version"),
+		table.IntegerColumn("has_logging"),
+		table.IntegerColumn("has_completions"),
+		table.TextColumn("instructions"),
+		table.TextColumn("tools"),     // JSON array of tool names
+		table.TextColumn("prompts"),   // JSON array of prompt names
+		table.TextColumn("resources"), // JSON array of resource URIs
+	}
+}
+
+// Generate connects to the running osqueryd over the provided socket and queries
+// the listening_ports and processes tables to find processes with listening ports,
+// then checks each port to see if an MCP server is responding.
+func Generate(ctx context.Context, queryContext table.QueryContext, socket string) ([]map[string]string, error) {
+	scanner := newMCPScanner()
+	return generateWithScanner(ctx, queryContext, socket, scanner)
+}
+
+// generateWithScanner is an internal helper that allows dependency injection for testing.
+func generateWithScanner(ctx context.Context, queryContext table.QueryContext, socket string, scanner *mcpScanner) ([]map[string]string, error) {
+	// Ensure we don't hang forever if osquery is unresponsive.
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	// Open an osquery client using the extension socket.
+	c, err := scanner.newClient(socket, 2*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("open osquery client: %w", err)
+	}
+	defer c.Close()
+
+	// Get the running processes with listening ports
+	sql := `
+	SELECT DISTINCT lp.pid, lp.port, lp.address, lp.family, p.name, p.cmdline
+	FROM listening_ports lp JOIN processes p ON lp.pid = p.pid
+	`
+
+	rows, err := c.QueryRowsContext(ctx, sql)
+	if err != nil {
+		return nil, err
+	}
+
+	// For each row, check if there's an MCP server listening on that port
+	results := make([]map[string]string, 0, len(rows))
+	for _, row := range rows {
+		port, ok := row["port"]
+		if !ok || port == "" {
+			continue
+		}
+
+		address, ok := row["address"]
+		if !ok {
+			address = "127.0.0.1"
+		}
+
+		// Check family and log a warning if it's not a known address family
+		if family, ok := row["family"]; ok {
+			if family != afUnix && family != afInet && family != afInet6 {
+				log.Warn().Str("family", family).Str("port", port).Str("pid", row["pid"]).Msg("unexpected family value")
+			}
+		}
+
+		// Check if MCP server is active on this port
+		mcpInfo := scanner.checkMCPServer(ctx, address, port)
+
+		// Only include rows where an MCP server is actually responding
+		if mcpInfo == nil {
+			continue
+		}
+
+		// Convert lists to JSON
+		toolsJSON, _ := json.Marshal(mcpInfo.Tools)
+		promptsJSON, _ := json.Marshal(mcpInfo.Prompts)
+		resourcesJSON, _ := json.Marshal(mcpInfo.Resources)
+
+		// Convert boolean to integer string (0 or 1)
+		hasLogging := "0"
+		if mcpInfo.HasLogging {
+			hasLogging = "1"
+		}
+		hasCompletions := "0"
+		if mcpInfo.HasCompletions {
+			hasCompletions = "1"
+		}
+
+		// Create result row with all required columns
+		result := map[string]string{
+			"pid":              row["pid"],
+			"name":             row["name"],
+			"cmdline":          row["cmdline"],
+			"port":             port,
+			"address":          address,
+			"protocol_version": mcpInfo.ProtocolVersion,
+			"server_name":      mcpInfo.ServerName,
+			"server_title":     mcpInfo.ServerTitle,
+			"server_version":   mcpInfo.ServerVersion,
+			"has_logging":      hasLogging,
+			"has_completions":  hasCompletions,
+			"instructions":     mcpInfo.Instructions,
+			"tools":            string(toolsJSON),
+			"prompts":          string(promptsJSON),
+			"resources":        string(resourcesJSON),
+		}
+		results = append(results, result)
+	}
+
+	return results, nil
+}
+
+const (
+	// Socket address family constants (as strings for osquery comparison)
+	afUnix  = "1"  // AF_UNIX - Unix domain sockets
+	afInet  = "2"  // AF_INET - IPv4
+	afInet6 = "10" // AF_INET6 - IPv6
+)
+
+// osqueryClient abstracts the osquery client for ease of testing.
+type osqueryClient interface {
+	QueryRowContext(ctx context.Context, sql string) (map[string]string, error)
+	QueryRowsContext(ctx context.Context, sql string) ([]map[string]string, error)
+	Close()
+}
+
+// mcpScanner holds the dependencies needed for scanning MCP servers.
+type mcpScanner struct {
+	httpClient *http.Client
+	newClient  func(socket string, timeout time.Duration) (osqueryClient, error)
+}
+
+// newMCPScanner creates a new MCP scanner with default dependencies.
+func newMCPScanner() *mcpScanner {
+	return &mcpScanner{
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return osqclient.NewClient(socket, timeout)
+		},
+	}
+}
+
+// mcpTool represents a tool with its metadata
+type mcpTool struct {
+	Name        string `json:"name"`
+	Description string `json:"description,omitempty"`
+}
+
+// mcpPrompt represents a prompt with its metadata
+type mcpPrompt struct {
+	Name        string `json:"name"`
+	Description string `json:"description,omitempty"`
+}
+
+// mcpResource represents a resource with its metadata
+type mcpResource struct {
+	URI         string `json:"uri"`
+	Name        string `json:"name,omitempty"`
+	Description string `json:"description,omitempty"`
+}
+
+// mcpServerInfo holds information about an MCP server
+type mcpServerInfo struct {
+	ProtocolVersion string
+	ServerName      string
+	ServerTitle     string
+	ServerVersion   string
+	HasPrompts      bool
+	HasResources    bool
+	HasTools        bool
+	HasLogging      bool
+	HasCompletions  bool
+	Instructions    string
+	Tools           []mcpTool     // List of tools with descriptions
+	Prompts         []mcpPrompt   // List of prompts with descriptions
+	Resources       []mcpResource // List of resources with descriptions
+}
+
+// mcpResponse represents the JSON-RPC response from an MCP server
+type mcpResponse struct {
+	JSONRPC string `json:"jsonrpc"`
+	ID      int    `json:"id"`
+	Result  struct {
+		ProtocolVersion string `json:"protocolVersion"`
+		Capabilities    struct {
+			Prompts     *json.RawMessage `json:"prompts"`
+			Resources   *json.RawMessage `json:"resources"`
+			Tools       *json.RawMessage `json:"tools"`
+			Logging     *json.RawMessage `json:"logging"`
+			Completions *json.RawMessage `json:"completions"`
+		} `json:"capabilities"`
+		ServerInfo struct {
+			Name    string `json:"name"`
+			Title   string `json:"title"`
+			Version string `json:"version"`
+		} `json:"serverInfo"`
+		Instructions string `json:"instructions"`
+	} `json:"result"`
+}
+
+// mcpListResponse represents responses to tools/list, prompts/list, resources/list
+type mcpListResponse struct {
+	JSONRPC string `json:"jsonrpc"`
+	ID      int    `json:"id"`
+	Result  struct {
+		Tools     []mcpTool     `json:"tools"`
+		Prompts   []mcpPrompt   `json:"prompts"`
+		Resources []mcpResource `json:"resources"`
+	} `json:"result"`
+	Error *struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+// terminateSession sends an HTTP DELETE to the MCP endpoint to explicitly terminate the session.
+// Per the MCP spec, the server MAY respond with 405 Method Not Allowed if it doesn't support
+// explicit session termination, which is acceptable.
+func (s *mcpScanner) terminateSession(ctx context.Context, url, sessionID string) {
+	if sessionID == "" {
+		return
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "DELETE", url, nil)
+	if err != nil {
+		return
+	}
+
+	req.Header.Set("Mcp-Session-Id", sessionID)
+
+	resp, err := s.httpClient.Do(req)
+	if err != nil {
+		// Best-effort cleanup, don't log errors
+		return
+	}
+	resp.Body.Close()
+}
+
+// makeMCPRequest makes an MCP JSON-RPC request and returns the response body and session ID
+func (s *mcpScanner) makeMCPRequest(ctx context.Context, url, method, sessionID string, params interface{}) ([]byte, string, error) {
+	reqBody := map[string]interface{}{
+		"jsonrpc": "2.0",
+		"id":      1,
+		"method":  method,
+	}
+	if params != nil {
+		reqBody["params"] = params
+	}
+
+	bodyJSON, err := json.Marshal(reqBody)
+	if err != nil {
+		return nil, "", err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(bodyJSON))
+	if err != nil {
+		return nil, "", err
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json,text/event-stream")
+	if sessionID != "" {
+		req.Header.Set("Mcp-Session-Id", sessionID)
+	}
+
+	resp, err := s.httpClient.Do(req)
+	if err != nil {
+		return nil, "", err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return nil, "", fmt.Errorf("HTTP %d", resp.StatusCode)
+	}
+
+	// Extract session ID from response header
+	responseSessionID := resp.Header.Get("Mcp-Session-Id")
+
+	bodyBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, "", err
+	}
+
+	// Handle SSE format (Server-Sent Events) - responses may contain lines like:
+	// event: message
+	// id: ...
+	// data: {...}
+	jsonData := bodyBytes
+
+	// Check if this looks like SSE format (contains "data: " on a line)
+	if bytes.Contains(bodyBytes, []byte("data: ")) {
+		// Extract JSON from SSE format
+		lines := bytes.Split(bodyBytes, []byte("\n"))
+		for _, line := range lines {
+			if bytes.HasPrefix(line, []byte("data: ")) {
+				jsonData = bytes.TrimPrefix(line, []byte("data: "))
+				break
+			}
+		}
+	}
+
+	return jsonData, responseSessionID, nil
+}
+
+// checkMCPServer attempts to connect to an MCP server at the given address and port.
+// It returns information about the MCP server if it responds successfully, or nil otherwise.
+func (s *mcpScanner) checkMCPServer(ctx context.Context, address, port string) *mcpServerInfo {
+	// Build the URL - handle localhost and IPv6 addresses specially
+	host := address
+
+	// Check if the address is IPv6
+	isIPv6 := false
+	if ip := net.ParseIP(address); ip != nil {
+		isIPv6 = ip.To4() == nil // If To4() returns nil, it's IPv6
+	}
+
+	// Handle wildcard and loopback addresses
+	if address == "0.0.0.0" || address == "127.0.0.1" || address == "::" || address == "::1" || address == "" {
+		host = "localhost"
+	} else if isIPv6 {
+		// IPv6 addresses must be wrapped in brackets in URLs
+		host = "[" + address + "]"
+	}
+
+	url := fmt.Sprintf("http://%s:%s/mcp", host, port)
+
+	// MCP initialize request
+	// 2025-06-18 is the latest version of the MCP protocol as of 2025-10-14
+	// TODO update this to the latest version of the MCP protocol when it is released
+	initParams := map[string]interface{}{
+		"protocolVersion": "2025-06-18",
+		"capabilities": map[string]interface{}{
+			"tools":     map[string]interface{}{},
+			"resources": map[string]interface{}{},
+			"prompts":   map[string]interface{}{},
+			"roots":     map[string]interface{}{},
+		},
+		"clientInfo": map[string]interface{}{
+			"name":    "fleetd",
+			"version": build.Version,
+		},
+	}
+
+	// Send initialize request without a session ID - server will provide one
+	jsonData, sessionID, err := s.makeMCPRequest(ctx, url, "initialize", "", initParams)
+	if err != nil {
+		return nil
+	}
+
+	// Terminate the session when we're done, per MCP spec
+	defer s.terminateSession(ctx, url, sessionID)
+
+	var mcpResp mcpResponse
+	if err := json.Unmarshal(jsonData, &mcpResp); err != nil {
+		// If JSON parsing fails, return nil so the row is not included
+		// (we only want rows where we can successfully identify an MCP server)
+		return nil
+	}
+
+	// Build the server info struct
+	info := &mcpServerInfo{
+		ProtocolVersion: mcpResp.Result.ProtocolVersion,
+		ServerName:      mcpResp.Result.ServerInfo.Name,
+		ServerTitle:     mcpResp.Result.ServerInfo.Title,
+		ServerVersion:   mcpResp.Result.ServerInfo.Version,
+		HasPrompts:      mcpResp.Result.Capabilities.Prompts != nil,
+		HasResources:    mcpResp.Result.Capabilities.Resources != nil,
+		HasTools:        mcpResp.Result.Capabilities.Tools != nil,
+		HasLogging:      mcpResp.Result.Capabilities.Logging != nil,
+		HasCompletions:  mcpResp.Result.Capabilities.Completions != nil,
+		Instructions:    mcpResp.Result.Instructions,
+		// Initialize slices to empty so they marshal to [] instead of null
+		Tools:     []mcpTool{},
+		Prompts:   []mcpPrompt{},
+		Resources: []mcpResource{},
+	}
+
+	// Fetch lists of tools, prompts, and resources if capabilities are available
+	if info.HasTools {
+		if listData, _, err := s.makeMCPRequest(ctx, url, "tools/list", sessionID, nil); err == nil {
+			var listResp mcpListResponse
+			if err := json.Unmarshal(listData, &listResp); err == nil {
+				// Check if the response contains an error
+				if listResp.Error != nil {
+					log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("tools/list returned error")
+				} else {
+					info.Tools = append(info.Tools, listResp.Result.Tools...)
+				}
+			}
+		}
+	}
+
+	if info.HasPrompts {
+		if listData, _, err := s.makeMCPRequest(ctx, url, "prompts/list", sessionID, nil); err == nil {
+			var listResp mcpListResponse
+			if err := json.Unmarshal(listData, &listResp); err == nil {
+				// Check if the response contains an error
+				if listResp.Error != nil {
+					log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("prompts/list returned error")
+				} else {
+					info.Prompts = append(info.Prompts, listResp.Result.Prompts...)
+				}
+			}
+		}
+	}
+
+	if info.HasResources {
+		if listData, _, err := s.makeMCPRequest(ctx, url, "resources/list", sessionID, nil); err == nil {
+			var listResp mcpListResponse
+			if err := json.Unmarshal(listData, &listResp); err == nil {
+				// Check if the response contains an error
+				if listResp.Error != nil {
+					log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("resources/list returned error")
+				} else {
+					info.Resources = append(info.Resources, listResp.Result.Resources...)
+				}
+			}
+		}
+	}
+
+	return info
+}
@@ -0,0 +1,427 @@
+package mcp_listening_servers
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/fleetdm/fleet/v4/pkg/fleethttp"
+	"github.com/osquery/osquery-go/plugin/table"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const mockSessionID = "test-session-id"
+
+type mockClient struct {
+	row  map[string]string
+	rows []map[string]string
+	err  error
+}
+
+func (m *mockClient) QueryRowContext(ctx context.Context, sql string) (map[string]string, error) {
+	return m.row, m.err
+}
+
+func (m *mockClient) QueryRowsContext(ctx context.Context, sql string) ([]map[string]string, error) {
+	return m.rows, m.err
+}
+func (m *mockClient) Close() {}
+
+func TestGenerate_WithMCPServerActive(t *testing.T) {
+	// Create a scanner with mock dependencies
+	scanner := &mcpScanner{
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return &mockClient{rows: []map[string]string{
+				{"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"},
+			}}, nil
+		},
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+	}
+
+	scanner.httpClient.Transport = &mockTransport{
+		t: t,
+		responses: map[string]string{
+			"initialize": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"protocolVersion": "2025-03-26",
+					"capabilities": {
+						"prompts": {},
+						"resources": {"subscribe": true},
+						"tools": {},
+						"logging": {},
+						"completions": {}
+					},
+					"serverInfo": {
+						"name": "example-servers/everything",
+						"title": "Everything Example Server",
+						"version": "1.0.0"
+					},
+					"instructions": "Testing and demonstration server for MCP protocol features."
+				}
+			}`,
+			"tools/list": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"tools": [
+						{"name": "get_weather", "description": "Get weather for a location"},
+						{"name": "search_web", "description": "Search the web"}
+					]
+				}
+			}`,
+			"prompts/list": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"prompts": [
+						{"name": "code_review", "description": "Review code for quality"},
+						{"name": "summarize", "description": "Summarize content"}
+					]
+				}
+			}`,
+			"resources/list": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"resources": [
+						{"uri": "file:///data/doc1.txt", "name": "Document 1", "description": "First document"},
+						{"uri": "file:///data/doc2.txt", "name": "Document 2", "description": "Second document"}
+					]
+				}
+			}`,
+		},
+	}
+
+	qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+	rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+	require.NoError(t, err)
+	require.Len(t, rows, 1)
+
+	assert.Equal(t, "3001", rows[0]["port"])
+	assert.Equal(t, "2025-03-26", rows[0]["protocol_version"])
+	assert.Equal(t, "example-servers/everything", rows[0]["server_name"])
+	assert.Equal(t, "Everything Example Server", rows[0]["server_title"])
+	assert.Equal(t, "1.0.0", rows[0]["server_version"])
+	assert.Equal(t, "1", rows[0]["has_logging"])
+	assert.Equal(t, "1", rows[0]["has_completions"])
+	assert.Equal(t, "Testing and demonstration server for MCP protocol features.", rows[0]["instructions"])
+	assert.Equal(t, `[{"name":"get_weather","description":"Get weather for a location"},{"name":"search_web","description":"Search the web"}]`, rows[0]["tools"])
+	assert.Equal(t, `[{"name":"code_review","description":"Review code for quality"},{"name":"summarize","description":"Summarize content"}]`, rows[0]["prompts"])
+	assert.Equal(t, `[{"uri":"file:///data/doc1.txt","name":"Document 1","description":"First document"},{"uri":"file:///data/doc2.txt","name":"Document 2","description":"Second document"}]`, rows[0]["resources"])
+}
+
+func TestGenerate_WithMCPServerInactive(t *testing.T) {
+	// Create a scanner with mock dependencies
+	scanner := &mcpScanner{
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return &mockClient{rows: []map[string]string{
+				{"pid": "5678", "port": "8080", "address": "0.0.0.0", "name": "nginx", "cmdline": "nginx"},
+			}}, nil
+		},
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+	}
+
+	scanner.httpClient.Transport = &mockTransport{
+		t:   t,
+		err: http.ErrServerClosed,
+	}
+
+	qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+	rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+	require.NoError(t, err)
+	// Should return 0 rows since no MCP server is active
+	assert.Empty(t, rows)
+}
+
+func TestGenerate_MultipleActiveServers(t *testing.T) {
+	// Create a scanner with mock dependencies
+	scanner := &mcpScanner{
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return &mockClient{rows: []map[string]string{
+				{"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp1.js"},
+				{"pid": "5678", "port": "3002", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp2.js"},
+			}}, nil
+		},
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+	}
+
+	scanner.httpClient.Transport = &mockTransport{
+		t: t,
+		responses: map[string]string{
+			"initialize": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"protocolVersion": "2025-03-26",
+					"capabilities": {
+						"prompts": {},
+						"resources": {"subscribe": true},
+						"tools": {},
+						"logging": {},
+						"completions": {}
+					},
+					"serverInfo": {
+						"name": "example-servers/everything",
+						"title": "Everything Example Server",
+						"version": "1.0.0"
+					},
+					"instructions": "Testing and demonstration server for MCP protocol features."
+				}
+			}`,
+			"tools/list":     `{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}`,
+			"prompts/list":   `{"jsonrpc":"2.0","id":1,"result":{"prompts":[]}}`,
+			"resources/list": `{"jsonrpc":"2.0","id":1,"result":{"resources":[]}}`,
+		},
+		statusCode: 200,
+	}
+
+	qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+	rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+	require.NoError(t, err)
+	assert.Len(t, rows, 2)
+}
+
+func TestGenerate_WithSSEResponse(t *testing.T) {
+	// Create a scanner with mock dependencies
+	scanner := &mcpScanner{
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return &mockClient{rows: []map[string]string{
+				{"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"},
+			}}, nil
+		},
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+	}
+
+	// Full SSE format with event, id, and data lines
+	scanner.httpClient.Transport = &mockTransport{
+		t: t,
+		responses: map[string]string{
+			"initialize": `event: message
+id: 4b74868e-307e-416c-951d-6f305856cb43_1760466849580_vmzdy66v
+data: {"result":{"protocolVersion":"2025-03-26","capabilities":{"prompts":{},"resources":{"subscribe":true},"tools":{},"logging":{},"completions":{}},"serverInfo":{"name":"example-servers/everything","title":"Everything Example Server","version":"1.0.0"},"instructions":"Testing and demonstration server for MCP protocol features."},"jsonrpc":"2.0","id":1}
+`,
+			"tools/list":     `{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"test_tool","description":"A test tool"}]}}`,
+			"prompts/list":   `{"jsonrpc":"2.0","id":1,"result":{"prompts":[{"name":"test_prompt","description":"A test prompt"}]}}`,
+			"resources/list": `{"jsonrpc":"2.0","id":1,"result":{"resources":[{"uri":"test://resource","name":"Test Resource","description":"A test resource"}]}}`,
+		},
+		statusCode: 200,
+	}
+
+	qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+	rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+	require.NoError(t, err)
+	require.Len(t, rows, 1)
+
+	assert.Equal(t, "2025-03-26", rows[0]["protocol_version"])
+	assert.Equal(t, "example-servers/everything", rows[0]["server_name"])
+}
+
+func TestGenerate_WithIPv6Address(t *testing.T) {
+	// Create a scanner with mock dependencies
+	scanner := &mcpScanner{
+		newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+			return &mockClient{rows: []map[string]string{
+				{"pid": "1234", "port": "3001", "address": "::1", "family": afInet6, "name": "node", "cmdline": "node mcp-server.js"},
+				{"pid": "5678", "port": "3002", "address": "2001:db8::1", "family": afInet6, "name": "node", "cmdline": "node mcp-server2.js"},
+				{"pid": "9999", "port": "3003", "address": "::", "family": afInet6, "name": "node", "cmdline": "node mcp-server3.js"},
+			}}, nil
+		},
+		httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+	}
+
+	// Track which URLs were actually requested to verify IPv6 bracket handling
+	requestedURLs := []string{}
+	scanner.httpClient.Transport = &mockTransport{
+		t:             t,
+		requestedURLs: &requestedURLs,
+		responses: map[string]string{
+			"initialize": `{
+				"jsonrpc": "2.0",
+				"id": 1,
+				"result": {
+					"protocolVersion": "2025-03-26",
+					"capabilities": {
+						"tools": {}
+					},
+					"serverInfo": {
+						"name": "test-server",
+						"title": "Test Server",
+						"version": "1.0.0"
+					}
+				}
+			}`,
+			"tools/list": `{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}`,
+		},
+	}
+
+	qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+	rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+	require.NoError(t, err)
+	require.Len(t, rows, 3)
+
+	// Verify that IPv6 loopback (::1) was converted to localhost
+	assert.Contains(t, requestedURLs, "http://localhost:3001/mcp")
+
+	// Verify that IPv6 wildcard (::) was converted to localhost
+	assert.Contains(t, requestedURLs, "http://localhost:3003/mcp")
+
+	// Verify that regular IPv6 addresses are wrapped in brackets
+	assert.Contains(t, requestedURLs, "http://[2001:db8::1]:3002/mcp")
+
+	// All should return the same server info
+	assert.Equal(t, "test-server", rows[0]["server_name"])
+	assert.Equal(t, "test-server", rows[1]["server_name"])
+	assert.Equal(t, "test-server", rows[2]["server_name"])
+}
+
+func TestSessionTermination(t *testing.T) {
+	tests := []struct {
+		name             string
+		deleteStatusCode int
+	}{
+		{
+			name:             "server supports termination (200 OK)",
+			deleteStatusCode: http.StatusOK,
+		},
+		{
+			name:             "server doesn't support termination (405 Method Not Allowed)",
+			deleteStatusCode: http.StatusMethodNotAllowed,
+		},
+		{
+			name:             "server returns 204 No Content",
+			deleteStatusCode: http.StatusNoContent,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			deleteRequestReceived := false
+
+			scanner := &mcpScanner{
+				newClient: func(socket string, timeout time.Duration) (osqueryClient, error) {
+					return &mockClient{rows: []map[string]string{
+						{"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"},
+					}}, nil
+				},
+				httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)),
+			}
+
+			scanner.httpClient.Transport = &mockTransport{
+				t:                     t,
+				deleteRequestReceived: &deleteRequestReceived,
+				deleteStatusCode:      tt.deleteStatusCode,
+				responses: map[string]string{
+					"initialize": `{
+						"jsonrpc": "2.0",
+						"id": 1,
+						"result": {
+							"protocolVersion": "2025-03-26",
+							"capabilities": {},
+							"serverInfo": {
+								"name": "test-server",
+								"title": "Test Server",
+								"version": "1.0.0"
+							}
+						}
+					}`,
+				},
+			}
+
+			qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}}
+
+			rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner)
+			require.NoError(t, err)
+			require.Len(t, rows, 1)
+
+			// Verify DELETE request was sent
+			assert.True(t, deleteRequestReceived, "DELETE request should have been sent for session termination")
+		})
+	}
+}
+
+// mockTransport is a unified HTTP transport for testing MCP servers
+type mockTransport struct {
+	t                     *testing.T
+	responses             map[string]string // method -> response for POST requests
+	statusCode            int               // status code for POST requests (default 200)
+	err                   error             // error to return instead of response
+	requestedURLs         *[]string         // if non-nil, track requested URLs
+	deleteRequestReceived *bool             // if non-nil, track DELETE requests
+	deleteStatusCode      int               // status code for DELETE requests (default 200)
+}
+
+func (m *mockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+
+	// Track the requested URL if tracking is enabled
+	if m.requestedURLs != nil {
+		*m.requestedURLs = append(*m.requestedURLs, req.URL.String())
+	}
+
+	// Handle DELETE requests for session termination
+	if req.Method == "DELETE" {
+		if m.deleteRequestReceived != nil {
+			*m.deleteRequestReceived = true
+		}
+		// Verify that the session ID header is correct
+		sessionID := req.Header.Get("Mcp-Session-Id")
+		assert.Equal(m.t, mockSessionID, sessionID, "DELETE request should have correct Mcp-Session-Id header")
+		deleteStatus := http.StatusOK
+		if m.deleteStatusCode != 0 {
+			deleteStatus = m.deleteStatusCode
+		}
+		return &http.Response{
+			StatusCode: deleteStatus,
+			Body:       io.NopCloser(bytes.NewBufferString("")),
+			Header:     http.Header{},
+		}, nil
+	}
+
+	// Parse the request to determine which method is being called
+	var reqBody map[string]interface{}
+	bodyBytes, _ := io.ReadAll(req.Body)
+	_ = json.Unmarshal(bodyBytes, &reqBody)
+
+	method, _ := reqBody["method"].(string)
+
+	// Verify session ID header requirements:
+	// - initialize request should NOT have a session ID (it's the first request)
+	// - all other requests (tools/list, prompts/list, resources/list) MUST have a session ID
+	sessionID := req.Header.Get("Mcp-Session-Id")
+	if method == "initialize" {
+		assert.Equal(m.t, "", sessionID, "initialize request should not have Mcp-Session-Id header")
+	} else if method != "" {
+		// All non-initialize methods should have the correct session ID
+		assert.Equal(m.t, mockSessionID, sessionID, "%s request should have correct Mcp-Session-Id header", method)
+	}
+
+	responseBody := m.responses[method]
+	if responseBody == "" {
+		responseBody = `{"jsonrpc":"2.0","id":1,"result":{}}`
+	}
+
+	postStatus := http.StatusOK
+	if m.statusCode != 0 {
+		postStatus = m.statusCode
+	}
+
+	return &http.Response{
+		StatusCode: postStatus,
+		Body:       io.NopCloser(bytes.NewBufferString(responseBody)),
+		Header:     http.Header{"Mcp-Session-Id": []string{mockSessionID}},
+	}, nil
+}
@@ -16369,6 +16369,112 @@
    ],
    "fleetRepoUrl": "https://github.com/fleetdm/fleet/blob/main/schema/tables/managed_policies.yml"
  },
+  {
+    "name": "mcp_listening_servers",
+    "notes": "This table is a fleetd table. Fleetd tables are built into Fleet's agent (fleetd).",
+    "description": "Lists processes with listening ports that are responding as [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers via HTTP (Many MCP servers use stdio and will not be detected by this table).  This table queries the `listening_ports` and `processes` tables to find listening ports, then probes each port to identify active MCP servers.",
+    "platforms": [
+      "darwin",
+      "windows",
+      "linux"
+    ],
+    "evented": false,
+    "examples": "List all MCP servers currently running on the system.\n\n```\nSELECT * FROM mcp_listening_servers;\n```",
+    "columns": [
+      {
+        "name": "pid",
+        "description": "The process ID of the process running the MCP server.",
+        "required": false,
+        "type": "bigint"
+      },
+      {
+        "name": "name",
+        "description": "The name of the process running the MCP server.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "cmdline",
+        "description": "The full command line of the process running the MCP server.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "port",
+        "description": "The port number on which the MCP server is listening.",
+        "required": false,
+        "type": "integer"
+      },
+      {
+        "name": "address",
+        "description": "The network address on which the MCP server is listening (e.g., \"127.0.0.1\", \"0.0.0.0\", \"::1\").",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "protocol_version",
+        "description": "The MCP protocol version supported by the server (e.g., \"2025-06-18\").",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "server_name",
+        "description": "The name identifier of the MCP server.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "server_title",
+        "description": "The human-readable title of the MCP server.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "server_version",
+        "description": "The version of the MCP server implementation.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "has_logging",
+        "description": "Indicates if the MCP server supports logging capabilities (1 = true, 0 = false).",
+        "required": false,
+        "type": "integer"
+      },
+      {
+        "name": "has_completions",
+        "description": "Indicates if the MCP server supports completion capabilities (1 = true, 0 = false).",
+        "required": false,
+        "type": "integer"
+      },
+      {
+        "name": "instructions",
+        "description": "Instructions or description text provided by the MCP server about its purpose and usage.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "tools",
+        "description": "JSON array of tools provided by the MCP server, including their names and descriptions.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "prompts",
+        "description": "JSON array of prompts provided by the MCP server, including their names and descriptions.",
+        "required": false,
+        "type": "text"
+      },
+      {
+        "name": "resources",
+        "description": "JSON array of resources provided by the MCP server, including their URIs, names, and descriptions.",
+        "required": false,
+        "type": "text"
+      }
+    ],
+    "url": "https://fleetdm.com/tables/mcp_listening_servers",
+    "fleetRepoUrl": "https://github.com/fleetdm/fleet/blob/main/schema/tables/mcp_listening_servers.yml"
+  },
  {
    "name": "md_devices",
    "description": "Software RAID array settings.",
@@ -0,0 +1,76 @@
+name: mcp_listening_servers
+notes: This table is a fleetd table. Fleetd tables are built into Fleet's agent (fleetd).
+description: Lists processes with listening ports that are responding as [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers via HTTP (Many MCP servers use stdio and will not be detected by this table).  This table queries the `listening_ports` and `processes` tables to find listening ports, then probes each port to identify active MCP servers.
+platforms:
+  - darwin
+  - windows
+  - linux
+evented: false
+examples: |-
+  List all MCP servers currently running on the system.
+
+  ```
+  SELECT * FROM mcp_listening_servers;
+  ```
+columns:
+  - name: pid
+    description: The process ID of the process running the MCP server.
+    required: false
+    type: bigint
+  - name: name
+    description: The name of the process running the MCP server.
+    required: false
+    type: text
+  - name: cmdline
+    description: The full command line of the process running the MCP server.
+    required: false
+    type: text
+  - name: port
+    description: The port number on which the MCP server is listening.
+    required: false
+    type: integer
+  - name: address
+    description: The network address on which the MCP server is listening (e.g., "127.0.0.1", "0.0.0.0", "::1").
+    required: false
+    type: text
+  - name: protocol_version
+    description: The MCP protocol version supported by the server (e.g., "2025-06-18").
+    required: false
+    type: text
+  - name: server_name
+    description: The name identifier of the MCP server.
+    required: false
+    type: text
+  - name: server_title
+    description: The human-readable title of the MCP server.
+    required: false
+    type: text
+  - name: server_version
+    description: The version of the MCP server implementation.
+    required: false
+    type: text
+  - name: has_logging
+    description: Indicates if the MCP server supports logging capabilities (1 = true, 0 = false).
+    required: false
+    type: integer
+  - name: has_completions
+    description: Indicates if the MCP server supports completion capabilities (1 = true, 0 = false).
+    required: false
+    type: integer
+  - name: instructions
+    description: Instructions or description text provided by the MCP server about its purpose and usage.
+    required: false
+    type: text
+  - name: tools
+    description: JSON array of tools provided by the MCP server, including their names and descriptions.
+    required: false
+    type: text
+  - name: prompts
+    description: JSON array of prompts provided by the MCP server, including their names and descriptions.
+    required: false
+    type: text
+  - name: resources
+    description: JSON array of resources provided by the MCP server, including their URIs, names, and descriptions.
+    required: false
+    type: text
+
				`@@ -0,0 +1 @@`
				* Add `mcp_listening_servers` table to find MCP servers listening over HTTP.