Class: McpToolkit::Gateway::Aggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/mcp_toolkit/gateway/aggregator.rb

Overview

Aggregates the tool lists of all configured upstream MCP servers for a gateway's tools/list, namespacing each tool as <app>__<tool>.

Per upstream the namespaced list is cached in config.cache_store (config.upstream_list_ttl, default 15 min) and bustable via flush! / flush!(key). On a cache miss the live pull runs per-upstream (one HTTP call each); a failing/timeout upstream is omitted + logged, never breaking the list.

Why the cache is safe to share globally: an upstream's tools/list is token-INDEPENDENT — it returns the same public tool definitions to every valid caller and enforces scope only when a tool is CALLED (per-call authorization is the upstream's job). So one caller's pull is a correct answer for all callers. This is a registration CONTRACT, not an assumption: an upstream that filters its list by the caller's privilege (e.g. hides superuser-only tools) MUST register public_tool_list: false, which opts it out of this cache (pulled live per request) so a privileged caller's list can never be served to an unprivileged one. See McpToolkit::Gateway::UpstreamRegistry::Upstream.

Only a NON-EMPTY pull is ever cached. An empty or failed pull is almost always a transient upstream hiccup (timeout, a session/handshake blip, a degenerate 200); caching it would freeze a whole app's tools out of tools/list for the full TTL for EVERY caller (a poisoned global cache), which is exactly the failure this guards against. A stale empty already in the cache is treated as a miss and re-pulled, so the aggregate self-heals as soon as the upstream returns tools.

Concurrency: upstreams are pulled CONCURRENTLY via concurrent-ruby futures. When running inside Rails, each future is wrapped in Rails.application.executor so it participates in the framework's per-request lifecycle (reloading, query cache, connection checkout); a non-Rails host runs plain futures. Output order follows the registry order.

Constant Summary collapse

CACHE_KEY_PREFIX = 
"mcp_toolkit:gateway:tools:"

Instance Method Summary collapse

Constructor Details

#initialize(config: McpToolkit.config) ⇒ Aggregator

Returns a new instance of Aggregator.



38
39
40
 
# File 'lib/mcp_toolkit/gateway/aggregator.rb', line 38

def initialize(config: McpToolkit.config)
  @config = config
end

Instance Method Details

#flush!(key = nil) ⇒ Object 



56
57
58
59
60
61
62
 
# File 'lib/mcp_toolkit/gateway/aggregator.rb', line 56

def flush!(key = nil)
  if key
    cache.delete(cache_key(key))
  else
    config.upstreams.all.each { |upstream| cache.delete(cache_key(upstream.key)) }
  end
end

#tool_definitions(bearer_token: nil) ⇒ Object

Namespaced tool definitions across all configured upstreams. bearer_token is used only on a cache miss, so the upstream can authenticate the list request the same way it would a call. Upstreams are fetched concurrently; output order follows the registry order.



46
47
48
49
50
51
52
53
54
 
# File 'lib/mcp_toolkit/gateway/aggregator.rb', line 46

def tool_definitions(bearer_token: nil)
  futures = config.upstreams.all.map do |upstream|
    Concurrent::Promises.future do
      within_executor { cached_or_live_definitions(upstream, bearer_token:) }
    end
  end

  futures.flat_map(&:value!)
end