The OpenAI Agents SDK

# script_id: day_056_openai_agents_sdk/sdk_minimal
# Install: uv add "openai-agents==0.24.0"   Env: none (keyless); OPENAI_API_KEY switches to a live run
import os
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")  # samples never upload traces
from agents import Agent, Runner, function_tool
from agents.testing import ScriptedModel, assistant_message, function_call

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")

@function_tool
def search_issues(query: str) -> str:
    """Search past Orbit issues by keyword. Returns matching ids and titles."""
    return '[{"id": 4, "title": "Monthly statement job runs twice, customers get two emails"}]'

# The paper trace above, as two scripted turns: call the tool, then answer.
fake = ScriptedModel([
    [function_call("search_issues", {"query": "runs twice lease"}, call_id="c1")],
    [assistant_message("#41 looks like a duplicate of #4 (lease expiry, ORB-1101).")],
])
agent = Agent(name="Atlas", instructions="Triage Orbit issues. Cite issue ids.",
              model=MODEL if os.environ.get("OPENAI_API_KEY") else fake, tools=[search_issues])
result = Runner.run_sync(agent, "Issue #41: nightly index rebuild runs twice.", max_turns=8)
print(result.final_output)
print("turns:", len(result.raw_responses), "| items:", [i.type for i in result.new_items])

#41 looks like a duplicate of #4 (lease expiry, ORB-1101).
turns: 2 | items: ['tool_call_item', 'tool_call_output_item', 'message_output_item']

# script_id: day_056_openai_agents_sdk/break_it
# Install: uv add "openai-agents==0.24.0"   Env: none (keyless)
import os
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from agents import Agent, MaxTurnsExceeded, Runner, function_tool
from agents.models import get_default_model
from agents.testing import ScriptedModel, assistant_message, function_call

# 1. Forget model= and the SDK picks its own default, not your ATLAS_MODEL.
print("Agent without model= runs on:", get_default_model())

@function_tool
def search_issues(query: str) -> str:
    """Search past Orbit issues by keyword."""
    return "[]"  # nothing found, so a stubborn model keeps searching

# 2. A model that never stops calling tools hits max_turns, and the SDK raises.
looping = ScriptedModel([[function_call("search_issues", {"query": f"try {i}"}, call_id=f"c{i}")]
                         for i in range(20)])
try:
    Runner.run_sync(Agent(name="Loop", model=looping, tools=[search_issues]), "find dupes", max_turns=3)
except MaxTurnsExceeded as e:
    print("MaxTurnsExceeded:", e, "| items so far:", len(e.run_data.new_items))

# 3. A crashing tool does NOT raise: the model gets a generic error string instead.
@function_tool
def get_doc_section(doc_id: str) -> str:
    """Read one Orbit doc page."""
    raise FileNotFoundError(f"docs/{doc_id}.md")

crashy = ScriptedModel([[function_call("get_doc_section", {"doc_id": "nope"}, call_id="d1")],
                        [assistant_message("Sorry, I could not read the docs.")]])
r = Runner.run_sync(Agent(name="Crash", model=crashy, tools=[get_doc_section]), "read nope")
print("tool output the model saw:", [i.output for i in r.new_items if i.type == "tool_call_output_item"])

Agent without model= runs on: gpt-5.6-luna
MaxTurnsExceeded: Max turns (3) exceeded | items so far: 6
tool output the model saw: ['An error occurred while running the tool. Please try again.']

# script_id: day_056_openai_agents_sdk/sdk_agent
# atlas/sdk_agent.py: the Day 48 agent on the OpenAI Agents SDK, behind a thin adapter.
# Install: uv add "openai-agents==0.24.0"   Env: ORBIT_DATA; OPENAI_API_KEY only for live runs
# Services: none (orbit.db is opened read-only). This is the ONLY Atlas module that imports `agents`.
from __future__ import annotations

import asyncio, json, os, re, sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Literal

os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")  # decide tracing per data class (Day 73)
from agents import Agent, FunctionTool, MaxTurnsExceeded, Model, RunConfig, Runner

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# USD per 1M tokens, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": {"input": 0.10, "cached": 0.01, "output": 0.50}}

# built on Day 48 (../Day_048_first_agent): compact copy, fields must match exactly
StopReason = Literal["done", "max_steps", "max_seconds", "max_tokens", "max_errors", "interrupted"]

@dataclass
class StepRecord:
    index: int
    tool_calls: list[str]
    errors: int
    input_tokens: int
    output_tokens: int
    ms: int = 0  # the SDK reports no per-step latency; read it from traces if you need it

