ops-bridge/src/bridge/health.py

32 lines
948 B
Python
Raw Normal View History

"""HTTP health checker for OpsBridge."""
from __future__ import annotations
feat(BRIDGE-WP-0003): MCP server, /bridge-status skill, cross-mode coverage enforcement Implements the full BRIDGE-WP-0003 workplan: 188 tests passing, 0 lint errors. ## What's added **Capability registry** (`src/bridge/capabilities.py`): - 10 capabilities with required_access_modes (cli/mcp/skill) - Single source of truth for what OpsBridge does and where **MCP server** (`src/bridge/mcp_server/server.py`): - 10 FastMCP tools: bridge_up/down/restart/status/logs + 5 catalog_* tools - 3 resources: bridge://status, catalog://domains, catalog://targets - `.mcp.json` for project-scope auto-registration - `scripts/register_mcp.py` for user-scope machine-global registration **Skill** (`~/.claude/plugins/ops-bridge/bridge-status.md`): - /bridge-status: health table with emoji indicators + remediation advice **Cross-mode test coverage enforcement**: - `tests/conftest.py`: capability/access_mode marks + collect_capability_coverage() - `tests/test_mcp.py`: 31 FastMCP in-process client tests (Client(mcp) pattern) - `tests/test_skill.py`: static skill lint against capability registry - `tests/test_coverage_completeness.py`: meta-test that fails if any required (capability × mode) pair lacks a test; also validates CLI commands and MCP tools are registered in the capability registry **ADR** (`architecture/adr-001-cross-mode-capability-registry.md`): - Documents the registry pattern and FastMCP 3.x testing approach Key implementation note: FastMCP 3.x in-process results are in result.content[0].text (JSON string), not result.data directly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-12 11:33:16 +01:00
from dataclasses import dataclass
from typing import Optional
import httpx
@dataclass
class HealthResult:
ok: bool
status_code: Optional[int] = None
error: Optional[str] = None
class HealthChecker:
def __init__(self, url: str, timeout_seconds: int = 5):
self._url = url
self._timeout = timeout_seconds
async def check(self) -> HealthResult:
try:
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.get(self._url)
response.raise_for_status()
return HealthResult(ok=True, status_code=response.status_code)
except httpx.HTTPStatusError as e:
return HealthResult(ok=False, status_code=e.response.status_code, error=str(e))
except Exception as e:
return HealthResult(ok=False, error=str(e))