Monitoring and logging

Lesson updated November 3, 2026 based on OpenAI’s Computer use guide and the Responses API documentation: logging now reads actions from a computer_call and elements from response.output (messages with their phase, reasoning summaries, calls), instead of the tool_use blocks from Anthropic’s API; hard-coded Claude Sonnet pricing is replaced by the pricing of the model in use, and the deprecated datetime.utcnow() (since Python 3.12) is replaced by datetime.now(timezone.utc). Sources: Computer use · Pricing.

Observing your agent in production

A Computer Use agent in production is a black box if you don’t set up proper monitoring. Unlike a standard API where you log requests and responses, here you must also trace visual actions, screenshots, and the model’s decisions. The difference is concrete: when a workflow fails at 3 a.m., the question that matters isn’t “what exception was raised,” but “what did the agent see, and why did it click there?” A log that doesn’t answer that question won’t help you the next morning.

The three levels of logging

Instrumentation therefore builds on three complementary layers: what the agent did, what it thought before doing it, and what it saw when deciding. Each writes to a different channel because neither the volume nor the useful lifespan is the same.

Level 1: Actions

Every action executed by the agent must be traced. The format below systematically prefixes the iteration number and the name of the current step, allowing you to filter a log of several thousand lines down to the single sequence that went wrong. Actions arrive as objects from the computer_call, with the same fields as in your handler (x, y, scroll_x, keys…). Note the handling of the type action: we record the length of the text entered, never its content. An agent filling out a login form types a password, and the rule from the previous lesson applies here without exception.

import logging

logger = logging.getLogger("computer_use")
logger.setLevel(logging.INFO)

handler = logging.FileHandler("agent.log")
handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
handler.addFilter(SensitiveFilter())  # filter from lesson 12
logger.addHandler(handler)


def log_action(action, iteration: int, step: str = ""):
    """Logs an action from a computer_call."""
    prefix = f"[iter={iteration}] [{step}]"
    match action.type:
        case "click" | "double_click" | "move":
            logger.info("%s %s at (%s, %s)", prefix, action.type.upper(), action.x, action.y)
        case "type":
            # Never the text itself: only its length
            logger.info("%s TYPE (%d chars)", prefix, len(action.text))
        case "scroll":
            logger.info("%s SCROLL dx=%s dy=%s", prefix, action.scroll_x, action.scroll_y)
        case "keypress":
            logger.info("%s KEYPRESS %s", prefix, "+".join(action.keys))
        case "drag":
            logger.info("%s DRAG (%d points)", prefix, len(action.path))
        case _:
            logger.info("%s %s", prefix, action.type.upper())

Level 2: Model decisions

Log what the model says about its choices to understand why it picked a specific action. A click at unexpected coordinates means nothing on its own; paired with the sentence that preceded it—“the submit button seems hidden by the cookie banner”—it becomes a diagnosis. The text is truncated to two hundred characters, more than enough to grasp the intent without exploding the file volume. Two sources of text exist in a Responses API response. Intermediate messages that the model writes before acting carry phase: "commentary", while the final response carries phase: "final_answer". And if the model in use supports reasoning summaries, you get them by requesting them (reasoning={"summary": "auto"}): they arrive in elements of type reasoning.

Token consumption is logged at the same pace, and it deserves close attention: even with previous_response_id, all input tokens from previous responses in the chain are re-billed as input, screenshots included. The cost of a turn therefore increases as the session lengthens.

def log_model_response(response, iteration: int):
    """Logs notes, summaries, and calls from a response (excluding images)."""
    for item in response.output:
        if item.type == "reasoning":
            for summary in item.summary:
                logger.info("[iter=%d] REASONING: %s", iteration, summary.text[:200])
        elif item.type == "message":
            text = "".join(c.text for c in item.content if c.type == "output_text")
            logger.info("[iter=%d] MESSAGE (%s): %s", iteration, item.phase or "-", text[:200])
        elif item.type == "computer_call":
            actions = ", ".join(a.type for a in item.actions or [])
            logger.info("[iter=%d] COMPUTER_CALL %s: %s", iteration, item.call_id, actions)
        elif item.type == "function_call":
            logger.info("[iter=%d] FUNCTION_CALL %s", iteration, item.name)

    # Tokens consumed
    logger.info(
        "[iter=%d] TOKENS: in=%d out=%d",
        iteration, response.usage.input_tokens, response.usage.output_tokens,
    )

Level 3: Screenshots

Save screenshots for post-mortem debugging, but watch your disk space: a PNG capture often weighs several hundred kilobytes, and a long workflow produces hundreds of them. The class below solves this with simple rotation. The filename starts with a timestamp, aligning alphabetical order with chronological order; sorting in _cleanup is then enough to identify the oldest captures and delete them until dropping back under the ceiling of one hundred files. The label field lets you annotate interesting moments, for example error or before_validation, and spot them at a glance in the folder.

from datetime import datetime, timezone
from pathlib import Path