@dataclass
class AgentResult:
    text: str
    stop_reason: StopReason
    steps: list[StepRecord]
    usage: dict[str, int]
    cost_usd: float
    partial: bool = False

# script_id: day_056_openai_agents_sdk/sdk_agent
class CostTracker:  # built on Day 12: compact
    def __init__(self) -> None:
        self.total_usd = 0.0

    def record(self, model: str, input_tokens: int, output_tokens: int, cached: int = 0) -> float:
        p = PRICES.get(model, PRICES["gpt-6-luna"])
        usd = ((input_tokens - cached) * p["input"] + cached * p["cached"] + output_tokens * p["output"]) / 1e6
        self.total_usd += usd
        return usd

tracker = CostTracker()

def search_issues(query: str, limit: int = 3) -> list[dict]:  # built on Day 35
    words = [w for w in re.findall(r"\w+", query.lower()) if len(w) > 2]
    db = sqlite3.connect(f"file:{DATA_DIR / 'issues' / 'orbit.db'}?mode=ro", uri=True)
    rows = db.execute("SELECT id, title, body, state, duplicate_of FROM issues").fetchall()
    db.close()
    hits = sorted(((sum(w in f"{t} {b}".lower() for w in words), i, t, s, d) for i, t, b, s, d in rows),
                  reverse=True)
    return [{"id": i, "title": t, "state": s, "duplicate_of": d} for n, i, t, s, d in hits[:limit] if n]

def get_doc_section(doc_id: str, heading: str) -> str:  # built on Day 35
    if not re.fullmatch(r"[a-z0-9-]+", doc_id):  # no path tricks like "../"
        raise ValueError(f"bad doc_id {doc_id!r}; use a page name like 'troubleshooting-duplicates'")
    for part in re.split(r"\n(?=## )", (DATA_DIR / "docs" / f"{doc_id}.md").read_text()):
        if part.startswith("## ") and heading.lower() in part.splitlines()[0].lower():
            return part[:2000]
    raise LookupError(f"no section matching {heading!r} in {doc_id}")

def _fn_spec(name: str, description: str, props: dict) -> dict:
    return {"type": "function", "name": name, "description": description, "strict": True,
            "parameters": {"type": "object", "properties": props, "required": list(props),
                           "additionalProperties": False}}

TOOLS: list[dict] = [
    _fn_spec("search_issues", "Keyword search over past Orbit issues; returns id, title, state, duplicate_of.",
             {"query": {"type": "string"}, "limit": {"type": "integer"}}),
    _fn_spec("get_doc_section", "Read one section of an Orbit doc page by page id and heading words.",
             {"doc_id": {"type": "string"}, "heading": {"type": "string"}}),
]
REGISTRY: dict[str, Callable[..., Any]] = {"search_issues": search_issues, "get_doc_section": get_doc_section}

# script_id: day_056_openai_agents_sdk/sdk_agent
def envelope_err(kind: str, message: str, hint: str | None = None) -> dict:  # built on Day 37
    return {"ok": False, "error": {"kind": kind, "message": message, "retryable": False, "hint": hint}}

def to_sdk_tool(spec: dict, fn: Callable[..., Any] | None = None) -> FunctionTool:
    """Wrap a Day 35 tool spec as an SDK FunctionTool; our schema and error envelope win."""
    fn = fn or REGISTRY[spec["name"]]

    async def invoke(ctx: Any, args_json: str) -> str:
        try:
            return json.dumps({"ok": True, "data": fn(**json.loads(args_json))})
        except Exception as exc:  # never let the SDK's generic error string reach the model
            kind = "bad_args" if isinstance(exc, (TypeError, ValueError)) else "permanent"
            return json.dumps(envelope_err(kind, str(exc), "fix the arguments or try another tool"))

    return FunctionTool(name=spec["name"], description=spec["description"],
                        params_json_schema=spec["parameters"], on_invoke_tool=invoke,
                        strict_json_schema=spec.get("strict", True))

AGENT_SYSTEM = ("You triage new Orbit issues. Search past issues and the docs before answering. "
                "Cite issue ids (#N) and doc page ids. If evidence is missing, say what is missing.")

def build_sdk_agent(model: str | Model = MODEL, *, tools: list[dict] = TOOLS, **agent_options: Any) -> Agent:
    """agent_options passes extra Agent fields through (input_guardrails, handoffs, ...)."""
    return Agent(name="Atlas investigator", instructions=AGENT_SYSTEM, model=model,
                 tools=[to_sdk_tool(s) for s in tools], **agent_options)

