Skip to content

Observability

Observability component — profiling, metrics, tracing, alerting.

Cluster: Core Infrastructure | Type: component | MCP Tools: 29

Overview

Runtime observability engine that profiles Python code execution, measures subprocess performance, analyses hotspots, records local SQLite-backed metrics, exports Prometheus text, and generates Grafana-importable dashboards. A drift-alert sub-module monitors metric windows and fires alerts when values exceed expected bounds, while the MCP block extends coverage to tracing, metric summaries, alert rules, anomaly detection, and dashboard-ready metric series.

When to use:

  • Profiling a component's infer() call to find CPU or memory hotspots before a release
  • Measuring wall-clock time and peak memory of arbitrary code snippets from an agent
  • Setting up drift alerts on inference latency or token usage across a rolling window
  • Giving a non-developer a plain-English answer to "why is my app slow or flaky?"
  • Exporting Prometheus metrics and Grafana dashboard JSON for first-class monitoring

Example:

from mvp.observability import ObservabilityBlock, ObservabilityInput

block = ObservabilityBlock(name="observability")
result = block.infer(ObservabilityInput(
    op="profile",
    code="import math; [math.sqrt(i) for i in range(100000)]",
    top_n=5,
))
# result.value.report -> str; result.value.hotspots -> list[dict]

Non-developer MCP validation:

python scripts/validate_observability_mcp_install.py

This starts the observability MCP entrypoint in a temporary home/database, records a metric, exports Prometheus text, creates a dashboard, and verifies the Grafana export includes a Prometheus datasource input.

Prometheus/Grafana flow:

  1. Record metrics with record_metric.
  2. Export Prometheus text with export_prometheus.
  3. Create a dashboard with create_dashboard.
  4. Export Grafana import JSON with export_dashboard.

Prometheus/Grafana are optional at runtime, but they are first-class outputs: persisted metrics export without pre-registration, labels are preserved as distinct series, and Grafana imports prompt for a Prometheus datasource.

Works well with: cicd, align_evals, core

Public API

CardinalityCheck

Result of a cardinality check for a metric + label combination.

Field Type Default
allowed bool required
metric_name str required
current_cardinality int required
limit int required

CardinalityLimiter

Limits unique label combinations per metric to prevent metric explosion.

Constructor:

Parameter Type Default
default_limit int 1000

Methods:

set_limit(metric_name: str, limit: int) -> None

Set a per-metric cardinality limit (overrides default).

check_cardinality(metric_name: str, labels: dict[str, str]) -> CardinalityCheck

Check whether labels are allowed for metric_name.

get_stats() -> dict[str, int]

Return current cardinality count per tracked metric.

reset(metric_name: str | None = None) -> None

Reset tracked label sets (all metrics or a single one).

WorkflowRisk

Field Type Default
domain_risk str LOW
side_effect_risk str NONE

Methods:

is_high_impact() -> bool

ConvergenceMetrics

The MEASURED convergence signals (spec ConvergenceMetrics).

Field Type Default
blocking_constraints_total int 0
blocking_constraints_passed int 0
eval_score float 0.0
source_grounding_score float 1.0
tool_policy_violations int 0
unresolved_high_risk_items int 0
regression_failures int 0
iteration int 1
max_iterations int 1
delta_eval_from_last float 0.0
necessary_property_unverified bool False

ConvergenceDecision

Field Type Default
decision str required
reliability_label str required
converged bool required
hard_blockers list[str] field(default_factory=list)
reasons list[str] field(default_factory=list)

Methods:

to_dict() -> dict

ObservabilityBlock(AIBlock[ObservabilityInput, ObservabilityOutput, None])

Stateless profiling / measurement block (4 ops).

Field Type Default
name str 'observability'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
state None field(default=None, init=False)

Methods:

infer(data: ObservabilityInput) -> Result[ObservabilityOutput]

health() -> dict

Return health status for production monitoring.

ObservabilityInput(BaseModel)

Input for stateless observability operations.

Field Type Default
op Literal['profile', 'measure', 'analyze', 'report', 'check_cardinality', 'tracing_context'] required
code str ''
name str ''
format Literal['text', 'json'] 'text'
top_n int 10
timeout_sec float 10.0
metric_name str ''
labels dict[str, str] Field(default_factory=dict)
cardinality_limit int \| None None
headers dict[str, str] Field(default_factory=dict)

ObservabilityOutput(BaseModel)

Output from stateless observability operations.

Field Type Default
op str required
report str ''
metrics dict Field(default_factory=dict)
hotspots list[dict] Field(default_factory=list)
duration_sec float 0.0
memory_peak_mb float 0.0
success bool True
error str ''
degraded bool False
degradation_reason str \| None None
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] \| None None
evidence dict[str, Any] Field(default_factory=dict)
request_id str ''
task_id str ''
run_id str ''

TracingContext

Lightweight carrier for W3C-style distributed trace state.

Field Type Default
trace_id str ''
span_id str ''
parent_span_id str ''
sampled bool True

Methods:

new() -> TracingContext

Create a brand-new root tracing context.

child_span() -> TracingContext

Derive a child span that shares the same trace_id.

Functions

assess_convergence(metrics: ConvergenceMetrics, risk: WorkflowRisk | None = None, min_eval_score: float = DEFAULT_MIN_EVAL_SCORE, min_grounding_score: float = DEFAULT_MIN_GROUNDING_SCORE, material_improvement_epsilon: float = DEFAULT_MATERIAL_IMPROVEMENT_EPSILON) -> ConvergenceDecision

Decide ship / iterate / escalate / not-shippable from MEASURED metrics — never from a

propagate_context(headers: dict[str, str]) -> TracingContext

Extract tracing context from a W3C traceparent header.

inject_context(ctx: TracingContext) -> dict[str, str]

Produce a W3C traceparent header dict from ctx.

MCP Tools

Operation Source
start_profile observability_mcp
stop_profile observability_mcp
get_profile_results observability_mcp
profile_function observability_mcp
get_hotspots observability_mcp
compare_profiles observability_mcp
register_metric observability_mcp
record_metric observability_mcp
query_metrics observability_mcp
export_prometheus observability_mcp
get_metric_summary observability_mcp
create_dashboard observability_mcp
update_dashboard observability_mcp
list_dashboards observability_mcp
export_dashboard observability_mcp
start_span observability_mcp
end_span observability_mcp
get_trace observability_mcp
query_traces observability_mcp
create_alert observability_mcp
check_alerts observability_mcp
list_alerts observability_mcp
analyze_performance observability_mcp
optimize_config observability_mcp
detect_anomalies observability_mcp
search observability_mcp
info observability_mcp
capabilities observability_mcp
list_patterns observability_mcp