MCP Sprut

MCPSprut

一种Go服务，通过多个MCP（模型上下文协议）服务器维护持久的SSE流，监听工具更改通知，并通过批写入将工具存储在本地数据库中。

建筑

flowchart TD
    main["cmd/sprut/main.go"] --> hub["Hub (orchestrator)"]
    hub --> connector["Connector pool"]
    connector --> c1["Conn: server 0"]
    connector --> c2["Conn: server 1"]
    connector --> cN["Conn: server N"]
    c1 --> mcpClient["MCP Client"]
    hub --> batcher["Batcher"]
    batcher --> storage["Storage (interface)"]
    storage --> bolt["BoltDB impl"]
    storage --> future["Future: PG, Redis..."]

运作原理

1. 启动时，从存储中加载MCP服务器列表
2. 对于每台服务器，执行MCP握手(initialize + tools/list)并持续使用工具
3. 向每个服务器打开GET SSE流，监听 notifications/tools/list_changed
4. 收到通知后（根据MCP规范，有效载荷为空），重新获取 tools/list 并更新存储
5. 批处理写入存储（可配置： 1 为了实时， 100 用于高通量）
6. 在运行时检测添加到存储中的新服务器——无需重新启动

数据流

sequenceDiagram
    participant Hub
    participant Connector
    participant MCPServer
    participant Batcher
    participant Storage

    Hub->>Storage: LoadServers()
    Hub->>Connector: Start(serverConfig)

    Connector->>MCPServer: POST initialize
    MCPServer-->>Connector: capabilities

    Connector->>MCPServer: POST tools/list
    MCPServer-->>Connector: tools[]
    Connector->>Batcher: Submit(serverID, tools)

    Connector->>MCPServer: GET /mcp (SSE stream)

    loop On notification
        MCPServer-->>Connector: notifications/tools/list_changed
        Connector->>MCPServer: POST tools/list
        MCPServer-->>Connector: tools[]
        Connector->>Batcher: Submit(serverID, tools)
    end

    Batcher->>Storage: SaveToolsBatch(updates)

项目结构

配置

所有设置都是从环境变量中读取的。

优雅地关闭

1. 信号/信号项→ 取消上下文
2. 所有连接器goroutines关闭SSE流并退出
3. 批处理程序刷新剩余缓冲区
4. BoltDB关闭
5. 进程退出

优雅降级

每个连接器独立运行。如果一台MCP服务器发生故障，其余服务器将继续工作。在SSE流断开连接或握手失败时，连接器会等待 SPRUT_RETRY_INTERVAL 并重试。

跑步

# Start the MCP simulator (separate repo)
cd ~/golang/mcp-simulator
go run ./cmd/simulator/ --servers 10 --port 9090

# Start MCPSprut
cd ~/golang/mcp-sprut
SPRUT_DB_PATH=sprut.db go run ./cmd/sprut/

调用树

main.go
├── config.Load()                              // env → Config
├── storage.NewBoltStorage(dbPath)             // open sprut.db
├── mcpclient.NewClient(timeout)               // HTTP client for MCP
├── batcher.NewBatcher(store, size, interval)
│   └── batcher.Start(ctx)                    // goroutine: buffer → flush to storage
├── hub.NewHub(store, client, batcher, retry)
│   └── hub.Start(ctx)
│       ├── store.LoadServers()                // load servers from BoltDB
│       ├── for each server → startConnector()
│       │   └── go connector.Run(ctx)          // goroutine per server
│       │       └── connect()                  // loop with retry
│       │           ├── client.Initialize()
│       │           ├── client.SendInitialized()
│       │           ├── client.ListTools() → batcher.Submit()
│       │           └── client.SubscribeNotifications() → SSE loop
│       │               └── on notification → client.ListTools() → batcher.Submit()
│       └── store.OnNewServer(callback)        // new servers at runtime
├── <-sigCh                                    // wait for SIGINT/SIGTERM
├── cancel()                                   // stop all goroutines
├── batcher.Wait()                             // wait for remaining flush
└── store.Close()                              // close BoltDB

设计模式