# script_id: day_056_openai_agents_sdk/sdk_agent
def _to_result(run: Any, model_name: str, stop: StopReason, text: str) -> AgentResult:
    """Map a RunResult (or MaxTurnsExceeded.run_data) to the Day 48 AgentResult."""
    failed = {i.raw_item["call_id"]: '"ok": false' in str(i.output)
              for i in run.new_items if i.type == "tool_call_output_item"}
    steps = []
    for n, resp in enumerate(run.raw_responses, start=1):
        calls = [o for o in resp.output if getattr(o, "type", "") == "function_call"]
        steps.append(StepRecord(n, [c.name for c in calls], sum(failed.get(c.call_id, False) for c in calls),
                                resp.usage.input_tokens, resp.usage.output_tokens))
    u = run.context_wrapper.usage
    cost = tracker.record(model_name, u.input_tokens, u.output_tokens, u.input_tokens_details.cached_tokens)
    usage = {"input_tokens": u.input_tokens, "output_tokens": u.output_tokens, "total_tokens": u.total_tokens}
    return AgentResult(text, stop, steps, usage, round(cost, 6), partial=stop != "done")

def run_sdk_agent(task: str, max_turns: int = 8, *, model: str | Model = MODEL,
                  max_seconds: float = 60.0, **agent_options: Any) -> AgentResult:
    agent = build_sdk_agent(model, **agent_options)
    name = model if isinstance(model, str) else MODEL  # test doubles are priced as MODEL
    config = RunConfig(workflow_name="atlas-investigator", trace_include_sensitive_data=False)
    try:  # inside async code (FastAPI, notebooks), await Runner.run(...) instead of asyncio.run
        run = asyncio.run(asyncio.wait_for(
            Runner.run(agent, task, max_turns=max_turns, run_config=config), timeout=max_seconds))
        return _to_result(run, name, "done", str(run.final_output))
    except MaxTurnsExceeded as exc:  # the SDK type stops here; callers get a partial AgentResult
        if exc.run_data is None:
            return AgentResult(str(exc), "max_steps", [], {}, 0.0, partial=True)
        return _to_result(exc.run_data, name, "max_steps", f"(stopped after {max_turns} turns)")
    except TimeoutError:  # wait_for cancelled the run, so its usage is lost
        return AgentResult(f"(stopped after {max_seconds}s)", "max_seconds", [], {}, 0.0, partial=True)

# script_id: day_056_openai_agents_sdk/sdk_agent
def scripted_investigation(issue: dict) -> Model:
    """Keyless stand-in: search, read a doc, answer. Token counts are illustrative, not measured."""
    from agents import Usage
    from agents.testing import ScriptedModel, assistant_message, function_call
    return ScriptedModel([
        [function_call("search_issues", {"query": issue["title"], "limit": 3}, call_id="s1")],
        [function_call("get_doc_section", {"doc_id": "troubleshooting-duplicates", "heading": "checklist"},
                       call_id="g1")],
        [assistant_message(f"#{issue['id']}: compared with past issues and the duplicates checklist.")],
    ], default_usage=Usage(requests=1, input_tokens=1500, output_tokens=120, total_tokens=1620))

def load_agent_set() -> list[int]:
    """Day 48's five issues (agent_set.json) if you have the file, else the same five by id."""
    path = Path("agent_set.json")
    return [int(i) for i in json.loads(path.read_text())] if path.exists() else [3, 6, 41, 62, 94]

if __name__ == "__main__":
    issues = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
    live = bool(os.environ.get("OPENAI_API_KEY"))
    for iid in load_agent_set():
        issue = issues[iid]
        task = f"New issue #{iid}: {issue['title']}\n\n{issue['body'][:1500]}"
        res = run_sdk_agent(task, model=MODEL if live else scripted_investigation(issue))
        tools = sum(len(s.tool_calls) for s in res.steps)
        print(f"#{iid:<3} {res.stop_reason:<9} steps={len(res.steps)} tools={tools} cost_usd={res.cost_usd:.5f}")
    print(f"total ${tracker.total_usd:.4f}", "(live)" if live else "(scripted, illustrative usage)")

