forked from nextlevelbuilder/goclaw
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathloop_mcp_user.go
More file actions
118 lines (105 loc) · 4.22 KB
/
Copy pathloop_mcp_user.go
File metadata and controls
118 lines (105 loc) · 4.22 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package agent
import (
"context"
"log/slog"
"maps"
mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp"
"github.com/nextlevelbuilder/goclaw/internal/tools"
)
// getUserMCPTools returns per-user MCP tools for servers requiring user credentials.
// Tools are cached per-user in mcpUserTools sync.Map and registered in the shared
// tool registry so ExecuteWithContext can resolve them. On first call for a user,
// connections are established via pool.AcquireUser() and BridgeTools created.
func (l *Loop) getUserMCPTools(ctx context.Context, userID string) []tools.Tool {
if len(l.mcpUserCredSrvs) == 0 || l.mcpPool == nil || l.mcpStore == nil || userID == "" {
if userID == "" && len(l.mcpUserCredSrvs) > 0 {
slog.Debug("mcp.user_tools_skipped", "reason", "empty_user_id", "servers", len(l.mcpUserCredSrvs))
}
return nil
}
if cached, ok := l.mcpUserTools.Load(userID); ok {
cachedTools := cached.([]tools.Tool)
// Check if any cached tool's connection was evicted by pool.
// If so, clear cache and re-acquire connections.
allConnected := true
for _, t := range cachedTools {
if bt, ok := t.(interface{ IsConnected() bool }); ok && !bt.IsConnected() {
allConnected = false
break
}
}
if allConnected {
return cachedTools
}
l.mcpUserTools.Delete(userID)
slog.Debug("mcp.user_tools_stale", "user", userID, "reason", "pool_evicted")
}
var userTools []tools.Tool
for _, info := range l.mcpUserCredSrvs {
srv := info.Server
// Check if user has credentials for this server
uc, err := l.mcpStore.GetUserCredentials(ctx, srv.ID, userID)
if err != nil || uc == nil || (uc.APIKey == "" && len(uc.Headers) == 0 && len(uc.Env) == 0) {
continue
}
// Resolve connection params: server defaults merged with user overrides
args := mcpbridge.ParseJSONBytesToStringSlice(srv.Args)
env := mcpbridge.ParseJSONBytesToStringMap(srv.Env)
if env == nil {
env = make(map[string]string)
}
headers := mcpbridge.ParseJSONBytesToStringMap(srv.Headers)
if headers == nil {
headers = make(map[string]string)
}
// Inject server-level API key into headers if present
if srv.APIKey != "" && headers["Authorization"] == "" {
headers["Authorization"] = "Bearer " + srv.APIKey
}
// Merge user credentials (user overrides server defaults)
if uc.APIKey != "" {
headers["Authorization"] = "Bearer " + uc.APIKey
}
maps.Copy(headers, uc.Headers)
maps.Copy(env, uc.Env)
// Acquire user-keyed pool connection
entry, err := l.mcpPool.AcquireUser(ctx, l.tenantID, srv.Name, userID,
srv.Transport, srv.Command, args, env, srv.URL, headers, srv.TimeoutSec)
if err != nil {
slog.Warn("mcp.user_pool_acquire_failed", "server", srv.Name, "user", userID, "error", err)
continue
}
// Release immediately — BridgeTools hold client pointer directly.
// This allows pool idle eviction to work (refCount=0 + lastUsed for TTL).
// When pool evicts the connection, BridgeTool.Execute detects connected=false.
l.mcpPool.ReleaseUser(mcpbridge.UserPoolKey(l.tenantID, srv.Name, userID))
// Create BridgeTools pointing to user's connection and register in the
// shared tool registry so ExecuteWithContext can resolve them by name.
reg, _ := l.tools.(*tools.Registry)
hints := mcpbridge.ParseToolHints(srv.Settings)
for _, mcpTool := range entry.MCPTools() {
bt := mcpbridge.NewBridgeTool(srv.Name, mcpTool, entry.ClientPtr(), srv.ToolPrefix, srv.TimeoutSec, entry.Connected(), srv.ID, l.mcpGrantChecker).
WithHints(hints.Global, hints.HintFor(mcpTool.Name))
// Register in registry so ExecuteWithContext can find them.
// Skip if already registered (another user loaded this server with same tool names).
if reg != nil {
if _, exists := reg.Get(bt.Name()); !exists {
reg.Register(bt)
}
}
userTools = append(userTools, bt)
}
}
if len(userTools) > 0 {
l.mcpUserTools.Store(userID, userTools)
// Update "mcp" tool group so policy expansion via alsoAllow includes
// per-user tools. MergeToolGroup is additive — safe for concurrent users.
var names []string
for _, t := range userTools {
names = append(names, t.Name())
}
l.registry.MergeToolGroup("mcp", names)
slog.Info("mcp.user_tools_loaded", "user", userID, "tools", len(userTools))
}
return userTools
}