Desktop Automation Reimagined
On March 24, 2026, Anthropic formally launched Computer Use for macOS, marking a fundamental shift in how we approach desktop automation. Unlike traditional scripting tools or RPA platforms, Computer Use harnesses Claude's natural language understanding and reasoning to handle the messy reality of real-world applications—unpredictable window focus, dialog boxes, network latency, and the chaos of actual human workflows.
This guide walks you through production-ready automation patterns that go beyond simple scripts. We'll cover overnight batch processing, multi-application orchestration, and the security practices that distinguish laboratory experiments from reliable systems.
Computer Use requires macOS 12+, 4GB RAM, and integration with Claude's Dispatch framework for background execution. All patterns in this guide prioritize observability, error recovery, and security.
Pattern 1: Overnight Batch Processing
The Scenario
Your sales team exports daily reports (revenue, customer acquisition, churn) as Excel files to a shared folder. Each night at 10 PM, you need to automatically:
- Detect new reports in
~/Downloads/reports/ - Validate file signatures (ensuring they're legitimate reports)
- Convert Excel → CSV for compatibility
- Upload to Google Sheets
- Archive to cloud storage
Security-First Architecture
The key to reliable batch processing is defensive programming:
Pre-execution Validation Checklist:
1. File location verification (whitelist only /Downloads/reports)
2. File signature validation (match known report patterns)
3. Pre-run simulation (show what will happen, wait for approval)
4. Comprehensive logging (screenshots + operation history)
5. Automatic rollback on failure + Slack notification to admins
Implementation Example
#!/usr/bin/env python3
# batch_processor.py - Runs nightly via Dispatch
import os
import json
import logging
from datetime import datetime
from pathlib import Path
import subprocess
import asyncio
# Configuration
MONITORED_DIR = Path.home() / "Downloads" / "reports"
ALLOWED_PATTERNS = ["sales_*.xlsx", "churn_*.xlsx", "acq_*.xlsx"]
LOG_DIR = Path("/var/log/batch_automation")
LOG_DIR.mkdir(exist_ok=True)
# Setup logging
logging.basicConfig(
filename=LOG_DIR / f"{datetime.now():%Y%m%d}.log",
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def validate_file(file_path: Path) -> bool:
"""Ensure file matches allowed report patterns"""
return any(
file_path.match(pattern)
for pattern in ALLOWED_PATTERNS
)
def dry_run(files: list) -> bool:
"""Display what will be processed and wait for approval"""
if not files:
print("ℹ️ No new files detected")
return False
print("\n" + "="*60)
print("DRY RUN: Files to process")
print("="*60)
for file in files:
size_kb = file.stat().st_size / 1024
print(f" • {file.name} ({size_kb:.1f} KB)")
print("="*60)
# In Computer Use context, Claude would wait for user confirmation
response = input("\nProceed with batch processing? (yes/no): ")
return response.lower() == 'yes'
async def process_files(files: list) -> dict:
"""Convert, validate, and upload files"""
results = {"success": [], "failed": []}
for file in files:
try:
logging.info(f"Processing: {file.name}")
print(f"🔄 Processing {file.name}...", end=" ")
# Step 1: Convert Excel to CSV
csv_file = file.with_suffix(".csv")
subprocess.run(
["ssconvert", str(file), str(csv_file)],
check=True,
capture_output=True,
timeout=30
)
# Step 2: Upload to Google Sheets
upload_cmd = [
"python3", "upload_to_sheets.py",
"--file", str(csv_file),
"--sheet-id", os.environ["GOOGLE_SHEET_ID"]
]
subprocess.run(upload_cmd, check=True, capture_output=True, timeout=60)
# Step 3: Archive original
archive_path = Path.home() / "Google Drive" / "Reports" / "Archive" / file.name
file.replace(archive_path)
results["success"].append(file.name)
logging.info(f"✅ Success: {file.name}")
print("✅ Done")
except subprocess.TimeoutExpired:
results["failed"].append((file.name, "Timeout"))
logging.error(f"❌ Timeout: {file.name}")
print("❌ Timeout")
except Exception as e:
results["failed"].append((file.name, str(e)))
logging.error(f"❌ Error: {file.name} - {str(e)}")
print(f"❌ {type(e).__name__}")
return results
def notify_admins(results: dict) -> None:
"""Send Slack notification with results"""
if not results["failed"]:
message = f"✅ Batch processing complete. Processed {len(results['success'])} files."
else:
failed_list = "\n".join([f" • {name}: {error}" for name, error in results["failed"]])
message = f"""
❌ Batch processing completed with errors
✅ Success: {len(results['success'])}
❌ Failed: {len(results['failed'])}
Failed files:
{failed_list}
"""
webhook = os.environ.get("SLACK_WEBHOOK_URL")
if webhook:
subprocess.run([
"curl", "-X", "POST", webhook,
"-H", "Content-Type: application/json",
"-d", json.dumps({"text": message})
])
async def main():
# Detect new files
files = [
f for f in MONITORED_DIR.glob("*.xlsx")
if validate_file(f)
]
if not files:
logging.info("No files to process")
return
# Dry run (human approval)
if not dry_run(files):
logging.info("User cancelled batch processing")
return
# Execute batch processing
results = await process_files(files)
# Notify admins
notify_admins(results)
if __name__ == "__main__":
asyncio.run(main())Deployment with Dispatch:
# scheduled_batch.py
from anthropic_dispatch import scheduled
import subprocess
@scheduled("0 22 * * *") # Daily at 10 PM
async def nightly_batch():
"""Claude Computer Use triggers overnight batch automation"""
subprocess.run(["python3", "/path/to/batch_processor.py"])Pattern 2: Multi-Application Orchestration
The Scenario
Before your weekly sales meeting, you need a seamless workflow:
- Slack: Pull new deal information from #sales_leads
- Notion: Add customer details to your CRM database
- Figma: Download the latest metrics dashboard
- Keynote: Open deck and update this week's numbers
- Google Meet: Create meeting link and post to #sales_team
Each application has different delays, authentication states, and UI patterns. The automation must be resilient.
Robustness Through Observation
Multi-App Orchestration Principles:
1. Screenshot before/after each step (visual confirmation)
2. OCR recognition (find buttons by content, not coordinates)
3. Explicit error detection (watch for error dialogs, loading spinners)
4. Flexible timeouts (network delays vary by application)
5. Graceful degradation (skip non-critical steps if they fail)
Implementation
#!/usr/bin/env python3
# meeting_prep_orchestration.py
import anthropic
import asyncio
import json
import time
from datetime import datetime
from typing import List, Dict
client = anthropic.Anthropic()
class OrchestratedTask:
def __init__(self, app: str, action: str, timeout: int = 60):
self.app = app
self.action = action
self.timeout = timeout
self.result = None
self.error = None
async def execute(self) -> bool:
"""Execute task via Computer Use"""
try:
print(f"\n🔄 [{self.app}] {self.action}")
response = await asyncio.wait_for(
self._call_computer_use(),
timeout=self.timeout
)
self.result = response
print(f"✅ [{self.app}] Complete")
return True
except asyncio.TimeoutError:
self.error = f"Timeout after {self.timeout}s"
print(f"⏱️ [{self.app}] {self.error}")
return False
except Exception as e:
self.error = str(e)
print(f"❌ [{self.app}] {self.error}")
return False
finally:
await asyncio.sleep(2) # Pause between apps
async def _call_computer_use(self):
"""Call Claude with Computer Use tool"""
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: client.messages.create(
model="claude-opus-4-6",
max_tokens=2048,
tools=[{"type": "computer_use", "name": "computer_use"}],
messages=[{
"role": "user",
"content": self.action
}]
)
)
return response
async def orchestrate_meeting_prep() -> Dict:
"""Orchestrate multi-step meeting preparation"""
tasks = [
OrchestratedTask(
"Slack",
"Open Slack app. In channel #sales_leads, scroll and read the last 5 messages posted in the last 24 hours. Summarize the new deal information.",
timeout=45
),
OrchestratedTask(
"Notion",
"Open Notion. Navigate to 'CRM Database'. Add a new row with the customer information from Slack. Set deal size, probability, and expected close date.",
timeout=60
),
OrchestratedTask(
"Figma",
"Open Figma. Find the file 'Weekly_Metrics' in your drafts. Download the current version of the 'Dashboard' board.",
timeout=45
),
OrchestratedTask(
"Keynote",
"Open Keynote. Open the 'Sales_Meeting_Deck' file. Find the 'Metrics' slide. Update the 'Total Revenue' number to this week's latest figure: $2,450,000.",
timeout=90
),
OrchestratedTask(
"Google Meet",
"Open Google Meet. Create a new meeting. Copy the meeting link. Open Slack. In #sales_team, post: 'Weekly Sales Meeting: [meeting link]'",
timeout=45
)
]
print(f"\n{'='*60}")
print(f"Meeting Prep Orchestration - {datetime.now().strftime('%H:%M:%S')}")
print(f"{'='*60}")
results = []
for task in tasks:
success = await task.execute()
results.append({
"app": task.app,
"status": "success" if success else "failed",
"error": task.error,
"result": task.result
})
# Summary
successful = sum(1 for r in results if r["status"] == "success")
print(f"\n{'='*60}")
print(f"Summary: {successful}/{len(results)} tasks completed")
print(f"{'='*60}\n")
return {"tasks": results, "timestamp": datetime.now().isoformat()}
# Run orchestration
if __name__ == "__main__":
results = asyncio.run(orchestrate_meeting_prep())
print(json.dumps(results, indent=2))Pattern 3: Spreadsheet Automation at Scale
The Scenario
Your Google Sheets sales pipeline needs to stay current. You want to:
- Auto-calculate deal rankings (Probability × Deal Size)
- Flag deals closing soon in red
- Generate pipeline summary charts
- Backup to CSV weekly
Key Considerations
Spreadsheet Automation Safeguards:
1. Never hardcode credentials (use environment variables)
2. Batch updates via API (avoid cell-by-cell edits)
3. Validate formulas (prevent circular references)
4. Conditional formatting automation
5. Screenshot verification before committing large changes
#!/usr/bin/env python3
# pipeline_automation.py
import gspread
import os
from datetime import datetime, date, timedelta
from typing import List, Dict
# Initialize Google Sheets API
gc = gspread.service_account(
filename=os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"]
)
class PipelineAutomation:
def __init__(self, sheet_id: str):
self.sh = gc.open_by_key(sheet_id)
self.ws = self.sh.worksheet("Pipeline")
self.records = self.ws.get_all_records()
def calculate_deal_rankings(self) -> List[Dict]:
"""Calculate Probability of Close (POC) for each deal"""
updates = []
for i, record in enumerate(self.records, 2):
try:
prob = float(record.get("Close_Probability", 0)) / 100
deal_size = float(record.get("Deal_Size", 0))
pot = prob * deal_size
# Find POT column
pot_col = self._get_column_index("POT")
updates.append({
"range": f"{chr(64 + pot_col)}{i}",
"values": [[f"{pot:,.0f}"]]
})
except ValueError:
pass
return updates
def flag_closing_soon(self, days_threshold: int = 7) -> None:
"""Apply conditional formatting for deals closing soon"""
today = date.today()
deadline = today + timedelta(days=days_threshold)
close_date_col = self._get_column_index("Close_Date")
col_letter = chr(64 + close_date_col)
# Apply red background to deals closing within threshold
self.ws.format(
f"{col_letter}2:{col_letter}1000",
{
"conditionalFormats": [
{
"ranges": [f"{col_letter}2:{col_letter}1000"],
"booleanRule": {
"condition": {
"type": "DATE_BEFORE",
"values": [deadline.isoformat()]
},
"format": {
"backgroundColor": {
"red": 1.0,
"green": 0.8,
"blue": 0.8
}
}
}
}
]
}
)
def generate_summary_metrics(self) -> Dict:
"""Calculate pipeline summary"""
total_pipeline = sum(
float(r.get("Deal_Size", 0)) for r in self.records
)
weighted_pipeline = sum(
float(r.get("Deal_Size", 0)) * (float(r.get("Close_Probability", 0)) / 100)
for r in self.records
)
return {
"total_pipeline": total_pipeline,
"weighted_pipeline": weighted_pipeline,
"deal_count": len(self.records),
"updated_at": datetime.now().isoformat()
}
def _get_column_index(self, header: str) -> int:
"""Find column index by header name"""
headers = self.ws.row_values(1)
try:
return headers.index(header) + 1
except ValueError:
raise ValueError(f"Column '{header}' not found")
def apply_updates(self, updates: List[Dict]) -> None:
"""Batch apply spreadsheet updates"""
if updates:
self.ws.batch_update(updates)
print(f"✅ Applied {len(updates)} updates")
# Main automation
if __name__ == "__main__":
sheet_id = os.environ["PIPELINE_SHEET_ID"]
automation = PipelineAutomation(sheet_id)
# Step 1: Calculate rankings
updates = automation.calculate_deal_rankings()
automation.apply_updates(updates)
# Step 2: Apply conditional formatting
automation.flag_closing_soon(days_threshold=7)
# Step 3: Generate and log summary
summary = automation.generate_summary_metrics()
print(f"\n📊 Pipeline Summary:")
print(f" Total: ${summary['total_pipeline']:,.0f}")
print(f" Weighted: ${summary['weighted_pipeline']:,.0f}")
print(f" Deals: {summary['deal_count']}")Pattern 4: Browser Automation with Ethics
The Scenario
Monitor competitor pricing across 5 SaaS platforms. Daily at 9 AM, fetch current pricing and store in a Google Sheet for analysis.
The Ethics & Legality Framework
Responsible Web Automation:
✅ DO:
• Check robots.txt before scraping
• Use official APIs when available
• Add delays between requests (2-5 seconds minimum)
• Identify yourself honestly (User-Agent)
• Cache results (don't re-fetch same page twice)
❌ DON'T:
• Bypass authentication mechanisms
• Ignore robots.txt restrictions
• Hammer servers with rapid requests
• Disguise your identity
• Violate terms of service
#!/usr/bin/env python3
# competitor_price_monitor.py
import asyncio
import os
import json
import random
import gspread
from datetime import datetime
from typing import List, Dict
import anthropic
client = anthropic.Anthropic()
class CompetitorMonitor:
def __init__(self, sheet_id: str):
gc = gspread.service_account(
filename=os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"]
)
self.sh = gc.open_by_key(sheet_id)
self.ws = self.sh.worksheet(datetime.now().strftime("%Y-%m"))
async def fetch_pricing(self, url: str, selectors: Dict[str, str]) -> Dict:
"""Fetch pricing from competitor using Computer Use"""
prompt = f"""
Please visit this URL and extract pricing information:
{url}
Look for these pricing elements and extract the current prices:
{json.dumps(selectors, indent=2)}
Return the results as a JSON object like:
{{"plan_name": "price_in_usd", ...}}
Important: Please wait for the page to fully load, and if you see any
disclaimers about scraping, stop and inform me.
"""
# Add delay to respect server load
await asyncio.sleep(random.uniform(3, 6))
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
tools=[{"type": "computer_use", "name": "computer_use"}],
messages=[{"role": "user", "content": prompt}]
)
return response
async def monitor_all(self, competitors: List[Dict]) -> List[Dict]:
"""Monitor all competitors"""
results = []
for comp in competitors:
try:
pricing = await self.fetch_pricing(comp["url"], comp["selectors"])
row = [
datetime.now().isoformat(),
comp["name"],
json.dumps(pricing)
]
self.ws.append_row(row)
results.append({
"competitor": comp["name"],
"status": "success",
"data": pricing
})
except Exception as e:
results.append({
"competitor": comp["name"],
"status": "failed",
"error": str(e)
})
return results
# Configuration
COMPETITORS = [
{
"name": "Competitor A",
"url": "https://competitor-a.com/pricing",
"selectors": {
"basic": ".plan-basic .price",
"pro": ".plan-pro .price"
}
},
{
"name": "Competitor B",
"url": "https://competitor-b.com/plans",
"selectors": {
"starter": "#plan-starter .amount",
"professional": "#plan-pro .amount"
}
}
]
if __name__ == "__main__":
monitor = CompetitorMonitor(os.environ["COMPETITOR_SHEET_ID"])
results = asyncio.run(monitor.monitor_all(COMPETITORS))
successful = sum(1 for r in results if r["status"] == "success")
print(f"\n✅ Monitoring complete: {successful}/{len(results)} successful")Pattern 5: Xcode Build Automation
The Scenario
Automate iOS app builds every night. Run tests, capture logs, notify the team of failures.
Build Pipeline Resilience
Robust Build Automation:
1. Pre-build verification (CocoaPods, signing certificates)
2. Real-time error monitoring (syntax errors, linker errors)
3. Automatic cleanup (derived data, old builds)
4. Detailed failure reports with actionable information
5. Rollback strategy (revert to last successful commit on failure)
#!/usr/bin/env python3
# ios_build_automation.py
import subprocess
import json
import os
import asyncio
from datetime import datetime
from typing import Dict, List
class iOSBuildAutomation:
def __init__(self, project_path: str, scheme: str):
self.project_path = project_path
self.scheme = scheme
self.results = []
async def run_prebuild_checks(self) -> bool:
"""Verify build environment before compilation"""
print(f"🔍 Pre-build checks for {self.scheme}...")
try:
# Check CocoaPods
result = subprocess.run(
["pod", "repo", "update"],
cwd=self.project_path,
capture_output=True,
timeout=120,
text=True
)
if result.returncode != 0:
print(f"❌ CocoaPods update failed")
return False
print("✅ CocoaPods updated")
return True
except Exception as e:
print(f"❌ Pre-build check failed: {e}")
return False
async def build(self) -> Dict:
"""Execute the build"""
print(f"🏗️ Building {self.scheme}...")
try:
result = subprocess.run(
[
"xcodebuild",
"-scheme", self.scheme,
"-configuration", "Debug",
"-derivedDataPath", "/tmp/xcode_build",
"build",
"2>&1"
],
cwd=self.project_path,
capture_output=True,
timeout=600,
text=True
)
if result.returncode == 0:
return {"status": "success", "message": "Build successful"}
# Parse errors from output
errors = [
line for line in result.stdout.split('\n')
if "error:" in line.lower()
]
return {
"status": "failed",
"message": "Build failed",
"errors": errors[:5], # First 5 errors
"full_log": result.stdout
}
except subprocess.TimeoutExpired:
return {
"status": "failed",
"message": "Build timeout (10 minutes exceeded)"
}
except Exception as e:
return {
"status": "failed",
"message": str(e)
}
async def cleanup(self) -> None:
"""Clean up build artifacts"""
try:
subprocess.run(
["rm", "-rf", "/tmp/xcode_build"],
capture_output=True
)
print("🧹 Cleaned up temporary build files")
except:
pass
async def execute(self) -> Dict:
"""Full build workflow"""
if not await self.run_prebuild_checks():
return {"status": "failed", "reason": "Pre-build checks failed"}
result = await self.build()
await self.cleanup()
return result
async def main():
projects = [
{"path": "/Users/dev/MyApp", "scheme": "MyApp"},
{"path": "/Users/dev/OtherApp", "scheme": "OtherApp"}
]
all_results = []
for project in projects:
builder = iOSBuildAutomation(project["path"], project["scheme"])
result = await builder.execute()
all_results.append({
"project": project["scheme"],
**result
})
# Slack notification
successful = sum(1 for r in all_results if r["status"] == "success")
webhook = os.environ.get("SLACK_WEBHOOK_URL")
if webhook:
message = f"🔨 Daily iOS Build Report: {successful}/{len(all_results)} passed"
subprocess.run([
"curl", "-X", "POST", webhook,
"-H", "Content-Type: application/json",
"-d", json.dumps({"text": message})
])
print(json.dumps(all_results, indent=2))
if __name__ == "__main__":
asyncio.run(main())Security Best Practices
1. Credential Management
# ❌ Wrong: credentials in code
API_KEY = "sk-xxx..."
# ✅ Right: environment variables
API_KEY = os.environ["SLACK_API_KEY"]
# Even better: macOS Keychain
import subprocess
api_key = subprocess.run(
["security", "find-generic-password", "-w", "-s", "slack_api"],
capture_output=True,
text=True
).stdout.strip()2. Audit Logging
Every automated operation should be logged:
{
"timestamp": "2026-03-25T22:00:15Z",
"operation": "batch_report_sync",
"initiated_by": "dispatch_scheduler",
"files_processed": 12,
"status": "success",
"duration_seconds": 145,
"checksum": "abc123def456"
}3. Permission Minimization
Limit Computer Use's access:
Allowed directories:
• ~/Documents/Reports
• ~/Downloads/Incoming
Blocked directories:
• ~/.ssh
• ~/Library/Keychains
• /private/var/db
Next Steps
You've learned the foundational patterns. Now explore:
- Claude Code Hooks Automation Guide for fine-grained GitHub Actions integration
- Claude Dispatch Guide for remote team setups
The future of desktop automation is here. Build with confidence.