Lenses MCP Server

Instructions

Implementation Reference

• src/lenses_mcp/tools/environments.py:81-126 (handler)

  The core handler function for the 'check_environment_health' tool. It is decorated with @mcp.tool() which registers it with the MCP server. The function retrieves the environment details, checks the agent connection status, scans for issues in metrics, computes overall health, and returns a structured health status report including summary metrics.
  @mcp.tool() async def check_environment_health(name: str) -> Dict[str, Any]: """ Checks the health status of a Lenses environment. Args: name: The name of the environment to check. Returns: Health status information including agent connection and any issues. """ env = await get_environment(name) health_status = { "environment": name, "healthy": False, "agent_connected": False, "issues": [] } if "status" in env: health_status["agent_connected"] = env["status"].get("agent_connected", False) if env["status"]["agent_connected"] and "agent" in env["status"]: agent_data = env["status"]["agent"] metrics = agent_data.get("metrics", {}) # Check for issues if "other" in metrics and metrics["other"].get("num_issues", 0) > 0: health_status["issues"].append(f"Found {metrics['other']['num_issues']} issues") # Basic health check health_status["healthy"] = ( health_status["agent_connected"] and len(health_status["issues"]) == 0 ) # Add summary metrics health_status["summary"] = { "kafka_brokers": metrics.get("kafka", {}).get("num_brokers", 0), "topics": metrics.get("data", {}).get("num_topics", 0), "consumers": metrics.get("apps", {}).get("num_consumers", 0), "connectors": metrics.get("connect", {}).get("num_connectors", 0) } return health_status