Foundations of AI Agents

import json, time
from openai import OpenAI

client = OpenAI()

TOOLS = {
    "get_weather": lambda city: {"temp_c": 18, "city": city},
}

SYSTEM = """You are a weather agent. Loop: think, then either call a tool
or emit FINAL. Always respond as JSON: {"thought": str, "action": str,
"args": object} where action is a tool name or "FINAL"."""

def run(goal: str, max_steps: int = 6, max_seconds: int = 30):
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": goal}]
    started = time.time()
    for step in range(max_steps):
        if time.time() - started > max_seconds:
            return {"status": "budget_exhausted", "reason": "time"}
        resp = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages,
            response_format={"type": "json_object"})
        out = json.loads(resp.choices[0].message.content)
        messages.append({"role": "assistant", "content": json.dumps(out)})
        if out["action"] == "FINAL":
            return {"status": "success", "answer": out.get("args", {})}
        tool = TOOLS.get(out["action"])
        if not tool:
            obs = {"error": "unknown_tool", "name": out["action"]}
        else:
            obs = tool(**out["args"])
        messages.append({"role": "user",
                         "content": f"Observation: {json.dumps(obs)}"})
    return {"status": "budget_exhausted", "reason": "steps"}

print(run("What is the weather in Paris right now?"))

{
  "name": "query_invoices",
  "description": "Search invoices in the billing system. Use this when the user asks about a specific invoice, payment status, or overdue amounts.",
  "parameters": {
    "type": "object",
    "required": ["customer_id"],
    "properties": {
      "customer_id": {
        "type": "string",
        "description": "Stripe customer ID, format cus_XXXX. NOT the email."
      },
      "status": {
        "type": "string",
        "enum": ["draft", "open", "paid", "void", "uncollectible"],
        "description": "Filter by invoice status. Omit to return all statuses."
      },
      "limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 20}
    }
  }
}

export type AgentResult =
  | { status: "success"; answer: unknown; steps: number; cost_usd: number; trace_id: string }
  | { status: "budget_exhausted"; budget: "steps" | "time" | "tokens" | "dollars"; partial?: unknown; trace_id: string }
  | { status: "guardrail_tripped"; rule: string; details: unknown; trace_id: string }
  | { status: "human_handoff"; reason: string; context: unknown; trace_id: string }
  | { status: "error"; error: { code: string; message: string }; trace_id: string };

{
  "trace_id": "run_01HZ...",
  "step": 3,
  "thought": "I need yesterday's revenue to answer the question.",
  "action": "query_db",
  "args": {"sql": "SELECT SUM(amount) FROM orders WHERE day = '2026-05-27'"},
  "observation": {"rows": [{"sum": 12431}], "latency_ms": 84},
  "budgets_remaining": {"steps": 22, "tokens": 184213, "dollars": 0.94},
  "ts": "2026-05-28T09:14:22.118Z"
}

from dataclasses import dataclass
import time

@dataclass
class Budget:
    max_steps: int = 25
    max_seconds: float = 300
    max_tokens: int = 200_000
    max_dollars: float = 1.00
    used_steps: int = 0
    used_tokens: int = 0
    used_dollars: float = 0.0
    started_at: float = 0.0
    def tick(self, tokens: int, dollars: float) -> str | None:
        self.used_steps += 1
        self.used_tokens += tokens
        self.used_dollars += dollars
        if self.used_steps > self.max_steps: return "steps"
        if time.time() - self.started_at > self.max_seconds: return "time"
        if self.used_tokens > self.max_tokens: return "tokens"
        if self.used_dollars > self.max_dollars: return "dollars"
        return None