Skip to content

mcp_registry_client.cache

Response caching for the MCP registry client.

CacheEntry

Cache entry with expiration support.

Source code in mcp_registry_client/cache.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class CacheEntry:
    """Cache entry with expiration support."""

    def __init__(self, value: T, ttl: int) -> None:
        """Initialize cache entry.

        Args:
            value: The cached value
            ttl: Time to live in seconds

        """
        self.value = value
        self.expires_at = time.time() + ttl

    def is_expired(self) -> bool:
        """Check if the cache entry has expired."""
        return time.time() > self.expires_at

__init__(value, ttl)

Initialize cache entry.

Parameters:

Name Type Description Default
value T

The cached value

 required
ttl int

Time to live in seconds

 required
Source code in mcp_registry_client/cache.py 
15
16
17
18
19
20
21
22
23
24
def __init__(self, value: T, ttl: int) -> None:
    """Initialize cache entry.

    Args:
        value: The cached value
        ttl: Time to live in seconds

    """
    self.value = value
    self.expires_at = time.time() + ttl

is_expired()

Check if the cache entry has expired.

Source code in mcp_registry_client/cache.py 
26
27
28
def is_expired(self) -> bool:
    """Check if the cache entry has expired."""
    return time.time() > self.expires_at

ResponseCache

In-memory response cache with TTL support.

Source code in mcp_registry_client/cache.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class ResponseCache:
    """In-memory response cache with TTL support."""

    def __init__(self, config: ClientConfig) -> None:
        """Initialize response cache.

        Args:
            config: Client configuration with cache settings

        """
        self.enabled = config.enable_cache
        self.ttl = config.cache_ttl
        self._cache: dict[str, CacheEntry] = {}
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> Any | None:  # noqa: ANN401
        """Get a value from the cache.

        Args:
            key: Cache key

        Returns:
            Cached value if found and not expired, None otherwise

        """
        if not self.enabled:
            return None

        async with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None

            if entry.is_expired():
                del self._cache[key]
                return None

            return entry.value

    async def set(self, key: str, value: Any) -> None:  # noqa: ANN401
        """Set a value in the cache.

        Args:
            key: Cache key
            value: Value to cache

        """
        if not self.enabled:
            return

        async with self._lock:
            self._cache[key] = CacheEntry(value, self.ttl)

    async def clear(self) -> None:
        """Clear all cached values."""
        async with self._lock:
            self._cache.clear()

    async def cleanup_expired(self) -> None:
        """Remove expired entries from the cache."""
        if not self.enabled:
            return

        async with self._lock:
            current_time = time.time()
            expired_keys = [
                key for key, entry in self._cache.items() if current_time > entry.expires_at
            ]
            for key in expired_keys:
                del self._cache[key]

    def cache_key_for_search(self, name: str) -> str:
        """Generate cache key for search requests.

        Args:
            name: Search name parameter

        Returns:
            Cache key

        """
        return f'search:{name.lower().strip()}'

    def cache_key_for_server(self, server_id: str) -> str:
        """Generate cache key for server requests.

        Args:
            server_id: Server ID

        Returns:
            Cache key

        """
        return f'server:{server_id}'

    def cache_key_for_server_by_name(self, name: str) -> str:
        """Generate cache key for server by name requests.

        Args:
            name: Server name

        Returns:
            Cache key

        """
        return f'server_by_name:{name.lower().strip()}'

__init__(config)

Initialize response cache.

Parameters:

Name Type Description Default
config ClientConfig

Client configuration with cache settings

 required
Source code in mcp_registry_client/cache.py 
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, config: ClientConfig) -> None:
    """Initialize response cache.

    Args:
        config: Client configuration with cache settings

    """
    self.enabled = config.enable_cache
    self.ttl = config.cache_ttl
    self._cache: dict[str, CacheEntry] = {}
    self._lock = asyncio.Lock()

Generate cache key for search requests.

Parameters:

Name Type Description Default
name str

Search name parameter

 required

Returns:

Type Description
str

Cache key
Source code in mcp_registry_client/cache.py 
102
103
104
105
106
107
108
109
110
111
112
def cache_key_for_search(self, name: str) -> str:
    """Generate cache key for search requests.

    Args:
        name: Search name parameter

    Returns:
        Cache key

    """
    return f'search:{name.lower().strip()}'

cache_key_for_server(server_id)

Generate cache key for server requests.

Parameters:

Name Type Description Default
server_id str

Server ID

 required

Returns:

Type Description
str

Cache key
Source code in mcp_registry_client/cache.py 
114
115
116
117
118
119
120
121
122
123
124
def cache_key_for_server(self, server_id: str) -> str:
    """Generate cache key for server requests.

    Args:
        server_id: Server ID

    Returns:
        Cache key

    """
    return f'server:{server_id}'

cache_key_for_server_by_name(name)

Generate cache key for server by name requests.

Parameters:

Name Type Description Default
name str

Server name

 required

Returns:

Type Description
str

Cache key
Source code in mcp_registry_client/cache.py 
126
127
128
129
130
131
132
133
134
135
136
def cache_key_for_server_by_name(self, name: str) -> str:
    """Generate cache key for server by name requests.

    Args:
        name: Server name

    Returns:
        Cache key

    """
    return f'server_by_name:{name.lower().strip()}'

cleanup_expired() async

Remove expired entries from the cache.

Source code in mcp_registry_client/cache.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def cleanup_expired(self) -> None:
    """Remove expired entries from the cache."""
    if not self.enabled:
        return

    async with self._lock:
        current_time = time.time()
        expired_keys = [
            key for key, entry in self._cache.items() if current_time > entry.expires_at
        ]
        for key in expired_keys:
            del self._cache[key]

clear() async

Clear all cached values.

Source code in mcp_registry_client/cache.py 
84
85
86
87
async def clear(self) -> None:
    """Clear all cached values."""
    async with self._lock:
        self._cache.clear()

get(key) async

Get a value from the cache.

Parameters:

Name Type Description Default
key str

Cache key

 required

Returns:

Type Description
Any | None

Cached value if found and not expired, None otherwise
Source code in mcp_registry_client/cache.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def get(self, key: str) -> Any | None:  # noqa: ANN401
    """Get a value from the cache.

    Args:
        key: Cache key

    Returns:
        Cached value if found and not expired, None otherwise

    """
    if not self.enabled:
        return None

    async with self._lock:
        entry = self._cache.get(key)
        if entry is None:
            return None

        if entry.is_expired():
            del self._cache[key]
            return None

        return entry.value

set(key, value) async

Set a value in the cache.

Parameters:

Name Type Description Default
key str

Cache key

 required
value Any

Value to cache

 required
Source code in mcp_registry_client/cache.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
async def set(self, key: str, value: Any) -> None:  # noqa: ANN401
    """Set a value in the cache.

    Args:
        key: Cache key
        value: Value to cache

    """
    if not self.enabled:
        return

    async with self._lock:
        self._cache[key] = CacheEntry(value, self.ttl)