Spaces:
Running
Running
SooryaKiranKakkarayil
Day 2: MCP server complete - 4 tools, TTL cache, retry/circuit-breaker- lookup_domain_age, check_ipqs, check_virustotal, check_local_blocklist- Shared cache.py (1h TTL) and resilience.py (retry + circuit breaker)- Local OpenPhish/URLhaus blocklist loader with real snapshot data- Mocked pytest suite (16 tests) plus live MCP Inspector verification- Fixes: follow_redirects on RDAP bootstrap redirect; sys.path fix for direct script invocation; load_dotenv() for stdio-launched subprocess
33391b8 | """In-memory TTL cache shared by all MCP tools. | |
| Why this exists: IPQS gives 1,000 lookups/month and VirusTotal caps at | |
| 4 req/min, 500/day. A single dev/test loop that re-scans the same handful | |
| of domains can burn through either budget in minutes without a cache. | |
| This is intentionally a plain dict, not Redis or anything external — | |
| the process is short-lived (one HF Space container) so in-memory is | |
| enough, and it keeps the MCP server dependency-free. | |
| """ | |
| import time | |
| CACHE_TTL_SECONDS = 3600 # 1 hour, per the design doc | |
| # key -> (stored_at_epoch_seconds, value) | |
| _store: dict[str, tuple[float, dict]] = {} | |
| def cache_get(key: str) -> dict | None: | |
| """Return the cached value for key if present and not expired, else None.""" | |
| hit = _store.get(key) | |
| if hit is None: | |
| return None | |
| stored_at, value = hit | |
| if time.time() - stored_at >= CACHE_TTL_SECONDS: | |
| _store.pop(key, None) | |
| return None | |
| return value | |
| def cache_set(key: str, value: dict) -> dict: | |
| """Store value under key and return it (so callers can `return cache_set(...)`).""" | |
| _store[key] = (time.time(), value) | |
| return value | |
| def cache_clear() -> None: | |
| """Wipe the cache. Used by tests; not called in normal operation.""" | |
| _store.clear() | |
| def cache_size() -> int: | |
| return len(_store) | |