Event-Driven Agents: Work That Starts Itself

from fastapi import FastAPI, Request
import asyncio, httpx, json, os

app = FastAPI()
SANDBOX_URL = os.environ["SANDBOX_URL"]
API_KEY = os.environ["SANDBOX_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


async def run_agent_task(task_name: str, command: str):
    """Create a sandbox, run a task, clean up."""
    async with httpx.AsyncClient(timeout=600) as client:
        # Create
        sandbox = (await client.post(f"{SANDBOX_URL}/api/sandboxes",
            headers=HEADERS,
            json={"name": task_name, "instance_type": "ab0t.micro", "auto_stop_minutes": 15}
        )).json()
        sid = sandbox["sandbox_id"]

        # Wait for boot
        for _ in range(24):
            s = (await client.get(f"{SANDBOX_URL}/api/sandboxes/{sid}", headers=HEADERS)).json()
            if s["status"] == "running": break
            await asyncio.sleep(5)

        # Execute
        result = (await client.post(f"{SANDBOX_URL}/api/sandboxes/{sid}/execute",
            headers=HEADERS, json={"command": command})).json()

        # Cleanup
        await client.delete(f"{SANDBOX_URL}/api/sandboxes/{sid}", headers=HEADERS)

        return result


# --- Slack: someone asks a question in #ask-agent ---
@app.post("/webhook/slack")
async def handle_slack(request: Request):
    payload = await request.json()
    if payload.get("type") == "event_callback":
        event = payload["event"]
        if event.get("type") == "message" and not event.get("bot_id"):
            question = event["text"]
            channel = event["channel"]
            asyncio.create_task(slack_research_agent(question, channel))
    return {"challenge": payload.get("challenge", "")}


async def slack_research_agent(question, channel):
    result = await run_agent_task(
        "slack-research",
        f"pip install httpx anthropic && python research.py '{question}'")
    # Post result back to Slack
    await post_to_slack(channel, result["stdout"])


# --- GitHub: PR opened ---
@app.post("/webhook/github")
async def handle_github(request: Request):
    payload = await request.json()
    event = request.headers.get("X-GitHub-Event")

    if event == "pull_request" and payload["action"] in ("opened", "synchronize"):
        pr = payload["pull_request"]
        repo = payload["repository"]["full_name"]
        asyncio.create_task(pr_review_agent(repo, pr["number"], pr["head"]["ref"]))

    return {"status": "ok"}


async def pr_review_agent(repo, pr_number, branch):
    result = await run_agent_task(
        f"pr-review-{pr_number}",
        f"git clone https://github.com/{repo} project && cd project && git checkout {branch} && npm test")
    # Post comment on PR
    await post_github_comment(repo, pr_number, result["stdout"])


# --- Stripe: payment failed ---
@app.post("/webhook/stripe")
async def handle_stripe(request: Request):
    payload = await request.json()
    if payload["type"] == "invoice.payment_failed":
        customer_id = payload["data"]["object"]["customer"]
        asyncio.create_task(payment_failure_agent(customer_id))
    return {"status": "ok"}


async def payment_failure_agent(customer_id):
    # Agent logs into billing portal, checks customer status,
    # posts summary to #billing-alerts Slack channel
    result = await run_agent_task(
        "payment-failure",
        f"python investigate_payment.py {customer_id}")
    await post_to_slack("billing-alerts", result["stdout"])

# Daily at 6am: competitive intelligence sweep
0 6 * * * /opt/agents/run-agent.sh competitive-intel

# Every Monday at 7am: AP vendor portal run
0 7 * * 1 /opt/agents/run-agent.sh weekly-ap

# Every 15 minutes: data quality check
*/15 * * * * /opt/agents/run-agent.sh data-quality-check

# 28th of every month: accounts reconciliation
0 8 28 * * /opt/agents/run-agent.sh monthly-reconciliation

# Every Friday at 5pm: weekly summary report
0 17 * * 5 /opt/agents/run-agent.sh weekly-summary

#!/bin/bash
# Generic agent runner: create sandbox, run task, cleanup
TASK_NAME=$1

SID=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"name\":\"$TASK_NAME\",\"instance_type\":\"ab0t.micro\",\"auto_stop_minutes\":15}" \
  | jq -r '.sandbox_id')

# Wait for boot
while [ "$(curl -s "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $API_KEY" | jq -r '.status')" != "running" ]; do
  sleep 5
done

# Run the task (each task has its own repo/script)
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/execute" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"command\":\"cd /workspace && git clone https://github.com/your-org/agent-tasks.git . && bash tasks/$TASK_NAME.sh\"}"