#3   done      steps=3 tools=2 cost_usd=0.00063
#6   done      steps=3 tools=2 cost_usd=0.00063
#41  done      steps=3 tools=2 cost_usd=0.00063
#62  done      steps=3 tools=2 cost_usd=0.00063
#94  done      steps=3 tools=2 cost_usd=0.00063
total $0.0032 (scripted, illustrative usage)

# script_id: day_056_openai_agents_sdk/session_followup
# Install: uv add "openai-agents==0.24.0"   Env: none (keyless); writes chat.db in the working directory
import asyncio, os
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from agents import Agent, Runner, SQLiteSession
from agents.testing import ScriptedModel, assistant_message

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
live = bool(os.environ.get("OPENAI_API_KEY"))
fake = ScriptedModel([[assistant_message("#41 is a duplicate of #4: lease expiry (ORB-1101).")],
                      [assistant_message("troubleshooting-duplicates, 'Cause 1: lease expiry on long jobs'.")]])
agent = Agent(name="Atlas", instructions="Answer Orbit maintainers briefly.", model=MODEL if live else fake)

session = SQLiteSession("issue-41", "chat.db")  # one session id per conversation
asyncio.run(session.clear_session())             # fresh start each time you run the demo
Runner.run_sync(agent, "Is #41 a duplicate?", session=session)
second = Runner.run_sync(agent, "Which doc explains the fix?", session=session)  # no history by hand
print(second.final_output)
print("items stored:", len(asyncio.run(session.get_items())))
if not live:
    print("input items the model saw on turn 2:", len(fake.last_call.input))

troubleshooting-duplicates, 'Cause 1: lease expiry on long jobs'.
items stored: 4
input items the model saw on turn 2: 3

# script_id: day_056_openai_agents_sdk/input_guardrail
# Install: uv add "openai-agents==0.24.0"   Env: ORBIT_DATA (keyless)
import json, os, re
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from pathlib import Path
from agents import Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, Runner, input_guardrail
from agents.testing import ScriptedModel, assistant_message

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
SECRET = re.compile(r"\bsk-[A-Za-z0-9-]{8,}")  # one key shape; a real scanner is Day 73's job

@input_guardrail(run_in_parallel=False)  # check BEFORE the model call, so a blocked run costs nothing
async def no_secrets(ctx, agent, user_input) -> GuardrailFunctionOutput:
    found = SECRET.findall(str(user_input))
    return GuardrailFunctionOutput(output_info={"secrets_found": len(found)}, tripwire_triggered=bool(found))

agent = Agent(name="Atlas", instructions="Triage Orbit issues.", input_guardrails=[no_secrets],
              model=ScriptedModel([[assistant_message("triaged: likely lease expiry")]]))
issues = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
for iid in (41, 13):  # #13 contains a fake Stripe key in its env dump
    try:
        print(f"#{iid} ->", Runner.run_sync(agent, issues[iid]["body"]).final_output)
    except InputGuardrailTripwireTriggered as e:
        print(f"#{iid} -> blocked:", e.guardrail_result.output.output_info)

#41 -> triaged: likely lease expiry
#13 -> blocked: {'secrets_found': 1}

# script_id: day_056_openai_agents_sdk/test_sdk_agent
# Offline adapter tests. Install: uv add --dev pytest   Env: ORBIT_DATA   Run: uv run pytest test_sdk_agent.py -q
import pytest
from agents.testing import ScriptedModel, assistant_message, function_call
from sdk_agent import TOOLS, AgentResult, build_sdk_agent, run_sdk_agent

def test_tool_call_then_answer():
    fake = ScriptedModel([[function_call("search_issues", {"query": "lease expiry", "limit": 2}, call_id="c1")],
                          [assistant_message("Duplicate of #4.")]])
    res = run_sdk_agent("Issue #41 runs twice", model=fake)
    assert (res.stop_reason, res.partial, res.text) == ("done", False, "Duplicate of #4.")
    assert [s.tool_calls for s in res.steps] == [["search_issues"], []]

def test_turn_overrun_returns_partial_instead_of_raising():
    loop = ScriptedModel([[function_call("search_issues", {"query": "x", "limit": 1}, call_id=f"c{i}")]
                          for i in range(5)])
    res = run_sdk_agent("loop forever", max_turns=3, model=loop)
    assert (res.stop_reason, res.partial, len(res.steps)) == ("max_steps", True, 3)

def test_bad_tool_args_become_an_envelope():
    fake = ScriptedModel([[function_call("get_doc_section", {"doc_id": "../etc", "heading": "x"}, call_id="g1")],
                          [assistant_message("Could not read that doc.")]])
    res = run_sdk_agent("read a doc", model=fake)
    assert res.steps[0].errors == 1 and res.stop_reason == "done"

