Documentation

cache

Search result caching for Context-Fabric MCP server.

Caches search results by (corpus, template) to avoid re-running expensive queries. All return_types (results, count, statistics, passages) can draw from the same cache.

Classes

class

CachedSearchResult

A cached search result.

A cached search result. Attributes: cursor_id: UUID for pagination/lookup corpus: The corpus name template: The search template results: Raw search results (list of node ID tuples) created_at: Unix timestamp when cached expires_at: Unix timestamp when cache entry expires last_accessed: Unix timestamp of last access (for LRU)

Attributes

NameTypeDescription
corpusstr—
created_atfloat—
cursor_idstr—
expires_atfloat—
is_expiredboolCheck if this cache entry has expired.
last_accessedfloat—
resultslist[tuple[(int, ...)]]—
templatestr—

Methods

__init__(self, cursor_id: str, corpus: str, template: str, results: list[tuple[(int, ...)]], created_at: float, expires_at: float, last_accessed: float = {'arguments': [], 'cls': 'ExprCall', 'function': {'cls': 'ExprAttribute', 'values': [{'cls': 'ExprName', 'member': None, 'name': 'time'}, {'cls': 'ExprName', 'member': None, 'name': 'time'}]}}) → None
Parameters
  • cursor_id: str
  • corpus: str
  • template: str
  • results: list[tuple[(int, ...)]]
  • created_at: float
  • expires_at: float
  • last_accessed: float
touch(self) → None

Update last accessed time.

class

SearchCache

In-memory cache for search results.

In-memory cache for search results. Caches by (corpus, template) key. Different return_types and parameters can reuse the same cached results. Thread-safe for concurrent access.

Attributes

NameTypeDescription
_by_cursordict[(str, CachedSearchResult)]—
_cachedict[(tuple[(str, str)], CachedSearchResult)]—
_lock——
default_ttl——
max_entries——
max_results_per_entry——

Methods

__init__(self, default_ttl: int = 300, max_entries: int = 100, max_results_per_entry: int = 10000) → None

Initialize the search cache.

Parameters
  • default_ttl: int= 300
  • max_entries: int= 100
  • max_results_per_entry: int= 10000
cleanup_expired(self) → int

Remove all expired cache entries.

clear(self) → None

Clear all cache entries.

get_by_cursor(self, cursor_id: str) → CachedSearchResult | None

Get cached result by cursor ID.

Parameters
  • cursor_id: str
get_or_execute(self, corpus: str, template: str, search_fn: Callable[([], list[tuple[(int, ...)]])], ttl: int | None = None) → CachedSearchResult

Get cached result or execute search and cache it.

Parameters
  • corpus: str
  • template: str
  • search_fn: Callable[([], list[tuple[(int, ...)]])]
  • ttl: int | None= None
get_page(self, cursor_id: str, offset: int, limit: int) → tuple[(list[tuple[(int, ...)]], bool, int)] | None

Get a page of results from a cached search.

Parameters
  • cursor_id: str
  • offset: int
  • limit: int
stats(self) → dict[(str, Any)]

Get cache statistics.

Functions

function
get_cache() → SearchCache

Get the global search cache instance.

Get the global search cache instance. Creates a new instance if none exists.
function
reset_cache() → None

Reset the global cache instance.

Reset the global cache instance. Useful for testing.