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 <noreply@anthropic.com>
This commit is contained in:
KoheiYamashita
parent
836fb6e864
commit
f6da055421
11 changed files with 739 additions and 4 deletions
|
|
@ -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": {
|
||||
|
|
|
|||
4
go.mod
4
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
|
||||
)
|
||||
|
||||
|
|
|
|||
12
go.sum
12
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=
|
||||
|
|
|
|||
|
|
@ -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 != "" {
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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":
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
MCP map[string]MCPServerConfig `json:"mcp,omitempty"`
|
||||
}
|
||||
|
||||
func DefaultConfig() *Config {
|
||||
|
|
|
|||
18
pkg/mcp/http_auth.go
Normal file
18
pkg/mcp/http_auth.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
457
pkg/mcp/manager.go
Normal file
457
pkg/mcp/manager.go
Normal file
|
|
@ -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("<mcp_servers>\n")
|
||||
for i, cfg := range enabled {
|
||||
sb.WriteString(" <server>\n")
|
||||
sb.WriteString(fmt.Sprintf(" <name>%s</name>\n", names[i]))
|
||||
if cfg.Description != "" {
|
||||
sb.WriteString(fmt.Sprintf(" <description>%s</description>\n", cfg.Description))
|
||||
}
|
||||
transport := "stdio"
|
||||
if cfg.URL != "" {
|
||||
transport = "http"
|
||||
}
|
||||
sb.WriteString(fmt.Sprintf(" <transport>%s</transport>\n", transport))
|
||||
sb.WriteString(" </server>\n")
|
||||
}
|
||||
sb.WriteString("</mcp_servers>")
|
||||
|
||||
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")
|
||||
}
|
||||
9
pkg/mcp/types.go
Normal file
9
pkg/mcp/types.go
Normal file
|
|
@ -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
|
||||
}
|
||||
144
pkg/tools/mcp.go
Normal file
144
pkg/tools/mcp.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue