Building Custom Analytics Reports: Scheduled Delivery of Agent Performance Data

Why Scheduled Reports Still Matter

Dashboards are powerful but passive. Stakeholders who do not log into Grafana daily miss important trends. Scheduled reports push insights to the people who need them, ensuring that performance changes are noticed and acted on without requiring anyone to remember to check a dashboard.

A well-designed weekly report becomes the heartbeat of your AI agent program, creating accountability and driving continuous improvement.

Report Data Aggregation

The first step is aggregating raw analytics data into report-ready summaries. A report typically covers a time period and compares it to the previous period.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

from dataclasses import dataclass
from datetime import datetime, timedelta
import psycopg2

@dataclass
class ReportPeriod:
    start: datetime
    end: datetime
    label: str

def get_report_periods(report_date: datetime) -> tuple:
    current_end = report_date
    current_start = report_date - timedelta(days=7)
    previous_end = current_start
    previous_start = previous_end - timedelta(days=7)
    return (
        ReportPeriod(current_start, current_end, "This Week"),
        ReportPeriod(previous_start, previous_end, "Last Week"),
    )

def aggregate_metrics(conn_string: str, period: ReportPeriod) -> dict:
    conn = psycopg2.connect(conn_string)
    cur = conn.cursor()

    cur.execute("""
        SELECT
            COUNT(DISTINCT conversation_id) AS total_conversations,
            COUNT(*) FILTER (WHERE event_type = 'resolution') AS resolutions,
            COUNT(*) FILTER (WHERE event_type = 'escalation') AS escalations,
            COUNT(*) FILTER (WHERE event_type = 'error') AS errors,
            SUM(token_count) AS total_tokens,
            AVG(latency_ms) AS avg_latency_ms
        FROM agent_events
        WHERE event_ts BETWEEN %s AND %s
    """, (period.start, period.end))

    row = cur.fetchone()
    total = row[0] or 0
    resolutions = row[1] or 0
    escalations = row[2] or 0

    cur.close()
    conn.close()

    return {
        "period": period.label,
        "total_conversations": total,
        "resolutions": resolutions,
        "escalations": escalations,
        "errors": row[3] or 0,
        "total_tokens": row[4] or 0,
        "avg_latency_ms": round(row[5] or 0, 1),
        "resolution_rate": round(
            resolutions / total * 100, 1
        ) if total else 0,
        "containment_rate": round(
            (total - escalations) / total * 100, 1
        ) if total else 0,
    }

Computing Period-over-Period Changes

Stakeholders care about trends, not just numbers. Comparing the current period to the previous one makes changes immediately visible.

def compute_changes(
    current: dict, previous: dict
) -> dict:
    changes = {}
    numeric_keys = [
        "total_conversations", "resolution_rate",
        "containment_rate", "errors", "avg_latency_ms",
    ]
    for key in numeric_keys:
        curr_val = current.get(key, 0)
        prev_val = previous.get(key, 0)
        if prev_val == 0:
            pct_change = 0 if curr_val == 0 else 100
        else:
            pct_change = round(
                (curr_val - prev_val) / prev_val * 100, 1
            )
        direction = "up" if pct_change > 0 else "down" if pct_change < 0 else "flat"
        changes[key] = {
            "current": curr_val,
            "previous": prev_val,
            "change_pct": pct_change,
            "direction": direction,
        }
    return changes

HTML Report Generation

Generate HTML reports that can be sent via email or converted to PDF. Use a template approach with inline styles for email compatibility.

def generate_html_report(
    metrics: dict, changes: dict, report_date: str
) -> str:
    def change_badge(key: str, higher_is_better: bool = True) -> str:
        info = changes.get(key, {})
        pct = info.get("change_pct", 0)
        direction = info.get("direction", "flat")
        if direction == "flat":
            color = "#6b7280"
            arrow = "~"
        elif (direction == "up" and higher_is_better) or \
             (direction == "down" and not higher_is_better):
            color = "#10b981"
            arrow = "+"
        else:
            color = "#ef4444"
            arrow = ""
        return (
            f'<span style="color:{color};font-weight:bold">'
            f'{arrow}{pct}%</span>'
        )

    html = f"""
    <html>
    <body style="font-family:Arial,sans-serif;max-width:600px;margin:0 auto">
    <h1 style="color:#1f2937">Agent Performance Report</h1>
    <p style="color:#6b7280">Week ending {report_date}</p>
    <table style="width:100%;border-collapse:collapse">
      <tr style="background:#f3f4f6">
        <th style="padding:12px;text-align:left">Metric</th>
        <th style="padding:12px;text-align:right">Value</th>
        <th style="padding:12px;text-align:right">vs Last Week</th>
      </tr>
      <tr>
        <td style="padding:12px">Conversations</td>
        <td style="padding:12px;text-align:right">
          {metrics['total_conversations']:,}</td>
        <td style="padding:12px;text-align:right">
          {change_badge('total_conversations')}</td>
      </tr>
      <tr style="background:#f9fafb">
        <td style="padding:12px">Resolution Rate</td>
        <td style="padding:12px;text-align:right">
          {metrics['resolution_rate']}%</td>
        <td style="padding:12px;text-align:right">
          {change_badge('resolution_rate')}</td>
      </tr>
      <tr>
        <td style="padding:12px">Containment Rate</td>
        <td style="padding:12px;text-align:right">
          {metrics['containment_rate']}%</td>
        <td style="padding:12px;text-align:right">
          {change_badge('containment_rate')}</td>
      </tr>
      <tr style="background:#f9fafb">
        <td style="padding:12px">Avg Latency</td>
        <td style="padding:12px;text-align:right">
          {metrics['avg_latency_ms']}ms</td>
        <td style="padding:12px;text-align:right">
          {change_badge('avg_latency_ms', higher_is_better=False)}</td>
      </tr>
      <tr>
        <td style="padding:12px">Errors</td>
        <td style="padding:12px;text-align:right">
          {metrics['errors']}</td>
        <td style="padding:12px;text-align:right">
          {change_badge('errors', higher_is_better=False)}</td>
      </tr>
    </table>
    </body></html>
    """
    return html

