Skip to content

Commit 55133e9

Browse files
jscottmillerjscottmiller
committed
feat(agent): run plugin MCP servers discovered by agentcontext
The agent now feeds each valid plugin's mcp.json into the MCP engine. mcpConfigSources composes the env-configured .mcp.json files followed by one plugin-scoped source per plugin, built by agentmcp.PluginSource with PLUGIN_DATA under ~/.coder/plugin-data. A plugin without an mcp.json contributes no servers, and the engine's config watcher picks the file up when it appears. The agent registers mcpConfigSources with Manager.SetSources, and the startup reload, a context snapshot subscriber, and config watcher reloads all compute the current sources from it. ReloadCurrent calls it on every attempt and re-checks the snapshot after joining a concurrent reload, so one trigger cannot drop another's servers. It returns an error if the sources keep changing after three reloads. The subscriber reloads on every snapshot change once the first resolve has finished; a reload with unchanged sources is a no-op. The subscription is taken before the context manager's Run loop starts, and the current snapshot is examined before the first wait. Snapshot.Plugins() returns the valid plugins in the order the resolver claimed their names, so the first plugin to claim a server name wins deterministically. MCPServerStatus gains PluginName, which buildMCPServerResources copies onto the resource. MCP server resources declared by a plugin carry the plugin name and use <plugin>/<server> as their source, so a plugin entry that lost a name collision with a legacy server still appears with its error. A plugin server whose <plugin>/<server> ID equals a legacy server name is not started, and the legacy server's resource carries a warning naming it. A plugin server whose source exceeds the 1024-byte source limit is left out and counted on an unreadable <plugin>/mcp.json resource, or on the plugin's server named mcp.json when there is one.
1 parent a003e25 commit 55133e9

12 files changed

Lines changed: 783 additions & 37 deletions

‎agent/agent.go‎

