check_spec_health
Read-only
Tests connectivity to the specification endpoint and returns a health status report with caching information.
Instructions
Checks the health and connectivity of the specification fetching service.
This tool performs connectivity tests specifically for the specification endpoint
and returns status information about spec fetching capabilities.
Returns:
str: Health status report with spec service connectivity and cache information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The main handler function for the check_spec_health tool. It uses the fetcher's health_check() to get service status, extracts specification-specific info, formats a markdown health report with endpoint status and cache info, and returns it.
async def check_spec_health(ctx: Context) -> str: """ Checks the health and connectivity of the specification fetching service. This tool performs connectivity tests specifically for the specification endpoint and returns status information about spec fetching capabilities. Returns: str: Health status report with spec service connectivity and cache information """ try: await ctx.info("Performing specification service health check") fetcher = get_fetcher() # Get overall health info and extract spec-specific information health_info = await fetcher.health_check() # Extract specification service status spec_service = health_info["services"].get("specification", {}) spec_status = spec_service.get("status", "unknown") # Format health report health_md = "# Specification Service Health Report\n\n" health_md += f"**Overall Status:** {spec_status.upper()}\n\n" # Specification endpoint status health_md += "## Specification Endpoint\n\n" status_emoji = "✅" if spec_status == "healthy" else "❌" health_md += f"{status_emoji} **Ilograph Spec Endpoint**: {spec_status.upper()}\n" health_md += ( f" - URL: {spec_service.get('url', 'https://www.ilograph.com/docs/spec/')}\n" ) if "error" in spec_service: health_md += f" - Error: {spec_service['error']}\n" # Cache information for spec cache_stats = health_info.get("cache_stats", {}) cached_keys = cache_stats.get("keys", []) spec_cached = "specification" in cached_keys health_md += "\n## Specification Cache\n\n" health_md += f"- **Cached:** {'Yes' if spec_cached else 'No'}\n" health_md += ( f"- **Total Cache Entries:** {cache_stats.get('total_entries', 'Unknown')}\n" ) health_md += ( f"- **Valid Cache Entries:** {cache_stats.get('valid_entries', 'Unknown')}\n" ) health_md += "\n---\n\n" if spec_status == "healthy": health_md += ( "*Specification service is operational and ready to fetch the latest spec.*" ) else: health_md += ( "*Specification service is experiencing issues. Spec fetching may be limited.*" ) await ctx.info(f"Spec health check completed - Status: {spec_status}") return health_md except Exception as e: error_msg = f"Error performing spec health check: {str(e)}" await ctx.error(error_msg) return f"Error: {error_msg}" - The FastMCP @mcp.tool decorator with annotations defining the tool's title, description, and readOnlyHint for check_spec_health.
@mcp.tool( annotations={ "title": "Check Specification Service Health", "readOnlyHint": True, "description": "Checks the health and connectivity of the specification fetching service", } ) - src/ilograph_mcp/server.py:64-65 (registration)Registration call: register_fetch_spec_tool(mcp) registers the check_spec_health tool as part of the fetch spec module on the server.
register_fetch_spec_tool(mcp) logger.info("Registered fetch_spec_tool") - The health_check() method on IlographContentFetcher which performs connectivity tests for specification (and other) endpoints and returns health status info used by check_spec_health.
async def health_check(self) -> Dict[str, Any]: """ Perform a health check of the fetcher by testing connectivity. Returns: Dictionary with health status information """ health: Dict[str, Any] = { "status": "healthy", "services": {}, "cache_stats": self.cache.stats(), } try: # Test documentation endpoint test_response = await self.http_client.fetch_with_retry( self.http_client.base_urls["docs"], method="HEAD" ) health["services"]["documentation"] = { "status": "healthy" if test_response else "unhealthy", "url": self.http_client.base_urls["docs"], } except Exception as e: health["services"]["documentation"] = { "status": "unhealthy", "error": str(e), "url": self.http_client.base_urls["docs"], } try: # Test specification endpoint test_response = await self.http_client.fetch_with_retry( self.http_client.base_urls["spec"], method="HEAD" ) health["services"]["specification"] = { "status": "healthy" if test_response else "unhealthy", "url": self.http_client.base_urls["spec"], } except Exception as e: health["services"]["specification"] = { "status": "unhealthy", "error": str(e), "url": self.http_client.base_urls["spec"], } try: # Test icons endpoint test_response = await self.http_client.fetch_with_retry( self.http_client.base_urls["icons"], method="HEAD" ) health["services"]["icons"] = { "status": "healthy" if test_response else "unhealthy", "url": self.http_client.base_urls["icons"], } except Exception as e: health["services"]["icons"] = { "status": "unhealthy", "error": str(e), "url": self.http_client.base_urls["icons"], } # Overall status unhealthy_services = [ service for service, info in health["services"].items() if info["status"] == "unhealthy" ] if unhealthy_services: health["status"] = "degraded" health["unhealthy_services"] = unhealthy_services return health # Global fetcher instance fetcher = IlographContentFetcher() def get_fetcher() -> IlographContentFetcher: """Get the global fetcher instance.""" return fetcher - The stats() method on MemoryCache that provides cache statistics (total_entries, valid_entries, keys) used in check_spec_health's cache report.
def stats(self) -> Dict[str, Any]: """ Get cache statistics. Returns: Dictionary with cache statistics """ total_entries = len(self._cache) expired_entries = sum(1 for entry in self._cache.values() if entry.is_expired()) valid_entries = total_entries - expired_entries return { "total_entries": total_entries, "valid_entries": valid_entries, "expired_entries": expired_entries, "keys": list(self._cache.keys()), }