summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorClaude <noreply@anthropic.com>2026-05-24 09:45:16 +0000
committerClaude <noreply@anthropic.com>2026-05-24 09:45:16 +0000
commit952b7623ee9dceec15099043086622aa2aab4741 (patch)
tree1cee247252bb9401139cb81f090b228e56c0f428
parent54f6631c28a8b85f6f874e17822549faba916a38 (diff)
feat(executor,api): wire agent MCP into ContainerRunner + mount /mcp (Phase 2)
ContainerRunner now mints a per-task MCP token (from an injected Registry), writes a claude mcp-config into the workspace pointing at the host agent MCP server over host.docker.internal with that bearer, and adds --mcp-config to the in-container claude invocation. The token is revoked when the run ends. After the run, a buffered ask_user (PendingQuestion on the channel) is converted into a BlockedError — the MCP path to BLOCKED — with the file-based question.json kept as a fallback for in-flight tasks started on the old wire. The API server mounts the StreamableHTTP MCP handler at POST/GET/DELETE /mcp when a registry is provided; serve.go constructs one Registry shared by the runners (mint) and the server (resolve). Minting is skipped for gemini agents. Tests: writeMCPConfig output shape, buildInnerCmd flag presence/absence, and an api-level end-to-end MCP tool call through NewServer/Handler proving the route is mounted (and absent without a registry). https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39
-rw-r--r--internal/api/agentmcp_endpoint_test.go104
-rw-r--r--internal/api/server.go13
-rw-r--r--internal/api/server_test.go4
-rw-r--r--internal/cli/serve.go8
-rw-r--r--internal/executor/channel.go15
-rw-r--r--internal/executor/container.go63
-rw-r--r--internal/executor/container_test.go58
7 files changed, 253 insertions, 12 deletions
diff --git a/internal/api/agentmcp_endpoint_test.go b/internal/api/agentmcp_endpoint_test.go
new file mode 100644
index 0000000..b2f77d3
--- /dev/null
+++ b/internal/api/agentmcp_endpoint_test.go
@@ -0,0 +1,104 @@
+package api
+
+import (
+ "context"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/modelcontextprotocol/go-sdk/mcp"
+ "github.com/thepeterstone/claudomator/internal/executor"
+ "github.com/thepeterstone/claudomator/internal/storage"
+)
+
+type fakeAgentChannel struct{ progress []string }
+
+func (f *fakeAgentChannel) AskUser(context.Context, string) (string, error) {
+ return "", executor.ErrAgentBlocked
+}
+func (f *fakeAgentChannel) ReportSummary(context.Context, string) error { return nil }
+func (f *fakeAgentChannel) SpawnSubtask(context.Context, executor.SubtaskSpec) (string, error) {
+ return "sub", nil
+}
+func (f *fakeAgentChannel) RecordProgress(_ context.Context, m string) error {
+ f.progress = append(f.progress, m)
+ return nil
+}
+
+type tokenRT struct {
+ token string
+ base http.RoundTripper
+}
+
+func (b tokenRT) RoundTrip(r *http.Request) (*http.Response, error) {
+ r = r.Clone(r.Context())
+ r.Header.Set("Authorization", "Bearer "+b.token)
+ return b.base.RoundTrip(r)
+}
+
+func TestServer_MountsAgentMCPEndpoint(t *testing.T) {
+ store, err := storage.Open(filepath.Join(t.TempDir(), "test.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+ pool := executor.NewPool(1, map[string]executor.Runner{}, store, logger)
+
+ reg := executor.NewRegistry()
+ fc := &fakeAgentChannel{}
+ token, _ := reg.Mint(fc)
+
+ srv := NewServer(store, pool, reg, logger, "claude", "gemini")
+ ts := httptest.NewServer(srv.Handler())
+ defer ts.Close()
+
+ client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil)
+ cs, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{
+ Endpoint: ts.URL + "/mcp",
+ HTTPClient: &http.Client{Transport: tokenRT{token: token, base: http.DefaultTransport}},
+ }, nil)
+ if err != nil {
+ t.Fatalf("connect to mounted /mcp: %v", err)
+ }
+ defer cs.Close()
+
+ if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{
+ Name: "record_progress",
+ Arguments: map[string]any{"message": "via api server"},
+ }); err != nil {
+ t.Fatalf("CallTool through api server: %v", err)
+ }
+ if len(fc.progress) != 1 || fc.progress[0] != "via api server" {
+ t.Errorf("tool call did not reach channel: %+v", fc.progress)
+ }
+}
+
+func TestServer_NoRegistry_NoMCPRoute(t *testing.T) {
+ store, err := storage.Open(filepath.Join(t.TempDir(), "test.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+ pool := executor.NewPool(1, map[string]executor.Runner{}, store, logger)
+
+ // nil registry → /mcp must not be mounted.
+ srv := NewServer(store, pool, nil, logger, "claude", "gemini")
+ ts := httptest.NewServer(srv.Handler())
+ defer ts.Close()
+
+ resp, err := http.Post(ts.URL+"/mcp", "application/json", http.NoBody)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ // Unmounted: the catch-all "GET /" route rejects POST with 405. A mounted
+ // MCP handler would instead respond (e.g. 400 for the missing token).
+ if resp.StatusCode == http.StatusBadRequest {
+ t.Errorf("expected /mcp to be absent without a registry, got 400 (handler ran)")
+ }
+}
diff --git a/internal/api/server.go b/internal/api/server.go
index ff3a111..1522b72 100644
--- a/internal/api/server.go
+++ b/internal/api/server.go
@@ -39,6 +39,7 @@ type Server struct {
taskLogStore taskLogStore // injectable for tests; defaults to store
questionStore questionStore // injectable for tests; defaults to store
pool *executor.Pool
+ registry *executor.Registry // per-task agent MCP token registry; mounts /mcp when set
hub *Hub
logger *slog.Logger
mux *http.ServeMux
@@ -100,7 +101,7 @@ func (s *Server) SetLLM(c *llm.Client) {
}
-func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server {
+func NewServer(store *storage.DB, pool *executor.Pool, registry *executor.Registry, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server {
wd, _ := os.Getwd()
s := &Server{
ctx: context.Background(),
@@ -109,6 +110,7 @@ func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, clau
taskLogStore: store,
questionStore: store,
pool: pool,
+ registry: registry,
hub: NewHub(),
logger: logger,
mux: http.NewServeMux(),
@@ -180,6 +182,15 @@ func (s *Server) routes() {
s.mux.HandleFunc("GET /api/drops", s.handleListDrops)
s.mux.HandleFunc("GET /api/drops/{filename}", s.handleGetDrop)
s.mux.HandleFunc("POST /api/drops", s.handlePostDrop)
+ if s.registry != nil {
+ mcpHandler := executor.NewAgentMCPHandler(s.registry)
+ // The streamable HTTP transport uses POST (messages), GET (SSE stream),
+ // and DELETE (session end). Register them explicitly so the patterns are
+ // more specific than the "GET /" catch-all and don't conflict.
+ s.mux.Handle("POST /mcp", mcpHandler)
+ s.mux.Handle("GET /mcp", mcpHandler)
+ s.mux.Handle("DELETE /mcp", mcpHandler)
+ }
s.mux.Handle("GET /", http.FileServerFS(webui.Files))
}
diff --git a/internal/api/server_test.go b/internal/api/server_test.go
index f902495..dd6eed5 100644
--- a/internal/api/server_test.go
+++ b/internal/api/server_test.go
@@ -98,7 +98,7 @@ func testServerWithRunner(t *testing.T, runner executor.Runner) (*Server, *stora
"gemini": runner,
}
pool := executor.NewPool(2, runners, store, logger)
- srv := NewServer(store, pool, logger, "claude", "gemini")
+ srv := NewServer(store, pool, nil, logger, "claude", "gemini")
return srv, store
}
@@ -194,7 +194,7 @@ func testServerWithGeminiMockRunner(t *testing.T) (*Server, *storage.DB) {
"gemini": mr,
}
pool := executor.NewPool(2, runners, store, logger)
- srv := NewServer(store, pool, logger, "claude", "gemini")
+ srv := NewServer(store, pool, nil, logger, "claude", "gemini")
return srv, store
}
diff --git a/internal/cli/serve.go b/internal/cli/serve.go
index 459c35b..b23304b 100644
--- a/internal/cli/serve.go
+++ b/internal/cli/serve.go
@@ -79,6 +79,9 @@ func serve(addr string) error {
claudeConfigDir := cfg.ClaudeConfigDir
repoDir, _ := os.Getwd()
+ // Shared per-task agent MCP token registry: runners mint tokens; the API
+ // server mounts /mcp and resolves them.
+ agentRegistry := executor.NewRegistry()
runners := map[string]executor.Runner{
// ContainerRunner: binaries are resolved via PATH inside the container image,
// so ClaudeBinary/GeminiBinary are left empty (host paths would not exist inside).
@@ -92,6 +95,7 @@ func serve(addr string) error {
ClaudeConfigDir: claudeConfigDir,
CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"),
Store: store,
+ Registry: agentRegistry,
},
"gemini": &executor.ContainerRunner{
Image: cfg.GeminiImage,
@@ -103,6 +107,7 @@ func serve(addr string) error {
ClaudeConfigDir: claudeConfigDir,
CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"),
Store: store,
+ Registry: agentRegistry,
},
"container": &executor.ContainerRunner{
Image: "claudomator-agent:latest",
@@ -114,6 +119,7 @@ func serve(addr string) error {
ClaudeConfigDir: claudeConfigDir,
CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"),
Store: store,
+ Registry: agentRegistry,
},
}
@@ -146,7 +152,7 @@ func serve(addr string) error {
pool.RecoverStaleQueued(context.Background())
pool.RecoverStaleBlocked()
- srv := api.NewServer(store, pool, logger, cfg.ClaudeBinaryPath, cfg.GeminiBinaryPath)
+ srv := api.NewServer(store, pool, agentRegistry, logger, cfg.ClaudeBinaryPath, cfg.GeminiBinaryPath)
// Configure notifiers: combine webhook (if set) with web push.
notifiers := []notify.Notifier{}
diff --git a/internal/executor/channel.go b/internal/executor/channel.go
index 541694b..8605ffa 100644
--- a/internal/executor/channel.go
+++ b/internal/executor/channel.go
@@ -50,6 +50,21 @@ type channelStore interface {
CreateEvent(e *event.Event) error
}
+// pendingAsker is implemented by channels that buffer an ask_user call so the
+// runner can convert it into a BlockedError after the agent subprocess exits.
+type pendingAsker interface {
+ PendingQuestion() (questionJSON string, blocked bool)
+}
+
+// channelPendingQuestion reports whether the agent asked a question via the
+// channel during the run (the MCP transport's ask_user path).
+func channelPendingQuestion(ch AgentChannel) (string, bool) {
+ if pa, ok := ch.(pendingAsker); ok {
+ return pa.PendingQuestion()
+ }
+ return "", false
+}
+
// storeChannel is the default AgentChannel backed by storage. Summary and
// question signals are buffered here under a mutex (they may arrive on an MCP
// handler goroutine mid-run); the pool flushes the summary to the execution and
diff --git a/internal/executor/container.go b/internal/executor/container.go
index 4269ef1..f0f728c 100644
--- a/internal/executor/container.go
+++ b/internal/executor/container.go
@@ -2,6 +2,7 @@ package executor
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"log/slog"
@@ -29,6 +30,9 @@ type ContainerRunner struct {
ClaudeConfigDir string // host path to ~/.claude; mounted into container for auth credentials
CredentialSyncCmd string // optional path to sync-credentials script for auth-error auto-recovery
Store Store // optional; used to look up stories and projects for story-aware cloning
+ // Registry mints the per-task MCP token; when set, the agent gets an
+ // mcp-config pointing at the host agent MCP server.
+ Registry *Registry
// Command allows mocking exec.CommandContext for tests.
Command func(ctx context.Context, name string, arg ...string) *exec.Cmd
}
@@ -307,8 +311,24 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto
return fmt.Errorf("writing instructions: %w", err)
}
+ // Per-task MCP back-channel: mint a scoped token bound to this run's channel
+ // and write an mcp-config pointing the agent at the host agent MCP server.
+ mcpEnabled := false
+ if r.Registry != nil && t.Agent.Type != "gemini" {
+ token, mintErr := r.Registry.Mint(ch)
+ if mintErr != nil {
+ return fmt.Errorf("minting agent mcp token: %w", mintErr)
+ }
+ defer r.Registry.Revoke(token)
+ mcpURL := strings.TrimRight(strings.ReplaceAll(r.APIURL, "localhost", "host.docker.internal"), "/") + "/mcp"
+ if err := writeMCPConfig(workspace, mcpURL, token); err != nil {
+ return fmt.Errorf("writing mcp config: %w", err)
+ }
+ mcpEnabled = true
+ }
+
args := r.buildDockerArgs(workspace, agentHome, e.TaskID)
- innerCmd := r.buildInnerCmd(t, e, isResume)
+ innerCmd := r.buildInnerCmd(t, e, isResume, mcpEnabled)
fullArgs := append(args, image)
fullArgs = append(fullArgs, innerCmd...)
@@ -365,7 +385,17 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto
e.SessionID = sessionID
}
- // Check whether the agent left a question before exiting.
+ // MCP transport: if the agent called ask_user, the question is buffered on
+ // the channel. Block so the task resumes with the user's answer.
+ if q, blocked := channelPendingQuestion(ch); blocked {
+ if e.SessionID == "" {
+ r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID)
+ }
+ return &BlockedError{QuestionJSON: q, SessionID: e.SessionID, SandboxDir: workspace}
+ }
+
+ // Check whether the agent left a question before exiting (file fallback for
+ // in-flight tasks started on the pre-MCP wire).
questionFile := filepath.Join(logDir, "question.json")
if data, readErr := os.ReadFile(questionFile); readErr == nil {
os.Remove(questionFile) // consumed
@@ -467,7 +497,30 @@ func (r *ContainerRunner) buildDockerArgs(workspace, claudeHome, taskID string)
return args
}
-func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume bool) []string {
+// mcpConfigContainerPath is where the agent mcp-config is mounted in-container
+// (the workspace is bind-mounted at /workspace).
+const mcpConfigContainerPath = "/workspace/.claudomator-mcp.json"
+
+// writeMCPConfig writes a claude CLI mcp-config that registers the per-task
+// agent MCP server over HTTP with a bearer token.
+func writeMCPConfig(workspace, mcpURL, token string) error {
+ cfg := map[string]any{
+ "mcpServers": map[string]any{
+ "claudomator": map[string]any{
+ "type": "http",
+ "url": mcpURL,
+ "headers": map[string]string{"Authorization": "Bearer " + token},
+ },
+ },
+ }
+ data, err := json.Marshal(cfg)
+ if err != nil {
+ return err
+ }
+ return os.WriteFile(filepath.Join(workspace, ".claudomator-mcp.json"), data, 0600)
+}
+
+func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume, mcpEnabled bool) []string {
// Claude CLI uses -p for prompt text. To pass a file, we use a shell to cat it.
// We use a shell variable to capture the expansion to avoid quoting issues with instructions contents.
// The outer single quotes around the sh -c argument prevent host-side expansion.
@@ -491,6 +544,9 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe
if isResume && e.ResumeSessionID != "" {
claudeCmd.WriteString(fmt.Sprintf(" --resume %s", e.ResumeSessionID))
}
+ if mcpEnabled {
+ claudeCmd.WriteString(" --mcp-config " + mcpConfigContainerPath)
+ }
claudeCmd.WriteString(" --output-format stream-json --verbose --permission-mode bypassPermissions")
return []string{"sh", "-c", claudeCmd.String()}
@@ -501,6 +557,7 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe
var scaffoldPrefixes = []string{
".claudomator-env",
".claudomator-instructions.txt",
+ ".claudomator-mcp.json",
".agent-home",
}
diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go
index 9cd80dc..86e95e2 100644
--- a/internal/executor/container_test.go
+++ b/internal/executor/container_test.go
@@ -2,6 +2,7 @@ package executor
import (
"context"
+ "encoding/json"
"fmt"
"io"
"log/slog"
@@ -53,13 +54,44 @@ func TestContainerRunner_BuildDockerArgs(t *testing.T) {
}
}
+func TestWriteMCPConfig(t *testing.T) {
+ dir := t.TempDir()
+ if err := writeMCPConfig(dir, "http://host.docker.internal:8484/mcp", "tok-abc"); err != nil {
+ t.Fatalf("writeMCPConfig: %v", err)
+ }
+ data, err := os.ReadFile(filepath.Join(dir, ".claudomator-mcp.json"))
+ if err != nil {
+ t.Fatalf("read config: %v", err)
+ }
+ var parsed struct {
+ MCPServers map[string]struct {
+ Type string `json:"type"`
+ URL string `json:"url"`
+ Headers map[string]string `json:"headers"`
+ } `json:"mcpServers"`
+ }
+ if err := json.Unmarshal(data, &parsed); err != nil {
+ t.Fatalf("config is not valid JSON: %v", err)
+ }
+ srv, ok := parsed.MCPServers["claudomator"]
+ if !ok {
+ t.Fatal("expected claudomator server entry")
+ }
+ if srv.Type != "http" || srv.URL != "http://host.docker.internal:8484/mcp" {
+ t.Errorf("unexpected server config: %+v", srv)
+ }
+ if srv.Headers["Authorization"] != "Bearer tok-abc" {
+ t.Errorf("expected bearer header, got %q", srv.Headers["Authorization"])
+ }
+}
+
func TestContainerRunner_BuildInnerCmd(t *testing.T) {
runner := &ContainerRunner{}
t.Run("claude-fresh", func(t *testing.T) {
tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}}
exec := &storage.Execution{}
- cmd := runner.buildInnerCmd(tk, exec, false)
+ cmd := runner.buildInnerCmd(tk, exec, false, false)
cmdStr := strings.Join(cmd, " ")
if strings.Contains(cmdStr, "--resume") {
@@ -73,7 +105,7 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) {
t.Run("claude-resume", func(t *testing.T) {
tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}}
exec := &storage.Execution{ResumeSessionID: "orig-session-123"}
- cmd := runner.buildInnerCmd(tk, exec, true)
+ cmd := runner.buildInnerCmd(tk, exec, true, false)
cmdStr := strings.Join(cmd, " ")
if !strings.Contains(cmdStr, "--resume orig-session-123") {
@@ -81,10 +113,26 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) {
}
})
+ t.Run("claude-mcp-enabled", func(t *testing.T) {
+ tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}}
+ cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, true), " ")
+ if !strings.Contains(cmdStr, "--mcp-config "+mcpConfigContainerPath) {
+ t.Errorf("expected --mcp-config flag when MCP enabled, got %q", cmdStr)
+ }
+ })
+
+ t.Run("claude-mcp-disabled", func(t *testing.T) {
+ tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}}
+ cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, false), " ")
+ if strings.Contains(cmdStr, "--mcp-config") {
+ t.Errorf("did not expect --mcp-config flag when MCP disabled, got %q", cmdStr)
+ }
+ })
+
t.Run("gemini", func(t *testing.T) {
tk := &task.Task{Agent: task.AgentConfig{Type: "gemini"}}
exec := &storage.Execution{}
- cmd := runner.buildInnerCmd(tk, exec, false)
+ cmd := runner.buildInnerCmd(tk, exec, false, false)
cmdStr := strings.Join(cmd, " ")
if !strings.Contains(cmdStr, "gemini -p \"$INST\"") {
@@ -99,13 +147,13 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) {
}
tkClaude := &task.Task{Agent: task.AgentConfig{Type: "claude"}}
- cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false)
+ cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false, false)
if !strings.Contains(strings.Join(cmdClaude, " "), "/usr/bin/claude-v2 -p") {
t.Errorf("expected custom claude binary, got %q", cmdClaude)
}
tkGemini := &task.Task{Agent: task.AgentConfig{Type: "gemini"}}
- cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false)
+ cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false, false)
if !strings.Contains(strings.Join(cmdGemini, " "), "/usr/local/bin/gemini-pro -p") {
t.Errorf("expected custom gemini binary, got %q", cmdGemini)
}