From f229c723931cde5421ced795689cbb24e4e65fe2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Fri, 30 Jan 2026 15:15:19 +0000 Subject: [PATCH] feat: migrate MCP server from backend to agentctl (#124) --- apps/backend/Makefile | 21 +- apps/backend/cmd/agentctl/main.go | 21 +- apps/backend/cmd/kandev/agents.go | 8 + apps/backend/cmd/kandev/main.go | 40 +- apps/backend/cmd/kandev/mcpserver.go | 31 - apps/backend/cmd/mcp-server/main.go | 120 ---- .../internal/agent/lifecycle/container.go | 3 + .../internal/agent/lifecycle/manager.go | 16 + .../internal/agent/lifecycle/runtime.go | 1 + .../agent/lifecycle/runtime_docker.go | 1 + .../agent/lifecycle/runtime_standalone.go | 2 + .../internal/agent/mcpconfig/resolve.go | 6 + .../agent/settings/controller/controller.go | 32 +- .../internal/agentctl/client/control.go | 2 + .../internal/agentctl/server/api/agent.go | 27 +- .../internal/agentctl/server/api/server.go | 27 +- .../internal/agentctl/server/config/config.go | 15 + .../agentctl/server/instance/instance.go | 7 + .../agentctl/server/instance/manager.go | 2 + .../internal/agentctl/server/mcp/handlers.go | 221 ++++++++ .../internal/agentctl/server/mcp/server.go | 170 ++++++ .../agentctl/server/wsclient/client.go | 177 ++++++ apps/backend/internal/common/config/config.go | 12 +- .../backend/internal/mcp/handlers/handlers.go | 296 ++++++++++ apps/backend/internal/mcpserver/provider.go | 50 -- apps/backend/internal/mcpserver/server.go | 179 ------ apps/backend/internal/mcpserver/tools.go | 533 ------------------ apps/backend/pkg/websocket/actions.go | 9 + .../[agentId]/profile-mcp-config-card.tsx | 28 +- .../task/chat/chat-input-container.tsx | 2 +- .../task/chat/clarification-input-overlay.tsx | 51 +- docs/architecture/mcp-agentctl-migration.md | 326 +++++++++++ 32 files changed, 1401 insertions(+), 1035 deletions(-) delete mode 100644 apps/backend/cmd/kandev/mcpserver.go delete mode 100644 apps/backend/cmd/mcp-server/main.go create mode 100644 apps/backend/internal/agentctl/server/mcp/handlers.go create mode 100644 apps/backend/internal/agentctl/server/mcp/server.go create mode 100644 apps/backend/internal/agentctl/server/wsclient/client.go create mode 100644 apps/backend/internal/mcp/handlers/handlers.go delete mode 100644 apps/backend/internal/mcpserver/provider.go delete mode 100644 apps/backend/internal/mcpserver/server.go delete mode 100644 apps/backend/internal/mcpserver/tools.go create mode 100644 docs/architecture/mcp-agentctl-migration.md diff --git a/apps/backend/Makefile b/apps/backend/Makefile index ea42b49f88..0f4fdf81f4 100644 --- a/apps/backend/Makefile +++ b/apps/backend/Makefile @@ -14,34 +14,28 @@ BUILD_TIME := $(shell date -u '+%Y-%m-%dT%H:%M:%SZ') LDFLAGS += -X main.Version=$(VERSION) -X main.Commit=$(COMMIT) -X main.BuildTime=$(BUILD_TIME) -.PHONY: all build build-all build-agentctl build-mcp-server clean run dev test lint fmt vet help +.PHONY: all build build-all build-agentctl clean run dev test lint fmt vet help ## Default target all: build ## Build the unified kandev binary -build: build-agentctl build-mcp-server +build: build-agentctl @echo "Building $(BINARY_NAME)..." @mkdir -p $(BUILD_DIR) $(GO) build $(GOFLAGS) -ldflags "$(LDFLAGS)" -o $(BUILD_DIR)/$(BINARY_NAME) ./cmd/kandev -## Build all binaries (unified + individual services + agentctl + mcp-server) -build-all: build build-agentctl build-mcp-server +## Build all binaries (unified + individual services + agentctl) +build-all: build build-agentctl @echo "Building individual services..." $(GO) build $(GOFLAGS) -ldflags "$(LDFLAGS)" -o $(BUILD_DIR)/agent-manager ./cmd/agent-manager -## Build the agentctl sidecar binary +## Build the agentctl sidecar binary (includes embedded MCP server) build-agentctl: @echo "Building agentctl..." @mkdir -p $(BUILD_DIR) $(GO) build $(GOFLAGS) -ldflags "$(LDFLAGS)" -o $(BUILD_DIR)/agentctl ./cmd/agentctl -## Build the standalone MCP server binary -build-mcp-server: - @echo "Building mcp-server..." - @mkdir -p $(BUILD_DIR) - $(GO) build $(GOFLAGS) -ldflags "$(LDFLAGS)" -o $(BUILD_DIR)/mcp-server ./cmd/mcp-server - ## Clean build artifacts clean: @echo "Cleaning..." @@ -112,9 +106,8 @@ help: @echo "" @echo "Targets:" @echo " build Build the unified kandev binary (default)" - @echo " build-all Build all binaries (unified + individual services + agentctl + mcp-server)" - @echo " build-agentctl Build the agentctl sidecar binary" - @echo " build-mcp-server Build the standalone MCP server binary" + @echo " build-all Build all binaries (unified + individual services + agentctl)" + @echo " build-agentctl Build the agentctl sidecar binary (includes embedded MCP server)" @echo " clean Remove build artifacts" @echo " run Build and run the application" @echo " dev Run in development mode with a local database" diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go index 49be04afbd..c45cfce003 100644 --- a/apps/backend/cmd/agentctl/main.go +++ b/apps/backend/cmd/agentctl/main.go @@ -20,7 +20,9 @@ import ( "github.com/kandev/kandev/internal/agentctl/server/api" "github.com/kandev/kandev/internal/agentctl/server/config" "github.com/kandev/kandev/internal/agentctl/server/instance" + mcpserver "github.com/kandev/kandev/internal/agentctl/server/mcp" "github.com/kandev/kandev/internal/agentctl/server/process" + "github.com/kandev/kandev/internal/agentctl/server/wsclient" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" @@ -82,7 +84,24 @@ func run(cfg *config.Config, log *logger.Logger) { // Set the server factory to create API servers for each instance instMgr.SetServerFactory(func(instCfg *config.InstanceConfig, procMgr *process.Manager, instLog *logger.Logger) http.Handler { - return api.NewServer(instCfg, procMgr, instLog).Router() + var mcpSrv *mcpserver.Server + + // Create MCP server if backend WS URL is configured + if instCfg.BackendWsURL != "" { + wsClient := wsclient.New(instCfg.BackendWsURL, instLog) + if err := wsClient.Connect(context.Background()); err != nil { + instLog.Error("failed to connect MCP WS client to backend", + zap.Error(err), + zap.String("backend_ws_url", instCfg.BackendWsURL)) + } else { + mcpSrv = mcpserver.New(wsClient, instCfg.SessionID, instLog) + instLog.Info("MCP server enabled", + zap.String("backend_ws_url", instCfg.BackendWsURL), + zap.String("session_id", instCfg.SessionID)) + } + } + + return api.NewServer(instCfg, procMgr, mcpSrv, instLog).Router() }) // Create control server diff --git a/apps/backend/cmd/kandev/agents.go b/apps/backend/cmd/kandev/agents.go index 03577afdf4..8220277a32 100644 --- a/apps/backend/cmd/kandev/agents.go +++ b/apps/backend/cmd/kandev/agents.go @@ -2,6 +2,7 @@ package main import ( "context" + "fmt" "os" "go.uber.org/zap" @@ -91,6 +92,13 @@ func provideLifecycleManager( lifecycle.RuntimeFallbackWarn, log, ) + + // Set backend WS URL for MCP tunneling + // agentctl instances will connect back to this URL to forward MCP tool calls + backendWsURL := fmt.Sprintf("ws://%s:%d/ws", cfg.Server.Host, cfg.Server.Port) + lifecycleMgr.SetBackendWsURL(backendWsURL) + log.Info("Backend WS URL configured for MCP tunneling", zap.String("url", backendWsURL)) + if err := lifecycleMgr.Start(ctx); err != nil { return nil, nil, err } diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 3774ac17da..61a00e6206 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -42,6 +42,9 @@ import ( "github.com/kandev/kandev/internal/clarification" + // MCP handlers + mcphandlers "github.com/kandev/kandev/internal/mcp/handlers" + // Task Service packages taskcontroller "github.com/kandev/kandev/internal/task/controller" taskhandlers "github.com/kandev/kandev/internal/task/handlers" @@ -218,30 +221,11 @@ func main() { }() } - // ============================================ - // MCP SERVER (embedded) - // ============================================ - mcpServerURL, mcpCleanup, err := provideMcpServer(ctx, cfg, log) - if err != nil { - log.Fatal("Failed to start MCP server", zap.Error(err)) - } - if mcpCleanup != nil { - cleanups = append(cleanups, mcpCleanup) - log.Info("MCP server started", zap.String("url", mcpServerURL)) - // Auto-set the MCP server URL if not explicitly configured - if cfg.Agent.McpServerURL == "" { - cfg.Agent.McpServerURL = mcpServerURL - } - // Set MCP server URL on controller and ensure default MCP config for all profiles - agentSettingsController.SetMcpServerURL(mcpServerURL) - if err := agentSettingsController.EnsureDefaultMcpConfig(ctx); err != nil { - log.Warn("Failed to ensure default MCP config", zap.Error(err)) - } - } - // ============================================ // AGENT MANAGER // ============================================ + // Note: MCP server is now embedded in agentctl (co-located with agents) + // and tunnels tool calls back to the backend via WebSocket. lifecycleMgr, agentRegistry, err := provideLifecycleManager(ctx, cfg, log, eventBus, dockerClient, agentSettingsRepo) if err != nil { log.Fatal("Failed to initialize agent manager", zap.Error(err)) @@ -563,6 +547,20 @@ func main() { clarification.RegisterRoutes(router, clarificationStore, gateway.Hub, msgCreator, taskRepo, log) log.Debug("Registered Clarification handlers (HTTP)") + // MCP handlers for agentctl WS tunnel (agentctl -> backend) + mcpHandlers := mcphandlers.NewHandlers( + workspaceController, + boardController, + taskController, + workflowCtrl, + clarificationStore, + msgCreator, + taskRepo, + log, + ) + mcpHandlers.RegisterHandlers(gateway.Dispatcher) + log.Debug("Registered MCP handlers (WebSocket)") + // Health check (simple HTTP for load balancers/monitoring) router.GET("/health", func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ diff --git a/apps/backend/cmd/kandev/mcpserver.go b/apps/backend/cmd/kandev/mcpserver.go deleted file mode 100644 index 904d1337fc..0000000000 --- a/apps/backend/cmd/kandev/mcpserver.go +++ /dev/null @@ -1,31 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/kandev/kandev/internal/common/config" - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/mcpserver" -) - -// provideMcpServer starts the embedded MCP server if enabled. -// Returns the SSE endpoint URL and a cleanup function. -func provideMcpServer(ctx context.Context, cfg *config.Config, log *logger.Logger) (string, func() error, error) { - if !cfg.Agent.McpServerEnabled { - return "", nil, nil - } - - mcpCfg := mcpserver.Config{ - Port: cfg.Agent.McpServerPort, - KandevURL: fmt.Sprintf("http://localhost:%d", cfg.Server.Port), - } - - srv, cleanup, err := mcpserver.Provide(ctx, mcpCfg, log) - if err != nil { - return "", nil, fmt.Errorf("failed to start MCP server: %w", err) - } - - return srv.SSEEndpoint(), cleanup, nil -} - diff --git a/apps/backend/cmd/mcp-server/main.go b/apps/backend/cmd/mcp-server/main.go deleted file mode 100644 index 4fbdb40dd3..0000000000 --- a/apps/backend/cmd/mcp-server/main.go +++ /dev/null @@ -1,120 +0,0 @@ -// Package main is the entry point for the standalone MCP server binary. -// mcp-server provides a Model Context Protocol server that exposes Kandev -// task management tools to MCP-compatible clients (Claude Desktop, Cursor, Codex, etc.) -// -// The server supports two transports: -// - SSE (Server-Sent Events) at /sse for Claude Desktop, Cursor -// - Streamable HTTP at /mcp for Codex -package main - -import ( - "context" - "flag" - "fmt" - "os" - "os/signal" - "syscall" - "time" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/mcpserver" - "go.uber.org/zap" -) - -// Command-line flags -var ( - portFlag = flag.Int("port", 9090, "MCP server port") - kandevURLFlag = flag.String("kandev-url", "http://localhost:8080", "Kandev API URL") - logLevelFlag = flag.String("log-level", "info", "Log level (debug, info, warn, error)") - logFormatFlag = flag.String("log-format", "console", "Log format (console, json)") -) - -func main() { - flag.Parse() - - // Build configuration from flags and environment - cfg := mcpserver.Config{ - Port: getEnvIntOrFlag("MCP_PORT", *portFlag), - KandevURL: getEnvOrFlag("KANDEV_API_URL", *kandevURLFlag), - } - - // Initialize logger - log, err := logger.NewLogger(logger.LoggingConfig{ - Level: getEnvOrFlag("MCP_LOG_LEVEL", *logLevelFlag), - Format: getEnvOrFlag("MCP_LOG_FORMAT", *logFormatFlag), - OutputPath: "stdout", - }) - if err != nil { - fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err) - os.Exit(1) - } - defer func() { _ = log.Sync() }() - - log.Info("starting mcp-server", - zap.Int("port", cfg.Port), - zap.String("kandev_url", cfg.KandevURL)) - - run(cfg, log) -} - -// run starts the MCP server and waits for shutdown. -func run(cfg mcpserver.Config, log *logger.Logger) { - ctx := context.Background() - srv, cleanup, err := mcpserver.Provide(ctx, cfg, log) - if err != nil { - log.Error("failed to start MCP server", zap.Error(err)) - os.Exit(1) - } - - log.Info("MCP server started", - zap.String("sse_endpoint", srv.SSEEndpoint()), - zap.String("streamable_http_endpoint", srv.StreamableHTTPEndpoint())) - - fmt.Printf("Kandev MCP server running on :%d\n", cfg.Port) - fmt.Printf("Kandev API URL: %s\n", cfg.KandevURL) - fmt.Printf("SSE endpoint: %s (for Claude Desktop, Cursor)\n", srv.SSEEndpoint()) - fmt.Printf("Streamable HTTP endpoint: %s (for Codex)\n", srv.StreamableHTTPEndpoint()) - - // Wait for shutdown signal - waitForShutdown(log, func(ctx context.Context) { - if err := cleanup(); err != nil { - log.Error("error during shutdown", zap.Error(err)) - } - }) -} - -// waitForShutdown waits for shutdown signal and calls cleanup -func waitForShutdown(log *logger.Logger, cleanup func(ctx context.Context)) { - quit := make(chan os.Signal, 1) - signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) - <-quit - - log.Info("shutting down mcp-server...") - - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - cleanup(ctx) - - log.Info("mcp-server stopped") -} - -// getEnvOrFlag returns the environment variable value if set, otherwise the flag value. -func getEnvOrFlag(envKey, flagValue string) string { - if v := os.Getenv(envKey); v != "" { - return v - } - return flagValue -} - -// getEnvIntOrFlag returns the environment variable value as int if set, otherwise the flag value. -func getEnvIntOrFlag(envKey string, flagValue int) int { - if v := os.Getenv(envKey); v != "" { - var i int - if _, err := fmt.Sscanf(v, "%d", &i); err == nil { - return i - } - } - return flagValue -} - diff --git a/apps/backend/internal/agent/lifecycle/container.go b/apps/backend/internal/agent/lifecycle/container.go index ca5614228a..db1290772b 100644 --- a/apps/backend/internal/agent/lifecycle/container.go +++ b/apps/backend/internal/agent/lifecycle/container.go @@ -30,6 +30,7 @@ type ContainerConfig struct { InstanceID string MainRepoGitDir string // Path to main repo's .git directory (for worktrees) McpServers []McpServerConfig + BackendWsURL string // WebSocket URL to Kandev backend for MCP tunneling } // ContainerManager handles Docker container lifecycle operations @@ -109,6 +110,8 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe Env: config.Credentials, AutoStart: false, McpServers: mcpServers, + BackendWsURL: config.BackendWsURL, + SessionID: config.SessionID, } resp, err := ctl.CreateInstance(ctx, createReq) diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 99d8624967..00eba1b992 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -72,6 +72,10 @@ type Manager struct { // Workspace info provider for on-demand instance creation workspaceInfoProvider WorkspaceInfoProvider + // backendWsURL is the WebSocket URL to the Kandev backend for MCP tunneling + // This is passed to agentctl instances so they can forward MCP tool calls + backendWsURL string + // Background cleanup cleanupInterval time.Duration stopCh chan struct{} @@ -170,6 +174,15 @@ func (m *Manager) SetWorktreeManager(worktreeMgr *worktree.Manager) { m.worktreeMgr = worktreeMgr } +// SetBackendWsURL sets the WebSocket URL to the Kandev backend for MCP tunneling. +// +// This URL is passed to agentctl instances so they can establish a WebSocket connection +// back to the backend and forward MCP tool calls through the tunnel. +// Format: ws://host:port/ws (e.g., ws://localhost:8080/ws) +func (m *Manager) SetBackendWsURL(url string) { + m.backendWsURL = url +} + // SetWorkspaceInfoProvider sets the provider for workspace information. // // The WorkspaceInfoProvider interface allows the lifecycle manager to dynamically create @@ -441,6 +454,8 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work AgentProfileID: info.AgentProfileID, WorkspacePath: info.WorkspacePath, Protocol: string(agentConfig.Protocol), + AgentConfig: agentConfig, + BackendWsURL: m.backendWsURL, } runtimeInstance, err := rt.CreateInstance(ctx, req) @@ -768,6 +783,7 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti Metadata: metadata, AgentConfig: agentConfig, McpServers: mcpServers, + BackendWsURL: m.backendWsURL, } runtimeInstance, err := rt.CreateInstance(ctx, runtimeReq) diff --git a/apps/backend/internal/agent/lifecycle/runtime.go b/apps/backend/internal/agent/lifecycle/runtime.go index be93c84923..6a7ebc3242 100644 --- a/apps/backend/internal/agent/lifecycle/runtime.go +++ b/apps/backend/internal/agent/lifecycle/runtime.go @@ -67,6 +67,7 @@ type RuntimeCreateRequest struct { Metadata map[string]interface{} McpServers []McpServerConfig AgentConfig *registry.AgentTypeConfig // Agent type info needed by runtimes + BackendWsURL string // WebSocket URL to Kandev backend for MCP tunneling } // RuntimeInstance represents an agentctl instance created by a runtime. diff --git a/apps/backend/internal/agent/lifecycle/runtime_docker.go b/apps/backend/internal/agent/lifecycle/runtime_docker.go index 2a6f5cc2b8..0aee70e314 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_docker.go +++ b/apps/backend/internal/agent/lifecycle/runtime_docker.go @@ -70,6 +70,7 @@ func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRe MainRepoGitDir: mainRepoGitDir, Credentials: req.Env, // Env contains credentials from the caller McpServers: req.McpServers, + BackendWsURL: req.BackendWsURL, } // Use ContainerManager to launch container diff --git a/apps/backend/internal/agent/lifecycle/runtime_standalone.go b/apps/backend/internal/agent/lifecycle/runtime_standalone.go index 21debd002f..180c20b6d2 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_standalone.go +++ b/apps/backend/internal/agent/lifecycle/runtime_standalone.go @@ -103,6 +103,8 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea Env: env, AutoStart: false, McpServers: mcpServers, + BackendWsURL: req.BackendWsURL, + SessionID: req.SessionID, } resp, err := r.ctl.CreateInstance(ctx, createReq) diff --git a/apps/backend/internal/agent/mcpconfig/resolve.go b/apps/backend/internal/agent/mcpconfig/resolve.go index d148d1df7e..1febeb5c27 100644 --- a/apps/backend/internal/agent/mcpconfig/resolve.go +++ b/apps/backend/internal/agent/mcpconfig/resolve.go @@ -16,6 +16,12 @@ func Resolve(config *ProfileConfig, policy Policy) ([]ResolvedServer, []string, resolved := make([]ResolvedServer, 0, len(config.Servers)) for name, server := range config.Servers { + // Skip kandev server - it's automatically injected by agentctl with the correct local URL. + // This handles existing profiles that still have the old kandev config in the database. + if name == "kandev" { + continue + } + if !policyAllowsServerName(policy, name) { warnings = append(warnings, fmt.Sprintf("mcp server %q skipped: server not allowed by policy", name)) continue diff --git a/apps/backend/internal/agent/settings/controller/controller.go b/apps/backend/internal/agent/settings/controller/controller.go index 2c222e870d..924c1ea151 100644 --- a/apps/backend/internal/agent/settings/controller/controller.go +++ b/apps/backend/internal/agent/settings/controller/controller.go @@ -37,7 +37,6 @@ type Controller struct { mcpService *mcpconfig.Service modelFetcher *modelfetcher.Fetcher logger *logger.Logger - mcpServerURL string // URL of the embedded Kandev MCP server } type SessionChecker interface { @@ -57,20 +56,11 @@ func NewController(repo store.Repository, discoveryRegistry *discovery.Registry, } } -// SetMcpServerURL sets the URL of the embedded Kandev MCP server. -// This should be called after the MCP server is started. -func (c *Controller) SetMcpServerURL(url string) { - c.mcpServerURL = url -} - -// EnsureDefaultMcpConfig ensures all agent profiles that support MCP have the -// Kandev MCP server configured and enabled by default. +// EnsureDefaultMcpConfig ensures all agent profiles that support MCP have +// MCP enabled by default. The kandev MCP server is automatically injected +// by agentctl at session creation time, so profiles don't need to explicitly +// configure it. func (c *Controller) EnsureDefaultMcpConfig(ctx context.Context) error { - if c.mcpServerURL == "" { - c.logger.Debug("MCP server URL not set, skipping default MCP config") - return nil - } - agents, err := c.repo.ListAgents(ctx) if err != nil { return err @@ -98,16 +88,13 @@ func (c *Controller) EnsureDefaultMcpConfig(ctx context.Context) error { continue } - // Create default MCP config with Kandev server enabled + // Create default MCP config with MCP enabled but no servers configured. + // The kandev MCP server is automatically injected by agentctl when + // creating a new session. Users can add additional external MCP servers. config := &models.AgentProfileMcpConfig{ ProfileID: profile.ID, Enabled: true, - Servers: map[string]interface{}{ - "kandev": map[string]interface{}{ - "type": "sse", - "url": c.mcpServerURL, - }, - }, + Servers: map[string]interface{}{}, Meta: map[string]interface{}{}, CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(), @@ -123,8 +110,7 @@ func (c *Controller) EnsureDefaultMcpConfig(ctx context.Context) error { c.logger.Info("created default MCP config for profile", zap.String("profile_id", profile.ID), - zap.String("profile_name", profile.Name), - zap.String("mcp_server_url", c.mcpServerURL)) + zap.String("profile_name", profile.Name)) } } diff --git a/apps/backend/internal/agentctl/client/control.go b/apps/backend/internal/agentctl/client/control.go index d7b46ff7e6..0db132fa49 100644 --- a/apps/backend/internal/agentctl/client/control.go +++ b/apps/backend/internal/agentctl/client/control.go @@ -42,6 +42,8 @@ type CreateInstanceRequest struct { Env map[string]string `json:"env,omitempty"` AutoStart bool `json:"auto_start,omitempty"` McpServers []McpServerConfig `json:"mcp_servers,omitempty"` + BackendWsURL string `json:"backend_ws_url,omitempty"` // WebSocket URL to Kandev backend for MCP tunneling + SessionID string `json:"session_id,omitempty"` // Task session ID for MCP tool calls } // CreateInstanceResponse contains the result of creating a new agent instance. diff --git a/apps/backend/internal/agentctl/server/api/agent.go b/apps/backend/internal/agentctl/server/api/agent.go index 7c400b7bc0..d7ebd24c3b 100644 --- a/apps/backend/internal/agentctl/server/api/agent.go +++ b/apps/backend/internal/agentctl/server/api/agent.go @@ -3,6 +3,7 @@ package api import ( "context" "encoding/json" + "fmt" "net/http" "time" @@ -114,7 +115,31 @@ func (s *Server) handleAgentNewSession(c *gin.Context) { return } - sessionID, err := adapter.NewSession(ctx, req.McpServers) + // If MCP server is enabled, prepend the local kandev MCP server to the list. + // This replaces the old external MCP server URL (http://localhost:9090/sse) with + // the local agentctl MCP server (http://localhost:{port}/sse). + mcpServers := req.McpServers + if s.mcpServer != nil { + localKandevMcp := types.McpServer{ + Name: "kandev", + Type: "sse", + URL: fmt.Sprintf("http://localhost:%d/sse", s.cfg.Port), + } + // Filter out any existing kandev server from the list (to avoid duplicates) + filtered := make([]types.McpServer, 0, len(mcpServers)+1) + filtered = append(filtered, localKandevMcp) + for _, srv := range mcpServers { + if srv.Name != "kandev" { + filtered = append(filtered, srv) + } + } + mcpServers = filtered + s.logger.Debug("injected local kandev MCP server", + zap.String("url", localKandevMcp.URL), + zap.Int("total_servers", len(mcpServers))) + } + + sessionID, err := adapter.NewSession(ctx, mcpServers) if err != nil { s.logger.Error("new session failed", zap.Error(err)) c.JSON(http.StatusInternalServerError, NewSessionResponse{ diff --git a/apps/backend/internal/agentctl/server/api/server.go b/apps/backend/internal/agentctl/server/api/server.go index 4e31cb5a78..16fc0e254f 100644 --- a/apps/backend/internal/agentctl/server/api/server.go +++ b/apps/backend/internal/agentctl/server/api/server.go @@ -8,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "github.com/kandev/kandev/internal/agentctl/server/config" + "github.com/kandev/kandev/internal/agentctl/server/mcp" "github.com/kandev/kandev/internal/agentctl/server/process" "github.com/kandev/kandev/internal/common/httpmw" "github.com/kandev/kandev/internal/common/logger" @@ -16,23 +17,26 @@ import ( // Server is the HTTP API server for a single agent instance. type Server struct { - cfg *config.InstanceConfig - procMgr *process.Manager - logger *logger.Logger - router *gin.Engine + cfg *config.InstanceConfig + procMgr *process.Manager + mcpServer *mcp.Server + logger *logger.Logger + router *gin.Engine upgrader websocket.Upgrader } // NewServer creates a new API server for an agent instance. -func NewServer(cfg *config.InstanceConfig, procMgr *process.Manager, log *logger.Logger) *Server { +// If mcpServer is provided, MCP routes will be registered. +func NewServer(cfg *config.InstanceConfig, procMgr *process.Manager, mcpServer *mcp.Server, log *logger.Logger) *Server { gin.SetMode(gin.ReleaseMode) s := &Server{ - cfg: cfg, - procMgr: procMgr, - logger: log.WithFields(zap.String("component", "api-server")), - router: gin.New(), + cfg: cfg, + procMgr: procMgr, + mcpServer: mcpServer, + logger: log.WithFields(zap.String("component", "api-server")), + router: gin.New(), upgrader: websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true // Allow all origins for container-local communication @@ -107,6 +111,11 @@ func (s *Server) setupRoutes() { api.POST("/git/create-pr", s.handleGitCreatePR) api.GET("/git/commit/:sha", s.handleGitShowCommit) } + + // MCP routes (if MCP server is configured) + if s.mcpServer != nil { + s.mcpServer.RegisterRoutes(s.router) + } } // Health check response diff --git a/apps/backend/internal/agentctl/server/config/config.go b/apps/backend/internal/agentctl/server/config/config.go index f1bf1c99ac..0cbffd4199 100644 --- a/apps/backend/internal/agentctl/server/config/config.go +++ b/apps/backend/internal/agentctl/server/config/config.go @@ -136,6 +136,13 @@ type InstanceConfig struct { // ProcessBufferMaxBytes caps per-process output buffer size ProcessBufferMaxBytes int64 + + // BackendWsURL is the WebSocket URL to connect back to the Kandev backend + // for MCP tool requests (e.g., ws://localhost:8080/ws) + BackendWsURL string + + // SessionID is the session ID for this agent instance (used in MCP tool calls) + SessionID string } // Load loads the configuration from environment variables. @@ -203,6 +210,12 @@ func (c *Config) NewInstanceConfig(port int, overrides *InstanceOverrides) *Inst if len(overrides.McpServers) > 0 { cfg.McpServers = overrides.McpServers } + if overrides.BackendWsURL != "" { + cfg.BackendWsURL = overrides.BackendWsURL + } + if overrides.SessionID != "" { + cfg.SessionID = overrides.SessionID + } } // Parse agent command into args @@ -226,6 +239,8 @@ type InstanceOverrides struct { ApprovalPolicy string AgentType string McpServers []McpServerConfig + BackendWsURL string + SessionID string } // ParseCommand splits a command string into arguments diff --git a/apps/backend/internal/agentctl/server/instance/instance.go b/apps/backend/internal/agentctl/server/instance/instance.go index 9854cb194c..5d55d42bcc 100644 --- a/apps/backend/internal/agentctl/server/instance/instance.go +++ b/apps/backend/internal/agentctl/server/instance/instance.go @@ -76,6 +76,13 @@ type CreateRequest struct { // McpServers is a list of MCP servers to configure for the agent. McpServers []McpServerConfig `json:"mcp_servers,omitempty"` + + // BackendWsURL is the WebSocket URL to the Kandev backend for MCP tunneling. + // If set, the MCP server will be enabled and forward tool calls to the backend. + BackendWsURL string `json:"backend_ws_url,omitempty"` + + // SessionID is the task session ID for MCP tool calls (used by ask_user_question). + SessionID string `json:"session_id,omitempty"` } // CreateResponse contains the result of creating a new agent instance. diff --git a/apps/backend/internal/agentctl/server/instance/manager.go b/apps/backend/internal/agentctl/server/instance/manager.go index 38adf399cc..809a5bf547 100644 --- a/apps/backend/internal/agentctl/server/instance/manager.go +++ b/apps/backend/internal/agentctl/server/instance/manager.go @@ -132,6 +132,8 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea AutoStart: &autoStart, Env: config.CollectAgentEnv(req.Env), McpServers: mcpServers, + BackendWsURL: req.BackendWsURL, + SessionID: req.SessionID, } // Create instance config using the unified method diff --git a/apps/backend/internal/agentctl/server/mcp/handlers.go b/apps/backend/internal/agentctl/server/mcp/handlers.go new file mode 100644 index 0000000000..704b055e03 --- /dev/null +++ b/apps/backend/internal/agentctl/server/mcp/handlers.go @@ -0,0 +1,221 @@ +package mcp + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/mark3labs/mcp-go/mcp" + "github.com/mark3labs/mcp-go/server" +) + +// MCP action constants for backend WS requests +const ( + ActionMCPListWorkspaces = "mcp.list_workspaces" + ActionMCPListBoards = "mcp.list_boards" + ActionMCPListWorkflowSteps = "mcp.list_workflow_steps" + ActionMCPListTasks = "mcp.list_tasks" + ActionMCPCreateTask = "mcp.create_task" + ActionMCPUpdateTask = "mcp.update_task" + ActionMCPAskUserQuestion = "mcp.ask_user_question" +) + +func (s *Server) listWorkspacesHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var result []map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPListWorkspaces, nil, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) listBoardsHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + workspaceID, err := req.RequireString("workspace_id") + if err != nil { + return mcp.NewToolResultError("workspace_id is required"), nil + } + payload := map[string]string{"workspace_id": workspaceID} + var result []map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPListBoards, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) listWorkflowStepsHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + boardID, err := req.RequireString("board_id") + if err != nil { + return mcp.NewToolResultError("board_id is required"), nil + } + payload := map[string]string{"board_id": boardID} + var result []map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPListWorkflowSteps, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) listTasksHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + boardID, err := req.RequireString("board_id") + if err != nil { + return mcp.NewToolResultError("board_id is required"), nil + } + payload := map[string]string{"board_id": boardID} + var result []map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPListTasks, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) createTaskHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + workspaceID, err := req.RequireString("workspace_id") + if err != nil { + return mcp.NewToolResultError("workspace_id is required"), nil + } + boardID, err := req.RequireString("board_id") + if err != nil { + return mcp.NewToolResultError("board_id is required"), nil + } + workflowStepID, err := req.RequireString("workflow_step_id") + if err != nil { + return mcp.NewToolResultError("workflow_step_id is required"), nil + } + title, err := req.RequireString("title") + if err != nil { + return mcp.NewToolResultError("title is required"), nil + } + description := req.GetString("description", "") + + payload := map[string]string{ + "workspace_id": workspaceID, + "board_id": boardID, + "workflow_step_id": workflowStepID, + "title": title, + "description": description, + } + var result map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPCreateTask, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) updateTaskHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + taskID, err := req.RequireString("task_id") + if err != nil { + return mcp.NewToolResultError("task_id is required"), nil + } + payload := map[string]interface{}{"task_id": taskID} + if title := req.GetString("title", ""); title != "" { + payload["title"] = title + } + if desc := req.GetString("description", ""); desc != "" { + payload["description"] = desc + } + if state := req.GetString("state", ""); state != "" { + payload["state"] = state + } + var result map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPUpdateTask, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + +func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + prompt, err := req.RequireString("prompt") + if err != nil { + return mcp.NewToolResultError("prompt is required"), nil + } + args := req.GetArguments() + optionsRaw, ok := args["options"] + if !ok { + return mcp.NewToolResultError("options is required"), nil + } + + // Parse options - can be array of strings or array of objects + optionsJSON, err := json.Marshal(optionsRaw) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("failed to parse options: %v", err)), nil + } + + // Try to parse as array of strings first (simple format from agents) + var stringOptions []string + if err := json.Unmarshal(optionsJSON, &stringOptions); err == nil { + // Convert string options to the expected format + formattedOptions := make([]map[string]interface{}, len(stringOptions)) + for i, opt := range stringOptions { + formattedOptions[i] = map[string]interface{}{ + "option_id": fmt.Sprintf("opt_%d", i+1), + "label": opt, + "description": opt, + } + } + optionsJSON, _ = json.Marshal(formattedOptions) + } + + var options []map[string]interface{} + if err := json.Unmarshal(optionsJSON, &options); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("failed to parse options: %v", err)), nil + } + + questionCtx := req.GetString("context", "") + + // Build the question object in the format expected by the backend + question := map[string]interface{}{ + "id": "q1", + "title": "Question", + "prompt": prompt, + "options": options, + } + + payload := map[string]interface{}{ + "session_id": s.sessionID, + "question": question, + "context": questionCtx, + } + var result map[string]interface{} + if err := s.wsClient.RequestPayload(ctx, ActionMCPAskUserQuestion, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + + // Extract the answer from the response + if answer, ok := result["answer"]; ok { + if answerMap, ok := answer.(map[string]interface{}); ok { + if selectedOptions, ok := answerMap["selected_options"].([]interface{}); ok && len(selectedOptions) > 0 { + return mcp.NewToolResultText(fmt.Sprintf("User selected: %v", selectedOptions[0])), nil + } + if customText, ok := answerMap["custom_text"].(string); ok && customText != "" { + return mcp.NewToolResultText(fmt.Sprintf("User answered: %s", customText)), nil + } + } + } + if rejected, ok := result["rejected"].(bool); ok && rejected { + reason, _ := result["reject_reason"].(string) + return mcp.NewToolResultText(fmt.Sprintf("User rejected the question: %s", reason)), nil + } + + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + diff --git a/apps/backend/internal/agentctl/server/mcp/server.go b/apps/backend/internal/agentctl/server/mcp/server.go new file mode 100644 index 0000000000..d37339a1cb --- /dev/null +++ b/apps/backend/internal/agentctl/server/mcp/server.go @@ -0,0 +1,170 @@ +// Package mcp provides MCP server functionality for agentctl. +// It exposes MCP tools that forward requests to the Kandev backend via WebSocket. +package mcp + +import ( + "context" + "net/http" + "sync" + + "github.com/gin-gonic/gin" + "github.com/kandev/kandev/internal/agentctl/server/wsclient" + "github.com/kandev/kandev/internal/common/logger" + "github.com/mark3labs/mcp-go/mcp" + "github.com/mark3labs/mcp-go/server" + "go.uber.org/zap" +) + +// Server wraps the MCP server with WebSocket client for backend communication. +type Server struct { + wsClient *wsclient.Client + sessionID string + mcpServer *server.MCPServer + sseServer *server.SSEServer + httpServer *server.StreamableHTTPServer + logger *logger.Logger + mu sync.Mutex + running bool +} + +// New creates a new MCP server for agentctl. +func New(wsClient *wsclient.Client, sessionID string, log *logger.Logger) *Server { + s := &Server{ + wsClient: wsClient, + sessionID: sessionID, + logger: log.WithFields(zap.String("component", "mcp-server")), + } + + // Create MCP server + s.mcpServer = server.NewMCPServer( + "kandev-mcp", + "1.0.0", + server.WithToolCapabilities(true), + ) + + // Register tools + s.registerTools() + + // Create SSE server for Claude Desktop, Cursor, etc. + s.sseServer = server.NewSSEServer(s.mcpServer) + + // Create Streamable HTTP server for Codex + s.httpServer = server.NewStreamableHTTPServer(s.mcpServer, + server.WithEndpointPath("/mcp"), + ) + + return s +} + +// RegisterRoutes adds MCP routes to the gin router. +func (s *Server) RegisterRoutes(router gin.IRouter) { + // SSE transport routes + router.GET("/sse", gin.WrapH(s.sseServer.SSEHandler())) + router.POST("/message", gin.WrapH(s.sseServer.MessageHandler())) + + // Streamable HTTP transport route + router.Any("/mcp", gin.WrapH(s.httpServer)) + + s.logger.Info("registered MCP routes", zap.String("sse", "/sse"), zap.String("http", "/mcp")) +} + +// Close shuts down the MCP server. +func (s *Server) Close(ctx context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.running { + return nil + } + s.running = false + + if s.sseServer != nil { + if err := s.sseServer.Shutdown(ctx); err != nil { + s.logger.Warn("failed to shutdown SSE server", zap.Error(err)) + } + } + if s.httpServer != nil { + if err := s.httpServer.Shutdown(ctx); err != nil { + s.logger.Warn("failed to shutdown HTTP server", zap.Error(err)) + } + } + + return nil +} + +// registerTools registers all MCP tools. +func (s *Server) registerTools() { + s.mcpServer.AddTool( + mcp.NewTool("list_workspaces", + mcp.WithDescription("List all workspaces. Use this first to get workspace IDs."), + ), + s.listWorkspacesHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("list_boards", + mcp.WithDescription("List all boards in a workspace."), + mcp.WithString("workspace_id", mcp.Required(), mcp.Description("The workspace ID")), + ), + s.listBoardsHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("list_workflow_steps", + mcp.WithDescription("List all workflow steps in a board."), + mcp.WithString("board_id", mcp.Required(), mcp.Description("The board ID")), + ), + s.listWorkflowStepsHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("list_tasks", + mcp.WithDescription("List all tasks on a board."), + mcp.WithString("board_id", mcp.Required(), mcp.Description("The board ID")), + ), + s.listTasksHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("create_task", + mcp.WithDescription("Create a new task on a board."), + mcp.WithString("workspace_id", mcp.Required(), mcp.Description("The workspace ID")), + mcp.WithString("board_id", mcp.Required(), mcp.Description("The board ID")), + mcp.WithString("workflow_step_id", mcp.Required(), mcp.Description("The workflow step ID")), + mcp.WithString("title", mcp.Required(), mcp.Description("The task title")), + mcp.WithString("description", mcp.Description("The task description")), + ), + s.createTaskHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("update_task", + mcp.WithDescription("Update an existing task."), + mcp.WithString("task_id", mcp.Required(), mcp.Description("The task ID")), + mcp.WithString("title", mcp.Description("New title")), + mcp.WithString("description", mcp.Description("New description")), + mcp.WithString("state", mcp.Description("New state: not_started, in_progress, etc.")), + ), + s.updateTaskHandler(), + ) + + s.mcpServer.AddTool( + mcp.NewTool("ask_user_question", + mcp.WithDescription("Ask the user a clarifying question."), + mcp.WithString("prompt", mcp.Required(), mcp.Description("The question to ask")), + mcp.WithArray("options", mcp.Required(), mcp.Description("Options for the user")), + mcp.WithString("context", mcp.Description("Context for the question")), + ), + s.askUserQuestionHandler(), + ) + + s.logger.Info("registered MCP tools", zap.Int("count", 7)) +} + +// WrapHandler wraps a handler function with error handling for gin. +func WrapHandler(h http.Handler) gin.HandlerFunc { + return func(c *gin.Context) { + h.ServeHTTP(c.Writer, c.Request) + } +} + diff --git a/apps/backend/internal/agentctl/server/wsclient/client.go b/apps/backend/internal/agentctl/server/wsclient/client.go new file mode 100644 index 0000000000..d03630f36f --- /dev/null +++ b/apps/backend/internal/agentctl/server/wsclient/client.go @@ -0,0 +1,177 @@ +// Package wsclient provides a WebSocket client for connecting back to the Kandev backend. +package wsclient + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +type Client struct { + url string + conn *websocket.Conn + logger *logger.Logger + pending map[string]chan *ws.Message + pendingMu sync.RWMutex + connected bool + connMu sync.RWMutex + writeMu sync.Mutex + reconnectInterval time.Duration + maxReconnectTries int +} + +func New(url string, log *logger.Logger) *Client { + return &Client{ + url: url, + logger: log.WithFields(zap.String("component", "wsclient")), + pending: make(map[string]chan *ws.Message), + reconnectInterval: 5 * time.Second, + maxReconnectTries: 10, + } +} + +func (c *Client) Connect(ctx context.Context) error { + c.connMu.Lock() + defer c.connMu.Unlock() + if c.connected { + return nil + } + conn, _, err := websocket.DefaultDialer.DialContext(ctx, c.url, nil) + if err != nil { + return fmt.Errorf("failed to connect to backend: %w", err) + } + c.conn = conn + c.connected = true + c.logger.Info("connected to backend WebSocket", zap.String("url", c.url)) + go c.readLoop() + return nil +} + +func (c *Client) Close() error { + c.connMu.Lock() + defer c.connMu.Unlock() + if !c.connected { + return nil + } + c.connected = false + if c.conn != nil { + return c.conn.Close() + } + return nil +} + +func (c *Client) IsConnected() bool { + c.connMu.RLock() + defer c.connMu.RUnlock() + return c.connected +} + +func (c *Client) Request(ctx context.Context, action string, payload interface{}) (*ws.Message, error) { + if !c.IsConnected() { + return nil, fmt.Errorf("not connected to backend") + } + id := uuid.New().String() + msg, err := ws.NewRequest(id, action, payload) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + respChan := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pending[id] = respChan + c.pendingMu.Unlock() + c.writeMu.Lock() + err = c.conn.WriteJSON(msg) + c.writeMu.Unlock() + if err != nil { + c.pendingMu.Lock() + delete(c.pending, id) + c.pendingMu.Unlock() + return nil, fmt.Errorf("failed to send request: %w", err) + } + c.logger.Debug("sent request", zap.String("action", action), zap.String("id", id)) + select { + case resp := <-respChan: + return resp, nil + case <-ctx.Done(): + c.pendingMu.Lock() + delete(c.pending, id) + c.pendingMu.Unlock() + return nil, ctx.Err() + } +} + +func (c *Client) RequestPayload(ctx context.Context, action string, payload, result interface{}) error { + resp, err := c.Request(ctx, action, payload) + if err != nil { + return err + } + if resp.Type == ws.MessageTypeError { + var ep struct{ Code, Message string } + if json.Unmarshal(resp.Payload, &ep) == nil { + return fmt.Errorf("backend error [%s]: %s", ep.Code, ep.Message) + } + return fmt.Errorf("backend error: %s", string(resp.Payload)) + } + if result != nil && len(resp.Payload) > 0 { + if err := json.Unmarshal(resp.Payload, result); err != nil { + return fmt.Errorf("failed to unmarshal response: %w", err) + } + } + return nil +} + +func (c *Client) readLoop() { + for { + c.connMu.RLock() + conn, connected := c.conn, c.connected + c.connMu.RUnlock() + if !connected || conn == nil { + return + } + var msg ws.Message + if err := conn.ReadJSON(&msg); err != nil { + if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Error("read error", zap.Error(err)) + } + c.handleDisconnect() + return + } + c.handleMessage(&msg) + } +} + +func (c *Client) handleMessage(msg *ws.Message) { + if msg.Type == ws.MessageTypeResponse || msg.Type == ws.MessageTypeError { + c.pendingMu.Lock() + ch, ok := c.pending[msg.ID] + delete(c.pending, msg.ID) + c.pendingMu.Unlock() + if ok { + ch <- msg + } + return + } +} + +func (c *Client) handleDisconnect() { + c.connMu.Lock() + c.connected = false + c.conn = nil + c.connMu.Unlock() + c.pendingMu.Lock() + for id, ch := range c.pending { + errMsg, _ := ws.NewError(id, "", ws.ErrorCodeInternalError, "connection lost", nil) + ch <- errMsg + delete(c.pending, id) + } + c.pendingMu.Unlock() +} + diff --git a/apps/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go index 12c6f9eb9f..2e05e87708 100644 --- a/apps/backend/internal/common/config/config.go +++ b/apps/backend/internal/common/config/config.go @@ -103,15 +103,17 @@ type AgentConfig struct { // StandalonePort is the control port for standalone agentctl (default: 9999) StandalonePort int `mapstructure:"standalonePort"` - // McpServerEnabled enables the embedded MCP server (default: true) + // McpServerEnabled enables the standalone MCP server (default: false) + // Note: MCP is now embedded in agentctl and tunnels to backend via WebSocket. + // This setting is only for running a separate standalone MCP server process. McpServerEnabled bool `mapstructure:"mcpServerEnabled"` - // McpServerPort is the port for the embedded MCP server (default: 9090) + // McpServerPort is the port for the standalone MCP server (default: 9090) McpServerPort int `mapstructure:"mcpServerPort"` // McpServerURL is the URL of the Kandev MCP server for task management // If set, agents with supports_mcp=true will be configured with this MCP server - // If McpServerEnabled is true and this is empty, it will be auto-set to http://localhost:{McpServerPort}/sse + // Note: With the new architecture, MCP is embedded in agentctl and this is typically not needed. McpServerURL string `mapstructure:"mcpServerUrl"` } @@ -183,9 +185,9 @@ func setDefaults(v *viper.Viper) { // Agent defaults (runtime selection is now per-task based on executor type) v.SetDefault("agent.standaloneHost", "localhost") v.SetDefault("agent.standalonePort", 9999) - v.SetDefault("agent.mcpServerEnabled", true) + v.SetDefault("agent.mcpServerEnabled", false) // MCP is now embedded in agentctl v.SetDefault("agent.mcpServerPort", 9090) - v.SetDefault("agent.mcpServerUrl", "") // Empty means auto-set when mcpServerEnabled is true + v.SetDefault("agent.mcpServerUrl", "") // Auth defaults v.SetDefault("auth.jwtSecret", "") diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go new file mode 100644 index 0000000000..57bfa3adf5 --- /dev/null +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -0,0 +1,296 @@ +// Package handlers provides WebSocket handlers for MCP tool requests. +// These handlers are called by agentctl via the WS tunnel and execute +// operations against the backend services directly. +package handlers + +import ( + "context" + "encoding/json" + + "github.com/kandev/kandev/internal/clarification" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/controller" + "github.com/kandev/kandev/internal/task/dto" + "github.com/kandev/kandev/internal/task/repository" + workflowctrl "github.com/kandev/kandev/internal/workflow/controller" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// Action constants for MCP WebSocket messages +const ( + ActionMCPListWorkspaces = "mcp.list_workspaces" + ActionMCPListBoards = "mcp.list_boards" + ActionMCPListWorkflowSteps = "mcp.list_workflow_steps" + ActionMCPListTasks = "mcp.list_tasks" + ActionMCPCreateTask = "mcp.create_task" + ActionMCPUpdateTask = "mcp.update_task" + ActionMCPAskUserQuestion = "mcp.ask_user_question" +) + +// ClarificationService defines the interface for clarification operations. +type ClarificationService interface { + CreateRequest(req *clarification.Request) string + WaitForResponse(ctx context.Context, pendingID string) (*clarification.Response, error) +} + +// MessageCreator creates messages for clarification requests. +type MessageCreator interface { + CreateClarificationRequestMessage(ctx context.Context, taskID, sessionID, pendingID string, question clarification.Question, clarificationContext string) (string, error) +} + +// Handlers provides MCP WebSocket handlers. +type Handlers struct { + workspaceCtrl *controller.WorkspaceController + boardCtrl *controller.BoardController + taskCtrl *controller.TaskController + workflowCtrl *workflowctrl.Controller + clarificationSvc ClarificationService + messageCreator MessageCreator + repo repository.Repository + logger *logger.Logger +} + +// NewHandlers creates new MCP handlers. +func NewHandlers( + workspaceCtrl *controller.WorkspaceController, + boardCtrl *controller.BoardController, + taskCtrl *controller.TaskController, + workflowCtrl *workflowctrl.Controller, + clarificationSvc ClarificationService, + messageCreator MessageCreator, + repo repository.Repository, + log *logger.Logger, +) *Handlers { + return &Handlers{ + workspaceCtrl: workspaceCtrl, + boardCtrl: boardCtrl, + taskCtrl: taskCtrl, + workflowCtrl: workflowCtrl, + clarificationSvc: clarificationSvc, + messageCreator: messageCreator, + repo: repo, + logger: log.WithFields(zap.String("component", "mcp-handlers")), + } +} + +// RegisterHandlers registers all MCP handlers with the dispatcher. +func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { + d.RegisterFunc(ActionMCPListWorkspaces, h.handleListWorkspaces) + d.RegisterFunc(ActionMCPListBoards, h.handleListBoards) + d.RegisterFunc(ActionMCPListWorkflowSteps, h.handleListWorkflowSteps) + d.RegisterFunc(ActionMCPListTasks, h.handleListTasks) + d.RegisterFunc(ActionMCPCreateTask, h.handleCreateTask) + d.RegisterFunc(ActionMCPUpdateTask, h.handleUpdateTask) + d.RegisterFunc(ActionMCPAskUserQuestion, h.handleAskUserQuestion) + + h.logger.Info("registered MCP handlers", zap.Int("count", 7)) +} + +// handleListWorkspaces lists all workspaces. +func (h *Handlers) handleListWorkspaces(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + resp, err := h.workspaceCtrl.ListWorkspaces(ctx, dto.ListWorkspacesRequest{}) + if err != nil { + h.logger.Error("failed to list workspaces", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list workspaces", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleListBoards lists boards for a workspace. +func (h *Handlers) handleListBoards(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + WorkspaceID string `json:"workspace_id"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.WorkspaceID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workspace_id is required", nil) + } + + resp, err := h.boardCtrl.ListBoards(ctx, dto.ListBoardsRequest{WorkspaceID: req.WorkspaceID}) + if err != nil { + h.logger.Error("failed to list boards", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list boards", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleListWorkflowSteps lists workflow steps for a board. +func (h *Handlers) handleListWorkflowSteps(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + BoardID string `json:"board_id"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.BoardID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "board_id is required", nil) + } + + resp, err := h.workflowCtrl.ListStepsByBoard(ctx, workflowctrl.ListStepsRequest{BoardID: req.BoardID}) + if err != nil { + h.logger.Error("failed to list workflow steps", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list workflow steps", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleListTasks lists tasks for a board. +func (h *Handlers) handleListTasks(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + BoardID string `json:"board_id"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.BoardID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "board_id is required", nil) + } + + resp, err := h.taskCtrl.ListTasks(ctx, dto.ListTasksRequest{BoardID: req.BoardID}) + if err != nil { + h.logger.Error("failed to list tasks", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list tasks", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleCreateTask creates a new task. +func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req dto.CreateTaskRequest + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.WorkspaceID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workspace_id is required", nil) + } + if req.BoardID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "board_id is required", nil) + } + if req.WorkflowStepID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workflow_step_id is required", nil) + } + if req.Title == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "title is required", nil) + } + + resp, err := h.taskCtrl.CreateTask(ctx, req) + if err != nil { + h.logger.Error("failed to create task", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to create task", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleUpdateTask updates an existing task. +func (h *Handlers) handleUpdateTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req dto.UpdateTaskRequest + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.ID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "id is required", nil) + } + + resp, err := h.taskCtrl.UpdateTask(ctx, req) + if err != nil { + h.logger.Error("failed to update task", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to update task", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// handleAskUserQuestion creates a clarification request and waits for response. +func (h *Handlers) handleAskUserQuestion(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + SessionID string `json:"session_id"` + TaskID string `json:"task_id"` + Question clarification.Question `json:"question"` + Context string `json:"context"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.SessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + } + if req.Question.Prompt == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "question.prompt is required", nil) + } + + // Generate question ID if missing + if req.Question.ID == "" { + req.Question.ID = "q1" + } + // Generate option IDs if missing + for i := range req.Question.Options { + if req.Question.Options[i].ID == "" { + req.Question.Options[i].ID = generateOptionID(0, i) + } + } + + // Look up task ID from session if not provided + taskID := req.TaskID + if taskID == "" && h.repo != nil { + session, err := h.repo.GetTaskSession(ctx, req.SessionID) + if err != nil { + h.logger.Warn("failed to look up session for task ID", + zap.String("session_id", req.SessionID), + zap.Error(err)) + } else { + taskID = session.TaskID + } + } + + // Create the clarification request + clarificationReq := &clarification.Request{ + SessionID: req.SessionID, + TaskID: taskID, + Question: req.Question, + Context: req.Context, + } + pendingID := h.clarificationSvc.CreateRequest(clarificationReq) + + // Create the message in the database (triggers WS event to frontend) + if h.messageCreator != nil { + _, err := h.messageCreator.CreateClarificationRequestMessage( + ctx, + taskID, + req.SessionID, + pendingID, + req.Question, + req.Context, + ) + if err != nil { + h.logger.Error("failed to create clarification request message", + zap.String("pending_id", pendingID), + zap.String("session_id", req.SessionID), + zap.Error(err)) + } + } + + // Wait for user response + resp, err := h.clarificationSvc.WaitForResponse(ctx, pendingID) + if err != nil { + h.logger.Error("failed waiting for clarification response", + zap.String("pending_id", pendingID), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get user response: "+err.Error(), nil) + } + + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +// generateOptionID generates an option ID for a question. +func generateOptionID(questionIndex, optionIndex int) string { + return "q" + string(rune('1'+questionIndex)) + "_opt" + string(rune('1'+optionIndex)) +} + diff --git a/apps/backend/internal/mcpserver/provider.go b/apps/backend/internal/mcpserver/provider.go deleted file mode 100644 index 2948f95361..0000000000 --- a/apps/backend/internal/mcpserver/provider.go +++ /dev/null @@ -1,50 +0,0 @@ -// Package mcpserver provides the MCP server for Kandev task management. -package mcpserver - -import ( - "context" - "sync" - "time" - - "github.com/kandev/kandev/internal/common/logger" - "go.uber.org/zap" -) - -// DefaultConfig returns the default configuration. -func DefaultConfig() Config { - return Config{ - Port: 9090, - KandevURL: "http://localhost:8080", - } -} - -// NewWithLogger creates a new MCP server with the given configuration and logger. -// This is useful for integration with dependency injection frameworks. -func NewWithLogger(cfg Config, log *logger.Logger) *Server { - srv := New(cfg) - srv.logger = log.WithFields(zap.String("component", "mcp-server")) - return srv -} - -// Provide starts the MCP server and returns a cleanup function to stop it. -// This is useful for integration with dependency injection frameworks. -func Provide(ctx context.Context, cfg Config, log *logger.Logger) (*Server, func() error, error) { - srv := NewWithLogger(cfg, log) - if err := srv.Start(ctx); err != nil { - return nil, nil, err - } - - var stopOnce sync.Once - cleanup := func() error { - var stopErr error - stopOnce.Do(func() { - stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - stopErr = srv.Stop(stopCtx) - }) - return stopErr - } - - return srv, cleanup, nil -} - diff --git a/apps/backend/internal/mcpserver/server.go b/apps/backend/internal/mcpserver/server.go deleted file mode 100644 index 9b06ec2895..0000000000 --- a/apps/backend/internal/mcpserver/server.go +++ /dev/null @@ -1,179 +0,0 @@ -// Package mcpserver provides a reusable MCP server for Kandev task management. -// It exposes tools for listing, creating, and updating tasks via SSE and Streamable HTTP transports. -package mcpserver - -import ( - "context" - "fmt" - "net" - "net/http" - "sync" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/mark3labs/mcp-go/server" - "go.uber.org/zap" -) - -// Config holds the MCP server configuration. -type Config struct { - Port int // Port to listen on - KandevURL string // Kandev API URL (e.g., http://localhost:8080) -} - -// Server wraps the SSE and Streamable HTTP servers with lifecycle management. -// It supports both transports for compatibility with different MCP clients: -// - SSE transport (/sse) for Claude Desktop, Cursor, etc. -// - Streamable HTTP transport (/mcp) for Codex -type Server struct { - cfg Config - sseServer *server.SSEServer - streamableHTTPServer *server.StreamableHTTPServer - httpServer *http.Server - mu sync.Mutex - running bool - logger *logger.Logger -} - -// New creates a new MCP server with the given configuration. -func New(cfg Config) *Server { - return &Server{ - cfg: cfg, - logger: logger.Default().WithFields(), - } -} - -// Start starts the MCP server in a goroutine and returns when it's listening. -// It starts both SSE and Streamable HTTP transports on the same port. -// It returns an error if the server fails to start. -func (s *Server) Start(ctx context.Context) error { - s.mu.Lock() - if s.running { - s.mu.Unlock() - return fmt.Errorf("server already running") - } - s.mu.Unlock() - - // Create MCP server (shared between both transports) - mcpServer := server.NewMCPServer( - "kandev-mcp", - "1.0.0", - server.WithToolCapabilities(true), - ) - - // Register tools (using internal logger) - registerTools(mcpServer, s.cfg, s.logger) - - // Create SSE server (for Claude Desktop, Cursor, etc.) - s.sseServer = server.NewSSEServer(mcpServer) - - // Create Streamable HTTP server (for Codex) - // Use WithEndpointPath to ensure it handles /mcp path - s.streamableHTTPServer = server.NewStreamableHTTPServer(mcpServer, - server.WithEndpointPath("/mcp"), - ) - - // Create HTTP mux to route requests to appropriate transport - mux := http.NewServeMux() - - // SSE transport routes - handle /sse and /message paths - mux.Handle("/sse", s.sseServer.SSEHandler()) - mux.Handle("/message", s.sseServer.MessageHandler()) - - // Streamable HTTP transport route - handle /mcp path - mux.Handle("/mcp", s.streamableHTTPServer) - - // Verify the port is available by creating a test listener - addr := fmt.Sprintf(":%d", s.cfg.Port) - listener, err := net.Listen("tcp", addr) - if err != nil { - return fmt.Errorf("failed to listen on %s: %w", addr, err) - } - if tcpAddr, ok := listener.Addr().(*net.TCPAddr); ok { - s.cfg.Port = tcpAddr.Port - } - - // Create HTTP server with the mux - s.httpServer = &http.Server{ - Handler: mux, - } - - // Ready channel to signal when server goroutine has started - ready := make(chan struct{}) - - // Start server in a goroutine - go func() { - s.mu.Lock() - s.running = true - s.mu.Unlock() - - // Signal that we're starting - close(ready) - - s.logger.Info("MCP server listening", - zap.Int("port", s.cfg.Port), - zap.String("sse_endpoint", "/sse"), - zap.String("streamable_http_endpoint", "/mcp")) - - if err := s.httpServer.Serve(listener); err != nil && err != http.ErrServerClosed { - s.logger.Error("MCP server error", zap.Error(err)) - } - - s.mu.Lock() - s.running = false - s.mu.Unlock() - }() - - // Wait for the goroutine to start or context cancellation - select { - case <-ready: - return nil - case <-ctx.Done(): - return ctx.Err() - } -} - -// Stop gracefully shuts down the server. -func (s *Server) Stop(ctx context.Context) error { - s.mu.Lock() - running := s.running - s.mu.Unlock() - - if !running { - return nil - } - - // Shutdown the HTTP server (this stops both transports) - if s.httpServer != nil { - if err := s.httpServer.Shutdown(ctx); err != nil { - return fmt.Errorf("failed to shutdown HTTP server: %w", err) - } - } - - // Also shutdown the SSE server to clean up any active sessions - if s.sseServer != nil { - if err := s.sseServer.Shutdown(ctx); err != nil { - s.logger.Warn("failed to shutdown SSE server", zap.Error(err)) - } - } - - // Shutdown the Streamable HTTP server to clean up any active sessions - if s.streamableHTTPServer != nil { - if err := s.streamableHTTPServer.Shutdown(ctx); err != nil { - s.logger.Warn("failed to shutdown Streamable HTTP server", zap.Error(err)) - } - } - - return nil -} - -// SSEEndpoint returns the full SSE URL for clients that use SSE transport -// (e.g., Claude Desktop, Cursor). -func (s *Server) SSEEndpoint() string { - return fmt.Sprintf("http://localhost:%d/sse", s.cfg.Port) -} - -// StreamableHTTPEndpoint returns the full Streamable HTTP URL for clients that use -// streamable HTTP transport (e.g., Codex). -func (s *Server) StreamableHTTPEndpoint() string { - return fmt.Sprintf("http://localhost:%d/mcp", s.cfg.Port) -} diff --git a/apps/backend/internal/mcpserver/tools.go b/apps/backend/internal/mcpserver/tools.go deleted file mode 100644 index 6ce9b93763..0000000000 --- a/apps/backend/internal/mcpserver/tools.go +++ /dev/null @@ -1,533 +0,0 @@ -package mcpserver - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "net/http" - "strings" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/mark3labs/mcp-go/mcp" - "github.com/mark3labs/mcp-go/server" - "go.uber.org/zap" -) - -func registerTools(s *server.MCPServer, cfg Config, log *logger.Logger) { - // List Workspaces tool - s.AddTool( - mcp.NewTool("list_workspaces", - mcp.WithDescription("List all workspaces. Use this first to get workspace IDs for other operations."), - ), - listWorkspacesHandler(cfg, log), - ) - - // List Boards tool - s.AddTool( - mcp.NewTool("list_boards", - mcp.WithDescription("List all boards in a workspace. Use this to get board IDs for task operations."), - mcp.WithString("workspace_id", - mcp.Required(), - mcp.Description("The workspace ID to list boards from"), - ), - ), - listBoardsHandler(cfg, log), - ) - - // List Workflow Steps tool - s.AddTool( - mcp.NewTool("list_workflow_steps", - mcp.WithDescription("List all workflow steps in a board. Use this to get workflow step IDs for creating tasks."), - mcp.WithString("board_id", - mcp.Required(), - mcp.Description("The board ID to list workflow steps from"), - ), - ), - listWorkflowStepsHandler(cfg, log), - ) - - // List Tasks tool - s.AddTool( - mcp.NewTool("list_tasks", - mcp.WithDescription("List all tasks on a board"), - mcp.WithString("board_id", - mcp.Required(), - mcp.Description("The board ID to list tasks from"), - ), - ), - listTasksHandler(cfg, log), - ) - - // Create Task tool - s.AddTool( - mcp.NewTool("create_task", - mcp.WithDescription("Create a new task on a board"), - mcp.WithString("workspace_id", - mcp.Required(), - mcp.Description("The workspace ID"), - ), - mcp.WithString("board_id", - mcp.Required(), - mcp.Description("The board ID"), - ), - mcp.WithString("workflow_step_id", - mcp.Required(), - mcp.Description("The workflow step ID to place the task in"), - ), - mcp.WithString("title", - mcp.Required(), - mcp.Description("The task title"), - ), - mcp.WithString("description", - mcp.Description("The task description (optional)"), - ), - ), - createTaskHandler(cfg, log), - ) - - // Update Task tool - s.AddTool( - mcp.NewTool("update_task", - mcp.WithDescription("Update an existing task"), - mcp.WithString("task_id", - mcp.Required(), - mcp.Description("The task ID to update"), - ), - mcp.WithString("title", - mcp.Description("New title (optional)"), - ), - mcp.WithString("description", - mcp.Description("New description (optional)"), - ), - mcp.WithString("state", - mcp.Description("New state: not_started, in_progress, paused, completed, cancelled (optional)"), - ), - ), - updateTaskHandler(cfg, log), - ) - - // Ask User Question tool - s.AddTool( - mcp.NewTool("ask_user_question", - mcp.WithDescription( - "Ask the user a clarifying question when you need more information to proceed. "+ - "Use this tool when you need to:\n"+ - "1. Gather user preferences or requirements\n"+ - "2. Clarify ambiguous instructions\n"+ - "3. Get decisions on implementation choices\n"+ - "4. Offer choices about what direction to take\n\n"+ - "The question includes 2-4 multiple-choice options. Users can also provide custom text input.\n"+ - "Returns the user's answer.", - ), - mcp.WithString("session_id", - mcp.Required(), - mcp.Description("The session ID for this clarification request"), - ), - mcp.WithString("prompt", - mcp.Required(), - mcp.Description("The question to ask the user"), - ), - mcp.WithArray("options", - mcp.Required(), - mcp.Description("2-4 options for the user to choose from. Each option should have: label (concise 1-5 words) and description (explanation of the option)"), - ), - mcp.WithString("context", - mcp.Description("Optional context explaining why this question is being asked"), - ), - ), - askUserQuestionHandler(cfg, log), - ) - - log.Info("registered MCP tools", zap.Int("count", 7)) -} - -func listWorkspacesHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - url := fmt.Sprintf("%s/api/v1/workspaces", cfg.KandevURL) - resp, err := http.Get(url) - if err != nil { - log.Error("failed to fetch workspaces", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to fetch workspaces: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func listBoardsHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - workspaceID, err := req.RequireString("workspace_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - - url := fmt.Sprintf("%s/api/v1/workspaces/%s/boards", cfg.KandevURL, workspaceID) - resp, err := http.Get(url) - if err != nil { - log.Error("failed to fetch boards", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to fetch boards: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func listWorkflowStepsHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - boardID, err := req.RequireString("board_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - - url := fmt.Sprintf("%s/api/v1/boards/%s/workflow/steps", cfg.KandevURL, boardID) - resp, err := http.Get(url) - if err != nil { - log.Error("failed to fetch workflow steps", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to fetch workflow steps: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func listTasksHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - boardID, err := req.RequireString("board_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - - url := fmt.Sprintf("%s/api/v1/boards/%s/tasks", cfg.KandevURL, boardID) - resp, err := http.Get(url) - if err != nil { - log.Error("failed to fetch tasks", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to fetch tasks: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func createTaskHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - workspaceID, err := req.RequireString("workspace_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - boardID, err := req.RequireString("board_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - workflowStepID, err := req.RequireString("workflow_step_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - title, err := req.RequireString("title") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - - payload := map[string]interface{}{ - "workspace_id": workspaceID, - "board_id": boardID, - "workflow_step_id": workflowStepID, - "title": title, - } - if desc := req.GetString("description", ""); desc != "" { - payload["description"] = desc - } - - body, _ := json.Marshal(payload) - url := fmt.Sprintf("%s/api/v1/tasks", cfg.KandevURL) - - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) - if err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to create request: %v", err)), nil - } - httpReq.Header.Set("Content-Type", "application/json") - - resp, err := http.DefaultClient.Do(httpReq) - if err != nil { - log.Error("failed to create task", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to create task: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - if resp.StatusCode >= 400 { - return mcp.NewToolResultError(fmt.Sprintf("API error (%d): %s", resp.StatusCode, string(result))), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func updateTaskHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - taskID, err := req.RequireString("task_id") - if err != nil { - return mcp.NewToolResultError(err.Error()), nil - } - - payload := make(map[string]interface{}) - if title := req.GetString("title", ""); title != "" { - payload["title"] = title - } - if desc := req.GetString("description", ""); desc != "" { - payload["description"] = desc - } - if state := req.GetString("state", ""); state != "" { - payload["state"] = state - } - - body, _ := json.Marshal(payload) - url := fmt.Sprintf("%s/api/v1/tasks/%s", cfg.KandevURL, taskID) - - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPatch, url, bytes.NewReader(body)) - if err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to create request: %v", err)), nil - } - httpReq.Header.Set("Content-Type", "application/json") - - resp, err := http.DefaultClient.Do(httpReq) - if err != nil { - log.Error("failed to update task", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to update task: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - var result json.RawMessage - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - if resp.StatusCode >= 400 { - return mcp.NewToolResultError(fmt.Sprintf("API error (%d): %s", resp.StatusCode, string(result))), nil - } - - formatted, _ := json.MarshalIndent(result, "", " ") - return mcp.NewToolResultText(string(formatted)), nil - } -} - -func askUserQuestionHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { - return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - sessionID, err := req.RequireString("session_id") - if err != nil || sessionID == "" { - return mcp.NewToolResultError("session_id is required - this tool can only be used within a Kandev task session"), nil - } - - prompt, err := req.RequireString("prompt") - if err != nil { - return mcp.NewToolResultError("prompt is required"), nil - } - - args := req.GetArguments() - optionsRaw, ok := args["options"] - if !ok { - return mcp.NewToolResultError("options is required"), nil - } - - // Parse options from the raw interface - optionsJSON, err := json.Marshal(optionsRaw) - if err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse options: %v", err)), nil - } - - var options []map[string]interface{} - if err := json.Unmarshal(optionsJSON, &options); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse options: %v", err)), nil - } - - if len(options) < 2 || len(options) > 4 { - return mcp.NewToolResultError("Must provide 2-4 options"), nil - } - - // Generate option IDs to ensure consistency between request and response - // The backend also generates these, but we need them here for matching - for i := range options { - options[i]["option_id"] = fmt.Sprintf("q1_opt%d", i+1) - } - - context := req.GetString("context", "") - - // Build the single question - question := map[string]interface{}{ - "id": "q1", - "title": "Question", - "prompt": prompt, - "options": options, - } - - // Create the clarification request (backend still expects questions array for now) - payload := map[string]interface{}{ - "session_id": sessionID, - "question": question, - "context": context, - } - - body, _ := json.Marshal(payload) - url := fmt.Sprintf("%s/api/v1/clarification/request", cfg.KandevURL) - - log.Debug("creating clarification request", - zap.String("session_id", sessionID)) - - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) - if err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to create request: %v", err)), nil - } - httpReq.Header.Set("Content-Type", "application/json") - - resp, err := http.DefaultClient.Do(httpReq) - if err != nil { - log.Error("failed to create clarification request", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to create clarification request: %v", err)), nil - } - defer func() { _ = resp.Body.Close() }() - - if resp.StatusCode >= 400 { - var errBody json.RawMessage - _ = json.NewDecoder(resp.Body).Decode(&errBody) - return mcp.NewToolResultError(fmt.Sprintf("API error (%d): %s", resp.StatusCode, string(errBody))), nil - } - - var createResp struct { - PendingID string `json:"pending_id"` - } - if err := json.NewDecoder(resp.Body).Decode(&createResp); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil - } - - // Now wait for the user's response - waitURL := fmt.Sprintf("%s/api/v1/clarification/%s/wait", cfg.KandevURL, createResp.PendingID) - - log.Debug("waiting for user response", - zap.String("pending_id", createResp.PendingID)) - - waitReq, err := http.NewRequestWithContext(ctx, http.MethodGet, waitURL, nil) - if err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to create wait request: %v", err)), nil - } - - waitResp, err := http.DefaultClient.Do(waitReq) - if err != nil { - log.Error("failed to wait for response", zap.Error(err)) - return mcp.NewToolResultError(fmt.Sprintf("Failed to wait for response: %v", err)), nil - } - defer func() { _ = waitResp.Body.Close() }() - - if waitResp.StatusCode >= 400 { - var errBody json.RawMessage - _ = json.NewDecoder(waitResp.Body).Decode(&errBody) - return mcp.NewToolResultError(fmt.Sprintf("Wait error (%d): %s", waitResp.StatusCode, string(errBody))), nil - } - - var userResponse struct { - PendingID string `json:"pending_id"` - Answer struct { - SelectedOptions []string `json:"selected_options"` - CustomText string `json:"custom_text"` - } `json:"answer"` - Rejected bool `json:"rejected"` - RejectReason string `json:"reject_reason"` - } - if err := json.NewDecoder(waitResp.Body).Decode(&userResponse); err != nil { - return mcp.NewToolResultError(fmt.Sprintf("Failed to parse user response: %v", err)), nil - } - - // Format the response for the agent - result := formatClarificationResponse(prompt, options, userResponse.Answer, userResponse.Rejected, userResponse.RejectReason) - - log.Debug("clarification response received", - zap.String("pending_id", createResp.PendingID), - zap.Bool("rejected", userResponse.Rejected)) - - return mcp.NewToolResultText(result), nil - } -} - -func formatClarificationResponse(prompt string, options []map[string]interface{}, answer struct { - SelectedOptions []string `json:"selected_options"` - CustomText string `json:"custom_text"` -}, rejected bool, rejectReason string) string { - if rejected { - if rejectReason != "" { - return fmt.Sprintf("The user declined to answer and said: %s\n\nProceed with your best judgment based on the available information.", rejectReason) - } - return "The user declined to answer this question. Proceed with your best judgment based on the available information." - } - - var result strings.Builder - result.WriteString("# User Response\n\n") - result.WriteString(fmt.Sprintf("**Question asked:** %s\n\n", prompt)) - - hasAnswer := false - - if len(answer.SelectedOptions) > 0 { - // Find the label and description for the selected option - for _, opt := range options { - optID, _ := opt["option_id"].(string) - if optID == "" { - optID, _ = opt["id"].(string) - } - for _, selected := range answer.SelectedOptions { - if optID == selected { - label, _ := opt["label"].(string) - description, _ := opt["description"].(string) - result.WriteString(fmt.Sprintf("**User's choice:** %s\n", label)) - if description != "" { - result.WriteString(fmt.Sprintf("**Choice description:** %s\n", description)) - } - hasAnswer = true - } - } - } - } - - if answer.CustomText != "" { - result.WriteString(fmt.Sprintf("\n**Additional user input:** %s\n", answer.CustomText)) - hasAnswer = true - } - - if !hasAnswer { - result.WriteString("The user did not provide an answer.\n") - } - - result.WriteString("\nProceed with the task based on the user's response above.") - - return result.String() -} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index b774aab530..4faecd3a62 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -198,6 +198,15 @@ const ( ActionUserGet = "user.get" ActionUserSettingsUpdate = "user.settings.update" ActionUserSettingsUpdated = "user.settings.updated" + + // MCP tool actions (agentctl -> backend via WS tunnel) + ActionMCPListWorkspaces = "mcp.list_workspaces" + ActionMCPListBoards = "mcp.list_boards" + ActionMCPListWorkflowSteps = "mcp.list_workflow_steps" + ActionMCPListTasks = "mcp.list_tasks" + ActionMCPCreateTask = "mcp.create_task" + ActionMCPUpdateTask = "mcp.update_task" + ActionMCPAskUserQuestion = "mcp.ask_user_question" ) // Error codes diff --git a/apps/web/app/settings/agents/[agentId]/profile-mcp-config-card.tsx b/apps/web/app/settings/agents/[agentId]/profile-mcp-config-card.tsx index 297a8d7475..013bb863a9 100644 --- a/apps/web/app/settings/agents/[agentId]/profile-mcp-config-card.tsx +++ b/apps/web/app/settings/agents/[agentId]/profile-mcp-config-card.tsx @@ -6,7 +6,6 @@ import { Switch } from '@kandev/ui/switch'; import { Textarea } from '@kandev/ui/textarea'; import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip'; import { UnsavedChangesBadge, UnsavedSaveButton } from '@/components/settings/unsaved-indicator'; -import { getBackendConfig } from '@/lib/config'; import { useProfileMcpConfig } from './use-profile-mcp-config'; import type { AgentProfileMcpConfig } from '@/lib/types/http'; @@ -64,14 +63,11 @@ export function ProfileMcpConfigCard({ const currentServers = isDraft ? draftState?.servers ?? '' : mcpServers; const currentError = isDraft ? draftState?.error ?? null : mcpError; const currentDirty = isDraft ? draftState?.dirty ?? false : mcpDirty; - const { mcpServerUrl } = getBackendConfig(); const kandevTools = 'Tools: list_workspaces, list_boards, list_workflow_steps, list_tasks, create_task, update_task'; + // Note: kandev MCP server is automatically injected by agentctl at session creation time. + // Users only need to configure external MCP servers here. const popularServers: Record> = { - kandev: { - type: 'sse', - url: mcpServerUrl, - }, playwright: { type: 'stdio', command: 'npx', @@ -204,22 +200,24 @@ export function ProfileMcpConfigCard({ />

MCP definitions are stored in the database and resolved per executor at runtime. This does not override your local agent config.

-

Popular servers

-
+

Built-in

+
- + ✓ Kandev MCP + - {kandevTools} +

Automatically available

+

{kandevTools}

 +
+

Popular servers

+
- ); - - if (option.description) { - return ( - - {button} - - {option.description} - - - ); - } - - return button; - })} + {question.options.map((option) => ( + + ))}
{/* Custom input row - indented to align with options */} diff --git a/docs/architecture/mcp-agentctl-migration.md b/docs/architecture/mcp-agentctl-migration.md new file mode 100644 index 0000000000..b9725157ae --- /dev/null +++ b/docs/architecture/mcp-agentctl-migration.md @@ -0,0 +1,326 @@ +# MCP Server Migration: Backend → agentctl + +This document describes the architectural change to move the MCP server from the Kandev backend to agentctl, enabling remote agents to access MCP tools. + +## Problem Statement + +Currently, the MCP server runs embedded in the Kandev backend on the user's machine. When agents run remotely (Docker containers, VPS, Kubernetes), they cannot access the MCP server because it's not reachable from their network location. + +## Solution + +Move the MCP server into agentctl, which always runs co-located with the agent. API calls from MCP tools are tunneled back to the Kandev backend through the existing WebSocket connection. + +--- + +## Current Architecture (MCP Server with Backend) + +```mermaid +flowchart TB + subgraph UserMachine["👤 USER'S MACHINE"] + subgraph Backend["KANDEV BACKEND"] + Orchestrator["Orchestrator"] + Lifecycle["Lifecycle Manager"] + Runtime["Runtime
(Docker/Standalone)"] + + subgraph MCPServer["EMBEDDED MCP SERVER :9090"] + SSE["/sse endpoint"] + MCP["/mcp endpoint"] + Tools["Tools:
list_workspaces
list_tasks
create_task
update_task"] + end + + API["REST API :8080
/api/v1/tasks
/api/v1/boards"] + + Orchestrator --> Lifecycle + Lifecycle --> Runtime + MCPServer -->|"HTTP calls"| API + end + end + + subgraph Remote["🖥️ REMOTE SERVER / CONTAINER"] + subgraph Agentctl["AGENTCTL"] + InstanceMgr["Instance Manager
Ports: 10000+"] + Agent["Agent Process
(Claude, Codex, etc)
stdin/stdout ACP"] + + InstanceMgr --> Agent + end + + NoMCP["🚫 NO MCP ACCESS
Port 9090 unreachable"] + Agent -.->|"❌ CANNOT CONNECT"| NoMCP + end + + Runtime -->|"WS connection"| Agentctl + NoMCP -.->|"❌ Network barrier"| MCPServer + + style NoMCP fill:#ffcccc,stroke:#cc0000 + style MCPServer fill:#ffffcc,stroke:#cccc00 +``` + +--- + +## Proposed Architecture (MCP Server in agentctl) + +```mermaid +flowchart TB + subgraph UserMachine["👤 USER'S MACHINE"] + subgraph Backend["KANDEV BACKEND"] + Orchestrator["Orchestrator"] + Lifecycle["Lifecycle Manager"] + Runtime["Runtime
(Docker/Standalone)"] + + subgraph WSGateway["WS Gateway :8080"] + Dispatcher["Dispatcher"] + MCPHandlers["MCP Handlers"] + end + + Services["Service Layer
(Task, Board, Workspace)"] + + Orchestrator --> Lifecycle + Lifecycle --> Runtime + Dispatcher --> MCPHandlers + MCPHandlers --> Services + end + end + + subgraph Remote["🖥️ REMOTE SERVER / CONTAINER"] + subgraph Agentctl["AGENTCTL"] + InstanceMgr["Instance Manager
Ports: 10000+"] + + subgraph MCPServer["EMBEDDED MCP SERVER"] + SSE["/sse endpoint"] + MCP["/mcp endpoint"] + Tools["Tools:
list_workspaces
list_tasks
create_task
update_task"] + end + + WSTunnel["WS Client
(connects to backend)"] + + Agent["Agent Process
(Claude, Codex, etc)
stdin/stdout ACP"] + + InstanceMgr --> Agent + MCPServer -->|"forward requests"| WSTunnel + Agent -->|"✅ localhost"| MCPServer + end + end + + Runtime <-->|"WS connection"| Agentctl + WSTunnel <-->|"mcp.* actions"| Dispatcher + + style MCPServer fill:#ccffcc,stroke:#00cc00 + style WSTunnel fill:#cce5ff,stroke:#0066cc + style MCPHandlers fill:#ccffcc,stroke:#00cc00 +``` + +--- + +## Sequence Diagram: Agent Using MCP (Proposed) + +```mermaid +sequenceDiagram + participant Agent as Agent Process + participant MCP as MCP Server
(in agentctl) + participant Tunnel as WS Tunnel
(in agentctl) + participant Dispatcher as WS Dispatcher
(in Backend) + participant Handler as MCP Handler
(in Backend) + participant Service as Service Layer
(in Backend) + participant DB as Database + + Note over Agent,DB: Agent wants to create a task + + Agent->>MCP: POST /mcp
tool: create_task + MCP->>MCP: Parse MCP request + + MCP->>Tunnel: Forward request + Tunnel->>Dispatcher: WS message
{action: "mcp.create_task", ...} + Dispatcher->>Handler: Dispatch to registered handler + Handler->>Service: taskService.Create(...) + Service->>DB: Insert task + DB-->>Service: Task created + Service-->>Handler: Task entity + Handler-->>Dispatcher: Response message + Dispatcher-->>Tunnel: WS message
{action: "mcp.create_task", result: {...}} + Tunnel-->>MCP: Response + + MCP-->>Agent: MCP tool result
{task_id: "..."} + + Note over Agent,DB: ✅ Uses existing WS Dispatcher pattern +``` + +--- + +## Component Comparison + +```mermaid +flowchart LR + subgraph Current["❌ CURRENT"] + direction TB + B1["Backend"] + M1["MCP Server"] + A1["agentctl"] + AG1["Agent"] + + B1 --- M1 + B1 -.->|"WS"| A1 + A1 --- AG1 + AG1 -.->|"❌ unreachable"| M1 + end + + subgraph Proposed["✅ PROPOSED"] + direction TB + B2["Backend"] + A2["agentctl"] + M2["MCP Server"] + AG2["Agent"] + + B2 <-->|"WS + Tunnel"| A2 + A2 --- M2 + A2 --- AG2 + AG2 -->|"✅ localhost"| M2 + end + + Current -->|"Migration"| Proposed + + style Current fill:#ffeeee + style Proposed fill:#eeffee +``` + +--- + +## Summary + +| Aspect | Current | Proposed | +|--------|---------|----------| +| **MCP Location** | Backend (user's machine) | agentctl (co-located with agent) | +| **Remote Agent Access** | ❌ Cannot reach MCP | ✅ Always localhost | +| **Network Dependency** | Requires direct network path | Works over existing WS tunnel | +| **Deployment Flexibility** | Limited to local agents | Works anywhere (Docker, VPS, K8s) | +| **API Communication** | MCP → localhost API | MCP → WS Tunnel → Backend API | + +--- + +## Implementation Notes + +### Changes Required + +**agentctl:** +1. Add embedded MCP server (adapt `internal/mcpserver` package) +2. Add WS client that connects back to Kandev backend +3. MCP tool handlers forward requests via WS client instead of HTTP + +**Backend:** +1. Add MCP handlers package (`internal/mcp/handlers/`) +2. Register `mcp.*` actions with WS Dispatcher +3. Handlers call existing services (TaskService, BoardService, etc.) +4. Remove embedded MCP server from `cmd/kandev` + +**Lifecycle Manager:** +1. Pass backend WS URL to agentctl during instance creation +2. agentctl connects back to backend on startup + +### Backend Internal Architecture + +The backend uses the existing **WS Dispatcher pattern** (not the event bus) for MCP requests. This is consistent with how other WS handlers work (tasks, messages, workflows, etc.): + +```mermaid +flowchart TB + subgraph Backend["KANDEV BACKEND"] + subgraph Gateway["WS Gateway"] + Hub["Hub
(client connections)"] + Dispatcher["Dispatcher
(action routing)"] + end + + subgraph Handlers["Registered Handlers"] + TaskH["Task Handlers
task.list, task.create..."] + MsgH["Message Handlers
message.add, message.list..."] + MCPH["MCP Handlers ⭐ NEW
mcp.create_task, mcp.list_tasks..."] + end + + subgraph Services["Service Layer"] + TaskSvc["TaskService"] + BoardSvc["BoardService"] + WorkspaceSvc["WorkspaceService"] + end + + Hub --> Dispatcher + Dispatcher --> TaskH + Dispatcher --> MsgH + Dispatcher --> MCPH + + TaskH --> TaskSvc + MsgH --> TaskSvc + MCPH --> TaskSvc + MCPH --> BoardSvc + MCPH --> WorkspaceSvc + end + + style MCPH fill:#ccffcc,stroke:#00cc00 +``` + +**Why WS Dispatcher instead of Event Bus?** + +| Aspect | WS Dispatcher | Event Bus | +|--------|---------------|-----------| +| **Pattern** | Request/Response | Pub/Sub (fire-and-forget) | +| **Use case** | Synchronous operations | Async notifications | +| **Existing usage** | All WS handlers (tasks, messages, etc.) | Agent lifecycle events, notifications | +| **MCP fit** | ✅ MCP tools need responses | ❌ Would need request-reply wrapper | + +### WS Message Format + +MCP requests use the standard Kandev WS message format with `mcp.*` action prefix: + +```json +{ + "id": "req-123", + "action": "mcp.create_task", + "payload": { + "workspace_id": "ws-1", + "board_id": "board-1", + "title": "New task" + }, + "timestamp": "2024-01-15T10:30:00Z" +} +``` + +Response: + +```json +{ + "id": "req-123", + "action": "mcp.create_task", + "payload": { + "task_id": "task-456", + "title": "New task" + }, + "timestamp": "2024-01-15T10:30:01Z" +} +``` + +Error response: + +```json +{ + "id": "req-123", + "action": "mcp.create_task", + "error": { + "code": "NOT_FOUND", + "message": "Board not found" + }, + "timestamp": "2024-01-15T10:30:01Z" +} +``` + +### MCP Configuration for Agents + +When configuring agents that support MCP, agentctl will provide the local MCP endpoint: + +```json +{ + "mcpServers": { + "kandev": { + "url": "http://localhost:10001/mcp" + } + } +} +``` + +Where `10001` is the instance's allocated port. +