Production Deployment & Safety

Monitoring and Observability

5 min read

Production Computer Use agents need comprehensive monitoring to ensure reliability, debug issues, and maintain security.

What to Monitor

CategoryMetrics
PerformanceResponse time, loop iterations, completion rate
CostsTokens used, screenshots processed, API calls
ReliabilityError rate, retry count, timeout frequency
SecurityBlocked actions, suspicious patterns, failed auth

Logging Framework

import logging
from datetime import datetime

logger = logging.getLogger("computer_use_agent")

class AgentLogger:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.start_time = datetime.now()

    def log_action(self, action: dict, result: dict):
        logger.info({
            "session": self.session_id,
            "timestamp": datetime.now().isoformat(),
            "action_type": action.get("type"),
            "coordinates": action.get("coordinate"),
            "success": result.get("success"),
            "duration_ms": result.get("duration_ms")
        })

    def log_screenshot(self, size_bytes: int):
        logger.info({
            "session": self.session_id,
            "event": "screenshot",
            "size_bytes": size_bytes
        })

Session Recording

Record full sessions for debugging:

class SessionRecorder:
    def __init__(self):
        self.screenshots = []
        self.actions = []

    def record_frame(self, screenshot_b64: str, action: dict):
        self.screenshots.append(screenshot_b64)
        self.actions.append({
            "timestamp": time.time(),
            "action": action
        })

    def save_recording(self, path: str):
        # Save as video or frame sequence
        with open(path, 'w') as f:
            json.dump({
                "actions": self.actions,
                "frame_count": len(self.screenshots)
            }, f)

Cost Tracking

The thing that surprises teams in production is not the price per token — it is that a computer-use conversation re-sends its screenshots. Step 20 carries the twenty screenshots before it, so naive cost grows with the square of the step count, not linearly.

Move the sliders. Watch what pruning to the last few screenshots does:

What a computer-use task actually costs

Set the prices for the model you deploy (read them from the pricing page). The gap between the two figures is context growth, not token prices.

Agent steps per task20
Image tokens per screenshot1,200
Output tokens per step200
Screenshots kept in context3
Input price3$/1M
Output price15$/1M
Tasks per day50
Cost per task, no pruning
$1
Cost per task, pruned
$0
Saved per day by pruning
$27

At the defaults — a 20-step task, 1200 image tokens per screenshot, 50 tasks a day — that is $0.82 per task unpruned against $0.28 pruned, or about $27 a day for one config change. The pruning itself is three lines: before each request, drop the image content from all but the last few tool_result blocks. Claude keeps the text history and only loses old pixels it has already acted on.

class CostTracker:
    # Rates are per-model and change: read them from the pricing page, don't trust
    # this constant. https://platform.claude.com/docs/en/pricing
    # Values below are placeholders in $ per 1M tokens — set them for YOUR model.
    COSTS = {
        "input_token": 0.003 / 1000,   # Per token ($3 / 1M)
        "output_token": 0.015 / 1000,  # Per token ($15 / 1M)
        "image_token": 0.003 / 1000,   # Per image token (same as input)
    }

    def __init__(self):
        self.total_cost = 0
        self.call_count = 0

    def add_usage(self, response):
        usage = response.usage
        cost = (
            usage.input_tokens * self.COSTS["input_token"] +
            usage.output_tokens * self.COSTS["output_token"]
        )
        self.total_cost += cost
        self.call_count += 1

        return {
            "call_cost": cost,
            "total_cost": self.total_cost,
            "call_count": self.call_count
        }

Health Checks

async def health_check():
    checks = {
        "api_connection": await test_api_connection(),
        "display_available": await test_display(),
        "disk_space": check_disk_space(),
        "memory_available": check_memory()
    }

    healthy = all(checks.values())
    return {"healthy": healthy, "checks": checks}

Alerting

def check_and_alert(metrics):
    alerts = []

    if metrics["error_rate"] > 0.1:
        alerts.append("High error rate: >10%")

    if metrics["avg_response_time"] > 30:
        alerts.append("Slow responses: >30s average")

    if metrics["cost_per_task"] > 1.0:
        alerts.append("High costs: >$1 per task")

    if alerts:
        send_alert(alerts)

Dashboard Metrics

Essential metrics for your dashboard:

MetricTargetAlert Threshold
Success rate>95%<90%
Avg completion time<60s>120s
Cost per task<$0.50>$1.00
Error rate<5%>10%

Debugging Tools

# Replay failed sessions
def replay_session(session_id: str):
    session = load_session(session_id)
    for i, (screenshot, action) in enumerate(session):
        print(f"Step {i}: {action}")
        display_screenshot(screenshot)
        input("Press Enter to continue...")

Tip: Store session recordings for 7-30 days to debug issues reported by users.

That is the last piece: an agent you can watch, price, and switch off. Take the capstone and point it at a form that matters to you. :::

Quiz

Module 5: Production Deployment & Safety

Take Quiz
Was this lesson helpful?

Sign in to rate