From f6da05542160b10700a7761ed5a8cb23404567e4 Mon Sep 17 00:00:00 2001 From: KoheiYamashita Date: Thu, 19 Feb 2026 14:50:24 +0900 Subject: [PATCH] feat: add MCP server support with official Go SDK and Streamable HTTP Integrate Model Context Protocol (MCP) support using the official Go SDK (modelcontextprotocol/go-sdk v1.3.1), enabling both stdio (local) and Streamable HTTP (remote) MCP server connections. - Add MCP Manager with lazy startup, crash rate limiting, and idle reaping - Support stdio transport (command subprocess) and HTTP transport (remote URL) - Add MCPBridgeTool for agent tool integration - Add MCP status indicators for WebSocket real-time updates - Extend config with URL/Headers fields for HTTP transport - Include headerTransport for Bearer token authentication Co-Authored-By: Claude Opus 4.6 --- config/config.example.json | 16 ++ go.mod | 4 + go.sum | 12 + pkg/agent/context.go | 20 ++ pkg/agent/loop.go | 16 ++ pkg/agent/status.go | 24 ++ pkg/config/config.go | 23 +- pkg/mcp/http_auth.go | 18 ++ pkg/mcp/manager.go | 457 +++++++++++++++++++++++++++++++++++++ pkg/mcp/types.go | 9 + pkg/tools/mcp.go | 144 ++++++++++++ 11 files changed, 739 insertions(+), 4 deletions(-) create mode 100644 pkg/mcp/http_auth.go create mode 100644 pkg/mcp/manager.go create mode 100644 pkg/mcp/types.go create mode 100644 pkg/tools/mcp.go diff --git a/config/config.example.json b/config/config.example.json index b4ccfb781..8ab4306db 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -135,6 +135,22 @@ "api_key": "YOUR_BRAVE_API_KEY", "max_results": 5 } + }, + "mcp": { + "example-stdio": { + "command": "npx", + "args": ["-y", "@example/mcp-server"], + "description": "Example local MCP server (stdio)", + "enabled": false + }, + "example-http": { + "url": "https://mcp.example.com/mcp", + "headers": { + "Authorization": "Bearer YOUR_TOKEN" + }, + "description": "Example remote MCP server (HTTP)", + "enabled": false + } } }, "heartbeat": { diff --git a/go.mod b/go.mod index 48997b899..d0a92b628 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 github.com/larksuite/oapi-sdk-go/v3 v3.5.3 + github.com/modelcontextprotocol/go-sdk v1.3.1 github.com/mymmrac/telego v1.6.0 github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1 github.com/openai/openai-go/v3 v3.22.0 @@ -23,6 +24,9 @@ require ( require ( github.com/davecgh/go-spew v1.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.3 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 0e95bf5cd..4658d7bdf 100644 --- a/go.sum +++ b/go.sum @@ -43,6 +43,8 @@ github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8= +github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= @@ -88,6 +90,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/larksuite/oapi-sdk-go/v3 v3.5.3 h1:xvf8Dv29kBXC5/DNDCLhHkAFW8l/0LlQJimO5Zn+JUk= github.com/larksuite/oapi-sdk-go/v3 v3.5.3/go.mod h1:ZEplY+kwuIrj/nqw5uSCINNATcH3KdxSN7y+UxYY5fI= +github.com/modelcontextprotocol/go-sdk v1.3.1 h1:TfqtNKOIWN4Z1oqmPAiWDC2Jq7K9OdJaooe0teoXASI= +github.com/modelcontextprotocol/go-sdk v1.3.1/go.mod h1:DgVX498dMD8UJlseK1S5i1T4tFz2fkBk4xogC3D15nw= github.com/mymmrac/telego v1.6.0 h1:Zc8rgyHozvd/7ZgyrigyHdAF9koHYMfilYfyB6wlFC0= github.com/mymmrac/telego v1.6.0/go.mod h1:xt6ZWA8zi8KmuzryE1ImEdl9JSwjHNpM4yhC7D8hU4Y= github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= @@ -108,6 +112,10 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.3 h1:OjMgICtcSFuNvQCdwqMCv9Tg7lEOXGwm1J5RPQccx6w= +github.com/segmentio/encoding v0.5.3/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/slack-go/slack v0.17.3 h1:zV5qO3Q+WJAQ/XwbGfNFrRMaJ5T/naqaonyPV/1TP4g= github.com/slack-go/slack v0.17.3/go.mod h1:X+UqOufi3LYQHDnMG1vxf0J8asC6+WllXrVrhl8/Prk= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -146,6 +154,8 @@ github.com/valyala/fastjson v1.6.7 h1:ZE4tRy0CIkh+qDc5McjatheGX2czdn8slQjomexVpB github.com/valyala/fastjson v1.6.7/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= +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.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= @@ -233,6 +243,8 @@ golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4f golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= +golang.org/x/tools v0.34.0 h1:qIpSLOxeCYGg9TrcJokLBG4KFA6d795g0xkBkiESGlo= +golang.org/x/tools v0.34.0/go.mod h1:pAP9OwEaY1CAW3HOmg3hLZC5Z0CCmzjAF2UQMSqNARg= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 4231241a0..568e4ff1e 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -9,6 +9,7 @@ import ( "time" "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/mcp" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/tools" @@ -20,6 +21,7 @@ type ContextBuilder struct { skillsLoader *skills.SkillsLoader memory *MemoryStore tools *tools.ToolRegistry // Direct reference to tool registry + mcpManager *mcp.Manager // MCP server manager } func getGlobalConfigDir() string { @@ -60,6 +62,11 @@ func (cb *ContextBuilder) SetToolsRegistry(registry *tools.ToolRegistry) { cb.tools = registry } +// SetMCPManager sets the MCP manager for system prompt integration. +func (cb *ContextBuilder) SetMCPManager(manager *mcp.Manager) { + cb.mcpManager = manager +} + func (cb *ContextBuilder) getIdentity() string { now := time.Now().Format("2006-01-02 15:04 (Monday)") workspacePath, _ := filepath.Abs(filepath.Join(cb.workspace)) @@ -141,6 +148,19 @@ The following skills extend your capabilities. To use a skill, call the skill_re %s`, skillsSummary)) } + // MCP Servers - show summary, AI uses mcp tool to discover and call + if cb.mcpManager != nil { + mcpSummary := cb.mcpManager.BuildSummary() + if mcpSummary != "" { + parts = append(parts, fmt.Sprintf(`# MCP Servers + +The following MCP servers provide additional tools. +Use the mcp tool to discover and call server tools. + +%s`, mcpSummary)) + } + } + // Memory context memoryContext := cb.memory.GetMemoryContext() if memoryContext != "" { diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 0271cd5bd..93cd33c2c 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -23,6 +23,7 @@ import ( "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/mcp" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/session" "github.com/sipeed/picoclaw/pkg/state" @@ -45,6 +46,7 @@ type AgentLoop struct { summarizing sync.Map // Tracks which sessions are currently being summarized channelManager *channels.Manager rateLimiter *rateLimiter + mcpManager *mcp.Manager } // processOptions configures how a message is processed @@ -156,6 +158,16 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers toolsRegistry.Register(skillTool) subagentTools.Register(skillTool) + // MCP server manager and bridge tool + var mcpManager *mcp.Manager + if len(cfg.Tools.MCP) > 0 { + mcpManager = mcp.NewManager(cfg.Tools.MCP) + mcpBridgeTool := tools.NewMCPBridgeTool(mcpManager) + toolsRegistry.Register(mcpBridgeTool) + subagentTools.Register(mcpBridgeTool) + contextBuilder.SetMCPManager(mcpManager) + } + return &AgentLoop{ bus: msgBus, provider: provider, @@ -169,6 +181,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers tools: toolsRegistry, summarizing: sync.Map{}, rateLimiter: newRateLimiter(cfg.RateLimits.MaxToolCallsPerMinute, cfg.RateLimits.MaxRequestsPerMinute), + mcpManager: mcpManager, } } @@ -223,6 +236,9 @@ func (al *AgentLoop) Run(ctx context.Context) error { func (al *AgentLoop) Stop() { al.running.Store(false) + if al.mcpManager != nil { + al.mcpManager.Stop() + } } func (al *AgentLoop) RegisterTool(tool tools.Tool) { diff --git a/pkg/agent/status.go b/pkg/agent/status.go index 43b5ba122..89e7ff0f7 100644 --- a/pkg/agent/status.go +++ b/pkg/agent/status.go @@ -56,6 +56,8 @@ func statusLabel(toolName string, args map[string]interface{}) string { return fmt.Sprintf("サブタスク実行中...（%s）", truncLabel(l, 20)) } return "サブタスク実行中..." + case "mcp": + return mcpStatusLabel(args) case "i2c": return i2cStatusLabel(args) case "spi": @@ -114,6 +116,28 @@ func cronStatusLabel(args map[string]interface{}) string { } } +func mcpStatusLabel(args map[string]interface{}) string { + switch strArg(args, "action") { + case "mcp_list": + return "MCPサーバー一覧取得中..." + case "mcp_tools": + if s := strArg(args, "server"); s != "" { + return fmt.Sprintf("MCPツール取得中...（%s）", s) + } + return "MCPツール取得中..." + case "mcp_call": + if t := strArg(args, "tool"); t != "" { + if s := strArg(args, "server"); s != "" { + return fmt.Sprintf("MCPツール実行中...（%s/%s）", s, t) + } + return fmt.Sprintf("MCPツール実行中...（%s）", t) + } + return "MCPツール実行中..." + default: + return "MCP操作中..." + } +} + func i2cStatusLabel(args map[string]interface{}) string { switch strArg(args, "action") { case "detect": diff --git a/pkg/config/config.go b/pkg/config/config.go index 259293902..bab5508a9 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -239,11 +239,26 @@ type SPIToolsConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_SPI_ENABLED"` } +type MCPServerConfig struct { + // Stdio transport + Command string `json:"command,omitempty"` + Args []string `json:"args,omitempty"` + Env map[string]string `json:"env,omitempty"` + // HTTP transport + URL string `json:"url,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + // Common + Description string `json:"description"` + Enabled bool `json:"enabled"` + IdleTimeout int `json:"idle_timeout,omitempty"` // seconds, default 300 +} + type ToolsConfig struct { - Web WebToolsConfig `json:"web"` - Exec ExecToolsConfig `json:"exec"` - I2C I2CToolsConfig `json:"i2c"` - SPI SPIToolsConfig `json:"spi"` + Web WebToolsConfig `json:"web"` + Exec ExecToolsConfig `json:"exec"` + I2C I2CToolsConfig `json:"i2c"` + SPI SPIToolsConfig `json:"spi"` + MCP map[string]MCPServerConfig `json:"mcp,omitempty"` } func DefaultConfig() *Config { diff --git a/pkg/mcp/http_auth.go b/pkg/mcp/http_auth.go new file mode 100644 index 000000000..a8428688f --- /dev/null +++ b/pkg/mcp/http_auth.go @@ -0,0 +1,18 @@ +package mcp + +import "net/http" + +// headerTransport wraps an http.RoundTripper to inject custom headers +// (e.g., Authorization: Bearer) into every outgoing request. +type headerTransport struct { + headers map[string]string + base http.RoundTripper +} + +func (t *headerTransport) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + for k, v := range t.headers { + req.Header.Set(k, v) + } + return t.base.RoundTrip(req) +} diff --git a/pkg/mcp/manager.go b/pkg/mcp/manager.go new file mode 100644 index 000000000..efd2a0c74 --- /dev/null +++ b/pkg/mcp/manager.go @@ -0,0 +1,457 @@ +package mcp + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "os/exec" + "strings" + "sync" + "time" + + sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" +) + +// ServerInstance manages a connected MCP server session. +type ServerInstance struct { + session *sdkmcp.ClientSession + done chan struct{} // closed when session ends; created once per session + tools []*sdkmcp.Tool + lastUsed time.Time + crashes []time.Time // track crash times for rate limiting + isHTTP bool + mu sync.Mutex +} + +// Manager manages the lifecycle of MCP servers. +type Manager struct { + configs map[string]config.MCPServerConfig + servers map[string]*ServerInstance + mu sync.RWMutex + stopCh chan struct{} + wg sync.WaitGroup +} + +// NewManager creates a new MCP Manager and starts the idle reaper. +func NewManager(configs map[string]config.MCPServerConfig) *Manager { + if configs == nil { + configs = make(map[string]config.MCPServerConfig) + } + + m := &Manager{ + configs: configs, + servers: make(map[string]*ServerInstance), + stopCh: make(chan struct{}), + } + + // Start idle reaper goroutine + m.wg.Add(1) + go m.idleReaper() + + return m +} + +// ListServers returns server names and descriptions without starting processes. +func (m *Manager) ListServers() []ServerSummary { + m.mu.RLock() + type entry struct { + name string + cfg config.MCPServerConfig + inst *ServerInstance + } + var entries []entry + for name, cfg := range m.configs { + if !cfg.Enabled { + continue + } + entries = append(entries, entry{name: name, cfg: cfg, inst: m.servers[name]}) + } + m.mu.RUnlock() + + var result []ServerSummary + for _, e := range entries { + status := "stopped" + if e.inst != nil { + e.inst.mu.Lock() + if e.inst.session != nil { + status = "running" + } + e.inst.mu.Unlock() + } + result = append(result, ServerSummary{ + Name: e.name, + Description: e.cfg.Description, + Status: status, + }) + } + return result +} + +// GetTools returns the tool list for a server, starting it if needed. +func (m *Manager) GetTools(ctx context.Context, serverName string) ([]*sdkmcp.Tool, error) { + inst, err := m.ensureRunning(ctx, serverName) + if err != nil { + return nil, err + } + + inst.mu.Lock() + defer inst.mu.Unlock() + + // Return cached tools if available + if len(inst.tools) > 0 { + inst.lastUsed = time.Now() + return inst.tools, nil + } + + // Fetch tools via SDK (handles pagination automatically) + result, err := inst.session.ListTools(ctx, nil) + if err != nil { + m.handleSessionError(serverName, inst, err) + return nil, fmt.Errorf("tools/list: %w", err) + } + + inst.tools = result.Tools + inst.lastUsed = time.Now() + + logger.InfoCF("mcp", fmt.Sprintf("Server %q: loaded %d tools", serverName, len(result.Tools)), + map[string]interface{}{ + "server": serverName, + "tools": len(result.Tools), + }) + + return result.Tools, nil +} + +// CallTool executes a tool on an MCP server. +func (m *Manager) CallTool(ctx context.Context, serverName, toolName string, args map[string]interface{}) (string, error) { + inst, err := m.ensureRunning(ctx, serverName) + if err != nil { + return "", err + } + + inst.mu.Lock() + defer inst.mu.Unlock() + + inst.lastUsed = time.Now() + + result, err := inst.session.CallTool(ctx, &sdkmcp.CallToolParams{ + Name: toolName, + Arguments: args, + }) + if err != nil { + m.handleSessionError(serverName, inst, err) + return "", fmt.Errorf("tools/call %s: %w", toolName, err) + } + + text := extractText(result) + + if result.IsError { + return "", fmt.Errorf("tool error: %s", text) + } + + return text, nil +} + +// BuildSummary generates XML for the system prompt using config only (no process start). +func (m *Manager) BuildSummary() string { + m.mu.RLock() + defer m.mu.RUnlock() + + var enabled []config.MCPServerConfig + var names []string + for name, cfg := range m.configs { + if cfg.Enabled { + enabled = append(enabled, cfg) + names = append(names, name) + } + } + + if len(enabled) == 0 { + return "" + } + + var sb strings.Builder + sb.WriteString("\n") + for i, cfg := range enabled { + sb.WriteString(" \n") + sb.WriteString(fmt.Sprintf(" %s\n", names[i])) + if cfg.Description != "" { + sb.WriteString(fmt.Sprintf(" %s\n", cfg.Description)) + } + transport := "stdio" + if cfg.URL != "" { + transport = "http" + } + sb.WriteString(fmt.Sprintf(" %s\n", transport)) + sb.WriteString(" \n") + } + sb.WriteString("") + + return sb.String() +} + +// Stop shuts down all running servers and the idle reaper. +func (m *Manager) Stop() { + close(m.stopCh) + + m.mu.Lock() + for name, inst := range m.servers { + inst.mu.Lock() + if inst.session != nil { + logger.InfoCF("mcp", fmt.Sprintf("Stopping server %q", name), nil) + inst.session.Close() + inst.session = nil + } + inst.mu.Unlock() + } + m.servers = make(map[string]*ServerInstance) + m.mu.Unlock() + + m.wg.Wait() +} + +// ensureRunning starts a server if not already running. +func (m *Manager) ensureRunning(ctx context.Context, serverName string) (*ServerInstance, error) { + m.mu.RLock() + cfg, ok := m.configs[serverName] + m.mu.RUnlock() + if !ok { + return nil, fmt.Errorf("unknown MCP server: %q", serverName) + } + if !cfg.Enabled { + return nil, fmt.Errorf("MCP server %q is disabled", serverName) + } + + m.mu.Lock() + inst, exists := m.servers[serverName] + if !exists { + inst = &ServerInstance{} + m.servers[serverName] = inst + } + m.mu.Unlock() + + inst.mu.Lock() + defer inst.mu.Unlock() + + // Already running — check if session is still alive + if inst.session != nil { + select { + case <-inst.done: + logger.WarnCF("mcp", fmt.Sprintf("Server %q session closed, restarting", serverName), nil) + inst.session = nil + inst.tools = nil + default: + return inst, nil + } + } + + // Check crash rate limit (max 3 in 60 seconds) + now := time.Now() + var recentCrashes []time.Time + for _, t := range inst.crashes { + if now.Sub(t) < 60*time.Second { + recentCrashes = append(recentCrashes, t) + } + } + inst.crashes = recentCrashes + if len(recentCrashes) >= 3 { + return nil, fmt.Errorf("MCP server %q crashed too frequently (3 times in 60s)", serverName) + } + + // Create SDK client + client := sdkmcp.NewClient( + &sdkmcp.Implementation{Name: "picoclaw", Version: "1.0.0"}, + nil, + ) + + // Create transport based on config + var transport sdkmcp.Transport + if cfg.URL != "" { + // HTTP (Streamable HTTP) transport + httpClient := &http.Client{} + if len(cfg.Headers) > 0 { + httpClient.Transport = &headerTransport{ + headers: cfg.Headers, + base: http.DefaultTransport, + } + } + transport = &sdkmcp.StreamableClientTransport{ + Endpoint: cfg.URL, + HTTPClient: httpClient, + } + inst.isHTTP = true + + logger.InfoCF("mcp", fmt.Sprintf("Connecting to HTTP server %q: %s", serverName, cfg.URL), + map[string]interface{}{ + "server": serverName, + "url": cfg.URL, + }) + } else { + // Stdio (Command) transport + var env []string + if len(cfg.Env) > 0 { + env = os.Environ() + for k, v := range cfg.Env { + env = append(env, fmt.Sprintf("%s=%s", k, v)) + } + } + + cmd := exec.Command(cfg.Command, cfg.Args...) + if len(env) > 0 { + cmd.Env = env + } + transport = &sdkmcp.CommandTransport{Command: cmd} + + logger.InfoCF("mcp", fmt.Sprintf("Starting server %q: %s %s", serverName, cfg.Command, strings.Join(cfg.Args, " ")), + map[string]interface{}{ + "server": serverName, + "command": cfg.Command, + }) + } + + // Connect performs the full MCP handshake (initialize + notifications/initialized) + session, err := client.Connect(ctx, transport, nil) + if err != nil { + inst.crashes = append(inst.crashes, now) + return nil, fmt.Errorf("connect MCP server %q: %w", serverName, err) + } + + inst.session = session + inst.lastUsed = now + inst.tools = nil // Clear cached tools for fresh fetch + + // Monitor session lifecycle — single goroutine per session + inst.done = make(chan struct{}) + go func() { + session.Wait() + close(inst.done) + }() + + initResult := session.InitializeResult() + logger.InfoCF("mcp", fmt.Sprintf("Server %q initialized (protocol: %s, server: %s %s)", + serverName, initResult.ProtocolVersion, initResult.ServerInfo.Name, initResult.ServerInfo.Version), + map[string]interface{}{ + "server": serverName, + "protocol": initResult.ProtocolVersion, + }) + + return inst, nil +} + +// handleSessionError records a crash and cleans up the session on transport errors. +func (m *Manager) handleSessionError(serverName string, inst *ServerInstance, err error) { + errStr := err.Error() + isTransportError := strings.Contains(errStr, "write") || strings.Contains(errStr, "read") || + strings.Contains(errStr, "pipe") || strings.Contains(errStr, "process") || + strings.Contains(errStr, "http") || strings.Contains(errStr, "connection") || + strings.Contains(errStr, "EOF") + + if isTransportError { + logger.WarnCF("mcp", fmt.Sprintf("Server %q transport error, marking for restart: %v", serverName, err), nil) + if inst.session != nil { + inst.session.Close() + inst.session = nil + } + inst.tools = nil + inst.crashes = append(inst.crashes, time.Now()) + } +} + +// idleReaper periodically checks for idle servers and stops them. +func (m *Manager) idleReaper() { + defer m.wg.Done() + + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-m.stopCh: + return + case <-ticker.C: + m.reapIdleServers() + } + } +} + +func (m *Manager) reapIdleServers() { + m.mu.RLock() + serverNames := make([]string, 0, len(m.servers)) + for name := range m.servers { + serverNames = append(serverNames, name) + } + m.mu.RUnlock() + + for _, name := range serverNames { + m.mu.RLock() + cfg := m.configs[name] + inst, ok := m.servers[name] + m.mu.RUnlock() + if !ok { + continue + } + + timeout := cfg.IdleTimeout + if timeout <= 0 { + timeout = 300 // default 5 minutes + } + + inst.mu.Lock() + if inst.session != nil && time.Since(inst.lastUsed) > time.Duration(timeout)*time.Second { + if inst.isHTTP { + logger.InfoCF("mcp", fmt.Sprintf("Closing idle HTTP session for %q (idle %v)", name, time.Since(inst.lastUsed).Round(time.Second)), nil) + } else { + logger.InfoCF("mcp", fmt.Sprintf("Stopping idle server %q (idle %v)", name, time.Since(inst.lastUsed).Round(time.Second)), nil) + } + inst.session.Close() + inst.session = nil + inst.tools = nil + } + inst.mu.Unlock() + } +} + +// extractText converts SDK content blocks and structured content into text. +func extractText(result *sdkmcp.CallToolResult) string { + var parts []string + + for _, content := range result.Content { + switch c := content.(type) { + case *sdkmcp.TextContent: + parts = append(parts, c.Text) + case *sdkmcp.ImageContent: + parts = append(parts, fmt.Sprintf("[image: %s, %d bytes]", c.MIMEType, len(c.Data))) + case *sdkmcp.AudioContent: + parts = append(parts, fmt.Sprintf("[audio: %s, %d bytes]", c.MIMEType, len(c.Data))) + case *sdkmcp.ResourceLink: + parts = append(parts, fmt.Sprintf("[resource_link: %s]", c.URI)) + case *sdkmcp.EmbeddedResource: + if c.Resource != nil { + if c.Resource.Text != "" { + parts = append(parts, c.Resource.Text) + } else if len(c.Resource.Blob) > 0 { + parts = append(parts, fmt.Sprintf("[embedded resource: %s, %s, %d bytes]", + c.Resource.URI, c.Resource.MIMEType, len(c.Resource.Blob))) + } else { + parts = append(parts, fmt.Sprintf("[embedded resource: %s]", c.Resource.URI)) + } + } + } + } + + if result.StructuredContent != nil { + if data, err := json.MarshalIndent(result.StructuredContent, "", " "); err == nil { + parts = append(parts, string(data)) + } + } + + if len(parts) == 0 { + return "(no content)" + } + + return strings.Join(parts, "\n") +} diff --git a/pkg/mcp/types.go b/pkg/mcp/types.go new file mode 100644 index 000000000..d87897d9b --- /dev/null +++ b/pkg/mcp/types.go @@ -0,0 +1,9 @@ +package mcp + +// ServerSummary is a lightweight view of a server for listing. +// This is a Manager-specific type (not part of the MCP SDK). +type ServerSummary struct { + Name string + Description string + Status string +} diff --git a/pkg/tools/mcp.go b/pkg/tools/mcp.go new file mode 100644 index 000000000..f3c5550f8 --- /dev/null +++ b/pkg/tools/mcp.go @@ -0,0 +1,144 @@ +package tools + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/sipeed/picoclaw/pkg/mcp" +) + +// MCPBridgeTool exposes MCP servers to the LLM as a single tool with actions. +type MCPBridgeTool struct { + manager *mcp.Manager +} + +// NewMCPBridgeTool creates a new MCP bridge tool. +func NewMCPBridgeTool(manager *mcp.Manager) *MCPBridgeTool { + return &MCPBridgeTool{manager: manager} +} + +func (t *MCPBridgeTool) Name() string { + return "mcp" +} + +func (t *MCPBridgeTool) Description() string { + return "Interact with MCP (Model Context Protocol) servers. Actions: mcp_list (list available servers), mcp_tools (get server's tool list), mcp_call (call a server tool)" +} + +func (t *MCPBridgeTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "action": map[string]interface{}{ + "type": "string", + "description": "The MCP action to perform", + "enum": []string{"mcp_list", "mcp_tools", "mcp_call"}, + }, + "server": map[string]interface{}{ + "type": "string", + "description": "MCP server name (required for mcp_tools and mcp_call)", + }, + "tool": map[string]interface{}{ + "type": "string", + "description": "Tool name to call (required for mcp_call)", + }, + "arguments": map[string]interface{}{ + "type": "object", + "description": "Arguments to pass to the tool (for mcp_call)", + }, + }, + "required": []string{"action"}, + } +} + +func (t *MCPBridgeTool) Execute(ctx context.Context, args map[string]interface{}) *ToolResult { + action, ok := args["action"].(string) + if !ok { + return ErrorResult("action is required") + } + + switch action { + case "mcp_list": + return t.listServers() + case "mcp_tools": + return t.getTools(ctx, args) + case "mcp_call": + return t.callTool(ctx, args) + default: + return ErrorResult(fmt.Sprintf("unknown action: %s", action)) + } +} + +func (t *MCPBridgeTool) listServers() *ToolResult { + servers := t.manager.ListServers() + if len(servers) == 0 { + return SilentResult("No MCP servers configured") + } + + var sb strings.Builder + sb.WriteString("Available MCP servers:\n") + for _, s := range servers { + sb.WriteString(fmt.Sprintf("- %s: %s [%s]\n", s.Name, s.Description, s.Status)) + } + return SilentResult(sb.String()) +} + +func (t *MCPBridgeTool) getTools(ctx context.Context, args map[string]interface{}) *ToolResult { + server, ok := args["server"].(string) + if !ok || server == "" { + return ErrorResult("server is required for mcp_tools") + } + + tools, err := t.manager.GetTools(ctx, server) + if err != nil { + return ErrorResult(fmt.Sprintf("failed to get tools from %q: %v", server, err)) + } + + if len(tools) == 0 { + return SilentResult(fmt.Sprintf("Server %q has no tools", server)) + } + + var sb strings.Builder + sb.WriteString(fmt.Sprintf("Tools from server %q:\n\n", server)) + for _, tool := range tools { + sb.WriteString(fmt.Sprintf("## %s\n", tool.Name)) + if tool.Description != "" { + sb.WriteString(fmt.Sprintf("%s\n", tool.Description)) + } + if tool.InputSchema != nil { + schema, _ := json.MarshalIndent(tool.InputSchema, "", " ") + sb.WriteString(fmt.Sprintf("Input schema:\n```json\n%s\n```\n", string(schema))) + } + sb.WriteString("\n") + } + return SilentResult(sb.String()) +} + +func (t *MCPBridgeTool) callTool(ctx context.Context, args map[string]interface{}) *ToolResult { + server, ok := args["server"].(string) + if !ok || server == "" { + return ErrorResult("server is required for mcp_call") + } + + toolName, ok := args["tool"].(string) + if !ok || toolName == "" { + return ErrorResult("tool is required for mcp_call") + } + + // Extract arguments (optional) + var toolArgs map[string]interface{} + if a, ok := args["arguments"]; ok && a != nil { + if m, ok := a.(map[string]interface{}); ok { + toolArgs = m + } + } + + result, err := t.manager.CallTool(ctx, server, toolName, toolArgs) + if err != nil { + return ErrorResult(fmt.Sprintf("mcp_call %s/%s failed: %v", server, toolName, err)) + } + + return SilentResult(result) +}