104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""Process-local MCP runtime supervision and recovery telemetry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import socket
|
|
import time
|
|
from datetime import UTC, datetime
|
|
from threading import Lock
|
|
from typing import Any
|
|
|
|
from mcp.server.context import CallNext, ServerRequestContext
|
|
from mcp.shared.exceptions import MCPError
|
|
|
|
|
|
class RuntimeSupervisor:
|
|
"""Track transport health without turning tool failures into process failures."""
|
|
|
|
def __init__(self) -> None:
|
|
self._started_monotonic = time.monotonic()
|
|
self._started_at = datetime.now(UTC).isoformat()
|
|
material = f"{socket.gethostname()}:{os.getpid()}:{self._started_at}".encode()
|
|
self.worker_id = hashlib.sha256(material).hexdigest()[:16]
|
|
self.generation = os.environ.get(
|
|
"HUB_RUNTIME_GENERATION",
|
|
os.environ.get("GITEA_CONNECTOR_BUILD_SHA", "development"),
|
|
)[:64]
|
|
self.restart_count = max(int(os.environ.get("HUB_RUNTIME_RESTART_COUNT", "0")), 0)
|
|
self._lock = Lock()
|
|
self._in_flight = 0
|
|
self._requests = 0
|
|
self._failures = 0
|
|
self._last_error_code: str | None = None
|
|
self._last_fatal_error_code: str | None = None
|
|
self._circuit_state = "CLOSED"
|
|
self._recovery_count = 0
|
|
|
|
def started(self) -> None:
|
|
with self._lock:
|
|
self._in_flight += 1
|
|
self._requests += 1
|
|
|
|
def completed(self) -> None:
|
|
with self._lock:
|
|
self._in_flight = max(self._in_flight - 1, 0)
|
|
if self._circuit_state == "DEGRADED":
|
|
self._circuit_state = "CLOSED"
|
|
self._recovery_count += 1
|
|
|
|
def failed(self, code: str, *, fatal: bool) -> None:
|
|
with self._lock:
|
|
self._in_flight = max(self._in_flight - 1, 0)
|
|
self._failures += 1
|
|
self._last_error_code = code[:128]
|
|
if fatal:
|
|
self._last_fatal_error_code = code[:128]
|
|
self._circuit_state = "DEGRADED"
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
return {
|
|
"workerId": self.worker_id,
|
|
"sessionId": None,
|
|
"sessionMode": "stateless-resumable-http",
|
|
"resumable": True,
|
|
"startedAt": self._started_at,
|
|
"uptimeSeconds": max(int(time.monotonic() - self._started_monotonic), 0),
|
|
"restartCount": self.restart_count,
|
|
"generation": self.generation,
|
|
"circuitState": self._circuit_state,
|
|
"recoveryCount": self._recovery_count,
|
|
"requests": self._requests,
|
|
"inFlight": self._in_flight,
|
|
"failures": self._failures,
|
|
"lastErrorCode": self._last_error_code,
|
|
"lastFatalErrorCode": self._last_fatal_error_code,
|
|
}
|
|
|
|
|
|
SUPERVISOR = RuntimeSupervisor()
|
|
|
|
|
|
class SupervisorMiddleware:
|
|
"""Keep per-operation errors observable while preserving the next request."""
|
|
|
|
async def __call__(self, ctx: ServerRequestContext[Any, Any], call_next: CallNext) -> Any:
|
|
SUPERVISOR.started()
|
|
try:
|
|
result = await call_next(ctx)
|
|
except MCPError as exc:
|
|
data = exc.data if isinstance(exc.data, dict) else {}
|
|
SUPERVISOR.failed(str(data.get("code", "MCP_ERROR")), fatal=False)
|
|
raise
|
|
except Exception as exc:
|
|
SUPERVISOR.failed(type(exc).__name__, fatal=True)
|
|
raise
|
|
SUPERVISOR.completed()
|
|
return result
|
|
|
|
|
|
def runtime_snapshot() -> dict[str, Any]:
|
|
return SUPERVISOR.snapshot()
|