feat(mcp): add embedded MCP server with dual transport support (#67)
This commit is contained in:
João Salavisa
@@ -0,0 +1,3 @@
|
||||
# Hoist @lezer/highlight for codemirror theme packages that expect it as a peer dependency
|
||||
public-hoist-pattern[]=@lezer/*
|
||||
|
||||
+14
-7
@@ -14,19 +14,19 @@ 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 clean run dev test lint fmt vet help
|
||||
.PHONY: all build build-all build-agentctl build-mcp-server clean run dev test lint fmt vet help
|
||||
|
||||
## Default target
|
||||
all: build
|
||||
|
||||
## Build the unified kandev binary
|
||||
build: build-agentctl
|
||||
build: build-agentctl build-mcp-server
|
||||
@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)
|
||||
build-all: build build-agentctl
|
||||
## Build all binaries (unified + individual services + agentctl + mcp-server)
|
||||
build-all: build build-agentctl build-mcp-server
|
||||
@echo "Building individual services..."
|
||||
$(GO) build $(GOFLAGS) -ldflags "$(LDFLAGS)" -o $(BUILD_DIR)/agent-manager ./cmd/agent-manager
|
||||
|
||||
@@ -36,6 +36,12 @@ build-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..."
|
||||
@@ -105,9 +111,10 @@ help:
|
||||
@echo " make [target]"
|
||||
@echo ""
|
||||
@echo "Targets:"
|
||||
@echo " build Build the unified kandev binary (default)"
|
||||
@echo " build-all Build all binaries (unified + individual services + agentctl)"
|
||||
@echo " build-agentctl Build the agentctl sidecar binary"
|
||||
@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 " clean Remove build artifacts"
|
||||
@echo " run Build and run the application"
|
||||
@echo " dev Run in development mode with a local database"
|
||||
|
||||
@@ -218,6 +218,22 @@ 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
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================
|
||||
// AGENT MANAGER
|
||||
// ============================================
|
||||
@@ -255,7 +271,7 @@ func main() {
|
||||
// WEBSOCKET GATEWAY (All communication via WebSocket)
|
||||
// ============================================
|
||||
log.Info("Initializing WebSocket Gateway...")
|
||||
gateway, notificationSvc, _, err := provideGateway(
|
||||
gateway, _, _, err := provideGateway(
|
||||
ctx,
|
||||
log,
|
||||
eventBus,
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
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
|
||||
}
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
// 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
|
||||
}
|
||||
|
||||
+8
-1
@@ -12,6 +12,7 @@ require (
|
||||
github.com/gin-gonic/gin v1.9.1
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/websocket v1.5.3
|
||||
github.com/mark3labs/mcp-go v0.43.2
|
||||
github.com/mattn/go-sqlite3 v1.14.33
|
||||
github.com/nats-io/nats.go v1.31.0
|
||||
github.com/spf13/viper v1.16.0
|
||||
@@ -21,6 +22,8 @@ require (
|
||||
|
||||
require (
|
||||
github.com/Microsoft/go-winio v0.4.21 // indirect
|
||||
github.com/bahlo/generic-list-go v0.2.0 // indirect
|
||||
github.com/buger/jsonparser v1.1.1 // indirect
|
||||
github.com/bytedance/sonic v1.9.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
|
||||
@@ -41,11 +44,13 @@ require (
|
||||
github.com/go-playground/validator/v10 v10.14.0 // indirect
|
||||
github.com/goccy/go-json v0.10.2 // indirect
|
||||
github.com/hashicorp/hcl v1.0.0 // indirect
|
||||
github.com/invopop/jsonschema v0.13.0 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/klauspost/compress v1.17.0 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.4 // indirect
|
||||
github.com/leodido/go-urn v1.2.4 // indirect
|
||||
github.com/magiconair/properties v1.8.7 // indirect
|
||||
github.com/mailru/easyjson v0.7.7 // indirect
|
||||
github.com/mattn/go-isatty v0.0.19 // indirect
|
||||
github.com/mitchellh/mapstructure v1.5.0 // indirect
|
||||
github.com/moby/docker-image-spec v1.3.1 // indirect
|
||||
@@ -62,12 +67,14 @@ require (
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/spf13/afero v1.9.5 // indirect
|
||||
github.com/spf13/cast v1.5.1 // indirect
|
||||
github.com/spf13/cast v1.7.1 // indirect
|
||||
github.com/spf13/jwalterweatherman v1.1.0 // indirect
|
||||
github.com/spf13/pflag v1.0.5 // indirect
|
||||
github.com/subosito/gotenv v1.4.2 // indirect
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
|
||||
github.com/ugorji/go/codec v1.2.11 // indirect
|
||||
github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
|
||||
github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.64.0 // indirect
|
||||
go.opentelemetry.io/otel v1.39.0 // indirect
|
||||
|
||||
+19
-4
@@ -42,6 +42,10 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03
|
||||
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
|
||||
github.com/Microsoft/go-winio v0.4.21 h1:+6mVbXh4wPzUrl1COX9A+ZCvEpYsOBZ6/+kwDnvLyro=
|
||||
github.com/Microsoft/go-winio v0.4.21/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
|
||||
github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk=
|
||||
github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg=
|
||||
github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
|
||||
github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
|
||||
github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
|
||||
github.com/bytedance/sonic v1.9.1 h1:6iJ6NqdoxCDr6mbY8h18oSO+cShGSMRGCEo7F2h0x8s=
|
||||
github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZXU064P/U=
|
||||
@@ -89,8 +93,8 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
|
||||
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
|
||||
github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0XL9UY=
|
||||
github.com/frankban/quicktest v1.14.4/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
|
||||
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
|
||||
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
|
||||
github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
|
||||
github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
|
||||
github.com/gabriel-vasile/mimetype v1.4.2 h1:w5qFW6JKBz9Y393Y4q372O9A7cUSequkh1Q7OhCmWKU=
|
||||
@@ -186,6 +190,9 @@ github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
|
||||
github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/invopop/jsonschema v0.13.0 h1:KvpoAJWEjR3uD9Kbm2HWJmqsEaHt8lBUpd0qHcIi21E=
|
||||
github.com/invopop/jsonschema v0.13.0/go.mod h1:ffZ5Km5SWWRAIN6wbDXItl95euhFz2uON45H2qjYt+0=
|
||||
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
|
||||
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
|
||||
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
|
||||
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
|
||||
@@ -208,6 +215,10 @@ github.com/leodido/go-urn v1.2.4 h1:XlAE/cm/ms7TE/VMVoduSpNBoyc2dOxHs5MZSwAN63Q=
|
||||
github.com/leodido/go-urn v1.2.4/go.mod h1:7ZrI8mTSeBSHl/UaRyKQW1qZeMgak41ANeCNaVckg+4=
|
||||
github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
|
||||
github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
|
||||
github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
|
||||
github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
|
||||
github.com/mark3labs/mcp-go v0.43.2 h1:21PUSlWWiSbUPQwXIJ5WKlETixpFpq+WBpbMGDSVy/I=
|
||||
github.com/mark3labs/mcp-go v0.43.2/go.mod h1:YnJfOL382MIWDx1kMY+2zsRHU/q78dBg9aFb8W6Thdw=
|
||||
github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
|
||||
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/mattn/go-sqlite3 v1.14.33 h1:A5blZ5ulQo2AtayQ9/limgHEkFreKj1Dv226a1K73s0=
|
||||
@@ -255,8 +266,8 @@ github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ
|
||||
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
|
||||
github.com/spf13/afero v1.9.5 h1:stMpOSZFs//0Lv29HduCmli3GUfpFoF3Y1Q/aXj/wVM=
|
||||
github.com/spf13/afero v1.9.5/go.mod h1:UBogFpq8E9Hx+xc5CNTTEpTnuHVmXDwZcZcE1eb/UhQ=
|
||||
github.com/spf13/cast v1.5.1 h1:R+kOtfhWQE6TVQzY+4D7wJLBgkdVasCEFxSUBYBYIlA=
|
||||
github.com/spf13/cast v1.5.1/go.mod h1:b9PdjNptOpzXr7Rq1q9gJML/2cdGQAo69NKzQ10KN48=
|
||||
github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y=
|
||||
github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
|
||||
github.com/spf13/jwalterweatherman v1.1.0 h1:ue6voC5bR5F8YxI5S67j9i582FU4Qvo2bmqnqMYADFk=
|
||||
github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
|
||||
github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
|
||||
@@ -284,6 +295,10 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
|
||||
github.com/ugorji/go/codec v1.2.11 h1:BMaWp1Bb6fHwEtbplGBGJ498wD+LKlNSl25MjdZY4dU=
|
||||
github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
|
||||
github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc=
|
||||
github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw=
|
||||
github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
|
||||
github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
|
||||
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
|
||||
@@ -28,6 +28,7 @@ type ContainerConfig struct {
|
||||
ProfileInfo *AgentProfileInfo
|
||||
InstanceID string
|
||||
MainRepoGitDir string // Path to main repo's .git directory (for worktrees)
|
||||
McpServers []McpServerConfig
|
||||
}
|
||||
|
||||
// ContainerManager handles Docker container lifecycle operations
|
||||
@@ -87,6 +88,18 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
|
||||
return "", nil, fmt.Errorf("agentctl health check failed: %w", err)
|
||||
}
|
||||
|
||||
// Convert MCP server configs
|
||||
var mcpServers []agentctl.McpServerConfig
|
||||
for _, mcp := range config.McpServers {
|
||||
mcpServers = append(mcpServers, agentctl.McpServerConfig{
|
||||
Name: mcp.Name,
|
||||
URL: mcp.URL,
|
||||
Type: mcp.Type,
|
||||
Command: mcp.Command,
|
||||
Args: mcp.Args,
|
||||
})
|
||||
}
|
||||
|
||||
// Create an instance via the control API (same flow as standalone mode)
|
||||
createReq := &agentctl.CreateInstanceRequest{
|
||||
ID: config.InstanceID,
|
||||
@@ -94,6 +107,7 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
|
||||
AgentCommand: "", // Agent command set via Configure endpoint later
|
||||
Env: config.Credentials,
|
||||
AutoStart: false,
|
||||
McpServers: mcpServers,
|
||||
}
|
||||
|
||||
resp, err := ctl.CreateInstance(ctx, createReq)
|
||||
|
||||
@@ -441,6 +441,25 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
|
||||
// Build environment variables
|
||||
env := m.buildEnvForRuntime(executionID, &reqWithWorktree, agentConfig)
|
||||
|
||||
// Resolve MCP servers early so they're available for protocols that need them at startup (e.g., Codex)
|
||||
acpMcpServers, err := m.resolveMcpServersWithParams(ctx, reqWithWorktree.AgentProfileID, reqWithWorktree.Metadata, agentConfig)
|
||||
if err != nil {
|
||||
m.logger.Warn("failed to resolve MCP servers for launch", zap.Error(err))
|
||||
// Continue without MCP servers - not a fatal error
|
||||
}
|
||||
|
||||
// Convert ACP MCP servers to runtime config format
|
||||
var mcpServers []McpServerConfig
|
||||
for _, srv := range acpMcpServers {
|
||||
mcpServers = append(mcpServers, McpServerConfig{
|
||||
Name: srv.Name,
|
||||
URL: srv.URL,
|
||||
Type: srv.Type,
|
||||
Command: srv.Command,
|
||||
Args: srv.Args,
|
||||
})
|
||||
}
|
||||
|
||||
// Create runtime request (agent command not included - started explicitly later)
|
||||
runtimeReq := &RuntimeCreateRequest{
|
||||
InstanceID: executionID,
|
||||
@@ -455,6 +474,7 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
|
||||
WorktreeBranch: worktreeBranch,
|
||||
MainRepoGitDir: mainRepoGitDir,
|
||||
AgentConfig: agentConfig,
|
||||
McpServers: mcpServers,
|
||||
}
|
||||
|
||||
runtimeInstance, err := m.runtime.CreateInstance(ctx, runtimeReq)
|
||||
@@ -717,14 +737,20 @@ func (m *Manager) getAgentConfigForExecution(execution *AgentExecution) (*regist
|
||||
// - applies executor-scoped transport rules, allow/deny lists, URL rewrites, and env injection,
|
||||
// - converts to ACP stdio server definitions used during session initialization.
|
||||
func (m *Manager) resolveMcpServers(ctx context.Context, execution *AgentExecution, agentConfig *registry.AgentTypeConfig) ([]agentctltypes.McpServer, error) {
|
||||
if execution == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return m.resolveMcpServersWithParams(ctx, execution.AgentProfileID, execution.Metadata, agentConfig)
|
||||
}
|
||||
|
||||
// resolveMcpServersWithParams resolves MCP servers with explicit parameters.
|
||||
// This is used by Launch() before the execution object exists.
|
||||
func (m *Manager) resolveMcpServersWithParams(ctx context.Context, profileID string, metadata map[string]interface{}, agentConfig *registry.AgentTypeConfig) ([]agentctltypes.McpServer, error) {
|
||||
if m.mcpProvider == nil || agentConfig == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
profileID := ""
|
||||
if execution != nil {
|
||||
profileID = strings.TrimSpace(execution.AgentProfileID)
|
||||
}
|
||||
profileID = strings.TrimSpace(profileID)
|
||||
if profileID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
@@ -745,11 +771,11 @@ func (m *Manager) resolveMcpServers(ctx context.Context, execution *AgentExecuti
|
||||
|
||||
policy := mcpconfig.DefaultPolicyForRuntime(runtimeName(m.runtime))
|
||||
executorID := ""
|
||||
if execution != nil && execution.Metadata != nil {
|
||||
if value, ok := execution.Metadata["executor_id"].(string); ok {
|
||||
if metadata != nil {
|
||||
if value, ok := metadata["executor_id"].(string); ok {
|
||||
executorID = value
|
||||
}
|
||||
if value, ok := execution.Metadata["executor_mcp_policy"]; ok {
|
||||
if value, ok := metadata["executor_mcp_policy"]; ok {
|
||||
updated, policyWarnings, err := mcpconfig.ApplyExecutorPolicy(policy, value)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid executor MCP policy: %w", err)
|
||||
|
||||
@@ -34,6 +34,15 @@ type Runtime interface {
|
||||
RecoverInstances(ctx context.Context) ([]*RuntimeInstance, error)
|
||||
}
|
||||
|
||||
// McpServerConfig holds configuration for an MCP server.
|
||||
type McpServerConfig struct {
|
||||
Name string `json:"name"`
|
||||
URL string `json:"url,omitempty"`
|
||||
Type string `json:"type,omitempty"`
|
||||
Command string `json:"command,omitempty"`
|
||||
Args []string `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// RuntimeCreateRequest contains parameters for creating an agentctl instance.
|
||||
type RuntimeCreateRequest struct {
|
||||
InstanceID string
|
||||
@@ -44,6 +53,7 @@ type RuntimeCreateRequest struct {
|
||||
Protocol string
|
||||
Env map[string]string
|
||||
Metadata map[string]interface{}
|
||||
McpServers []McpServerConfig
|
||||
// Docker-specific
|
||||
AgentConfig *registry.AgentTypeConfig
|
||||
MainRepoGitDir string
|
||||
|
||||
@@ -52,6 +52,7 @@ func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRe
|
||||
InstanceID: req.InstanceID,
|
||||
MainRepoGitDir: req.MainRepoGitDir,
|
||||
Credentials: req.Env, // Env contains credentials from the caller
|
||||
McpServers: req.McpServers,
|
||||
}
|
||||
|
||||
// Use ContainerManager to launch container
|
||||
|
||||
@@ -79,6 +79,18 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea
|
||||
env["KANDEV_TASK_ID"] = req.TaskID
|
||||
env["KANDEV_SESSION_ID"] = req.SessionID
|
||||
|
||||
// Convert MCP server configs
|
||||
var mcpServers []agentctl.McpServerConfig
|
||||
for _, mcp := range req.McpServers {
|
||||
mcpServers = append(mcpServers, agentctl.McpServerConfig{
|
||||
Name: mcp.Name,
|
||||
URL: mcp.URL,
|
||||
Type: mcp.Type,
|
||||
Command: mcp.Command,
|
||||
Args: mcp.Args,
|
||||
})
|
||||
}
|
||||
|
||||
// Create instance via control API
|
||||
// Agent command is NOT set - workspace access only. Agent is started explicitly via agentctl client.
|
||||
createReq := &agentctl.CreateInstanceRequest{
|
||||
@@ -88,6 +100,7 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea
|
||||
Protocol: req.Protocol,
|
||||
Env: env,
|
||||
AutoStart: false,
|
||||
McpServers: mcpServers,
|
||||
}
|
||||
|
||||
resp, err := r.ctl.CreateInstance(ctx, createReq)
|
||||
|
||||
@@ -2,19 +2,26 @@ package mcpconfig
|
||||
|
||||
import "github.com/kandev/kandev/internal/agentctl/types"
|
||||
|
||||
// ToACPServers converts resolved MCP servers into ACP stdio server list because
|
||||
// ACP currently accepts only stdio MCP definitions in session/new.
|
||||
// ToACPServers converts resolved MCP servers into ACP server list.
|
||||
// Supports both stdio and SSE transports.
|
||||
func ToACPServers(resolved []ResolvedServer) []types.McpServer {
|
||||
servers := make([]types.McpServer, 0, len(resolved))
|
||||
for _, server := range resolved {
|
||||
if server.Type != ServerTypeStdio {
|
||||
continue
|
||||
switch server.Type {
|
||||
case ServerTypeStdio:
|
||||
servers = append(servers, types.McpServer{
|
||||
Name: server.Name,
|
||||
Type: "stdio",
|
||||
Command: server.Command,
|
||||
Args: append([]string{}, server.Args...),
|
||||
})
|
||||
case ServerTypeSSE:
|
||||
servers = append(servers, types.McpServer{
|
||||
Name: server.Name,
|
||||
Type: "sse",
|
||||
URL: server.URL,
|
||||
})
|
||||
}
|
||||
servers = append(servers, types.McpServer{
|
||||
Name: server.Name,
|
||||
Command: server.Command,
|
||||
Args: append([]string{}, server.Args...),
|
||||
})
|
||||
}
|
||||
return servers
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
"id": "auggie-agent",
|
||||
"name": "Augment Coding Agent",
|
||||
"display_name": "Auggie",
|
||||
"description": "Auggie CLI-powered autonomous coding agent. Requires AUGMENT_SESSION_AUTH for authentication.",
|
||||
"description": "Auggie CLI-powered autonomous coding agent.",
|
||||
"enabled": true,
|
||||
"image": "kandev/augment-agent",
|
||||
"tag": "latest",
|
||||
@@ -29,7 +29,7 @@
|
||||
"workspace_flag": "--workspace-root",
|
||||
"session_config": {
|
||||
"resume_via_acp": false,
|
||||
"resume_flag": "",
|
||||
"resume_flag": "--resume",
|
||||
"can_recover": false,
|
||||
"session_dir_template": "{home}/.augment/sessions",
|
||||
"session_dir_target": "/root/.augment/sessions"
|
||||
@@ -45,6 +45,11 @@
|
||||
},
|
||||
"discovery": {
|
||||
"supports_mcp": true,
|
||||
"mcp_config_path": {
|
||||
"linux": [],
|
||||
"windows": [],
|
||||
"macos": []
|
||||
},
|
||||
"installation_path": {
|
||||
"linux": ["~/.augment/.auggie.json"],
|
||||
"windows": ["~/.augment/.auggie.json"],
|
||||
@@ -141,7 +146,12 @@
|
||||
"tools_requiring_permission": []
|
||||
},
|
||||
"discovery": {
|
||||
"supports_mcp": false,
|
||||
"supports_mcp": true,
|
||||
"mcp_config_path": {
|
||||
"linux": [],
|
||||
"windows": [],
|
||||
"macos": []
|
||||
},
|
||||
"installation_path": {
|
||||
"linux": ["~/.codex/auth.json"],
|
||||
"windows": ["~/.codex/auth.json"],
|
||||
@@ -189,3 +199,4 @@
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@@ -23,6 +23,15 @@ type ControlClient struct {
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
// McpServerConfig holds configuration for an MCP server.
|
||||
type McpServerConfig struct {
|
||||
Name string `json:"name"`
|
||||
URL string `json:"url,omitempty"`
|
||||
Type string `json:"type,omitempty"`
|
||||
Command string `json:"command,omitempty"`
|
||||
Args []string `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// CreateInstanceRequest contains the parameters for creating a new agent instance.
|
||||
type CreateInstanceRequest struct {
|
||||
ID string `json:"id,omitempty"`
|
||||
@@ -32,6 +41,7 @@ type CreateInstanceRequest struct {
|
||||
WorkspaceFlag string `json:"workspace_flag,omitempty"` // CLI flag for workspace path (e.g., "--workspace-root")
|
||||
Env map[string]string `json:"env,omitempty"`
|
||||
AutoStart bool `json:"auto_start,omitempty"`
|
||||
McpServers []McpServerConfig `json:"mcp_servers,omitempty"`
|
||||
}
|
||||
|
||||
// CreateInstanceResponse contains the result of creating a new agent instance.
|
||||
|
||||
@@ -17,12 +17,12 @@ import (
|
||||
// ACPAdapter implements AgentAdapter for agents using the ACP protocol.
|
||||
// ACP (Agent Communication Protocol) uses JSON-RPC 2.0 over stdin/stdout.
|
||||
// The subprocess is managed externally (by process.Manager) and stdin/stdout
|
||||
// are passed to the adapter.
|
||||
// are connected via the Connect method after the process starts.
|
||||
type ACPAdapter struct {
|
||||
cfg *Config
|
||||
logger *logger.Logger
|
||||
|
||||
// Subprocess stdin/stdout (managed externally)
|
||||
// Subprocess stdin/stdout (set via Connect)
|
||||
stdin io.Writer
|
||||
stdout io.Reader
|
||||
|
||||
@@ -46,17 +46,35 @@ type ACPAdapter struct {
|
||||
}
|
||||
|
||||
// NewACPAdapter creates a new ACP protocol adapter.
|
||||
// stdin and stdout are the subprocess's stdin/stdout pipes (managed by process.Manager).
|
||||
func NewACPAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.Logger) *ACPAdapter {
|
||||
// Call Connect() after starting the subprocess to wire up stdin/stdout.
|
||||
func NewACPAdapter(cfg *Config, log *logger.Logger) *ACPAdapter {
|
||||
return &ACPAdapter{
|
||||
cfg: cfg,
|
||||
logger: log.WithFields(zap.String("adapter", "acp")),
|
||||
stdin: stdin,
|
||||
stdout: stdout,
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
}
|
||||
}
|
||||
|
||||
// PrepareEnvironment is a no-op for ACP.
|
||||
// ACP passes MCP servers through the protocol during session creation.
|
||||
func (a *ACPAdapter) PrepareEnvironment() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Connect wires up the stdin/stdout pipes from the running agent subprocess.
|
||||
func (a *ACPAdapter) Connect(stdin io.Writer, stdout io.Reader) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.stdin != nil || a.stdout != nil {
|
||||
return fmt.Errorf("adapter already connected")
|
||||
}
|
||||
|
||||
a.stdin = stdin
|
||||
a.stdout = stdout
|
||||
return nil
|
||||
}
|
||||
|
||||
// Initialize establishes the ACP connection with the agent subprocess.
|
||||
// The subprocess should already be running (started by process.Manager).
|
||||
func (a *ACPAdapter) Initialize(ctx context.Context) error {
|
||||
@@ -137,13 +155,25 @@ func toACPMcpServers(servers []types.McpServer) []acp.McpServer {
|
||||
}
|
||||
out := make([]acp.McpServer, 0, len(servers))
|
||||
for _, server := range servers {
|
||||
out = append(out, acp.McpServer{
|
||||
Stdio: &acp.McpServerStdio{
|
||||
Name: server.Name,
|
||||
Command: server.Command,
|
||||
Args: append([]string{}, server.Args...),
|
||||
},
|
||||
})
|
||||
switch server.Type {
|
||||
case "sse":
|
||||
out = append(out, acp.McpServer{
|
||||
Sse: &acp.McpServerSse{
|
||||
Name: server.Name,
|
||||
Url: server.URL,
|
||||
Type: "sse",
|
||||
Headers: []acp.HttpHeader{},
|
||||
},
|
||||
})
|
||||
default: // stdio
|
||||
out = append(out, acp.McpServer{
|
||||
Stdio: &acp.McpServerStdio{
|
||||
Name: server.Name,
|
||||
Command: server.Command,
|
||||
Args: append([]string{}, server.Args...),
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/types"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
@@ -47,7 +48,26 @@ type AgentInfo struct {
|
||||
// AgentAdapter defines the interface for protocol adapters.
|
||||
// Each adapter translates a specific protocol (ACP, REST, MCP, etc.) into the
|
||||
// normalized AgentEvent format that agentctl exposes via its HTTP API.
|
||||
//
|
||||
// Lifecycle:
|
||||
// 1. Create adapter with New*Adapter(cfg, logger)
|
||||
// 2. Call PrepareEnvironment() before starting the agent process
|
||||
// 3. Start the agent process and obtain stdin/stdout pipes
|
||||
// 4. Call Connect(stdin, stdout) to wire up communication
|
||||
// 5. Call Initialize() to establish protocol handshake
|
||||
// 6. Use NewSession/LoadSession/Prompt/Cancel for agent interactions
|
||||
// 7. Call Close() when done
|
||||
type AgentAdapter interface {
|
||||
// PrepareEnvironment performs protocol-specific setup before the agent process starts.
|
||||
// For Codex, this writes MCP servers to ~/.codex/config.toml.
|
||||
// For ACP, this is a no-op (MCP servers are passed through the protocol).
|
||||
// Must be called before the agent subprocess is started.
|
||||
PrepareEnvironment() error
|
||||
|
||||
// Connect wires up the stdin/stdout pipes from the running agent subprocess.
|
||||
// Must be called after the subprocess is started and before Initialize.
|
||||
Connect(stdin io.Writer, stdout io.Reader) error
|
||||
|
||||
// Initialize establishes the connection with the agent and exchanges capabilities.
|
||||
// For subprocess-based agents (ACP), this sends the initialize request.
|
||||
// For HTTP-based agents (REST), this might do a health check.
|
||||
@@ -89,6 +109,20 @@ type AgentAdapter interface {
|
||||
Close() error
|
||||
}
|
||||
|
||||
// McpServerConfig holds configuration for an MCP server.
|
||||
type McpServerConfig struct {
|
||||
// Name is the human-readable name of the MCP server
|
||||
Name string `json:"name"`
|
||||
// URL is the URL for HTTP/SSE transport
|
||||
URL string `json:"url,omitempty"`
|
||||
// Type is the transport type: "sse" or "http"
|
||||
Type string `json:"type,omitempty"`
|
||||
// Command is the command for stdio transport
|
||||
Command string `json:"command,omitempty"`
|
||||
// Args are the arguments for stdio transport
|
||||
Args []string `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// Config holds configuration for creating adapters
|
||||
type Config struct {
|
||||
// WorkDir is the working directory for the agent
|
||||
@@ -102,6 +136,9 @@ type Config struct {
|
||||
// Defaults to "on-request" if empty.
|
||||
ApprovalPolicy string
|
||||
|
||||
// McpServers is a list of MCP servers to configure for the agent
|
||||
McpServers []McpServerConfig
|
||||
|
||||
// For HTTP-based adapters (REST)
|
||||
BaseURL string // Base URL of the agent's HTTP API
|
||||
AuthHeader string // Optional auth header name
|
||||
|
||||
@@ -286,6 +286,64 @@ func TestPlanEntry_StatusValues(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertSSEToStreamableHTTP(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
input string
|
||||
expected string
|
||||
}{
|
||||
{
|
||||
name: "standard SSE endpoint",
|
||||
input: "http://localhost:9090/sse",
|
||||
expected: "http://localhost:9090/mcp",
|
||||
},
|
||||
{
|
||||
name: "SSE endpoint with different port",
|
||||
input: "http://example.com:8080/sse",
|
||||
expected: "http://example.com:8080/mcp",
|
||||
},
|
||||
{
|
||||
name: "trailing slash not converted",
|
||||
input: "http://localhost:9090/sse/",
|
||||
expected: "http://localhost:9090/sse/",
|
||||
},
|
||||
{
|
||||
name: "non-SSE endpoint unchanged",
|
||||
input: "http://localhost:9090/other",
|
||||
expected: "http://localhost:9090/other",
|
||||
},
|
||||
{
|
||||
name: "empty string",
|
||||
input: "",
|
||||
expected: "",
|
||||
},
|
||||
{
|
||||
name: "already streamable HTTP endpoint",
|
||||
input: "http://localhost:9090/mcp",
|
||||
expected: "http://localhost:9090/mcp",
|
||||
},
|
||||
{
|
||||
name: "sse in path but not at end",
|
||||
input: "http://localhost:9090/sse/something",
|
||||
expected: "http://localhost:9090/sse/something",
|
||||
},
|
||||
{
|
||||
name: "https endpoint",
|
||||
input: "https://secure.example.com/sse",
|
||||
expected: "https://secure.example.com/mcp",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
result := convertSSEToStreamableHTTP(tc.input)
|
||||
if result != tc.expected {
|
||||
t.Errorf("convertSSEToStreamableHTTP(%q) = %q, want %q", tc.input, result, tc.expected)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionUpdate_WithComplexToolArgs(t *testing.T) {
|
||||
// Test with complex nested ToolArgs
|
||||
update := AgentEvent{
|
||||
|
||||
@@ -5,12 +5,15 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/types"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/pkg/codex"
|
||||
"github.com/pelletier/go-toml/v2"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
@@ -20,7 +23,7 @@ type CodexAdapter struct {
|
||||
cfg *Config
|
||||
logger *logger.Logger
|
||||
|
||||
// Subprocess stdin/stdout (managed externally)
|
||||
// Subprocess stdin/stdout (set via Connect)
|
||||
stdin io.Writer
|
||||
stdout io.Reader
|
||||
|
||||
@@ -64,20 +67,181 @@ type turnCompleteResult struct {
|
||||
}
|
||||
|
||||
// NewCodexAdapter creates a new Codex protocol adapter.
|
||||
// stdin and stdout are the subprocess's stdin/stdout pipes (managed by process.Manager).
|
||||
func NewCodexAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.Logger) *CodexAdapter {
|
||||
// Call Connect() after starting the subprocess to wire up stdin/stdout.
|
||||
func NewCodexAdapter(cfg *Config, log *logger.Logger) *CodexAdapter {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &CodexAdapter{
|
||||
cfg: cfg,
|
||||
logger: log.WithFields(zap.String("adapter", "codex")),
|
||||
stdin: stdin,
|
||||
stdout: stdout,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
}
|
||||
}
|
||||
|
||||
// PrepareEnvironment writes MCP servers to the Codex config file.
|
||||
// Codex reads MCP configuration from ~/.codex/config.toml at startup time.
|
||||
func (a *CodexAdapter) PrepareEnvironment() error {
|
||||
a.logger.Info("PrepareEnvironment called",
|
||||
zap.Int("mcp_server_count", len(a.cfg.McpServers)))
|
||||
for i, srv := range a.cfg.McpServers {
|
||||
a.logger.Info("MCP server config",
|
||||
zap.Int("index", i),
|
||||
zap.String("name", srv.Name),
|
||||
zap.String("url", srv.URL),
|
||||
zap.String("type", srv.Type),
|
||||
zap.String("command", srv.Command))
|
||||
}
|
||||
return WriteCodexMcpConfig(a.cfg.McpServers, "", a.logger)
|
||||
}
|
||||
|
||||
// Connect wires up the stdin/stdout pipes from the running agent subprocess.
|
||||
func (a *CodexAdapter) Connect(stdin io.Writer, stdout io.Reader) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.stdin != nil || a.stdout != nil {
|
||||
return fmt.Errorf("adapter already connected")
|
||||
}
|
||||
|
||||
a.stdin = stdin
|
||||
a.stdout = stdout
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteCodexMcpConfig writes MCP server configuration to Codex's config.toml.
|
||||
// It merges with existing config, preserving other settings and existing MCP servers.
|
||||
// Codex reads MCP servers from ~/.codex/config.toml at startup time, not through the protocol.
|
||||
// This function should be called before starting the Codex process.
|
||||
// If homeDir is empty, it uses os.UserHomeDir().
|
||||
func WriteCodexMcpConfig(mcpServers []McpServerConfig, homeDir string, log *logger.Logger) error {
|
||||
if len(mcpServers) == 0 {
|
||||
return nil // Nothing to configure
|
||||
}
|
||||
|
||||
// Determine config directory
|
||||
if homeDir == "" {
|
||||
var err error
|
||||
homeDir, err = os.UserHomeDir()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get user home directory: %w", err)
|
||||
}
|
||||
}
|
||||
configDir := filepath.Join(homeDir, ".codex")
|
||||
configPath := filepath.Join(configDir, "config.toml")
|
||||
|
||||
// Create config directory if it doesn't exist
|
||||
if err := os.MkdirAll(configDir, 0755); err != nil {
|
||||
return fmt.Errorf("failed to create codex config directory: %w", err)
|
||||
}
|
||||
|
||||
// Read existing config if it exists
|
||||
existingData, err := os.ReadFile(configPath)
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("failed to read existing codex config: %w", err)
|
||||
}
|
||||
|
||||
// Parse existing config into a generic map to preserve all fields
|
||||
var rawConfig map[string]interface{}
|
||||
if len(existingData) > 0 {
|
||||
if err := toml.Unmarshal(existingData, &rawConfig); err != nil {
|
||||
if log != nil {
|
||||
log.Warn("failed to parse existing codex config, will create new",
|
||||
zap.String("path", configPath),
|
||||
zap.Error(err))
|
||||
}
|
||||
rawConfig = make(map[string]interface{})
|
||||
}
|
||||
} else {
|
||||
rawConfig = make(map[string]interface{})
|
||||
}
|
||||
|
||||
// Get or create mcp_servers section
|
||||
mcpServersSection, ok := rawConfig["mcp_servers"].(map[string]interface{})
|
||||
if !ok {
|
||||
mcpServersSection = make(map[string]interface{})
|
||||
}
|
||||
|
||||
// Add/update our MCP servers
|
||||
for _, server := range mcpServers {
|
||||
safeName := sanitizeCodexServerName(server.Name)
|
||||
serverConfig := make(map[string]interface{})
|
||||
|
||||
if server.Type == "sse" || server.Type == "http" {
|
||||
// HTTP/SSE transport - use url field
|
||||
// Codex doesn't support SSE transport - it uses streamable HTTP which requires POST requests.
|
||||
// Convert SSE URLs (/sse) to streamable HTTP URLs (/mcp) for Codex compatibility.
|
||||
url := server.URL
|
||||
if url != "" {
|
||||
url = convertSSEToStreamableHTTP(url)
|
||||
serverConfig["url"] = url
|
||||
}
|
||||
} else {
|
||||
// STDIO transport - use command and args
|
||||
if server.Command != "" {
|
||||
serverConfig["command"] = server.Command
|
||||
}
|
||||
if len(server.Args) > 0 {
|
||||
serverConfig["args"] = server.Args
|
||||
}
|
||||
}
|
||||
|
||||
mcpServersSection[safeName] = serverConfig
|
||||
}
|
||||
|
||||
// Update the mcp_servers section in the config
|
||||
rawConfig["mcp_servers"] = mcpServersSection
|
||||
|
||||
// Marshal back to TOML
|
||||
output, err := toml.Marshal(rawConfig)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal codex config: %w", err)
|
||||
}
|
||||
|
||||
// Write config file
|
||||
if err := os.WriteFile(configPath, output, 0644); err != nil {
|
||||
return fmt.Errorf("failed to write codex config: %w", err)
|
||||
}
|
||||
|
||||
if log != nil {
|
||||
log.Info("wrote Codex MCP config",
|
||||
zap.String("path", configPath),
|
||||
zap.Int("server_count", len(mcpServers)),
|
||||
zap.Int("total_mcp_servers", len(mcpServersSection)))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sanitizeCodexServerName converts a server name to a valid TOML table name.
|
||||
// Replaces spaces and special characters with underscores.
|
||||
func sanitizeCodexServerName(name string) string {
|
||||
var result strings.Builder
|
||||
for _, r := range name {
|
||||
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '-' {
|
||||
result.WriteRune(r)
|
||||
} else {
|
||||
result.WriteRune('_')
|
||||
}
|
||||
}
|
||||
sanitized := result.String()
|
||||
if sanitized == "" {
|
||||
return "server"
|
||||
}
|
||||
return sanitized
|
||||
}
|
||||
|
||||
// convertSSEToStreamableHTTP converts an SSE endpoint URL to a streamable HTTP endpoint URL.
|
||||
// Codex doesn't support SSE transport - it uses streamable HTTP which requires POST requests.
|
||||
// This converts URLs ending in /sse to /mcp for Kandev MCP server compatibility.
|
||||
// Example: http://localhost:9090/sse -> http://localhost:9090/mcp
|
||||
func convertSSEToStreamableHTTP(url string) string {
|
||||
if strings.HasSuffix(url, "/sse") {
|
||||
return strings.TrimSuffix(url, "/sse") + "/mcp"
|
||||
}
|
||||
return url
|
||||
}
|
||||
|
||||
// Initialize establishes the Codex connection with the agent subprocess.
|
||||
func (a *CodexAdapter) Initialize(ctx context.Context) error {
|
||||
a.logger.Info("initializing Codex adapter",
|
||||
@@ -139,6 +303,9 @@ func (a *CodexAdapter) GetAgentInfo() *AgentInfo {
|
||||
}
|
||||
|
||||
// NewSession creates a new Codex thread (session).
|
||||
// Note: The mcpServers parameter is ignored because Codex reads MCP configuration from
|
||||
// ~/.codex/config.toml at startup time, not through the protocol. MCP servers are written
|
||||
// to the config file by PrepareEnvironment() before the Codex process starts.
|
||||
func (a *CodexAdapter) NewSession(ctx context.Context, _ []types.McpServer) (string, error) {
|
||||
// Check client under lock, but don't hold lock during Call() to avoid deadlock
|
||||
// with handleNotification which also needs the lock
|
||||
|
||||
@@ -69,6 +69,20 @@ type InstanceDefaults struct {
|
||||
HealthCheckInterval int
|
||||
}
|
||||
|
||||
// McpServerConfig holds configuration for an MCP server.
|
||||
type McpServerConfig struct {
|
||||
// Name is the human-readable name of the MCP server
|
||||
Name string `json:"name"`
|
||||
// URL is the URL for HTTP/SSE transport
|
||||
URL string `json:"url,omitempty"`
|
||||
// Type is the transport type: "sse" or "http"
|
||||
Type string `json:"type,omitempty"`
|
||||
// Command is the command for stdio transport
|
||||
Command string `json:"command,omitempty"`
|
||||
// Args are the arguments for stdio transport
|
||||
Args []string `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// InstanceConfig holds configuration for a single agent instance.
|
||||
// This is passed to the process manager and API server.
|
||||
type InstanceConfig struct {
|
||||
@@ -113,6 +127,9 @@ type InstanceConfig struct {
|
||||
// AgentType identifies the agent (e.g., "auggie", "codex", "claude")
|
||||
// Used for agent-specific adapter selection
|
||||
AgentType string
|
||||
|
||||
// McpServers is a list of MCP servers to configure for the agent
|
||||
McpServers []McpServerConfig
|
||||
}
|
||||
|
||||
// Load loads the configuration from environment variables.
|
||||
@@ -175,6 +192,9 @@ func (c *Config) NewInstanceConfig(port int, overrides *InstanceOverrides) *Inst
|
||||
if overrides.AgentType != "" {
|
||||
cfg.AgentType = overrides.AgentType
|
||||
}
|
||||
if len(overrides.McpServers) > 0 {
|
||||
cfg.McpServers = overrides.McpServers
|
||||
}
|
||||
}
|
||||
|
||||
// Parse agent command into args
|
||||
@@ -197,6 +217,7 @@ type InstanceOverrides struct {
|
||||
Env []string
|
||||
ApprovalPolicy string
|
||||
AgentType string
|
||||
McpServers []McpServerConfig
|
||||
}
|
||||
|
||||
// ParseCommand splits a command string into arguments
|
||||
|
||||
@@ -41,6 +41,15 @@ type Instance struct {
|
||||
server *http.Server
|
||||
}
|
||||
|
||||
// McpServerConfig holds configuration for an MCP server.
|
||||
type McpServerConfig struct {
|
||||
Name string `json:"name"`
|
||||
URL string `json:"url,omitempty"`
|
||||
Type string `json:"type,omitempty"`
|
||||
Command string `json:"command,omitempty"`
|
||||
Args []string `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// CreateRequest contains the parameters for creating a new agent instance.
|
||||
type CreateRequest struct {
|
||||
// ID is an optional identifier for the instance. If empty, one will be generated.
|
||||
@@ -64,6 +73,9 @@ type CreateRequest struct {
|
||||
|
||||
// AutoStart indicates whether to start the agent automatically after creation.
|
||||
AutoStart bool `json:"auto_start,omitempty"`
|
||||
|
||||
// McpServers is a list of MCP servers to configure for the agent.
|
||||
McpServers []McpServerConfig `json:"mcp_servers,omitempty"`
|
||||
}
|
||||
|
||||
// CreateResponse contains the result of creating a new agent instance.
|
||||
|
||||
@@ -83,12 +83,26 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea
|
||||
|
||||
// Build overrides from request
|
||||
autoStart := req.AutoStart
|
||||
|
||||
// Convert MCP server configs
|
||||
var mcpServers []config.McpServerConfig
|
||||
for _, mcp := range req.McpServers {
|
||||
mcpServers = append(mcpServers, config.McpServerConfig{
|
||||
Name: mcp.Name,
|
||||
URL: mcp.URL,
|
||||
Type: mcp.Type,
|
||||
Command: mcp.Command,
|
||||
Args: mcp.Args,
|
||||
})
|
||||
}
|
||||
|
||||
overrides := &config.InstanceOverrides{
|
||||
Protocol: agent.Protocol(req.Protocol),
|
||||
AgentCommand: agentCmd,
|
||||
WorkDir: req.WorkspacePath,
|
||||
AutoStart: &autoStart,
|
||||
Env: config.CollectAgentEnv(req.Env),
|
||||
McpServers: mcpServers,
|
||||
}
|
||||
|
||||
// Create instance config using the unified method
|
||||
|
||||
@@ -79,7 +79,8 @@ type Manager struct {
|
||||
shell *shell.Session
|
||||
|
||||
// Protocol adapter for agent communication
|
||||
adapter adapter.AgentAdapter
|
||||
adapter adapter.AgentAdapter
|
||||
adapterCfg *adapter.Config
|
||||
|
||||
// Agent event notifications (protocol-agnostic)
|
||||
updatesCh chan adapter.AgentEvent
|
||||
@@ -201,6 +202,35 @@ func (m *Manager) Start(ctx context.Context) error {
|
||||
return fmt.Errorf("failed to create stderr pipe: %w", err)
|
||||
}
|
||||
|
||||
// Build adapter config
|
||||
mcpServers := make([]adapter.McpServerConfig, len(m.cfg.McpServers))
|
||||
for i, mcp := range m.cfg.McpServers {
|
||||
mcpServers[i] = adapter.McpServerConfig{
|
||||
Name: mcp.Name,
|
||||
URL: mcp.URL,
|
||||
Type: mcp.Type,
|
||||
Command: mcp.Command,
|
||||
Args: mcp.Args,
|
||||
}
|
||||
}
|
||||
m.adapterCfg = &adapter.Config{
|
||||
WorkDir: m.cfg.WorkDir,
|
||||
AutoApprove: m.cfg.AutoApprovePermissions,
|
||||
ApprovalPolicy: m.cfg.ApprovalPolicy,
|
||||
McpServers: mcpServers,
|
||||
}
|
||||
|
||||
// Create adapter before starting the process so we can call PrepareEnvironment
|
||||
if err := m.createAdapter(); err != nil {
|
||||
m.status.Store(StatusError)
|
||||
return fmt.Errorf("failed to create adapter: %w", err)
|
||||
}
|
||||
|
||||
// Prepare protocol-specific environment before starting the process
|
||||
if err := m.adapter.PrepareEnvironment(); err != nil {
|
||||
m.logger.Warn("failed to prepare protocol environment", zap.Error(err))
|
||||
}
|
||||
|
||||
// Start the process
|
||||
m.logger.Info("starting agent process",
|
||||
zap.Strings("args", m.cfg.AgentArgs),
|
||||
@@ -214,10 +244,10 @@ func (m *Manager) Start(ctx context.Context) error {
|
||||
m.stopCh = make(chan struct{})
|
||||
m.doneCh = make(chan struct{})
|
||||
|
||||
// Create protocol adapter based on configuration
|
||||
if err := m.createAdapter(); err != nil {
|
||||
// Connect adapter to the process stdin/stdout pipes
|
||||
if err := m.adapter.Connect(m.stdin, m.stdout); err != nil {
|
||||
m.status.Store(StatusError)
|
||||
return fmt.Errorf("failed to create adapter: %w", err)
|
||||
return fmt.Errorf("failed to connect adapter: %w", err)
|
||||
}
|
||||
|
||||
// Start stderr reader and exit waiter
|
||||
@@ -307,24 +337,20 @@ func (m *Manager) Configure(command string, env map[string]string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// createAdapter creates the appropriate protocol adapter based on configuration
|
||||
// createAdapter creates the appropriate protocol adapter based on configuration.
|
||||
// This should be called before starting the process so PrepareEnvironment can run.
|
||||
func (m *Manager) createAdapter() error {
|
||||
protocol := m.cfg.Protocol
|
||||
if protocol == "" {
|
||||
protocol = agent.ProtocolACP // Default to ACP
|
||||
}
|
||||
|
||||
adapterCfg := &adapter.Config{
|
||||
WorkDir: m.cfg.WorkDir,
|
||||
AutoApprove: m.cfg.AutoApprovePermissions,
|
||||
ApprovalPolicy: m.cfg.ApprovalPolicy,
|
||||
}
|
||||
|
||||
// Use the adapter config built in Start()
|
||||
switch protocol {
|
||||
case agent.ProtocolACP:
|
||||
m.adapter = adapter.NewACPAdapter(m.stdin, m.stdout, adapterCfg, m.logger)
|
||||
m.adapter = adapter.NewACPAdapter(m.adapterCfg, m.logger)
|
||||
case agent.ProtocolCodex:
|
||||
m.adapter = adapter.NewCodexAdapter(m.stdin, m.stdout, adapterCfg, m.logger)
|
||||
m.adapter = adapter.NewCodexAdapter(m.adapterCfg, m.logger)
|
||||
default:
|
||||
return fmt.Errorf("unsupported protocol: %s", protocol)
|
||||
}
|
||||
|
||||
@@ -115,11 +115,14 @@ type PermissionHandler func(ctx context.Context, req *PermissionRequest) (*Permi
|
||||
// WorkspaceMessageType represents the type of workspace stream message
|
||||
type WorkspaceMessageType string
|
||||
|
||||
// McpServer represents a stdio MCP server configuration.
|
||||
// McpServer represents an MCP server configuration.
|
||||
// Supports both stdio (command+args) and SSE (url) transports.
|
||||
type McpServer struct {
|
||||
Name string `json:"name"`
|
||||
Command string `json:"command"`
|
||||
Args []string `json:"args,omitempty"`
|
||||
Command string `json:"command,omitempty"` // For stdio transport
|
||||
Args []string `json:"args,omitempty"` // For stdio transport
|
||||
URL string `json:"url,omitempty"` // For SSE transport
|
||||
Type string `json:"type,omitempty"` // "stdio" or "sse"
|
||||
}
|
||||
|
||||
const (
|
||||
|
||||
@@ -100,6 +100,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 in standalone mode)
|
||||
McpServerEnabled bool `mapstructure:"mcpServerEnabled"`
|
||||
|
||||
// McpServerPort is the port for the embedded 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
|
||||
McpServerURL string `mapstructure:"mcpServerUrl"`
|
||||
}
|
||||
|
||||
// ReadTimeoutDuration returns the read timeout as a time.Duration.
|
||||
@@ -152,6 +163,9 @@ func setDefaults(v *viper.Viper) {
|
||||
v.SetDefault("agent.runtime", "standalone")
|
||||
v.SetDefault("agent.standaloneHost", "localhost")
|
||||
v.SetDefault("agent.standalonePort", 9999)
|
||||
v.SetDefault("agent.mcpServerEnabled", true)
|
||||
v.SetDefault("agent.mcpServerPort", 9090)
|
||||
v.SetDefault("agent.mcpServerUrl", "") // Empty means auto-set when mcpServerEnabled is true
|
||||
|
||||
// Auth defaults
|
||||
v.SetDefault("auth.jwtSecret", "")
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
// 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
|
||||
}
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
// 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)
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
package mcpserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
|
||||
"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 Columns tool
|
||||
s.AddTool(
|
||||
mcp.NewTool("list_columns",
|
||||
mcp.WithDescription("List all columns in a board. Use this to get column IDs for creating tasks."),
|
||||
mcp.WithString("board_id",
|
||||
mcp.Required(),
|
||||
mcp.Description("The board ID to list columns from"),
|
||||
),
|
||||
),
|
||||
listColumnsHandler(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("column_id",
|
||||
mcp.Required(),
|
||||
mcp.Description("The column 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),
|
||||
)
|
||||
|
||||
log.Info("registered MCP tools", zap.Int("count", 6))
|
||||
}
|
||||
|
||||
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 listColumnsHandler(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/columns", cfg.KandevURL, boardID)
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
log.Error("failed to fetch columns", zap.Error(err))
|
||||
return mcp.NewToolResultError(fmt.Sprintf("Failed to fetch columns: %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
|
||||
}
|
||||
columnID, err := req.RequireString("column_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,
|
||||
"column_id": columnID,
|
||||
"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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -102,10 +102,13 @@ type SessionNewParams struct {
|
||||
}
|
||||
|
||||
// McpServer configuration for MCP servers
|
||||
// Supports both stdio (command+args) and remote (url+type) transports
|
||||
type McpServer struct {
|
||||
Name string `json:"name"`
|
||||
Command string `json:"command"`
|
||||
Args []string `json:"args,omitempty"`
|
||||
Name string `json:"name"`
|
||||
Command string `json:"command,omitempty"` // For stdio transport
|
||||
Args []string `json:"args,omitempty"` // For stdio transport
|
||||
URL string `json:"url,omitempty"` // For HTTP/SSE transport
|
||||
Type string `json:"type,omitempty"` // "sse" or "http" for remote transport
|
||||
}
|
||||
|
||||
// SessionNewResult from session/new method
|
||||
|
||||
Generated
+5
-2
@@ -96,7 +96,7 @@ importers:
|
||||
version: 9.39.2(jiti@2.6.1)
|
||||
eslint-config-next:
|
||||
specifier: 16.1.1
|
||||
version: 16.1.1(@typescript-eslint/parser@8.53.0(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3))(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3)
|
||||
version: 16.1.1(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3)
|
||||
tailwindcss:
|
||||
specifier: ^4.0.0
|
||||
version: 4.1.18
|
||||
@@ -293,6 +293,9 @@ importers:
|
||||
'@kandev/ui':
|
||||
specifier: workspace:*
|
||||
version: link:../packages/ui
|
||||
'@lezer/highlight':
|
||||
specifier: ^1.2.3
|
||||
version: 1.2.3
|
||||
'@tabler/icons-react':
|
||||
specifier: ^3.36.1
|
||||
version: 3.36.1(react@19.2.3)
|
||||
@@ -401,7 +404,7 @@ importers:
|
||||
version: 9.39.2(jiti@2.6.1)
|
||||
eslint-config-next:
|
||||
specifier: 16.1.1
|
||||
version: 16.1.1(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3)
|
||||
version: 16.1.1(@typescript-eslint/parser@8.53.0(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3))(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3)
|
||||
happy-dom:
|
||||
specifier: ^20.1.0
|
||||
version: 20.1.0
|
||||
|
||||
@@ -64,6 +64,10 @@ export function ProfileMcpConfigCard({
|
||||
const currentDirty = isDraft ? draftState?.dirty ?? false : mcpDirty;
|
||||
|
||||
const popularServers: Record<string, Record<string, unknown>> = {
|
||||
kandev: {
|
||||
type: 'sse',
|
||||
url: 'http://localhost:9090/sse',
|
||||
},
|
||||
playwright: {
|
||||
type: 'stdio',
|
||||
command: 'npx',
|
||||
@@ -198,6 +202,13 @@ export function ProfileMcpConfigCard({
|
||||
MCP definitions are stored in the database and resolved per executor at runtime. This does not override your local agent config.</p>
|
||||
<p className="text-xs font-medium text-muted-foreground">Popular servers</p>
|
||||
<div className="flex flex-wrap gap-2">
|
||||
<button
|
||||
type="button"
|
||||
className="text-xs rounded-full border border-muted-foreground/30 px-2 py-1 hover:bg-muted cursor-pointer"
|
||||
onClick={() => applyPopularServer('kandev')}
|
||||
>
|
||||
+ Kandev MCP
|
||||
</button>
|
||||
<button
|
||||
type="button"
|
||||
className="text-xs rounded-full border border-muted-foreground/30 px-2 py-1 hover:bg-muted cursor-pointer"
|
||||
|
||||
@@ -33,6 +33,7 @@
|
||||
"@kandev/theme": "workspace:*",
|
||||
"@kandev/types": "workspace:*",
|
||||
"@kandev/ui": "workspace:*",
|
||||
"@lezer/highlight": "^1.2.3",
|
||||
"@tabler/icons-react": "^3.36.1",
|
||||
"@tanstack/react-virtual": "^3.13.18",
|
||||
"@uiw/codemirror-theme-vscode": "^4.25.4",
|
||||
|
||||
@@ -0,0 +1,247 @@
|
||||
# Cancel/Stop Agent Turn Implementation
|
||||
|
||||
This document describes how to implement end-to-end functionality for cancelling the current agent turn for both Codex and Auggie agents.
|
||||
|
||||
## Overview
|
||||
|
||||
Users need the ability to stop an agent mid-execution without killing the entire agent process. This allows for quick iteration—cancelling a wrong path and sending a new prompt—without waiting for the agent to finish or restarting the session.
|
||||
|
||||
## Protocol Summary
|
||||
|
||||
### Auggie (ACP Protocol)
|
||||
- **Method**: `session/cancel` (notification)
|
||||
- **Transport**: JSON-RPC 2.0 over stdin/stdout
|
||||
- **Message format**:
|
||||
```json
|
||||
{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"<session-id>"}}
|
||||
```
|
||||
- **Key characteristics**:
|
||||
- It's a notification (no `id` field, no response expected)
|
||||
- Only requires `sessionId`
|
||||
- The pending `session/prompt` request returns with `{"stopReason":"cancelled"}`
|
||||
- Works immediately—agent stops current operation
|
||||
|
||||
### Codex (App-Server Protocol)
|
||||
- **Method**: `turn/interrupt` (request)
|
||||
- **Transport**: JSON-RPC variant over stdin/stdout (omits `jsonrpc` field)
|
||||
- **Message format**:
|
||||
```json
|
||||
{"id":N,"method":"turn/interrupt","params":{"threadId":"<thread-id>","turnId":"<turn-id>"}}
|
||||
```
|
||||
- **Key characteristics**:
|
||||
- It's a request (has `id` field, expects response)
|
||||
- Requires both `threadId` AND `turnId`
|
||||
- `threadId` = session ID (from `thread/start` response)
|
||||
- `turnId` = operation ID (from `turn/start` response)
|
||||
|
||||
## Current State
|
||||
|
||||
### Already Implemented ✅
|
||||
|
||||
1. **AgentAdapter interface** has `Cancel(ctx context.Context) error` method
|
||||
2. **ACPAdapter.Cancel()** is fully implemented using `acpConn.Cancel()`
|
||||
3. **CodexAdapter.Cancel()** is fully implemented using `turn/interrupt`
|
||||
4. **AgentAdapter.GetOperationID()** returns `turnID` for Codex (needed for cancel)
|
||||
5. **WebSocket action constant** `ActionAgentCancel = "agent.cancel"` exists
|
||||
|
||||
### Not Implemented ❌
|
||||
|
||||
1. **agentctl HTTP endpoint** for cancel (`POST /api/v1/acp/cancel`)
|
||||
2. **agentctl Client method** to call the cancel endpoint
|
||||
3. **Backend orchestrator/lifecycle integration** to route cancel requests
|
||||
4. **WebSocket handler** for `agent.cancel` action
|
||||
5. **Frontend UI** cancel button and hook
|
||||
|
||||
## High-Level Flow
|
||||
|
||||
```
|
||||
Frontend (Cancel Button)
|
||||
|
|
||||
| agent.cancel (WS request)
|
||||
v
|
||||
Backend (WS Gateway → Dispatcher)
|
||||
|
|
||||
| orchestrator handler
|
||||
v
|
||||
Backend (Executor → AgentManager)
|
||||
|
|
||||
| CancelAgent(ctx, sessionID)
|
||||
v
|
||||
Lifecycle Manager
|
||||
|
|
||||
| HTTP POST /api/v1/acp/cancel
|
||||
v
|
||||
agentctl (api server)
|
||||
|
|
||||
| adapter.Cancel(ctx)
|
||||
v
|
||||
Agent subprocess (ACP or Codex protocol)
|
||||
|
|
||||
v
|
||||
Agent stops current turn
|
||||
|
|
||||
| Returns stopReason=cancelled (ACP) or turn/completed with cancelled (Codex)
|
||||
v
|
||||
Prompt() returns → lifecycle emits session.cancelled event → UI updates
|
||||
```
|
||||
|
||||
## Implementation Plan
|
||||
|
||||
### Layer 1: agentctl Server (HTTP API)
|
||||
|
||||
**File**: `apps/backend/internal/agentctl/server/api/acp.go`
|
||||
|
||||
Add a new endpoint:
|
||||
- `POST /api/v1/acp/cancel` → `handleACPCancel`
|
||||
|
||||
Handler implementation:
|
||||
1. Get the adapter from process manager
|
||||
2. Call `adapter.Cancel(ctx)`
|
||||
3. Return success/error response
|
||||
|
||||
**File**: `apps/backend/internal/agentctl/server/api/server.go`
|
||||
|
||||
Register the new route in the router group.
|
||||
|
||||
### Layer 2: agentctl Client
|
||||
|
||||
**File**: `apps/backend/internal/agentctl/client/acp.go`
|
||||
|
||||
Add `Cancel(ctx context.Context) error` method:
|
||||
1. Make HTTP POST to `/api/v1/acp/cancel`
|
||||
2. Parse response and return error if failed
|
||||
|
||||
### Layer 3: Lifecycle Manager
|
||||
|
||||
**File**: `apps/backend/internal/agent/lifecycle/manager.go`
|
||||
|
||||
Add `CancelAgent(ctx context.Context, executionID string) error`:
|
||||
1. Look up execution by ID
|
||||
2. Validate execution is in running state
|
||||
3. Call `execution.agentctl.Cancel(ctx)`
|
||||
4. The running Prompt() call will return with cancelled stop reason
|
||||
|
||||
**File**: `apps/backend/internal/agent/lifecycle/session.go`
|
||||
|
||||
Handle cancelled stop reason in `SendPrompt()`:
|
||||
- When prompt returns with `stopReason=cancelled`, don't treat as error
|
||||
- Emit appropriate event for UI notification
|
||||
|
||||
### Layer 4: Executor & Orchestrator
|
||||
|
||||
**File**: `apps/backend/internal/orchestrator/executor/executor.go`
|
||||
|
||||
Add interface method to `AgentManagerClient`:
|
||||
```go
|
||||
CancelAgent(ctx context.Context, sessionID string) error
|
||||
```
|
||||
|
||||
Add `Cancel(ctx, taskID, sessionID string) error`:
|
||||
1. Look up session by ID
|
||||
2. Get agent execution ID
|
||||
3. Call `agentManager.CancelAgent(ctx, executionID)`
|
||||
|
||||
**File**: `apps/backend/cmd/kandev/adapters.go`
|
||||
|
||||
Implement `CancelAgent` on `lifecycleAdapter`:
|
||||
- Delegate to `mgr.CancelAgent()`
|
||||
|
||||
### Layer 5: WebSocket Handler
|
||||
|
||||
**File**: `apps/backend/internal/orchestrator/handlers/handlers.go`
|
||||
|
||||
Add handler for `agent.cancel`:
|
||||
```go
|
||||
func (h *Handlers) wsCancelAgent(ctx context.Context, msg *ws.Message) (*ws.Message, error)
|
||||
```
|
||||
|
||||
Request payload:
|
||||
```go
|
||||
type wsCancelAgentRequest struct {
|
||||
TaskID string `json:"task_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
}
|
||||
```
|
||||
|
||||
Register in `RegisterHandlers()`:
|
||||
```go
|
||||
d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent)
|
||||
```
|
||||
|
||||
**File**: `apps/backend/internal/orchestrator/controller/controller.go`
|
||||
|
||||
Add controller method:
|
||||
```go
|
||||
func (c *Controller) CancelAgent(ctx context.Context, req dto.CancelAgentRequest) (dto.CancelAgentResponse, error)
|
||||
```
|
||||
|
||||
**File**: `apps/backend/internal/orchestrator/controller/dto/dto.go`
|
||||
|
||||
Add request/response types:
|
||||
```go
|
||||
type CancelAgentRequest struct {
|
||||
TaskID string `json:"task_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
}
|
||||
|
||||
type CancelAgentResponse struct {
|
||||
Success bool `json:"success"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
```
|
||||
|
||||
### Layer 6: Frontend
|
||||
|
||||
**File**: `apps/web/hooks/use-session-agent.ts`
|
||||
|
||||
Add `handleCancelTurn` callback:
|
||||
```typescript
|
||||
const handleCancelTurn = useCallback(async () => {
|
||||
if (!task?.id || !agentSessionId) return;
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return;
|
||||
await client.request('agent.cancel', {
|
||||
task_id: task.id,
|
||||
session_id: agentSessionId
|
||||
});
|
||||
}, [task?.id, agentSessionId]);
|
||||
```
|
||||
|
||||
Return from hook: `{ ..., handleCancelTurn }`
|
||||
|
||||
**UI Integration** (location TBD—likely task chat panel or top bar):
|
||||
- Add "Stop" button visible when agent is running
|
||||
- Wire to `handleCancelTurn()`
|
||||
- Show loading state during cancel
|
||||
- Button disabled when no active turn
|
||||
|
||||
## Event Flow After Cancel
|
||||
|
||||
1. `adapter.Cancel()` sends protocol message
|
||||
2. Agent stops and responds with cancelled status
|
||||
3. `Prompt()` returns with `stopReason = "cancelled"`
|
||||
4. Lifecycle manager:
|
||||
- Updates execution status to READY (not COMPLETED)
|
||||
- Emits `session.agent.cancelled` or similar event
|
||||
5. Frontend receives event and updates UI:
|
||||
- Stops spinner
|
||||
- Shows "Cancelled" indicator
|
||||
- Re-enables input for new prompt
|
||||
|
||||
## Error Handling
|
||||
|
||||
| Scenario | Behavior |
|
||||
|----------|----------|
|
||||
| No active turn | Return success (no-op) |
|
||||
| Agent not running | Return error "agent not running" |
|
||||
| Session not found | Return error "session not found" |
|
||||
| Cancel timeout | Return error, but cancel still in flight |
|
||||
| Cancel during permission request | Cancel the permission, return cancelled |
|
||||
|
||||
## Testing
|
||||
|
||||
1. **Unit tests** for each layer (adapter, client, handler)
|
||||
2. **Integration test**: Start agent, send prompt, cancel mid-execution
|
||||
3. **E2E test**: UI cancel button flow
|
||||
4. **Protocol tests**: Verify correct JSON-RPC messages for ACP and Codex
|
||||
|
||||
Reference in New Issue
Block a user