diff --git a/agent/agent.go b/agent/agent.go index c9c3cc71c8..017f942e71 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -513,6 +513,14 @@ func (a *agent) init() { Clock: a.clock, WorkingDir: workingDirFn, InitialSources: initialContextSources(a.contextConfig, workingDirFn), + // The manager runs its own self-contained MCP runner: it + // connects to the .mcp.json servers it discovers, lists + // their tools, and pushes them to coderd as KindMCPServer + // resources. This is independent of a.mcpManager, which + // serves the agent's MCP HTTP API; the two MCP paths share + // no state during the rollout. + MCPExecer: a.execer, + MCPUpdateEnv: a.updateCommandEnv, }) a.contextAPI = agentcontext.NewAPI(a.contextManager) a.reconnectingPTYServer = reconnectingpty.NewServer( diff --git a/agent/agentcontext/doc.go b/agent/agentcontext/doc.go index b9a34653ad..5ed7ebf043 100644 --- a/agent/agentcontext/doc.go +++ b/agent/agentcontext/doc.go @@ -18,7 +18,12 @@ // to coderd without coupling this package to any particular // drpc client version. // -// The package is purely additive: existing agent code paths -// (agent/agentcontextconfig and agent/x/agentmcp) continue to -// operate unchanged. +// Live MCP server tool lists are produced by this package's own +// self-contained MCP runner: it connects to the MCP servers declared in +// the .mcp.json files the resolver discovers, lists their tools, and +// surfaces them as KindMCPServer resources so MCP servers and their +// tools are pushed to coderd alongside instruction files and skills. +// This runs independently of agent/x/agentmcp, which owns the agent's +// MCP HTTP proxy; the two MCP paths share no state and both continue to +// operate unchanged during the rollout. package agentcontext diff --git a/agent/agentcontext/manager.go b/agent/agentcontext/manager.go index decefdd1c4..e61367bdc7 100644 --- a/agent/agentcontext/manager.go +++ b/agent/agentcontext/manager.go @@ -9,6 +9,7 @@ import ( "golang.org/x/xerrors" "cdr.dev/slog/v3" + "github.com/coder/coder/v2/agent/agentexec" "github.com/coder/quartz" ) @@ -35,9 +36,21 @@ type ManagerOptions struct { // directly; production callers leave it unset. AllowedRoots []string // Resolver, when non-nil, replaces the default resolver. - // Tests use this to inject MCP providers and tighten - // caps. + // Tests use this to inject MCP resources (via + // Resolver.MCPResources) and tighten caps. Resolver *Resolver + // MCPExecer, when non-nil, enables the self-contained MCP + // runner: the Manager connects to the MCP servers declared + // in the .mcp.json files it discovers, lists their tools, + // and surfaces them as KindMCPServer resources in every + // snapshot. The runner uses this Execer to launch stdio MCP + // servers. It is ignored when the resolver already has an + // MCP provider (e.g. a test injecting one via Resolver). + MCPExecer agentexec.Execer + // MCPUpdateEnv optionally enriches the environment handed to + // stdio MCP servers (typically the agent's per-command env). + // Used only when MCPExecer is set; may be nil. + MCPUpdateEnv func([]string) ([]string, error) // Debounce overrides the watcher's debounce window. Debounce time.Duration } @@ -60,7 +73,11 @@ type Manager struct { workingDir func() string allowedRoots []string resolver *Resolver - debounce time.Duration + // mcpRunner, when non-nil, owns the agent's self-contained + // MCP connection lifecycle and feeds the resolver's MCP + // provider. runMCPSync (started by Run) drives its reloads. + mcpRunner *mcpRunner + debounce time.Duration mu sync.Mutex sources []Source @@ -135,6 +152,20 @@ func NewManager(opts ManagerOptions) *Manager { runStartedCh: make(chan struct{}), } + // Enable the self-contained MCP runner unless the resolver + // already has a provider (tests inject one via Resolver). The + // runner connects to the .mcp.json servers the resolver + // discovers and surfaces their tools as KindMCPServer + // resources; runMCPSync (started in Run) drives its reloads. + // The provider must be wired before the eager first resolve + // below so the seam is present from the first snapshot. + if resolver.MCPResources == nil && opts.MCPExecer != nil { + m.mcpRunner = newMCPRunner(m.logger.Named("mcp"), opts.MCPExecer, opts.MCPUpdateEnv, m.Trigger) + resolver.MCPResources = func() []Resource { + return buildMCPServerResources(m.mcpRunner.Servers()) + } + } + for _, s := range opts.InitialSources { canonical, err := CanonicalizePath(s.Path) if err != nil { @@ -205,6 +236,14 @@ func (m *Manager) Run(ctx context.Context) error { defer watcher.Close() + // Drive MCP server reloads from discovered .mcp.json files for + // the lifetime of Run. Started here (not in NewManager) so it + // runs alongside the trigger loop that consumes its re-resolve + // signals. + if m.mcpRunner != nil { + go m.runMCPSync(ctx) + } + for { select { case <-ctx.Done(): diff --git a/agent/agentcontext/manager_test.go b/agent/agentcontext/manager_test.go index ce96490a56..08ce803ead 100644 --- a/agent/agentcontext/manager_test.go +++ b/agent/agentcontext/manager_test.go @@ -25,6 +25,13 @@ import ( // Claude config files into snapshots and breaks every // Len(Resources, N) assertion. func TestMain(m *testing.M) { + // The MCP runner re-execs this test binary as a fake stdio MCP + // server (TEST_MCP_FAKE_SERVER=1). Serve and exit before any test + // setup runs. + if maybeServeFakeMCPServer() { + os.Exit(0) + } + home, err := os.MkdirTemp("", "agentcontext-test-home-") if err != nil { panic(err) @@ -395,3 +402,38 @@ func TestManager_SubscribeBroadcastOnChange(t *testing.T) { t.Fatal("expected subscriber to be notified") } } + +// TestManager_MCPResourcesAppliesToSnapshot verifies that MCP resources +// supplied via the resolver contribute KindMCPServer resources (with +// their tools) to the resolved snapshot. +func TestManager_MCPResourcesAppliesToSnapshot(t *testing.T) { + t.Parallel() + dir := t.TempDir() + + m := newTestManager(t, agentcontext.ManagerOptions{ + WorkingDir: func() string { return dir }, + Resolver: &agentcontext.Resolver{ + MCPResources: func() []agentcontext.Resource { + return []agentcontext.Resource{{ + ID: "mcp_server:fs", + Kind: agentcontext.KindMCPServer, + Source: "fs", + Name: "fs", + Status: agentcontext.StatusOK, + Tools: []agentcontext.MCPTool{{Name: "read", Description: "Read"}}, + }} + }, + }, + }) + + snap := m.Snapshot() + var found bool + for _, r := range snap.Resources { + if r.Kind == agentcontext.KindMCPServer && r.Source == "fs" { + found = true + require.Len(t, r.Tools, 1) + require.Equal(t, "read", r.Tools[0].Name) + } + } + require.True(t, found, "expected MCP server resource in snapshot") +} diff --git a/agent/agentcontext/mcp.go b/agent/agentcontext/mcp.go index 5efaf0bc1f..93901ef64e 100644 --- a/agent/agentcontext/mcp.go +++ b/agent/agentcontext/mcp.go @@ -1,30 +1,133 @@ package agentcontext -// MCPProvider supplies the live MCP server portion of a -// snapshot. Implementations typically wrap an existing MCP -// manager (e.g. agent/x/agentmcp.Manager) and translate each -// server's tool list into a KindMCPServer resource. +import ( + "crypto/sha256" + "encoding/json" + "slices" + "strings" +) + +// MCPServerStatus is a non-blocking, point-in-time view of a single MCP +// server the runner has attempted to connect to. It is the data +// buildMCPServerResources turns into a KindMCPServer resource. The +// runner owns the connection lifecycle; this type carries only the +// resolved result. +type MCPServerStatus struct { + // Name is the server name declared in .mcp.json. + Name string + // Connected reports whether the runner reached the server and + // listed its tools during the most recent reload. + Connected bool + // Err carries the connect/list failure when Connected is false. + Err string + // Tools is the server's tool list, with the tool names exactly + // as the server reported them (no server prefix), when + // Connected; empty otherwise. + Tools []MCPTool +} + +// buildMCPServerResources turns a per-server MCP snapshot into one +// KindMCPServer resource per server. Servers are emitted in name +// order, and tools within a server in name order, so the resource ID +// list and content hashes are deterministic across resolves. // -// The interface is intentionally minimal so the existing MCP -// lifecycle code can be reused without refactoring; a follow-up -// change absorbs the lifecycle into this package. -type MCPProvider interface { - // MCPResources returns one Resource per MCP server known - // to the provider. Each Resource must: - // - // - Have Kind == KindMCPServer. - // - Use the server name as Source. - // - Set Name to the server name (matches Source today; - // reserved for the case where a future provider scheme - // decouples them). - // - Populate ContentHash over a canonical encoding of the - // server name plus the tool list (proto Tools field) - // so any tool-set change flips the dirty bit. - // - Carry a Description summarizing the server. - // - Populate Tools with the structured tool list; Payload - // is unused for this kind and should be left empty. - // - // Implementations should never block; the resolver calls - // this on every re-resolve. - MCPResources() []Resource +// A connected server that exposes at least one tool becomes a +// StatusOK resource carrying its tools. A server that failed to +// connect becomes a StatusUnreadable resource carrying the connection +// error, so it appears in the snapshot's issues instead of vanishing. +// A connected server with no tools yet is skipped until its tools +// arrive (a later re-resolve, driven by the runner's reload, surfaces +// it). A server's .mcp.json entry still appears separately as a +// KindMCPConfig resource from the filesystem pass. +// +// Tool names are emitted exactly as the server reported them; flattening +// them into a single namespace (e.g. "server__tool") is the control +// plane's concern, since the resource already carries the server name. +func buildMCPServerResources(servers []MCPServerStatus) []Resource { + if len(servers) == 0 { + return nil + } + sorted := slices.Clone(servers) + slices.SortFunc(sorted, func(a, b MCPServerStatus) int { + return strings.Compare(a.Name, b.Name) + }) + + resources := make([]Resource, 0, len(sorted)) + for _, s := range sorted { + if s.Name == "" { + continue + } + if !s.Connected { + errMsg := s.Err + if errMsg == "" { + errMsg = "failed to connect" + } + resources = append(resources, Resource{ + ID: resourceID(KindMCPServer, s.Name), + Kind: KindMCPServer, + Source: s.Name, + Name: s.Name, + Status: StatusUnreadable, + Error: errMsg, + ContentHash: hashMCPServerError(s.Name, errMsg), + }) + continue + } + if len(s.Tools) == 0 { + continue + } + serverTools := slices.Clone(s.Tools) + slices.SortFunc(serverTools, func(a, b MCPTool) int { + return strings.Compare(a.Name, b.Name) + }) + resources = append(resources, Resource{ + ID: resourceID(KindMCPServer, s.Name), + Kind: KindMCPServer, + Source: s.Name, + Name: s.Name, + Status: StatusOK, + ContentHash: hashMCPServer(s.Name, serverTools), + Tools: serverTools, + }) + } + if len(resources) == 0 { + return nil + } + return resources +} + +// hashMCPServer produces a deterministic content hash over a server's +// identity and full tool set (name, description, and input schema) so +// any tool-set change flips the resource's content hash. The schema is +// encoded with encoding/json, which sorts map keys. +func hashMCPServer(server string, tools []MCPTool) [32]byte { + h := sha256.New() + writeLengthPrefixed(h, server) + for _, t := range tools { + writeLengthPrefixed(h, t.Name) + writeLengthPrefixed(h, t.Description) + if len(t.InputSchema) > 0 { + if schema, err := json.Marshal(t.InputSchema); err == nil { + writeLengthPrefixed(h, string(schema)) + } + } + } + var sum [32]byte + copy(sum[:], h.Sum(nil)) + return sum +} + +// hashMCPServerError produces a deterministic content hash for a +// failed-to-connect server. The "unreadable" discriminator keeps a +// failed server's hash distinct from an OK server's, so a server that +// transitions between connected and failed (or whose error text +// changes) flips its content hash. +func hashMCPServerError(server, errMsg string) [32]byte { + h := sha256.New() + writeLengthPrefixed(h, "unreadable") + writeLengthPrefixed(h, server) + writeLengthPrefixed(h, errMsg) + var sum [32]byte + copy(sum[:], h.Sum(nil)) + return sum } diff --git a/agent/agentcontext/mcp_internal_test.go b/agent/agentcontext/mcp_internal_test.go new file mode 100644 index 0000000000..9f101d0b3e --- /dev/null +++ b/agent/agentcontext/mcp_internal_test.go @@ -0,0 +1,154 @@ +package agentcontext + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestBuildMCPServerResources(t *testing.T) { + t.Parallel() + + t.Run("Empty", func(t *testing.T) { + t.Parallel() + require.Nil(t, buildMCPServerResources(nil)) + require.Nil(t, buildMCPServerResources([]MCPServerStatus{})) + }) + + t.Run("GroupsByServerSortedWithTools", func(t *testing.T) { + t.Parallel() + // Tool names are whatever the server reported; the runner no + // longer prefixes them with the server name. + servers := []MCPServerStatus{ + {Name: "github", Connected: true, Tools: []MCPTool{ + {Name: "search", Description: "Search"}, + {Name: "create", Description: "Create"}, + }}, + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "Read", InputSchema: map[string]any{"type": "object"}}, + }}, + // Dropped: a server with no name cannot be addressed. + {Name: "", Connected: true, Tools: []MCPTool{{Name: "orphan"}}}, + } + got := buildMCPServerResources(servers) + require.Len(t, got, 2) + + // Servers are emitted in name order: fs, then github. + require.Equal(t, "fs", got[0].Source) + require.Equal(t, "fs", got[0].Name) + require.Equal(t, KindMCPServer, got[0].Kind) + require.Equal(t, "mcp_server:fs", got[0].ID) + require.Equal(t, StatusOK, got[0].Status) + require.NotEqual(t, [32]byte{}, got[0].ContentHash) + require.Len(t, got[0].Tools, 1) + require.Equal(t, "read", got[0].Tools[0].Name) + require.Equal(t, map[string]any{"type": "object"}, got[0].Tools[0].InputSchema) + + require.Equal(t, "github", got[1].Source) + require.Len(t, got[1].Tools, 2) + // Tools within a server are sorted by name: create, then search. + require.Equal(t, "create", got[1].Tools[0].Name) + require.Equal(t, "search", got[1].Tools[1].Name) + }) + + t.Run("ConnectedWithoutToolsSkipped", func(t *testing.T) { + t.Parallel() + // A connected server that has not yet reported any tools is + // not surfaced; a later re-resolve picks it up once tools + // arrive. + require.Nil(t, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true}, + })) + }) + + t.Run("FailedServerSurfacesAsIssue", func(t *testing.T) { + t.Parallel() + got := buildMCPServerResources([]MCPServerStatus{ + {Name: "broken", Connected: false, Err: "initialize \"broken\": exec: no such file"}, + }) + require.Len(t, got, 1) + require.Equal(t, KindMCPServer, got[0].Kind) + require.Equal(t, "broken", got[0].Source) + require.Equal(t, "broken", got[0].Name) + require.Equal(t, "mcp_server:broken", got[0].ID) + require.Equal(t, StatusUnreadable, got[0].Status) + require.Equal(t, "initialize \"broken\": exec: no such file", got[0].Error) + require.Empty(t, got[0].Tools) + require.NotEqual(t, [32]byte{}, got[0].ContentHash) + }) + + t.Run("FailedServerWithoutErrorGetsDefault", func(t *testing.T) { + t.Parallel() + got := buildMCPServerResources([]MCPServerStatus{ + {Name: "broken", Connected: false}, + }) + require.Len(t, got, 1) + require.Equal(t, StatusUnreadable, got[0].Status) + require.Equal(t, "failed to connect", got[0].Error) + }) + + t.Run("ContentHashStableAndToolSensitive", func(t *testing.T) { + t.Parallel() + base := []MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "Read"}, + }}, + } + h1 := buildMCPServerResources(base)[0].ContentHash + // Identical input is hashed identically. + require.Equal(t, h1, buildMCPServerResources(base)[0].ContentHash) + // A description change flips the hash. + require.NotEqual(t, h1, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "Read files"}, + }}, + })[0].ContentHash) + // Adding a tool flips the hash. + require.NotEqual(t, h1, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "Read"}, + {Name: "write", Description: "Write"}, + }}, + })[0].ContentHash) + // A schema change flips the hash. + require.NotEqual(t, h1, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "Read", InputSchema: map[string]any{"type": "object"}}, + }}, + })[0].ContentHash) + }) + + t.Run("FailedServerHashErrorSensitive", func(t *testing.T) { + t.Parallel() + h1 := buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: false, Err: "boom"}, + })[0].ContentHash + // The error text participates in the hash so a changed error + // is detectable. + require.NotEqual(t, h1, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: false, Err: "different"}, + })[0].ContentHash) + // A failed server hashes differently from a connected one, so + // the connected->failed transition is detectable. + require.NotEqual(t, h1, buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{ + {Name: "read", Description: "boom"}, + }}, + })[0].ContentHash) + }) + + t.Run("MixedServersSortedByName", func(t *testing.T) { + t.Parallel() + // Failed and connected servers are emitted together in name + // order: broken (failed) before fs (ok). + got := buildMCPServerResources([]MCPServerStatus{ + {Name: "fs", Connected: true, Tools: []MCPTool{{Name: "read"}}}, + {Name: "broken", Connected: false, Err: "nope"}, + }) + require.Len(t, got, 2) + require.Equal(t, "broken", got[0].Source) + require.Equal(t, StatusUnreadable, got[0].Status) + require.Equal(t, "fs", got[1].Source) + require.Equal(t, StatusOK, got[1].Status) + }) +} diff --git a/agent/agentcontext/mcpexec_test.go b/agent/agentcontext/mcpexec_test.go new file mode 100644 index 0000000000..3b57790668 --- /dev/null +++ b/agent/agentcontext/mcpexec_test.go @@ -0,0 +1,189 @@ +package agentcontext_test + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/agent/agentcontext" + "github.com/coder/coder/v2/agent/agentexec" + "github.com/coder/coder/v2/testutil" +) + +// TestManager_MCPServerToolsInSnapshot exercises the MCP runner end to +// end against a real subprocess: a .mcp.json in the working directory is +// discovered by the resolver, the runner connects the declared stdio +// server, lists its tools, and they surface as a KindMCPServer resource +// in the manager's snapshot (the same snapshot that is pushed to coderd). +func TestManager_MCPServerToolsInSnapshot(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeMCPConfig(t, dir, "fake", map[string]string{"TEST_MCP_FAKE_SERVER": "1"}) + + m := newTestManager(t, agentcontext.ManagerOptions{ + WorkingDir: func() string { return dir }, + MCPExecer: agentexec.DefaultExecer, + }) + ctx := testutil.Context(t, testutil.WaitLong) + go func() { _ = m.Run(ctx) }() + + require.Eventually(t, func() bool { + return findMCPServer(m.Snapshot(), "fake") != nil + }, testutil.WaitLong, testutil.IntervalMedium, + "the connected MCP server's tools should surface in the snapshot") + + got := findMCPServer(m.Snapshot(), "fake") + require.NotNil(t, got) + require.Equal(t, agentcontext.StatusOK, got.Status) + require.Len(t, got.Tools, 1) + require.Equal(t, "echo", got.Tools[0].Name) + require.Equal(t, "echoes input", got.Tools[0].Description) +} + +// TestManager_MCPServerHangingCloseDoesNotStall is a regression test for +// a server that ignores stdin-close. mcp-go's stdio Close() closes stdin +// and then blocks on cmd.Wait(); without a force-kill the runner's +// per-server connect (and thus the whole reload) would hang and the +// tools would never be published. The runner force-kills the subprocess, +// so the tool still surfaces in the snapshot. +func TestManager_MCPServerHangingCloseDoesNotStall(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeMCPConfig(t, dir, "hang", map[string]string{ + "TEST_MCP_FAKE_SERVER": "1", + "TEST_MCP_HANG_AFTER_LIST": "1", + }) + + m := newTestManager(t, agentcontext.ManagerOptions{ + WorkingDir: func() string { return dir }, + MCPExecer: agentexec.DefaultExecer, + }) + ctx := testutil.Context(t, testutil.WaitLong) + go func() { _ = m.Run(ctx) }() + + require.Eventually(t, func() bool { + got := findMCPServer(m.Snapshot(), "hang") + return got != nil && got.Status == agentcontext.StatusOK + }, testutil.WaitLong, testutil.IntervalMedium, + "a hanging MCP server must not stall the reload; its tool should still surface") +} + +// findMCPServer returns the KindMCPServer resource for the named server, +// or nil if absent. +func findMCPServer(snap agentcontext.Snapshot, name string) *agentcontext.Resource { + for i := range snap.Resources { + if r := snap.Resources[i]; r.Kind == agentcontext.KindMCPServer && r.Source == name { + return &r + } + } + return nil +} + +// writeMCPConfig writes a .mcp.json into dir declaring a single stdio MCP +// server that re-execs this test binary into serveFakeMCPServer (via the +// TEST_MCP_FAKE_SERVER env, which TestMain handles). +func writeMCPConfig(t *testing.T, dir, name string, env map[string]string) { + t.Helper() + testBin, err := os.Executable() + require.NoError(t, err) + cfg := map[string]any{ + "mcpServers": map[string]any{ + name: map[string]any{ + "command": testBin, + "env": env, + }, + }, + } + data, err := json.Marshal(cfg) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(dir, ".mcp.json"), data, 0o600)) +} + +// maybeServeFakeMCPServer serves the fake stdio MCP server when +// TEST_MCP_FAKE_SERVER=1 and reports whether it handled the process so +// the caller (TestMain) can exit. The runner re-execs the test binary +// into this, so it must run at the very top of TestMain. When +// TEST_MCP_HANG_AFTER_LIST=1 the server blocks after serving instead of +// returning, simulating a server that ignores stdin-close so a test can +// exercise the runner's force-kill (the process is then killed by the +// parent and never returns here). +func maybeServeFakeMCPServer() (served bool) { + if os.Getenv("TEST_MCP_FAKE_SERVER") != "1" { + return false + } + serveFakeMCPServer() + if os.Getenv("TEST_MCP_HANG_AFTER_LIST") == "1" { + select {} + } + return true +} + +// serveFakeMCPServer serves a minimal MCP protocol over stdin/stdout: it +// answers initialize and advertises a single "echo" tool, then returns +// when the client closes stdin (EOF). +func serveFakeMCPServer() { + scanner := bufio.NewScanner(os.Stdin) + for scanner.Scan() { + line := scanner.Bytes() + + var req struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + if err := json.Unmarshal(line, &req); err != nil { + continue + } + + var resp any + switch req.Method { + case "initialize": + resp = map[string]any{ + "jsonrpc": "2.0", + "id": req.ID, + "result": map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{"tools": map[string]any{}}, + "serverInfo": map[string]any{"name": "fake-server", "version": "0.0.1"}, + }, + } + case "notifications/initialized": + // Notifications take no response. + continue + case "tools/list": + resp = map[string]any{ + "jsonrpc": "2.0", + "id": req.ID, + "result": map[string]any{ + "tools": []map[string]any{ + { + "name": "echo", + "description": "echoes input", + "inputSchema": map[string]any{ + "type": "object", + "properties": map[string]any{}, + }, + }, + }, + }, + } + default: + resp = map[string]any{ + "jsonrpc": "2.0", + "id": req.ID, + "error": map[string]any{"code": -32601, "message": "method not found"}, + } + } + + out, err := json.Marshal(resp) + if err != nil { + continue + } + _, _ = fmt.Fprintf(os.Stdout, "%s\n", out) + } +} diff --git a/agent/agentcontext/mcprunner.go b/agent/agentcontext/mcprunner.go new file mode 100644 index 0000000000..1469adc7b5 --- /dev/null +++ b/agent/agentcontext/mcprunner.go @@ -0,0 +1,502 @@ +package agentcontext + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "io/fs" + "os" + "os/exec" + "reflect" + "slices" + "strings" + "sync" + "time" + + "github.com/mark3labs/mcp-go/client" + "github.com/mark3labs/mcp-go/client/transport" + "github.com/mark3labs/mcp-go/mcp" + "golang.org/x/sync/errgroup" + "golang.org/x/xerrors" + + "cdr.dev/slog/v3" + "github.com/coder/coder/v2/agent/agentexec" + "github.com/coder/coder/v2/agent/usershell" + "github.com/coder/coder/v2/buildinfo" +) + +// mcpConnectTimeout bounds how long the runner waits for a single MCP +// server to start its transport, initialize, and report its tools. +const mcpConnectTimeout = 30 * time.Second + +// mcpConnectConcurrency bounds how many MCP servers the runner connects +// to at once. .mcp.json files rarely declare many servers, but the cap +// keeps a pathological config from spawning an unbounded number of +// subprocesses simultaneously. +const mcpConnectConcurrency = 8 + +// mcpServerConfig is a single MCP server declaration parsed from a +// .mcp.json file. It is the runner's self-contained equivalent of the +// agent/x/agentmcp ServerConfig: agentcontext deliberately does not +// import that package so the two MCP paths stay completely separate. +type mcpServerConfig struct { + Name string + Transport string + Command string + Args []string + Env map[string]string + URL string + Headers map[string]string +} + +// mcpConfigFile mirrors the on-disk .mcp.json schema. +type mcpConfigFile struct { + MCPServers map[string]json.RawMessage `json:"mcpServers"` +} + +// mcpServerEntry is a single server block inside mcpServers. +type mcpServerEntry struct { + Command string `json:"command"` + Args []string `json:"args"` + Env map[string]string `json:"env"` + Type string `json:"type"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` +} + +// parseMCPConfig reads a .mcp.json file at path and returns the declared +// MCP servers sorted by name. It returns an empty slice when the +// mcpServers key is missing or empty. It is a self-contained copy of the +// agent/x/agentmcp parser so agentcontext can discover and start its own +// MCP servers without importing that package. +func parseMCPConfig(path string) ([]mcpServerConfig, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, xerrors.Errorf("read mcp config %q: %w", path, err) + } + + var cfg mcpConfigFile + if err := json.Unmarshal(data, &cfg); err != nil { + return nil, xerrors.Errorf("parse mcp config %q: %w", path, err) + } + + if len(cfg.MCPServers) == 0 { + return []mcpServerConfig{}, nil + } + + servers := make([]mcpServerConfig, 0, len(cfg.MCPServers)) + for name, raw := range cfg.MCPServers { + var entry mcpServerEntry + if err := json.Unmarshal(raw, &entry); err != nil { + return nil, xerrors.Errorf("parse server %q in %q: %w", name, path, err) + } + + tr := inferMCPTransport(entry) + if tr == "" { + return nil, xerrors.Errorf("server %q in %q has no command or url", name, path) + } + + resolveMCPEnvVars(entry.Env) + + servers = append(servers, mcpServerConfig{ + Name: name, + Transport: tr, + Command: entry.Command, + Args: entry.Args, + Env: entry.Env, + URL: entry.URL, + Headers: entry.Headers, + }) + } + + slices.SortFunc(servers, func(a, b mcpServerConfig) int { + return strings.Compare(a.Name, b.Name) + }) + + return servers, nil +} + +// inferMCPTransport determines the transport type for a server entry. +// An explicit "type" field takes priority; otherwise the presence of +// "command" implies stdio and "url" implies http. +func inferMCPTransport(e mcpServerEntry) string { + if e.Type != "" { + return e.Type + } + if e.Command != "" { + return "stdio" + } + if e.URL != "" { + return "http" + } + return "" +} + +// resolveMCPEnvVars expands ${VAR} references in env map values using +// the current process environment. +func resolveMCPEnvVars(env map[string]string) { + for k, v := range env { + env[k] = os.Expand(v, os.Getenv) + } +} + +// mcpRunner connects to the MCP servers declared in the .mcp.json files +// the context resolver discovers, lists each server's tools, and caches +// a non-blocking per-server snapshot that buildMCPServerResources turns +// into KindMCPServer resources. It owns its own connection lifecycle and +// does not share state with agent/x/agentmcp: the two MCP paths run +// independently during the rollout. Connections are one-shot +// (connect, initialize, list tools, close) because the runner only needs +// each server's tool list to push to coderd, not a live tool-call proxy. +type mcpRunner struct { + logger slog.Logger + execer agentexec.Execer + updateEnv func([]string) ([]string, error) + onChange func() + + // reloadMu serializes Reload so a slow reload cannot interleave + // with a newer one and publish a stale cache. The sole production + // caller (runMCPSync) already calls Reload sequentially; the mutex + // is defensive. + reloadMu sync.Mutex + + mu sync.Mutex + cache []MCPServerStatus +} + +// newMCPRunner constructs a runner. onChange is invoked (outside the +// cache lock) after a Reload that changes the per-server snapshot, so +// the manager can re-resolve and push the updated KindMCPServer +// resources. updateEnv may be nil. +func newMCPRunner(logger slog.Logger, execer agentexec.Execer, updateEnv func([]string) ([]string, error), onChange func()) *mcpRunner { + return &mcpRunner{ + logger: logger, + execer: execer, + updateEnv: updateEnv, + onChange: onChange, + } +} + +// Servers returns a deep copy of the current per-server MCP snapshot. It +// never blocks on I/O: the resolver calls it on every re-resolve. +func (r *mcpRunner) Servers() []MCPServerStatus { + r.mu.Lock() + defer r.mu.Unlock() + return cloneMCPServers(r.cache) +} + +// Reload reparses the supplied .mcp.json paths, connects to every +// declared server in parallel, lists its tools, and replaces the cached +// snapshot with the fresh result (including per-server failures). It +// fires onChange when the snapshot changed. Reload is best-effort: a +// server that fails to connect or list tools is recorded as a +// disconnected entry rather than aborting the whole reload. +func (r *mcpRunner) Reload(ctx context.Context, paths []string) { + r.reloadMu.Lock() + defer r.reloadMu.Unlock() + + configs := r.parseConfigs(ctx, paths) + statuses := r.connectAll(ctx, configs) + + r.mu.Lock() + changed := !reflect.DeepEqual(r.cache, statuses) + r.cache = statuses + r.mu.Unlock() + + if changed && r.onChange != nil { + r.onChange() + } +} + +// parseConfigs parses every path and returns the union of declared +// servers, deduplicated by name (first occurrence wins). Missing files +// are skipped silently; other parse errors are logged and skipped so one +// broken .mcp.json does not drop the servers declared in sibling files. +func (r *mcpRunner) parseConfigs(ctx context.Context, paths []string) []mcpServerConfig { + var all []mcpServerConfig + for _, path := range paths { + configs, err := parseMCPConfig(path) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + continue + } + r.logger.Warn(ctx, "failed to parse MCP config", + slog.F("path", path), + slog.Error(err), + ) + continue + } + all = append(all, configs...) + } + + seen := make(map[string]struct{}, len(all)) + deduped := make([]mcpServerConfig, 0, len(all)) + for _, cfg := range all { + if _, ok := seen[cfg.Name]; ok { + continue + } + seen[cfg.Name] = struct{}{} + deduped = append(deduped, cfg) + } + return deduped +} + +// connectAll connects to each server in parallel (bounded) and returns +// one status per server in name order. Per-server failures are isolated: +// a failed connect or list becomes a disconnected status carrying the +// error instead of failing the batch. +func (r *mcpRunner) connectAll(ctx context.Context, configs []mcpServerConfig) []MCPServerStatus { + if len(configs) == 0 { + return nil + } + statuses := make([]MCPServerStatus, len(configs)) + var eg errgroup.Group + eg.SetLimit(mcpConnectConcurrency) + for i, cfg := range configs { + eg.Go(func() error { + st := MCPServerStatus{Name: cfg.Name} + tools, err := r.connectAndList(ctx, cfg) + if err != nil { + r.logger.Warn(ctx, "failed to connect MCP server", + slog.F("server", cfg.Name), + slog.Error(err), + ) + st.Err = err.Error() + } else { + st.Connected = true + st.Tools = tools + } + statuses[i] = st + return nil + }) + } + _ = eg.Wait() + return statuses +} + +// connectAndList starts a single MCP server, completes the initialize +// handshake, lists its tools, and closes the connection. Tool names are +// returned exactly as the server reported them; the resource carries the +// server name separately, so any flattening into a single namespace is +// left to the control plane. +func (r *mcpRunner) connectAndList(ctx context.Context, cfg mcpServerConfig) ([]MCPTool, error) { + tr, err := r.createTransport(ctx, cfg) + if err != nil { + return nil, xerrors.Errorf("create transport for %q: %w", cfg.Name, err) + } + + c := client.NewClient(tr) + + // Tie the subprocess to cmdCtx. mcp-go's stdio Close() closes stdin + // and then blocks on cmd.Wait() with no kill: a server that ignores + // stdin-close would stall this reload indefinitely (the deferred + // Close runs before connectAndList returns, so it would block the + // errgroup and hold the reload lock). Canceling cmdCtx force-kills + // the process via exec.CommandContext, so Close's Wait returns. The + // deferred cleanup cancels before closing, and runs before the + // connectCtx cancel because defers are LIFO. + cmdCtx, cmdCancel := context.WithCancel(ctx) + connectCtx, cancel := context.WithTimeout(cmdCtx, mcpConnectTimeout) + defer cancel() + + if err := c.Start(cmdCtx); err != nil { + cmdCancel() + _ = c.Close() + return nil, xerrors.Errorf("start %q: %w", cfg.Name, err) + } + defer func() { + cmdCancel() + _ = c.Close() + }() + + if _, err := c.Initialize(connectCtx, mcp.InitializeRequest{ + Params: mcp.InitializeParams{ + ProtocolVersion: mcp.LATEST_PROTOCOL_VERSION, + ClientInfo: mcp.Implementation{ + Name: "coder-agent", + Version: buildinfo.Version(), + }, + }, + }); err != nil { + return nil, xerrors.Errorf("initialize %q: %w", cfg.Name, err) + } + + result, err := c.ListTools(connectCtx, mcp.ListToolsRequest{}) + if err != nil { + return nil, xerrors.Errorf("list tools from %q: %w", cfg.Name, err) + } + + tools := make([]MCPTool, 0, len(result.Tools)) + for _, tool := range result.Tools { + tools = append(tools, MCPTool{ + Name: tool.Name, + Description: tool.Description, + InputSchema: toolInputSchema(tool.InputSchema), + }) + } + return tools, nil +} + +// createTransport builds the mcp-go transport for a server config. +func (r *mcpRunner) createTransport(ctx context.Context, cfg mcpServerConfig) (transport.Interface, error) { + switch cfg.Transport { + case "stdio": + env := r.buildEnv(ctx, cfg.Env) + return transport.NewStdioWithOptions( + cfg.Command, + env, + cfg.Args, + transport.WithCommandFunc(func(ctx context.Context, command string, cmdEnv []string, args []string) (*exec.Cmd, error) { + cmd := r.execer.CommandContext(ctx, command, args...) + cmd.Env = cmdEnv + return cmd, nil + }), + ), nil + case "http", "": + return transport.NewStreamableHTTP(cfg.URL, transport.WithHTTPHeaders(cfg.Headers)) + case "sse": + return transport.NewSSE(cfg.URL, transport.WithHeaders(cfg.Headers)) + default: + return nil, xerrors.Errorf("unsupported transport %q", cfg.Transport) + } +} + +// buildEnv enriches the process environment via the agent's updateEnv +// callback, then merges explicit overrides from the server config on +// top. Note: env enrichment is captured per Reload; an env change alone +// (without a .mcp.json change) does not trigger a re-list. +func (r *mcpRunner) buildEnv(ctx context.Context, explicit map[string]string) []string { + env := usershell.SystemEnvInfo{}.Environ() + if r.updateEnv != nil { + updated, err := r.updateEnv(env) + if err != nil { + r.logger.Warn(ctx, "failed to enrich MCP server environment", slog.Error(err)) + env = usershell.SystemEnvInfo{}.Environ() + } else { + env = updated + } + } + if len(explicit) == 0 { + return env + } + + existing := make(map[string]int, len(env)) + for i, kv := range env { + if k, _, ok := strings.Cut(kv, "="); ok { + existing[k] = i + } + } + for k, v := range explicit { + entry := k + "=" + v + if idx, ok := existing[k]; ok { + env[idx] = entry + } else { + env = append(env, entry) + } + } + return env +} + +// toolInputSchema converts an mcp-go tool input schema into the +// JSON-Schema-shaped map MCPTool carries. Required is converted to +// []any (not []string) so the downstream structpb encoding accepts it. +// An empty schema yields nil so the tool ships with InputSchema unset. +func toolInputSchema(s mcp.ToolInputSchema) map[string]any { + out := map[string]any{} + if s.Type != "" { + out["type"] = s.Type + } + if len(s.Properties) > 0 { + out["properties"] = s.Properties + } + if len(s.Required) > 0 { + required := make([]any, len(s.Required)) + for i, req := range s.Required { + required[i] = req + } + out["required"] = required + } + if len(out) == 0 { + return nil + } + return out +} + +// cloneMCPServers deep-copies a per-server snapshot so callers cannot +// mutate the runner's cache. Tool input schemas are treated as immutable +// and shared by reference. +func cloneMCPServers(in []MCPServerStatus) []MCPServerStatus { + if len(in) == 0 { + return nil + } + out := make([]MCPServerStatus, len(in)) + for i, s := range in { + s.Tools = slices.Clone(s.Tools) + out[i] = s + } + return out +} + +// runMCPSync keeps the runner's connected servers in sync with the +// .mcp.json files the resolver discovers. It subscribes to snapshot +// changes, extracts the set of KindMCPConfig paths (keyed by content +// hash so in-place edits are detected), and reloads the runner only when +// that set changes. A reload fires the runner's onChange, which +// re-resolves and surfaces the updated KindMCPServer resources; that +// re-resolve does not change the config set, so it does not loop. +func (m *Manager) runMCPSync(ctx context.Context) { + changes, unsubscribe := m.SubscribeChanges() + defer unsubscribe() + + var lastKey string + reload := func() { + paths, key := mcpConfigSet(m.Snapshot()) + if key == lastKey { + return + } + lastKey = key + m.mcpRunner.Reload(ctx, paths) + } + + // Pick up any .mcp.json discovered before we subscribed. + reload() + for { + select { + case <-ctx.Done(): + return + case <-m.closedCh: + return + case <-changes: + reload() + } + } +} + +// mcpConfigSet extracts the .mcp.json config files from a snapshot's +// KindMCPConfig resources. It returns the sorted unique source paths +// plus a key encoding path:contenthash pairs, so callers detect both +// path-set changes and in-place content edits. An empty set yields an +// empty key. +func mcpConfigSet(snap Snapshot) (paths []string, key string) { + hashes := make(map[string]string, len(snap.Resources)) + for _, r := range snap.Resources { + if r.Kind != KindMCPConfig || r.Source == "" { + continue + } + hashes[r.Source] = hex.EncodeToString(r.ContentHash[:]) + } + if len(hashes) == 0 { + return nil, "" + } + paths = make([]string, 0, len(hashes)) + for p := range hashes { + paths = append(paths, p) + } + slices.Sort(paths) + parts := make([]string, len(paths)) + for i, p := range paths { + parts[i] = p + ":" + hashes[p] + } + return paths, strings.Join(parts, "\n") +} diff --git a/agent/agentcontext/mcprunner_internal_test.go b/agent/agentcontext/mcprunner_internal_test.go new file mode 100644 index 0000000000..eba6d88a79 --- /dev/null +++ b/agent/agentcontext/mcprunner_internal_test.go @@ -0,0 +1,148 @@ +package agentcontext + +import ( + "os" + "path/filepath" + "testing" + + "github.com/mark3labs/mcp-go/mcp" + "github.com/stretchr/testify/require" +) + +func TestParseMCPConfig(t *testing.T) { + t.Parallel() + + write := func(t *testing.T, body string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, ".mcp.json") + require.NoError(t, os.WriteFile(path, []byte(body), 0o600)) + return path + } + + t.Run("InfersTransportAndSorts", func(t *testing.T) { + t.Parallel() + path := write(t, `{"mcpServers": { + "zebra": {"command": "zebra-bin", "args": ["--flag"]}, + "alpha": {"url": "https://example.com/mcp"} + }}`) + got, err := parseMCPConfig(path) + require.NoError(t, err) + require.Len(t, got, 2) + // Sorted by name. + require.Equal(t, "alpha", got[0].Name) + require.Equal(t, "http", got[0].Transport) + require.Equal(t, "https://example.com/mcp", got[0].URL) + require.Equal(t, "zebra", got[1].Name) + require.Equal(t, "stdio", got[1].Transport) + require.Equal(t, "zebra-bin", got[1].Command) + require.Equal(t, []string{"--flag"}, got[1].Args) + }) + + t.Run("ExplicitTypeWins", func(t *testing.T) { + t.Parallel() + path := write(t, `{"mcpServers": {"s": {"type": "sse", "url": "https://x"}}}`) + got, err := parseMCPConfig(path) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, "sse", got[0].Transport) + }) + + t.Run("EmptyServers", func(t *testing.T) { + t.Parallel() + path := write(t, `{"mcpServers": {}}`) + got, err := parseMCPConfig(path) + require.NoError(t, err) + require.Empty(t, got) + }) + + t.Run("RejectsServerWithoutCommandOrURL", func(t *testing.T) { + t.Parallel() + path := write(t, `{"mcpServers": {"s": {}}}`) + _, err := parseMCPConfig(path) + require.Error(t, err) + }) + + t.Run("InvalidJSON", func(t *testing.T) { + t.Parallel() + path := write(t, `{not json`) + _, err := parseMCPConfig(path) + require.Error(t, err) + }) +} + +// TestParseMCPConfig_ExpandsEnv is a standalone (non-parallel) test +// because t.Setenv cannot be used under a parallel parent test. +func TestParseMCPConfig_ExpandsEnv(t *testing.T) { + t.Setenv("AGENTCONTEXT_MCP_TEST_TOKEN", "secret") + dir := t.TempDir() + path := filepath.Join(dir, ".mcp.json") + require.NoError(t, os.WriteFile(path, + []byte(`{"mcpServers": {"s": {"command": "x", "env": {"TOKEN": "${AGENTCONTEXT_MCP_TEST_TOKEN}"}}}}`), 0o600)) + got, err := parseMCPConfig(path) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, "secret", got[0].Env["TOKEN"]) +} + +func TestToolInputSchema(t *testing.T) { + t.Parallel() + + t.Run("FullSchema", func(t *testing.T) { + t.Parallel() + got := toolInputSchema(mcp.ToolInputSchema{ + Type: "object", + Properties: map[string]any{"q": map[string]any{"type": "string"}}, + Required: []string{"q"}, + }) + require.Equal(t, "object", got["type"]) + require.Equal(t, map[string]any{"q": map[string]any{"type": "string"}}, got["properties"]) + // Required is converted to []any so structpb.NewStruct accepts it. + require.Equal(t, []any{"q"}, got["required"]) + }) + + t.Run("EmptyYieldsNil", func(t *testing.T) { + t.Parallel() + require.Nil(t, toolInputSchema(mcp.ToolInputSchema{})) + }) + + t.Run("TypeOnly", func(t *testing.T) { + t.Parallel() + got := toolInputSchema(mcp.ToolInputSchema{Type: "object"}) + require.Equal(t, map[string]any{"type": "object"}, got) + }) +} + +func TestMCPConfigSet(t *testing.T) { + t.Parallel() + + t.Run("Empty", func(t *testing.T) { + t.Parallel() + paths, key := mcpConfigSet(Snapshot{}) + require.Empty(t, paths) + require.Empty(t, key) + }) + + t.Run("SortedAndKeyedByContentHash", func(t *testing.T) { + t.Parallel() + snap := Snapshot{Resources: []Resource{ + {Kind: KindMCPConfig, Source: "/b/.mcp.json", ContentHash: [32]byte{0x01}}, + {Kind: KindMCPConfig, Source: "/a/.mcp.json", ContentHash: [32]byte{0x02}}, + // Non-config and empty-source resources are ignored. + {Kind: KindInstructionFile, Source: "/a/AGENTS.md"}, + {Kind: KindMCPServer, Source: "fs"}, + {Kind: KindMCPConfig, Source: ""}, + }} + paths, key := mcpConfigSet(snap) + require.Equal(t, []string{"/a/.mcp.json", "/b/.mcp.json"}, paths) + require.NotEmpty(t, key) + + // An in-place content edit (same path, new hash) changes the key. + snap2 := Snapshot{Resources: []Resource{ + {Kind: KindMCPConfig, Source: "/b/.mcp.json", ContentHash: [32]byte{0x01}}, + {Kind: KindMCPConfig, Source: "/a/.mcp.json", ContentHash: [32]byte{0x09}}, + }} + _, key2 := mcpConfigSet(snap2) + require.NotEqual(t, key, key2) + }) +} diff --git a/agent/agentcontext/resolve.go b/agent/agentcontext/resolve.go index d55680b8dd..86d3f4002d 100644 --- a/agent/agentcontext/resolve.go +++ b/agent/agentcontext/resolve.go @@ -1,8 +1,10 @@ package agentcontext import ( + "bytes" "context" "crypto/sha256" + "encoding/json" "errors" "fmt" "io" @@ -15,6 +17,8 @@ import ( "strconv" "strings" + "golang.org/x/xerrors" + "github.com/coder/coder/v2/codersdk/workspacesdk" ) @@ -100,10 +104,13 @@ type Resolver struct { // MaxDepth caps the directory walk depth. Use // DefaultMaxScanDepth if zero. MaxDepth int - // MCP, when non-nil, is consulted after the filesystem - // pass and contributes any KindMCPServer resources for - // live MCP servers. - MCP MCPProvider + // MCPResources, when non-nil, is consulted after the + // filesystem pass and returns the KindMCPServer resources + // for live MCP servers. It must not block: the resolver + // calls it on every re-resolve. In production the manager + // wires this to its MCP runner's snapshot; tests inject a + // closure directly. + MCPResources func() []Resource } // ScanRoot describes a single directory or file the resolver @@ -144,8 +151,8 @@ func (r *Resolver) ResolveContext(ctx context.Context, roots []ScanRoot) Snapsho // Append MCP server resources after the filesystem caps // are applied so a runaway MCP server cannot crowd out // instruction files. - if r.MCP != nil { - mcp := r.MCP.MCPResources() + if r.MCPResources != nil { + mcp := r.MCPResources() startIdx := len(resources) resources = append(resources, mcp...) // MCP resources may push the aggregate over the @@ -164,7 +171,10 @@ func (r *Resolver) ResolveContext(ctx context.Context, roots []ScanRoot) Snapsho payloadBytes += uint64(len(r.Payload)) } - hash := ComputeAggregateHash(resources) + // The drift hash covers only pinned prompt content; MCP resources are + // excluded (see driftResources). Snapshot.Resources still carries the + // full set so MCP servers stay visible in the chat-context snapshot. + hash := ComputeAggregateHash(driftResources(resources)) snap := Snapshot{ Resources: resources, @@ -439,8 +449,8 @@ func (r *Resolver) readInstructionFile(scanRoot, path string, info fs.FileInfo, // .mcp.json fragments frequently embed secret-bearing fields // (Env tokens, Authorization headers). The resolver hashes the // file for change detection but intentionally does not ship -// the bytes; the live MCP server's tool list arrives via the -// MCPProvider as a KindMCPServer resource, which is what +// the bytes; the live MCP server's tool list arrives +// separately as a KindMCPServer resource, which is what // downstream consumers actually need. func (r *Resolver) readMCPConfig(scanRoot, path string, info fs.FileInfo, userSource string) Resource { res := Resource{ @@ -472,9 +482,46 @@ func (r *Resolver) readMCPConfig(scanRoot, path string, info fs.FileInfo, userSo return res } res.ContentHash = sha256.Sum256(data) + // A .mcp.json with broken JSON yields no MCP servers at all; the + // MCP manager logs and skips it, so the failure is otherwise + // invisible. Flag structural problems here as StatusInvalid so the + // chat context surfaces them as an issue rather than silently + // dropping every server in the file. + if err := validateMCPConfig(data); err != nil { + res.Status = StatusInvalid + res.Error = err.Error() + } return res } +// validateMCPConfig performs lightweight structural validation of a +// .mcp.json document so syntactically broken files surface as +// StatusInvalid instead of silently producing no MCP servers. It is +// deliberately self-contained and does not import the MCP package: it +// only checks that the document is valid JSON shaped like +// {"mcpServers": {: {...}}}. Individual server fields +// (command/url/env/...) are not validated here; the MCP manager owns +// that when it connects. An absent or empty mcpServers map is valid. +func validateMCPConfig(data []byte) error { + var shape struct { + MCPServers map[string]json.RawMessage `json:"mcpServers"` + } + if err := json.Unmarshal(data, &shape); err != nil { + return err + } + // Each server entry must be a JSON object; a scalar or array + // entry is a structural error the MCP manager would reject. + // The top-level Unmarshal above already rejects malformed JSON, + // so a well-formed value starting with '{' is a complete object. + for name, raw := range shape.MCPServers { + trimmed := bytes.TrimSpace(raw) + if len(trimmed) == 0 || trimmed[0] != '{' { + return xerrors.Errorf("server %q must be a JSON object", name) + } + } + return nil +} + // readFileResource is the shared plumbing for kinds whose only // difference is the enum stamped on the Resource: build the // Resource header, enforce the per-resource size cap, read the @@ -785,8 +832,8 @@ const ( // more MCP servers. KindMCPConfig // KindMCPServer is a live MCP server's resolved tool list, - // populated by an MCPProvider after the server has been - // connected. + // populated from the MCP runner's snapshot after the server + // has been connected. KindMCPServer // KindPlugin is reserved for Claude Code plugin manifests. // Not emitted by v1. @@ -938,8 +985,12 @@ type Snapshot struct { Version uint64 // AggregateHash is sha256 over a canonical encoding of // (ID, Kind, Source, ContentHash, Status) for every - // resource. Identical inputs always produce identical - // hashes; see ComputeAggregateHash. + // drift-relevant resource. MCP resources (KindMCPConfig and + // KindMCPServer) are excluded because they describe live, + // agent-global runtime capabilities discovered at turn time, + // not pinned prompt content; see driftResources. Identical + // inputs always produce identical hashes; see + // ComputeAggregateHash. AggregateHash [32]byte // Resources is sorted by ID for deterministic encoding. Resources []Resource @@ -952,6 +1003,28 @@ type Snapshot struct { SnapshotError string } +// driftResources returns the subset of resources that participate in +// chat-context drift detection. MCP resources (the .mcp.json config and +// connected MCP servers) are deliberately excluded: an agent connects to +// its MCP servers asynchronously after startup, and the chat model +// discovers their tools live at turn time, not from pinned prompt +// content. Hashing them would dirty an already-hydrated chat the moment +// a server finished connecting, even though nothing the user pinned +// changed. Instruction files and skills, whose content is pinned into +// the chat, stay drift-relevant. +func driftResources(resources []Resource) []Resource { + out := make([]Resource, 0, len(resources)) + for _, r := range resources { + switch r.Kind { + case KindMCPConfig, KindMCPServer: + continue + default: + out = append(out, r) + } + } + return out +} + // ComputeAggregateHash produces the deterministic snapshot // aggregate hash for the supplied resources. The caller does // not need to pre-sort; the function sorts a copy of the slice diff --git a/agent/agentcontext/resolve_test.go b/agent/agentcontext/resolve_test.go index aa7d6090a7..9c6213d21b 100644 --- a/agent/agentcontext/resolve_test.go +++ b/agent/agentcontext/resolve_test.go @@ -354,7 +354,7 @@ func TestResolver_DuplicateRootsDeduplicated(t *testing.T) { require.Len(t, snap.Resources, 1) } -func TestResolver_MCPProviderResources(t *testing.T) { +func TestResolver_MCPResources(t *testing.T) { t.Parallel() dir := t.TempDir() @@ -368,7 +368,7 @@ func TestResolver_MCPProviderResources(t *testing.T) { Description: "GitHub MCP server", } r := &agentcontext.Resolver{ - MCP: &fakeMCPProvider{resources: []agentcontext.Resource{mcpRes}}, + MCPResources: func() []agentcontext.Resource { return []agentcontext.Resource{mcpRes} }, } snap := r.Resolve([]agentcontext.ScanRoot{{Path: dir}}) @@ -377,10 +377,10 @@ func TestResolver_MCPProviderResources(t *testing.T) { require.Equal(t, "GitHub MCP server", got.Description) } -// TestResolver_MCPProviderRespectsAggregateByteCap guards the +// TestResolver_MCPResourcesRespectAggregateByteCap guards the // contract that a single oversized MCP payload cannot blow past // MaxSnapshotBytes with StatusOK. -func TestResolver_MCPProviderRespectsAggregateByteCap(t *testing.T) { +func TestResolver_MCPResourcesRespectAggregateByteCap(t *testing.T) { t.Parallel() dir := t.TempDir() @@ -398,7 +398,7 @@ func TestResolver_MCPProviderRespectsAggregateByteCap(t *testing.T) { } r := &agentcontext.Resolver{ MaxSnapshotBytes: 512, - MCP: &fakeMCPProvider{resources: []agentcontext.Resource{mcpRes}}, + MCPResources: func() []agentcontext.Resource { return []agentcontext.Resource{mcpRes} }, } snap := r.Resolve([]agentcontext.ScanRoot{{Path: dir}}) @@ -409,12 +409,37 @@ func TestResolver_MCPProviderRespectsAggregateByteCap(t *testing.T) { require.NotEmpty(t, snap.SnapshotError, "snapshot must surface the cap breach") } -type fakeMCPProvider struct { - resources []agentcontext.Resource -} +// TestResolver_MCPExcludedFromAggregateHash verifies that MCP resources +// (config and live servers) are carried in the snapshot but excluded +// from the drift/aggregate hash, so an MCP server connecting (or its +// tools changing) does not flip already-hydrated chats to dirty. +func TestResolver_MCPExcludedFromAggregateHash(t *testing.T) { + t.Parallel() + dir := t.TempDir() + // An instruction file provides drift-relevant pinned content. + mustWriteFile(t, filepath.Join(dir, "AGENTS.md"), "workspace rules") -func (f *fakeMCPProvider) MCPResources() []agentcontext.Resource { - return f.resources + base := (&agentcontext.Resolver{}).Resolve([]agentcontext.ScanRoot{{Path: dir}}) + + mcpRes := agentcontext.Resource{ + ID: "mcp_server:github", + Kind: agentcontext.KindMCPServer, + Source: "github", + Name: "github", + Status: agentcontext.StatusOK, + ContentHash: sha256.Sum256([]byte("tool-list")), + Tools: []agentcontext.MCPTool{{Name: "search"}}, + } + withMCP := (&agentcontext.Resolver{ + MCPResources: func() []agentcontext.Resource { return []agentcontext.Resource{mcpRes} }, + }).Resolve([]agentcontext.ScanRoot{{Path: dir}}) + + // The MCP server resource is present in the snapshot... + got := findResource(t, withMCP.Resources, agentcontext.KindMCPServer, "github") + require.Len(t, got.Tools, 1) + // ...but does not change the drift/aggregate hash. + require.Equal(t, base.AggregateHash, withMCP.AggregateHash, + "MCP resources must not participate in the drift hash") } // TestResolver_UnreadableInstructionFile verifies the