class ScreenshotLogger:
    """Saves screenshots with rotation."""

    def __init__(self, base_dir: str = "logs/screenshots", max_files: int = 100):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)
        self.max_files = max_files
        self.counter = 0

    def save(self, screenshot_bytes: bytes, iteration: int, label: str = ""):
        """Saves a screenshot with a descriptive name."""
        ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        filename = f"{ts}_iter{iteration:03d}_{label}.png"
        filepath = self.base_dir / filename

        with open(filepath, "wb") as f:
            f.write(screenshot_bytes)

        self.counter += 1
        self._cleanup()

        return filepath

    def _cleanup(self):
        """Deletes old screenshots."""
        files = sorted(self.base_dir.glob("*.png"))
        while len(files) > self.max_files:
            files[0].unlink()
            files.pop(0)

Metrics to monitor

These four indicators are only valuable when compared to themselves over time: a workflow dropping from ninety-five to eighty percent success tells you an interface has moved, well before the first user complains. Therefore, implement a metrics collector attached to each execution. The WorkflowMetrics dataclass accumulates tokens iteration after iteration, keeps track of errors with their step and timestamp, and exposes two calculated properties: duration, which works even if the workflow isn’t finished yet, and a cost estimate based on the pricing of the model in use, which you take from the OpenAI pricing page rather than hard-coding. The summary method produces the compact dictionary you will archive at each execution.

from dataclasses import dataclass, field
from datetime import datetime, timezone

def now() -> datetime:
    return datetime.now(timezone.utc)

@dataclass
class WorkflowMetrics:
    """Collects metrics for a workflow."""
    workflow_id: str
    input_price_per_million: float  # pricing for the model in use
    output_price_per_million: float
    start_time: datetime = field(default_factory=now)
    end_time: datetime | None = None
    iterations: int = 0
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    errors: list = field(default_factory=list)
    steps_completed: int = 0
    steps_total: int = 0

    def record_iteration(self, response):
        self.iterations += 1
        self.total_input_tokens += response.usage.input_tokens
        self.total_output_tokens += response.usage.output_tokens

    def record_error(self, error: str, step: str):
        self.errors.append({
            "step": step,
            "error": error,
            "timestamp": now().isoformat(),
        })

    def finalize(self):
        self.end_time = now()

    @property
    def duration_seconds(self) -> float:
        end = self.end_time or now()
        return (end - self.start_time).total_seconds()

    @property
    def estimated_cost_usd(self) -> float:
        # Estimate: prompt caching and a hosted environment are not counted
        input_cost = self.total_input_tokens / 1_000_000 * self.input_price_per_million
        output_cost = self.total_output_tokens / 1_000_000 * self.output_price_per_million
        return input_cost + output_cost

    def summary(self) -> dict:
        return {
            "workflow_id": self.workflow_id,
            "duration_s": round(self.duration_seconds, 1),
            "iterations": self.iterations,
            "tokens": self.total_input_tokens + self.total_output_tokens,
            "cost_usd": round(self.estimated_cost_usd, 4),
            "success": len(self.errors) == 0,
            "errors": len(self.errors),
            "steps": f"{self.steps_completed}/{self.steps_total}",
        }

Monitoring dashboard

For real-time monitoring, send metrics to a monitoring service rather than letting them sleep in a file. The function below posts the timestamped summary to an HTTP endpoint, with two precautions that matter as much as the sending itself: a maximum delay of five seconds, and a caught exception that settles for a warning. An unavailable metrics collector must never bring down the workflow it observes—the opposite would be the definition of counterproductive supervision.

import requests  # pip install requests

def push_metrics(metrics: WorkflowMetrics, endpoint: str):
    """Sends metrics to a monitoring service."""
    payload = metrics.summary()
    payload["timestamp"] = now().isoformat()

    try:
        requests.post(
            endpoint,
            json=payload,  # requests sets the JSON header itself
            timeout=5,
        )
    except requests.RequestException as e:
        logger.warning(f"Metrics push error: {e}")

Automatic alerts

A dashboard that no one watches is just an archive. Therefore, configure alerts based on thresholds so the anomaly comes to you. The four conditions retained here cover classic drifts: a cost higher than two dollars, a duration exceeding three minutes, more than thirty iterations, or more than two errors in the same execution. These values are starting points, not truths: calibrate them on measurements from your own successful executions, placing the threshold well above your observed median. An alert that triggers every day will be ignored by the second week.

def check_alerts(metrics: WorkflowMetrics):
    """Checks thresholds and triggers alerts."""
    alerts = []

    if metrics.estimated_cost_usd > 2.0:
        alerts.append(f"High cost: ${metrics.estimated_cost_usd:.2f}")

    if metrics.duration_seconds > 180:
        alerts.append(f"Excessive duration: {metrics.duration_seconds:.0f}s")

    if metrics.iterations > 30:
        alerts.append(f"Too many iterations: {metrics.iterations}")

    if len(metrics.errors) > 2:
        alerts.append(f"Multiple errors: {len(metrics.errors)}")

    for alert in alerts:
        logger.warning(f"ALERT: {alert}")
        # Send notification (email, Slack, etc.)

    return alerts

Make it a habit to call check_alerts right after finalize, then push_metrics: you get an archived trace at each execution, a signal sent to the dashboard, and a notification only when something deserves your attention.

Key takeaways

• Log actions, model messages and summaries, and captures on three separate levels
• Monitor success rate, duration, token cost, and errors; the chained history is re-billed as input each turn
• Implement screenshot rotation to avoid saturating the disk
• Set up automatic alerts on critical thresholds
• Never log sensitive data (passwords, tokens, personal data)