Skip to main content

resq_mcp.core.telemetry

Telemetry subsystem for the ResQ MCP server. Provides unified OpenTelemetry tracing, Prometheus-compatible metrics, and structured logging with automatic PII redaction.

annotations

functools

logging

re

time

contextmanager

TYPE_CHECKING

Any

ParamSpec

TypeVar

settings

P

R

logger

tracer

meter

setup_telemetry

def setup_telemetry() -> None
Initialize OpenTelemetry tracing and metrics.

metrics

trace

def trace(_func_or_name: Callable[P, R] | str | None = None,
          name: str | None = None,
          *,
          record_args: bool = False,
          record_result: bool = False) -> Any
Instrument a function with an OpenTelemetry span.

span

@contextmanager
def span(name: str,
         attributes: dict[str, Any] | None = None) -> Generator[Any]

log_event

def log_event(event: str, level: int = logging.INFO, **attrs: Any) -> None

shutdown_telemetry

def shutdown_telemetry(timeout_ms: int = 5_000) -> None