1// bridge.go — MCP tool → zot tool registration and routing.
2//
3// Converts MCP tools into zot-registered tools with namespaced names:
4//
5// mcp__<server>__<tool>
6//
7// The double underscore separates the server name from the tool name,
8// avoiding collisions with zot's built-in tools (read, write, edit, bash, skill).
9package main
10
11import (
12 "context"
13 "encoding/json"
14 "errors"
15 "fmt"
16 "log"
17 "regexp"
18 "sort"
19 "strings"
20 "sync"
21 "time"
22
23 "github.com/mark3labs/mcp-go/mcp"
24 "github.com/patriceckhart/zot/packages/agent/ext"
25)
26
27// toolMapping tracks which zot tool name maps to which MCP server + tool.
28type toolMapping struct {
29 serverName string // e.g. "filesystem"
30 mcpTool string // e.g. "read_file"
31}
32
33// bridge connects MCP servers to zot's extension protocol.
34type bridge struct {
35 e *ext.Extension
36 // servers is written once in loadServers before any goroutine
37 // starts, and read-only afterwards — that invariant is what makes
38 // the lock-free reads in handleToolCall and the idle reaper safe.
39 // Do not add or remove entries after startup.
40 servers map[string]*managedServer
41 mapping map[string]toolMapping // zot tool name → MCP server + tool; guarded by mu
42 logger *log.Logger
43
44 mu sync.Mutex // guards mapping
45}
46
47// newBridge creates a new MCP→zot bridge.
48func newBridge(e *ext.Extension, logger *log.Logger) *bridge {
49 return &bridge{
50 e: e,
51 servers: make(map[string]*managedServer),
52 mapping: make(map[string]toolMapping),
53 logger: logger,
54 }
55}
56
57// sanitizeName converts a string into a valid zot tool name component.
58// Zot tool names must match [a-zA-Z][a-zA-Z0-9_]*.
59var invalidChars = regexp.MustCompile(`[^a-zA-Z0-9]`)
60
61func sanitizeName(s string) string {
62 s = invalidChars.ReplaceAllString(s, "_")
63 // Ensure it starts with a letter
64 if len(s) > 0 && (s[0] >= '0' && s[0] <= '9') {
65 s = "t_" + s
66 }
67 if s == "" {
68 s = "unnamed"
69 }
70 return s
71}
72
73// toolName builds the namespaced zot tool name for an MCP tool.
74func toolName(serverName, mcpToolName string) string {
75 return fmt.Sprintf("mcp__%s__%s", sanitizeName(serverName), sanitizeName(mcpToolName))
76}
77
78// loadServers reads the config and creates managed servers.
79func (b *bridge) loadServers(cfg Config) {
80 for name, srvCfg := range cfg.MCPServers {
81 srv := newManagedServer(name, srvCfg, b.logger)
82 b.servers[name] = srv
83 }
84}
85
86// registerCachedTools registers previously discovered tool definitions without
87// starting MCP servers. Tool calls still lazy-start the owning server on demand.
88func (b *bridge) registerCachedTools(cache toolCache) int {
89 count := 0
90 for serverName, srv := range b.servers {
91 cached, ok := cache.Servers[serverName]
92 if !ok || cached.Fingerprint != serverFingerprint(srv.config) {
93 continue
94 }
95 cachedNames := make([]mcp.Tool, 0, len(cached.Tools))
96 for _, tool := range cached.Tools {
97 b.registerCachedTool(serverName, tool)
98 cachedNames = append(cachedNames, mcp.Tool{Name: tool.Name})
99 count++
100 }
101 srv.mu.Lock()
102 if len(srv.tools) == 0 {
103 srv.tools = cachedNames
104 }
105 srv.mu.Unlock()
106 }
107 return count
108}
109
110// refreshToolCache discovers live tools and updates the on-disk cache.
111// It reports whether the cache changed; callers should only ask for /reload-ext
112// when changed is true.
113func (b *bridge) refreshToolCache(ctx context.Context, path string) (bool, error) {
114 previous, err := readToolCache(path)
115 if err != nil {
116 b.logger.Printf("read existing tool cache: %v", err)
117 }
118 cache := toolCache{Version: toolCacheVersion, Servers: map[string]cachedServer{}}
119 var wg sync.WaitGroup
120 errCh := make(chan error, len(b.servers))
121 var mu sync.Mutex
122
123 for name, srv := range b.servers {
124 wg.Add(1)
125 go func(n string, s *managedServer) {
126 defer wg.Done()
127 if err := s.start(ctx); err != nil {
128 b.logger.Printf("[%s] lazy discovery failed: %v", n, err)
129 errCh <- fmt.Errorf("%s: %w", n, err)
130 return
131 }
132
133 s.mu.Lock()
134 tools := append([]mcp.Tool(nil), s.tools...)
135 s.mu.Unlock()
136
137 cachedTools := make([]cachedTool, 0, len(tools))
138 for _, tool := range tools {
139 ct, err := cachedToolFromMCP(n, tool)
140 if err != nil {
141 b.logger.Printf("[%s] tool %s: schema marshal error: %v", n, tool.Name, err)
142 continue
143 }
144 cachedTools = append(cachedTools, ct)
145 }
146
147 sort.Slice(cachedTools, func(i, j int) bool { return cachedTools[i].Name < cachedTools[j].Name })
148
149 mu.Lock()
150 cache.Servers[n] = cachedServer{Fingerprint: serverFingerprint(s.config), Tools: cachedTools}
151 mu.Unlock()
152 }(name, srv)
153 }
154
155 wg.Wait()
156 close(errCh)
157
158 changed := !toolCachesEqual(previous, cache)
159 if changed {
160 if err := writeToolCache(path, cache); err != nil {
161 return false, err
162 }
163 }
164
165 var errs []string
166 for err := range errCh {
167 errs = append(errs, err.Error())
168 }
169 if len(errs) > 0 {
170 return changed, fmt.Errorf("%s", strings.Join(errs, "; "))
171 }
172 return changed, nil
173}
174
175func (b *bridge) registerCachedTool(serverName string, tool cachedTool) {
176 zotName := toolName(serverName, tool.Name)
177
178 b.mu.Lock()
179 if existing, exists := b.mapping[zotName]; exists {
180 b.mu.Unlock()
181 if existing.serverName != serverName || existing.mcpTool != tool.Name {
182 b.logger.Printf("tool name collision: %s already maps to %s/%s; skipping %s/%s",
183 zotName, existing.serverName, existing.mcpTool, serverName, tool.Name)
184 }
185 return
186 }
187 b.mapping[zotName] = toolMapping{serverName: serverName, mcpTool: tool.Name}
188 b.mu.Unlock()
189
190 b.e.Tool(zotName, tool.Description, json.RawMessage(tool.Schema), func(args json.RawMessage) ext.ToolResult {
191 return b.handleToolCall(zotName, args)
192 })
193
194 b.logger.Printf("registered cached tool: %s → %s/%s", zotName, serverName, tool.Name)
195}
196
197func mcpToolDescription(serverName string, tool mcp.Tool) string {
198 desc := tool.Description
199 if tool.Annotations.Title != "" {
200 desc = tool.Annotations.Title + ": " + desc
201 }
202
203 var hints []string
204 if tool.Annotations.ReadOnlyHint != nil && *tool.Annotations.ReadOnlyHint {
205 hints = append(hints, "read-only")
206 }
207 if tool.Annotations.IdempotentHint != nil && *tool.Annotations.IdempotentHint {
208 hints = append(hints, "idempotent")
209 }
210 if tool.Annotations.OpenWorldHint != nil && !*tool.Annotations.OpenWorldHint {
211 hints = append(hints, "closed-world")
212 }
213 if tool.Annotations.DestructiveHint != nil && *tool.Annotations.DestructiveHint {
214 hints = append(hints, "destructive")
215 }
216 if len(hints) > 0 {
217 desc += " [" + strings.Join(hints, ", ") + "]"
218 }
219 if desc == "" {
220 desc = fmt.Sprintf("MCP tool from server %q", serverName)
221 }
222 return desc
223}
224
225// mcpToolSchema converts an MCP Tool's input schema to a JSON Schema value.
226func mcpToolSchema(tool mcp.Tool) any {
227 if len(tool.RawInputSchema) > 0 {
228 return json.RawMessage(tool.RawInputSchema)
229 }
230
231 schema := map[string]any{
232 "type": tool.InputSchema.Type,
233 "properties": tool.InputSchema.Properties,
234 }
235 if len(tool.InputSchema.Required) > 0 {
236 schema["required"] = tool.InputSchema.Required
237 }
238 if len(tool.InputSchema.Defs) > 0 {
239 schema["$defs"] = tool.InputSchema.Defs
240 }
241 if tool.InputSchema.AdditionalProperties != nil {
242 schema["additionalProperties"] = tool.InputSchema.AdditionalProperties
243 }
244 return schema
245}
246
247// handleToolCall routes a zot tool call to the appropriate MCP server.
248func (b *bridge) handleToolCall(zotName string, args json.RawMessage) ext.ToolResult {
249 b.mu.Lock()
250 mapping, ok := b.mapping[zotName]
251 b.mu.Unlock()
252
253 if !ok {
254 return ext.TextErrorResult(fmt.Sprintf(
255 "Tool '%s' not found. This tool was registered but is no longer available. "+
256 "The MCP server may have been stopped. Try running '/mcp' to check server status.",
257 zotName))
258 }
259
260 srv, ok := b.servers[mapping.serverName]
261 if !ok {
262 return ext.TextErrorResult(fmt.Sprintf(
263 "MCP server '%s' not found. The server configuration may have been removed. "+
264 "Check your mcp.json configuration file.",
265 mapping.serverName))
266 }
267
268 // managedServer owns all timeouts (connect + request); pass a
269 // plain context here.
270 result, err := srv.callTool(context.Background(), mapping.mcpTool, args)
271 if err != nil {
272 if errors.Is(err, context.DeadlineExceeded) {
273 return ext.TextErrorResult(timeoutErrorMessage(mapping.serverName, srv.config.RequestTimeout))
274 }
275 errMsg := err.Error()
276 if strings.Contains(errMsg, "connection") || strings.Contains(errMsg, "transport") {
277 return ext.TextErrorResult(fmt.Sprintf(
278 "Connection to MCP server '%s' failed: %v. The server may have crashed or been stopped. "+
279 "Try running '/mcp start %s' to restart that server, or '/mcp restart' to restart all servers.",
280 mapping.serverName, err, mapping.serverName))
281 }
282 return ext.TextErrorResult(fmt.Sprintf(
283 "MCP tool call failed: %v. Server: %s, Tool: %s. "+
284 "Check '/mcp %s' for server status.",
285 err, mapping.serverName, mapping.mcpTool, mapping.serverName))
286 }
287
288 // Convert MCP result to zot ToolResult
289 return mcpResultToZot(result)
290}
291
292// timeoutErrorMessage formats the user-facing text for a deadline error.
293func timeoutErrorMessage(serverName string, requestTimeout int) string {
294 return fmt.Sprintf(
295 "Tool call timed out after %d seconds. The MCP server '%s' may be slow or unresponsive. "+
296 "You can increase the timeout in your mcp.json config with 'requestTimeout'.",
297 requestTimeout, serverName)
298}
299
300// mcpResultToZot converts an MCP CallToolResult to a zot ToolResult.
301func mcpResultToZot(result *mcp.CallToolResult) ext.ToolResult {
302 if result == nil {
303 return ext.TextResult("(no result)")
304 }
305
306 var contents []ext.ToolContent
307 for _, c := range result.Content {
308 switch v := c.(type) {
309 case mcp.TextContent:
310 contents = append(contents, ext.Text(v.Text))
311 case mcp.ImageContent:
312 contents = append(contents, ext.Image(v.MIMEType, v.Data))
313 default:
314 // Fallback: try to marshal as JSON
315 data, _ := json.Marshal(v)
316 contents = append(contents, ext.Text(string(data)))
317 }
318 }
319
320 if len(contents) == 0 {
321 contents = append(contents, ext.Text("(empty result)"))
322 }
323
324 tr := ext.ToolResult{Content: contents}
325 if result.IsError {
326 tr.IsError = true
327 }
328 return tr
329}
330
331// startAll starts all configured servers without re-registering their tools with zot.
332func (b *bridge) startAll(ctx context.Context) error {
333 var wg sync.WaitGroup
334 errCh := make(chan error, len(b.servers))
335
336 for name, srv := range b.servers {
337 wg.Add(1)
338 go func(n string, s *managedServer) {
339 defer wg.Done()
340 if err := s.start(ctx); err != nil {
341 b.logger.Printf("[%s] failed to start: %v", n, err)
342 errCh <- fmt.Errorf("%s: %w", n, err)
343 }
344 }(name, srv)
345 }
346
347 wg.Wait()
348 close(errCh)
349
350 var errs []string
351 for err := range errCh {
352 errs = append(errs, err.Error())
353 }
354 if len(errs) > 0 {
355 return fmt.Errorf("%s", strings.Join(errs, "; "))
356 }
357 return nil
358}
359
360// startIdleReaper runs a background goroutine that kills idle servers.
361// It has no shutdown mechanism by design: it lives exactly as long as
362// the extension process, and zot reaps the whole process on exit.
363func (b *bridge) startIdleReaper() {
364 go func() {
365 ticker := time.NewTicker(30 * time.Second)
366 defer ticker.Stop()
367 for range ticker.C {
368 for _, srv := range b.servers {
369 // Use per-server idle timeout
370 idleTimeout := time.Duration(srv.config.IdleTimeout) * time.Second
371 if srv.isIdle(idleTimeout) {
372 b.logger.Printf("[%s] idle timeout, stopping", srv.name)
373 srv.stop()
374 }
375 }
376 }
377 }()
378}
379
380// stopAll shuts down all MCP servers.
381func (b *bridge) stopAll() {
382 for _, srv := range b.servers {
383 srv.stop()
384 }
385}
386
387// serverStatus returns compact status info for all servers in stable order.
388func (b *bridge) serverStatus() []string {
389 names := make([]string, 0, len(b.servers))
390 for name := range b.servers {
391 names = append(names, name)
392 }
393 sort.Strings(names)
394
395 lines := make([]string, 0, len(names))
396 for _, name := range names {
397 lines = append(lines, b.servers[name].status())
398 }
399 return lines
400}
401
402func (b *bridge) notifyLevel() string {
403 if len(b.servers) == 0 {
404 return "info"
405 }
406 ready := 0
407 errored := 0
408 for _, srv := range b.servers {
409 srv.mu.Lock()
410 state := srv.state
411 srv.mu.Unlock()
412 switch state {
413 case stateReady:
414 ready++
415 case stateError:
416 errored++
417 }
418 }
419 if errored > 0 && ready == 0 {
420 return "error"
421 }
422 if errored > 0 {
423 return "warn"
424 }
425 return "success"
426}
427
428// startServer manually starts a specific server.
429func (b *bridge) startServer(name string) error {
430 srv, ok := b.servers[name]
431 if !ok {
432 return fmt.Errorf("unknown server: %s", name)
433 }
434 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
435 defer cancel()
436 return srv.start(ctx)
437}
438
439// stopServer manually stops a specific server.
440func (b *bridge) stopServer(name string) error {
441 srv, ok := b.servers[name]
442 if !ok {
443 return fmt.Errorf("unknown server: %s", name)
444 }
445 srv.stop()
446 return nil
447}