Deploying and Governing Agentic AI in Practice

# Generated from ch17_governance_monitoring/01_monitoring_dashboard.py for in-browser execution.
# Agno agent wiring is removed so the tool runs as a plain function;
# the full version is on GitHub. Do not edit: regenerate with
# scripts/build_demos.py.
# Monitoring dashboard tool.
# Book reference: Chapter 17, §17.5 "Monitoring and Observability"
# Repo note: read_thresholds, read_metrics, check_against_thresholds and
# aggregate_status live in support.py; the registry ships in
# data/metrics_registry.json. For an agent exercising this tool, see
# 02_governance_agent.py.

from support import (
    aggregate_status,
    check_against_thresholds,
    read_metrics,
    read_thresholds,
)


def monitoring_dashboard(
    agent_name: str,
    window_days: int = 7,
) -> dict:
    """Return operational metrics for the named agent over the window.

    Read-only against the operational metrics store. Returns a
    structured-status dict. Threshold values come from the firm's
    monitoring configuration registry, not from the agent's runtime
    state.
    """
    # Read pinned thresholds from the operational metrics registry.
    thresholds = read_thresholds(agent_name)            # firm-maintained
    metrics    = read_metrics(agent_name, window_days)  # rolling window

    # Threshold check produces the diagnostic surface (cells out of band).
    diagnostic = check_against_thresholds(metrics, thresholds)

    # Status is the worst case across the six metrics.
    status = aggregate_status(diagnostic)
    # Status values: nominal, degraded, out_of_range, incident.

    return {
        "status":     status,
        "data":       metrics,
        "diagnostic": diagnostic,
        "version":    thresholds["registry_version"],
    }


if __name__ == "__main__":
    # data_quality_agent runs nominal; reserving_agent trips thresholds.
    print(monitoring_dashboard("data_quality_agent"))
    print(monitoring_dashboard("reserving_agent"))

# Repo demo (not a printed listing): a simple governance agent
# exercising the Chapter 17 monitoring_dashboard tool. The tool itself,
# and its direct test, live in 01_monitoring_dashboard.py — run that
# first to see the raw structured-status dict the agent consumes here.
import importlib

from agno.agent import Agent
import os
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))  # for common/
from common.config import get_model

# Numeric module names can't be imported with a plain import statement.
monitoring_dashboard = importlib.import_module(
    "01_monitoring_dashboard").monitoring_dashboard

# The reviewer asks for an agent's operational status in plain language
# and gets the diagnostic surface summarised, with escalation advice
# when out of band.
governance_agent = Agent(
    model=get_model(),
    tools=[monitoring_dashboard],
    tool_call_limit=4,
    markdown=True,
    instructions=(
        "Report the operational status of the named agent using the "
        "monitoring_dashboard tool. Summarise any breached thresholds "
        "and recommend escalation when the status is not nominal. "
        "Only call the tools you have been given."
    ),
)

if __name__ == "__main__":
    governance_agent.print_response(
        "Check the 7-day operational status of reserving_agent and "
        "summarise any threshold breaches for the governance committee.",
        stream=True,
    )

Runs the unmodified chapter script on the server and streams the agent's tool calls and reasoning here.