summaryrefslogtreecommitdiff
path: root/internal/api
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 /internal/api
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
Diffstat (limited to 'internal/api')
-rw-r--r--internal/api/agentmcp_endpoint_test.go104
-rw-r--r--internal/api/server.go13
-rw-r--r--internal/api/server_test.go4
3 files changed, 118 insertions, 3 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
}