Lines changed: 62 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -497,6 +497,15 @@ func (a *agent) init() {
497497
// Re-resolve and re-push KindMCPServer resources whenever the MCP
498498
// engine's catalog changes (startup connect, .mcp.json edits).
499499
a.mcpManager.SetOnReload(a.contextManager.Trigger)
500+
a.mcpManager.SetSources(a.mcpConfigSources)
501+
pluginChanges, unsubPluginChanges := a.contextManager.SubscribeChanges()
502+
if err := a.trackGoroutine(func() {
503+
defer unsubPluginChanges()
504+
a.runPluginMCPSync(a.gracefulCtx, pluginChanges)
505+
}); err != nil {
506+
unsubPluginChanges()
507+
a.logger.Warn(a.gracefulCtx, "plugin MCP sync not started", slog.Error(err))
508+
}
500509
a.reconnectingPTYServer = reconnectingpty.NewServer(
501510
a.logger.Named("reconnecting-pty"),
502511
a.sshServer,
@@ -524,6 +533,57 @@ func (a *agent) init() {
524533
go a.runLoop()
525534
}
526535

536+
// mcpConfigSources omits plugin servers when the home directory, which
537+
// holds their data directories, cannot be determined.
538+
func (a *agent) mcpConfigSources() []agentmcp.ConfigSource {
539+
sources := agentmcp.LegacySources(a.contextConfigAPI.MCPConfigFiles())
540+
plugins := a.contextManager.Snapshot().Plugins()
541+
if len(plugins) == 0 {
542+
return sources
543+
}
544+
home, err := a.envInfo.HomeDir()
545+
if err != nil {
546+
a.logger.Warn(a.gracefulCtx, "skipping plugin MCP servers: home directory unavailable", slog.Error(err))
547+
return sources
548+
}
549+
for _, p := range plugins {
550+
src, err := agentmcp.PluginSource(home, p.Name, p.Root)
551+
if err != nil {
552+
a.logger.Warn(a.gracefulCtx, "skipping plugin MCP servers", slog.F("plugin", p.Name), slog.Error(err))
553+
continue
554+
}
555+
sources = append(sources, src)
556+
}
557+
return sources
558+
}
559+
560+
// runPluginMCPSync reloads the MCP engine on every context snapshot
561+
// change after the first real resolve. A reload whose sources are
562+
// unchanged is a no-op, and a failed reload is retried on the next
563+
// snapshot. The snapshot is read before the first wait, so a change
564+
// published before the loop starts is not lost.
565+
func (a *agent) runPluginMCPSync(ctx context.Context, changes <-chan struct{}) {
566+
for {
567+
if ctx.Err() != nil {
568+
return
569+
}
570+
if a.contextManager.Snapshot().Version > 0 {
571+
switch err := a.mcpManager.ReloadCurrent(ctx); {
572+
case err == nil:
573+
case errors.Is(err, context.Canceled), errors.Is(err, agentmcp.ErrManagerClosed):
574+
return
575+
default:
576+
a.logger.Warn(ctx, "failed to reload MCP servers", slog.F("trigger", "context_change"), slog.Error(err))
577+
}
578+
}
579+
select {
580+
case <-ctx.Done():
581+
return
582+
case <-changes:
583+
}
584+
}
585+
}
586+
527587
// initSocketServer initializes server that allows direct communication with a workspace agent using IPC.
528588
func (a *agent) initSocketServer() {
529589
if !a.socketServerEnabled {
@@ -1430,8 +1490,8 @@ func (a *agent) handleManifest(manifestOK *checkpoint) func(ctx context.Context,
14301490
// lifecycle transition to avoid delaying Ready.
14311491
// This runs inside the tracked goroutine so it
14321492
// is properly awaited on shutdown.
1433-
if mcpErr := a.mcpManager.Reload(a.gracefulCtx, a.contextConfigAPI.MCPConfigFiles()); mcpErr != nil {
1434-
a.logger.Warn(ctx, "failed to reload workspace MCP servers", slog.Error(mcpErr))
1493+
if mcpErr := a.mcpManager.ReloadCurrent(a.gracefulCtx); mcpErr != nil {
1494+
a.logger.Warn(ctx, "failed to reload MCP servers", slog.F("trigger", "manifest"), slog.Error(mcpErr))
14351495
}
14361496
})
14371497
if err != nil {

‎agent/agent_context_test.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,160 @@ func TestAgent_ContextStatePluginsDroppedAfterReconnect(t *testing.T) {
181181
}
182182
}
183183

184+
// The declared command does not exist, so the server is unreadable;
185+
// only the attribution and plugin-prefixed source are checked.
186+
func TestAgent_PluginMCPServersLoaded(t *testing.T) {
187+
t.Parallel()
188+
189+
dir := t.TempDir()
190+
pluginDir := filepath.Join(dir, ".agents", "plugins", "acme")
191+
require.NoError(t, os.MkdirAll(pluginDir, 0o700))
192+
require.NoError(t, os.WriteFile(filepath.Join(pluginDir, "plugin.json"), []byte(`{
193+
"$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json",
194+
"name": "acme"
195+
}`), 0o600))
196+
require.NoError(t, os.WriteFile(filepath.Join(pluginDir, "mcp.json"), []byte(`{
197+
"$schema": "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json",
198+
"mcpServers": {
199+
"srv": {"type": "stdio", "command": "coder-test-missing-mcp-binary"}
200+
}
201+
}`), 0o600))
202+
203+
//nolint:dogsled // setupAgent returns a wide tuple; we only care about the client.
204+
_, client, _, _, _ := setupAgent(t,
205+
agentsdk.Manifest{Directory: dir, PluginsSupported: true},
206+
0,
207+
func(_ *agenttest.Client, opts *agent.Options) {
208+
opts.ContextConfig = agentcontextconfig.Config{}
209+
},
210+
)
211+
212+
var (
213+
plugin *agentproto.ContextResource
214+
server *agentproto.ContextResource
215+
)
216+
pushed := assert.Eventually(t, func() bool {
217+
plugin, server = nil, nil
218+
for _, push := range client.ContextStatePushes() {
219+
for _, r := range push.GetResources() {
220+
if r.GetPlugin() != nil {
221+
plugin = r
222+
}
223+
if r.GetMcpServer() != nil && r.GetMcpServer().GetPluginName() == "acme" {
224+
server = r
225+
}
226+
}
227+
}
228+
return plugin != nil && server != nil
229+
}, testutil.WaitSuperLong, testutil.IntervalFast)
230+
if !pushed {
231+
t.Fatalf("expected plugin and plugin MCP server resources to be pushed; got %s",
232+
describePushes(client.ContextStatePushes()))
233+
}
234+
235+
assert.Equal(t, "acme", plugin.GetPlugin().GetName())
236+
assert.Equal(t, agentproto.ContextResource_OK, plugin.GetStatus())
237+
assert.Equal(t, "acme/srv", server.GetSource())
238+
assert.Equal(t, "srv", server.GetMcpServer().GetServerName())
239+
assert.Equal(t, agentproto.ContextResource_UNREADABLE, server.GetStatus())
240+
}
241+
242+
func TestAgent_PluginMCPServersFollowPluginChanges(t *testing.T) {
243+
t.Parallel()
244+
245+
dir := t.TempDir()
246+
pluginsDir := filepath.Join(dir, ".agents", "plugins")
247+
require.NoError(t, os.MkdirAll(pluginsDir, 0o700))
248+
249+
//nolint:dogsled // setupAgent returns a wide tuple; we only care about the client.
250+
_, client, _, _, _ := setupAgent(t,
251+
agentsdk.Manifest{Directory: dir, PluginsSupported: true},
252+
0,
253+
func(_ *agenttest.Client, opts *agent.Options) {
254+
opts.ContextConfig = agentcontextconfig.Config{}
255+
},
256+
)
257+
258+
latestHasServer := func() (pushed, found bool) {
259+
pushes := client.ContextStatePushes()
260+
if len(pushes) == 0 {
261+
return false, false
262+
}
263+
for _, r := range pushes[len(pushes)-1].GetResources() {
264+
if r.GetMcpServer() != nil && r.GetSource() == "acme/srv" {
265+
return true, true
266+
}
267+
}
268+
return true, false
269+
}
270+
require.Eventually(t, func() bool {
271+
pushed, _ := latestHasServer()
272+
return pushed
273+
}, testutil.WaitSuperLong, testutil.IntervalFast)
274+
275+
pluginDir := filepath.Join(pluginsDir, "acme")
276+
require.NoError(t, os.Mkdir(pluginDir, 0o700))
277+
require.NoError(t, os.WriteFile(filepath.Join(pluginDir, "plugin.json"), []byte(`{
278+
"$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json",
279+
"name": "acme"
280+
}`), 0o600))
281+
require.Eventually(t, func() bool {
282+
pushes := client.ContextStatePushes()
283+
for _, r := range pushes[len(pushes)-1].GetResources() {
284+
if r.GetPlugin().GetName() == "acme" {
285+
return true
286+
}
287+
}
288+
return false
289+
}, testutil.WaitSuperLong, testutil.IntervalFast)
290+
291+
// The plugin's mcp.json arrives after the plugin itself.
292+
require.NoError(t, os.WriteFile(filepath.Join(pluginDir, "mcp.json"), []byte(`{
293+
"$schema": "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json",
294+
"mcpServers": {
295+
"srv": {"type": "stdio", "command": "coder-test-missing-mcp-binary"}
296+
}
297+
}`), 0o600))
298+
299+
if !assert.Eventually(t, func() bool {
300+
_, found := latestHasServer()
301+
return found
302+
}, testutil.WaitSuperLong, testutil.IntervalFast) {
303+
t.Fatalf("expected the added plugin's MCP server to be pushed; got %s",
304+
describePushes(client.ContextStatePushes()))
305+
}
306+
307+
require.NoError(t, os.RemoveAll(pluginDir))
308+
if !assert.Eventually(t, func() bool {
309+
_, found := latestHasServer()
310+
return !found
311+
}, testutil.WaitSuperLong, testutil.IntervalFast) {
312+
t.Fatalf("expected the removed plugin's MCP server to leave the push; got %s",
313+
describePushes(client.ContextStatePushes()))
314+
}
315+
}
316+
317+
func describePushes(pushes []*agentproto.PushContextStateRequest) string {
318+
var sb strings.Builder
319+
for _, push := range pushes {
320+
for _, r := range push.GetResources() {
321+
kind := "other"
322+
switch {
323+
case r.GetPlugin() != nil:
324+
kind = "plugin"
325+
case r.GetMcpServer() != nil:
326+
kind = "mcp_server"
327+
case r.GetSkill() != nil:
328+
kind = "skill"
329+
case r.GetInstructionFile() != nil:
330+
kind = "instruction_file"
331+
}
332+
_, _ = fmt.Fprintf(&sb, "[v%d %s %s %s] ", push.GetVersion(), kind, r.GetSource(), r.GetStatus())
333+
}
334+
}
335+
return sb.String()
336+
}
337+
184338
// TestAgent_MissingDirectoryNotice verifies that an agent logs an Info
185339
// notice naming the working-directory context it skips only when its manifest
186340
// has no directory.

0 commit comments

Comments
 (0)