# Cleanup
curl -s -X DELETE "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $API_KEY"

import json, os, urllib3

SANDBOX_URL = os.environ["SANDBOX_URL"]
API_KEY = os.environ["SANDBOX_API_KEY"]
http = urllib3.PoolManager()


def handler(event, context):
    for record in event["Records"]:
        body = json.loads(record["body"])

        # Route by event type
        if body["type"] == "new_s3_object":
            task = f"aws s3 cp {body['s3_uri']} /workspace/input.csv && python process.py"
        elif body["type"] == "customer_signup":
            task = f"python onboard_customer.py {body['customer_id']}"
        elif body["type"] == "alarm_triggered":
            task = f"python investigate_alarm.py '{body['alarm_name']}'"
        else:
            continue

        # Create sandbox and run
        resp = http.request("POST", f"{SANDBOX_URL}/api/sandboxes",
            headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
            body=json.dumps({"name": body["type"], "instance_type": "ab0t.micro", "auto_stop_minutes": 10}))
        sandbox = json.loads(resp.data)

        # Execute (fire and forget — auto_stop handles cleanup)
        # In production, poll for completion before processing next message

from datetime import datetime, timedelta
import json

# Agent's task calendar
CALENDAR = [
    {"trigger": "day_of_month", "day": 28, "task": "monthly-reconciliation",
     "description": "Run month-end accounts reconciliation"},
    {"trigger": "day_of_week", "day": "wednesday", "task": "board-metrics",
     "description": "Update board metrics dashboard (board meets Thursday)"},
    {"trigger": "date", "date": "2026-03-31", "task": "q1-compliance",
     "description": "File Q1 compliance report (deadline: March 31)"},
    {"trigger": "days_before", "reference": "2026-03-31", "days": 3,
     "task": "compliance-prep",
     "description": "Prepare Q1 compliance data (deadline in 3 days)"},
]


def get_todays_tasks() -> list:
    """Check what tasks are due today."""
    today = datetime.utcnow()
    tasks = []

    for entry in CALENDAR:
        trigger = entry["trigger"]
        if trigger == "day_of_month" and today.day == entry["day"]:
            tasks.append(entry)
        elif trigger == "day_of_week" and today.strftime("%A").lower() == entry["day"]:
            tasks.append(entry)
        elif trigger == "date" and today.strftime("%Y-%m-%d") == entry["date"]:
            tasks.append(entry)
        elif trigger == "days_before":
            ref = datetime.strptime(entry["reference"], "%Y-%m-%d")
            if (ref - today).days == entry["days"]:
                tasks.append(entry)

    return tasks

# Run daily at 6am: check calendar, execute due tasks
tasks = get_todays_tasks()
for task in tasks:
    print(f"Running: {task['task']} — {task['description']}")
    # run_agent_task(task["task"], ...)

# Event routing table: event type → agent task
ROUTES = {
    "slack.message.ask-agent":    {"task": "slack-research",     "instance": "ab0t.micro"},
    "github.pull_request":        {"task": "pr-review",          "instance": "ab0t.small"},
    "stripe.payment_failed":      {"task": "payment-investigate", "instance": "ab0t.micro"},
    "s3.object_created":          {"task": "process-upload",     "instance": "ab0t.medium"},
    "cron.daily.competitive":     {"task": "competitive-intel",  "instance": "ab0t.micro"},
    "cron.weekly.ap":             {"task": "weekly-ap",         "instance": "ab0t.micro"},
    "cron.monthly.reconciliation": {"task": "reconciliation",    "instance": "ab0t.medium"},
    "calendar.board-prep":        {"task": "board-metrics",     "instance": "ab0t.micro"},
}

async def dispatch(event_type: str, payload: dict):
    route = ROUTES.get(event_type)
    if not route:
        print(f"Unknown event: {event_type}")
        return

    await run_agent_task(
        task_name=route["task"],
        instance_type=route["instance"],
        command=f"cd /workspace && git clone $AGENT_REPO . && bash tasks/{route['task']}.sh",
        env=payload,  # Pass event payload as env vars
    )

async def run_with_retry(event_type, payload, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            result = await dispatch(event_type, payload)
            if result and result.get("exit_code") == 0:
                return result
        except Exception as e:
            print(f"Attempt {attempt+1} failed: {e}")

        if attempt < max_retries:
            await asyncio.sleep(2 ** attempt * 5)

    # All retries exhausted — dead letter
    await post_to_slack("agent-failures",
        f"*Agent failed after {max_retries+1} attempts*\nEvent: {event_type}\nPayload: {json.dumps(payload)[:500]}")
    # Optionally write to SQS dead letter queue for manual review