feat: mcp servers + executors (#60)
* feat: mcp servers * feat: mcp executors * feat: mcp per agent profile * feat: improve side bar * feat: add reactive executors crud
This commit is contained in:
@@ -14,9 +14,12 @@ import (
|
||||
"github.com/google/uuid"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/mcpconfig"
|
||||
"github.com/kandev/kandev/internal/agent/registry"
|
||||
"github.com/kandev/kandev/internal/agent/runtime"
|
||||
"github.com/kandev/kandev/internal/agent/worktree"
|
||||
agentctl "github.com/kandev/kandev/internal/agentctl/client"
|
||||
agentctltypes "github.com/kandev/kandev/internal/agentctl/types"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/events"
|
||||
@@ -31,6 +34,7 @@ type Manager struct {
|
||||
credsMgr CredentialsManager
|
||||
profileResolver ProfileResolver
|
||||
worktreeMgr *worktree.Manager
|
||||
mcpProvider McpConfigProvider
|
||||
logger *logger.Logger
|
||||
|
||||
// Agent runtime abstraction (Docker, Standalone, K8s, SSH, etc.)
|
||||
@@ -63,6 +67,7 @@ func NewManager(
|
||||
containerManager *ContainerManager,
|
||||
credsMgr CredentialsManager,
|
||||
profileResolver ProfileResolver,
|
||||
mcpProvider McpConfigProvider,
|
||||
log *logger.Logger,
|
||||
) *Manager {
|
||||
componentLogger := log.WithFields(zap.String("component", "lifecycle-manager"))
|
||||
@@ -85,6 +90,7 @@ func NewManager(
|
||||
runtime: runtime,
|
||||
credsMgr: credsMgr,
|
||||
profileResolver: profileResolver,
|
||||
mcpProvider: mcpProvider,
|
||||
logger: componentLogger,
|
||||
executionStore: executionStore,
|
||||
commandBuilder: commandBuilder,
|
||||
@@ -108,7 +114,7 @@ func NewManager(
|
||||
sessionManager.SetDependencies(eventPublisher, mgr.streamManager, executionStore)
|
||||
|
||||
if runtime != nil {
|
||||
mgr.logger.Info("initialized with runtime", zap.String("runtime", runtime.Name()))
|
||||
mgr.logger.Info("initialized with runtime", zap.String("runtime", string(runtime.Name())))
|
||||
}
|
||||
|
||||
return mgr
|
||||
@@ -204,7 +210,7 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work
|
||||
zap.String("execution_id", executionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("workspace_path", info.WorkspacePath),
|
||||
zap.String("runtime", m.runtime.Name()))
|
||||
zap.String("runtime", string(m.runtime.Name())))
|
||||
|
||||
return execution, nil
|
||||
}
|
||||
@@ -213,7 +219,7 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work
|
||||
func (m *Manager) Start(ctx context.Context) error {
|
||||
runtimeName := "none"
|
||||
if m.runtime != nil {
|
||||
runtimeName = m.runtime.Name()
|
||||
runtimeName = string(m.runtime.Name())
|
||||
}
|
||||
m.logger.Info("starting lifecycle manager", zap.String("runtime", runtimeName))
|
||||
|
||||
@@ -491,7 +497,7 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
|
||||
|
||||
runtimeName := "unknown"
|
||||
if m.runtime != nil {
|
||||
runtimeName = m.runtime.Name()
|
||||
runtimeName = string(m.runtime.Name())
|
||||
}
|
||||
m.logger.Info("agentctl execution created (agent not started)",
|
||||
zap.String("execution_id", executionID),
|
||||
@@ -563,8 +569,14 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
|
||||
return fmt.Errorf("failed to get agent config: %w", err)
|
||||
}
|
||||
|
||||
mcpServers, err := m.resolveMcpServers(ctx, execution, agentConfig)
|
||||
if err != nil {
|
||||
m.updateExecutionError(executionID, "failed to resolve MCP config: "+err.Error())
|
||||
return fmt.Errorf("failed to resolve MCP config: %w", err)
|
||||
}
|
||||
|
||||
// Initialize ACP session
|
||||
if err := m.initializeACPSession(ctx, execution, agentConfig, taskDescription); err != nil {
|
||||
if err := m.initializeACPSession(ctx, execution, agentConfig, taskDescription, mcpServers); err != nil {
|
||||
m.updateExecutionError(executionID, "failed to initialize ACP: "+err.Error())
|
||||
return fmt.Errorf("failed to initialize ACP: %w", err)
|
||||
}
|
||||
@@ -700,9 +712,81 @@ func (m *Manager) getAgentConfigForExecution(execution *AgentExecution) (*regist
|
||||
return agentConfig, nil
|
||||
}
|
||||
|
||||
// resolveMcpServers centralizes MCP resolution for a session:
|
||||
// - loads per-agent MCP config,
|
||||
// - 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 m.mcpProvider == nil || agentConfig == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
profileID := ""
|
||||
if execution != nil {
|
||||
profileID = strings.TrimSpace(execution.AgentProfileID)
|
||||
}
|
||||
if profileID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
config, err := m.mcpProvider.GetConfigByProfileID(ctx, profileID)
|
||||
if err != nil {
|
||||
if errors.Is(err, mcpconfig.ErrAgentMcpUnsupported) || errors.Is(err, mcpconfig.ErrAgentProfileNotFound) {
|
||||
return nil, nil
|
||||
}
|
||||
m.logger.Warn("failed to load MCP config",
|
||||
zap.String("profile_id", profileID),
|
||||
zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
if config == nil || !config.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
policy := mcpconfig.DefaultPolicyForRuntime(runtimeName(m.runtime))
|
||||
executorID := ""
|
||||
if execution != nil && execution.Metadata != nil {
|
||||
if value, ok := execution.Metadata["executor_id"].(string); ok {
|
||||
executorID = value
|
||||
}
|
||||
if value, ok := execution.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)
|
||||
}
|
||||
policy = updated
|
||||
for _, warning := range policyWarnings {
|
||||
m.logger.Warn("mcp policy warning",
|
||||
zap.String("profile_id", profileID),
|
||||
zap.String("executor_id", executorID),
|
||||
zap.String("warning", warning))
|
||||
}
|
||||
}
|
||||
}
|
||||
resolved, warnings, err := mcpconfig.Resolve(config, policy)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, warning := range warnings {
|
||||
m.logger.Warn("mcp config warning",
|
||||
zap.String("profile_id", profileID),
|
||||
zap.String("executor_id", executorID),
|
||||
zap.String("warning", warning))
|
||||
}
|
||||
|
||||
return mcpconfig.ToACPServers(resolved), nil
|
||||
}
|
||||
|
||||
func runtimeName(rt Runtime) runtime.Name {
|
||||
if rt == nil {
|
||||
return runtime.NameUnknown
|
||||
}
|
||||
return rt.Name()
|
||||
}
|
||||
|
||||
// initializeACPSession delegates to SessionManager for full ACP session initialization and prompting
|
||||
func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExecution, agentConfig *registry.AgentTypeConfig, taskDescription string) error {
|
||||
return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, m.MarkReady)
|
||||
func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExecution, agentConfig *registry.AgentTypeConfig, taskDescription string, mcpServers []agentctltypes.McpServer) error {
|
||||
return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, mcpServers, m.MarkReady)
|
||||
}
|
||||
|
||||
// emitSessionStatusEvent emits a session_status event indicating whether the session was resumed or new.
|
||||
@@ -949,7 +1033,7 @@ func (m *Manager) StopAgent(ctx context.Context, executionID string, force bool)
|
||||
|
||||
runtimeName := "unknown"
|
||||
if m.runtime != nil {
|
||||
runtimeName = m.runtime.Name()
|
||||
runtimeName = string(m.runtime.Name())
|
||||
}
|
||||
m.logger.Info("stopping agent",
|
||||
zap.String("execution_id", executionID),
|
||||
|
||||
Reference in New Issue
Block a user