Delivery via Email and Slack

Schedule report delivery using email for formal distribution and Slack for team awareness.

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import httpx
import os

def send_email_report(
    html: str, recipients: list[str], subject: str
) -> None:
    msg = MIMEMultipart("alternative")
    msg["Subject"] = subject
    msg["From"] = os.environ["SMTP_FROM"]
    msg["To"] = ", ".join(recipients)
    msg.attach(MIMEText(html, "html"))

    with smtplib.SMTP(
        os.environ["SMTP_HOST"],
        int(os.environ.get("SMTP_PORT", 587)),
    ) as server:
        server.starttls()
        server.login(
            os.environ["SMTP_USER"],
            os.environ["SMTP_PASSWORD"],
        )
        server.send_message(msg)

def send_slack_summary(
    metrics: dict, changes: dict, webhook_url: str
) -> None:
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": "Weekly Agent Performance Report",
            },
        },
        {
            "type": "section",
            "fields": [
                {
                    "type": "mrkdwn",
                    "text": (
                        f"*Conversations:* {metrics['total_conversations']:,}"
                    ),
                },
                {
                    "type": "mrkdwn",
                    "text": (
                        f"*Resolution Rate:* {metrics['resolution_rate']}%"
                    ),
                },
                {
                    "type": "mrkdwn",
                    "text": (
                        f"*Containment:* {metrics['containment_rate']}%"
                    ),
                },
                {
                    "type": "mrkdwn",
                    "text": f"*Errors:* {metrics['errors']}",
                },
            ],
        },
    ]
    httpx.post(webhook_url, json={"blocks": blocks})

Scheduling with APScheduler

Automate the entire pipeline to run weekly without manual intervention.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime

scheduler = AsyncIOScheduler()

async def weekly_report_job():
    report_date = datetime.utcnow()
    current_period, previous_period = get_report_periods(report_date)

    conn_string = os.environ["DATABASE_URL"]
    current_metrics = aggregate_metrics(conn_string, current_period)
    previous_metrics = aggregate_metrics(conn_string, previous_period)
    changes = compute_changes(current_metrics, previous_metrics)

    html = generate_html_report(
        current_metrics, changes, report_date.strftime("%Y-%m-%d")
    )

    send_email_report(
        html,
        recipients=["[email protected]", "[email protected]"],
        subject=f"Agent Report - Week of {report_date.strftime('%b %d')}",
    )
    send_slack_summary(
        current_metrics, changes,
        webhook_url=os.environ["SLACK_WEBHOOK_URL"],
    )

scheduler.add_job(
    weekly_report_job,
    trigger="cron",
    day_of_week="mon",
    hour=9,
    minute=0,
)
scheduler.start()

FAQ

Should I send the same report to engineers and executives?

No. Engineers want granular data: error types, latency percentiles, token usage breakdowns, and specific failure examples. Executives want outcomes: resolution rate trends, cost savings, and volume growth. Create two report templates from the same data, or use a single report with an executive summary at the top and detailed appendices below.

What is the best format for emailed reports?

HTML with inline styles works most reliably across email clients. Avoid external CSS, JavaScript, or embedded images that need to load from your server. For stakeholders who prefer documents, generate a PDF attachment alongside the HTML email. The Python weasyprint library converts HTML to PDF cleanly.

How do I handle reports when data is incomplete or delayed?

Include a data quality section in every report. Show the percentage of expected events that were actually received and flag any gaps. If data completeness drops below 95%, add a visible warning banner to the report. This prevents stakeholders from making decisions based on incomplete data and builds trust in the reporting system.

---

#Reporting #Automation #Scheduling #Analytics #AIAgents #AgenticAI #LearnAI #AIEngineering