def test_sdk_types_stay_inside_the_adapter():
    res = run_sdk_agent("hi", model=ScriptedModel([[assistant_message("hello")]]))
    assert isinstance(res, AgentResult) and len(build_sdk_agent().tools) == len(TOOLS)

if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))

....                                                                     [100%]
4 passed in 0.05s

# script_id: day_056_openai_agents_sdk/compare_runtimes
# Same five issues, three runtimes. Needs OPENAI_API_KEY (≈ $0.02 on gpt-6-luna, as of 2026-11, verify).
# Install: uv add "openai-agents==0.24.0" langchain langchain-openai   Env: OPENAI_API_KEY, ORBIT_DATA
import inspect, json, time
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from openai import OpenAI
from sdk_agent import (AGENT_SYSTEM, DATA_DIR, MODEL, PRICES, REGISTRY, TOOLS, _to_result, build_sdk_agent,
                       load_agent_set, run_sdk_agent, to_sdk_tool)

client = OpenAI()

def usd(inp: int, out: int) -> float:  # list price, cache discounts ignored for all three runtimes
    return (inp * PRICES["gpt-6-luna"]["input"] + out * PRICES["gpt-6-luna"]["output"]) / 1e6

def own_loop(task: str, max_steps: int = 8) -> tuple[str, int, float]:  # built on Day 48: compact
    items, inp, out = [{"role": "user", "content": task}], 0, 0
    for step in range(1, max_steps + 1):
        resp = client.responses.create(model=MODEL, instructions=AGENT_SYSTEM, input=items, tools=TOOLS)
        inp, out = inp + resp.usage.input_tokens, out + resp.usage.output_tokens
        calls = [o for o in resp.output if o.type == "function_call"]
        if not calls:
            return resp.output_text, step, usd(inp, out)
        items += resp.output
        for c in calls:
            try:
                data = {"ok": True, "data": REGISTRY[c.name](**json.loads(c.arguments))}
            except Exception as exc:
                data = {"ok": False, "error": {"kind": "permanent", "message": str(exc)}}
            items.append({"type": "function_call_output", "call_id": c.call_id, "output": json.dumps(data)})
    return "(max_steps)", max_steps, usd(inp, out)

# script_id: day_056_openai_agents_sdk/compare_runtimes
# built on Day 53: the create_agent version (explicit model and Responses API, see Day 53's traps)
# tool() needs a description; the tools have no docstrings, so reuse the Day 35 spec text
lc_tools = [tool(REGISTRY[s["name"]], description=s["description"]) for s in TOOLS]
graph = create_agent(ChatOpenAI(model=MODEL, use_responses_api=True), tools=lc_tools, system_prompt=AGENT_SYSTEM)

def langgraph_run(task: str, max_steps: int = 8) -> tuple[str, int, float]:
    out = graph.invoke({"messages": [{"role": "user", "content": task}]},
                       {"recursion_limit": 2 * max_steps + 1})
    ai = [m for m in out["messages"] if m.type == "ai"]
    inp = sum((m.usage_metadata or {}).get("input_tokens", 0) for m in ai)
    outp = sum((m.usage_metadata or {}).get("output_tokens", 0) for m in ai)
    return out["messages"][-1].text, len(ai), usd(inp, outp)

def sdk_run(task: str) -> tuple[str, int, float]:
    r = run_sdk_agent(task)
    return r.text, len(r.steps), r.cost_usd

RUNTIMES = {"own loop": (own_loop, [own_loop]),
            "LangGraph": (langgraph_run, [langgraph_run]),
            "Agents SDK": (sdk_run, [to_sdk_tool, build_sdk_agent, run_sdk_agent, _to_result])}
issues = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
print(f"{'runtime':<11} {'loop LoC':>8} {'avg steps':>9} {'total $':>8} {'secs':>6}")
for name, (run, parts) in RUNTIMES.items():
    loc = sum(len(inspect.getsource(p).splitlines()) for p in parts)
    steps = cost = 0.0
    t0 = time.perf_counter()
    for iid in load_agent_set():
        _, n, c = run(f"New issue #{iid}: {issues[iid]['title']}\n\n{issues[iid]['body'][:1500]}")
        steps, cost = steps + n, cost + c
    print(f"{name:<11} {loc:>8} {steps / 5:>9.1f} {cost:>8.4f} {time.perf_counter() - t0:>6.1f}")