Built-in Tools

# script_id: day_038_built_in_tools/recorded_response
# Read a recorded web-search response offline: no API key, no network.
# Install: uv add "openai>=3.46,<4"
import re
from dataclasses import dataclass

from openai.types.responses import Response

AWS = "https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/"
TEXT = ("FIFO queues have lower per-queue quotas than standard queues, and AWS offers a "
        "high-throughput mode for FIFO queues "
        f"([docs.aws.amazon.com]({AWS}high-throughput-fifo.html)). "
        "The current limits are published per Region "
        f"([docs.aws.amazon.com]({AWS}quotas-messages.html)).")
RECORDED = {"id": "resp_recorded", "object": "response", "model": "gpt-6-luna", "output": [
    {"type": "web_search_call", "id": "ws_1", "status": "completed",
     "action": {"type": "search", "query": "SQS FIFO throttling quotas high throughput",
                "sources": [{"type": "url", "url": AWS + "high-throughput-fifo.html"},
                            {"type": "url", "url": AWS + "quotas-messages.html"}]}},
    {"type": "message", "id": "msg_1", "role": "assistant", "status": "completed",
     "content": [{"type": "output_text", "text": TEXT, "annotations": [
         {"type": "url_citation", "start_index": 116, "end_index": 241,
          "url": AWS + "high-throughput-fifo.html",
          "title": "High throughput for FIFO queues in Amazon SQS"},
         {"type": "url_citation", "start_index": 287, "end_index": 407,
          "url": AWS + "quotas-messages.html", "title": "Amazon SQS message quotas"}]}]}]}

@dataclass(frozen=True)
class WebCitation:
    url: str
    title: str
    start_index: int
    end_index: int

def url_citations(response) -> list[WebCitation]:
    """Every url_citation on every output_text part; [] when the model didn't search."""
    return [WebCitation(a.url, a.title, a.start_index, a.end_index)
            for item in response.output if item.type == "message"
            for part in item.content if part.type == "output_text"
            for a in part.annotations if a.type == "url_citation"]

if __name__ == "__main__":
    resp = Response.construct(**RECORDED)  # recorded JSON -> the SDK's typed objects
    for c in url_citations(resp):
        claim = re.split(r"[,.] ", resp.output_text[:c.start_index].rstrip())[-1]  # last clause
        print(f"[{c.start_index}:{c.end_index}] {c.title}\n    backs: {claim}")
    searches = [i for i in resp.output if i.type == "web_search_call" and i.action.type == "search"]
    print(f"searches: {len(searches)} · sources read: {len(searches[0].action.sources)}")

[116:241] High throughput for FIFO queues in Amazon SQS
    backs: and AWS offers a high-throughput mode for FIFO queues
[287:407] Amazon SQS message quotas
    backs: The current limits are published per Region
searches: 1 · sources read: 2

# script_id: day_038_built_in_tools/web_search_minimal
# One hosted web search with citations. Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY · ATLAS_MODEL (default gpt-6-luna) · Cost: one search ($0.01) + tokens
import os

from openai import OpenAI

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

resp = client.responses.create(
    model=MODEL,
    tools=[{"type": "web_search"}],  # the whole integration
    input="Is Python 3.13 still receiving bugfix releases upstream? Cite sources.",
)
print(resp.output_text, "\n")
cites = [a for item in resp.output if item.type == "message"
         for part in item.content if part.type == "output_text"
         for a in part.annotations if a.type == "url_citation"]
for n, a in enumerate(cites, 1):
    print(f"[{n}] {a.title} — {a.url}")

(today's answer about the 3.13 branch, with inline links; your wording and sources will differ)

[1] <page title> — https://<a python.org page>
[2] <page title> — https://<another python.org page>

# script_id: day_038_built_in_tools/web_search_knobs
plain = client.responses.create(model=MODEL, tools=[{"type": "web_search"}],
                                input="What does FIFO stand for?")
msg = next(i for i in plain.output if i.type == "message")
try:
    print(msg.content[0].annotations[0].url)  # the naive version
except IndexError as exc:
    print(f"IndexError: {exc}  (searches: {sum(i.type == 'web_search_call' for i in plain.output)})")

IndexError: list index out of range  (searches: 0)

# script_id: day_038_built_in_tools/research_helper
# research(): hosted web_search + Atlas's own docs tools in one bounded loop.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA
# Cost: ≈ $0.01 per search + luna tokens (as of 2026-11, verify on the pricing page)
import json
import os
import re
from dataclasses import dataclass, field
from pathlib import Path

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()

# built on Day 36 (search_docs, get_doc_section), compact keyword versions
DOCS = {p.stem: p.read_text(encoding="utf-8").split("---\n", 2)[2]  # drop the front matter
        for p in sorted((DATA_DIR / "docs").glob("*.md"))}
STOP = {"the", "and", "for", "what", "which", "does", "with", "are", "still", "orbit"}

def search_docs(query: str, k: int = 3) -> list[dict]:
    terms = {t for t in re.findall(r"[a-z0-9.]+", query.lower()) if len(t) > 2} - STOP
    scored = sorted(((sum(body.lower().count(t) for t in terms), doc_id)
                     for doc_id, body in DOCS.items()), reverse=True)
    return [{"doc_id": d, "title": DOCS[d].strip().splitlines()[0].lstrip("# ")}
            for score, d in scored[:k] if score]

def get_doc_section(doc_id: str, heading: str | None = None) -> dict:
    if doc_id not in DOCS:
        return {"error": f"Unknown doc_id {doc_id!r}. Call search_docs first."}
    sections = re.split(r"\n(?=## )", DOCS[doc_id])
    hit = next((s for s in sections if heading and heading.lower() in s.splitlines()[0].lower()),
               None)
    return {"doc_id": doc_id, "text": (hit or DOCS[doc_id]).strip()[:3000]}

# script_id: day_038_built_in_tools/research_helper
def fn_tool(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}}

DOC_TOOLS = [
    fn_tool("search_docs", "Find Orbit doc pages by keyword. Use for anything about Orbit itself.",
            {"query": {"type": "string"}}),
    fn_tool("get_doc_section", "Read an Orbit doc page by doc_id (from search_docs); "
            "heading picks one '## ' section, or null for the whole page.",
            {"doc_id": {"type": "string"}, "heading": {"type": ["string", "null"]}}),
]
REGISTRY = {"search_docs": search_docs, "get_doc_section": get_doc_section}

@dataclass(frozen=True)
class Citation:      # Day 15's Citation(doc_id, quote), compact
    doc_id: str
    quote: str

@dataclass(frozen=True)
class WebCitation:   # one url_citation, kept with its character offsets
    url: str
    title: str
    start_index: int
    end_index: int

@dataclass
class ResearchAnswer:
    text: str
    doc_citations: list[Citation]
    web_citations: list[WebCitation]
    web_calls: int
    cost_usd: float

PRICE_PER_1M = {"input": 0.10, "output": 0.50}  # gpt-6-luna, as of 2026-11; verify
SEARCH_FEE_USD = 0.01                            # $10 / 1k searches, as of 2026-11; verify

@dataclass
class CostTracker:   # Day 12's tracker, compact (ignores the cached-token discount)
    total_usd: float = 0.0
    fees: dict[str, float] = field(default_factory=dict)

    def record(self, response) -> None:
        u = response.usage
        self.total_usd += (u.input_tokens * PRICE_PER_1M["input"]
                           + u.output_tokens * PRICE_PER_1M["output"]) / 1e6

    def add_tool_fee(self, tool: str, usd: float) -> None:
        self.fees[tool] = self.fees.get(tool, 0.0) + usd
        self.total_usd += usd

# script_id: day_038_built_in_tools/research_helper
INSTRUCTIONS = ("You are Atlas, an assistant for Orbit maintainers. Use search_docs and "
                "get_doc_section for facts about Orbit; use web search only for facts outside "
                "Orbit's docs. Keep the answer short and say which half came from where.")

def url_citations(response) -> list[WebCitation]:
    return [WebCitation(a.url, a.title, a.start_index, a.end_index)
            for item in response.output if item.type == "message"
            for part in item.content if part.type == "output_text"
            for a in part.annotations if a.type == "url_citation"]

def research(question: str, *, allowed_domains: list[str] | None = None, max_rounds: int = 4,
             tracker: CostTracker | None = None) -> ResearchAnswer:
    tracker = tracker or CostTracker()
    start = tracker.total_usd
    web = {"type": "web_search", "search_context_size": "low"}
    if allowed_domains:
        web["filters"] = {"allowed_domains": allowed_domains}
    items: list = [{"role": "user", "content": question}]
    doc_cites: list[Citation] = []
    web_calls = 0
    for _ in range(max_rounds):
        resp = client.responses.create(model=MODEL, instructions=INSTRUCTIONS, input=items,
                                       tools=[web, *DOC_TOOLS], max_tool_calls=3)
        tracker.record(resp)
        n = sum(1 for i in resp.output if i.type == "web_search_call" and i.action.type == "search")
        web_calls += n
        tracker.add_tool_fee("web_search", n * SEARCH_FEE_USD)
        items += resp.output  # hosted calls go back as history, already done
        calls = [i for i in resp.output if i.type == "function_call"]
        if not calls:         # nothing left for our code to run: this is the answer
            return ResearchAnswer(resp.output_text, doc_cites, url_citations(resp),
                                  web_calls, round(tracker.total_usd - start, 4))
        for call in calls:    # ONLY function_call items get a function_call_output
            result = REGISTRY[call.name](**json.loads(call.arguments))
            if "text" in result:
                doc_cites.append(Citation(result["doc_id"], result["text"][:120]))
            items.append({"type": "function_call_output", "call_id": call.call_id,
                          "output": json.dumps(result)})
    raise RuntimeError(f"no final answer after {max_rounds} rounds")

# script_id: day_038_built_in_tools/export_issue_stats
# Export one row per (bug, component) from orbit.db to CSV over a read-only connection.
# Env: ORBIT_DATA (default datasets/orbit) · no API key needed
import csv
import os
import sqlite3
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
DB = DATA_DIR / "issues" / "orbit.db"
SQL = """
SELECT i.id, substr(i.created_at, 1, 10), l.value, i.severity, i.state
FROM issues i
JOIN issue_labels il ON il.issue_id = i.id
JOIN labels l ON l.id = il.label_id AND l.kind = 'comp'
WHERE i.type = 'bug'
ORDER BY i.created_at
"""

def connect_ro() -> sqlite3.Connection:  # built on Day 33: writes through it fail
    return sqlite3.connect(f"file:{DB.resolve()}?mode=ro", uri=True)

def export_issue_stats(path: Path = Path("issues_by_component.csv")) -> Path:
    conn = connect_ro()
    try:
        rows = conn.execute(SQL).fetchall()
    finally:
        conn.close()
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["issue_id", "created", "component", "severity", "state"])
        writer.writerows(rows)
    print(f"wrote {len(rows)} rows to {path}")
    return path

if __name__ == "__main__":
    export_issue_stats()

wrote 128 rows to issues_by_component.csv

# script_id: day_038_built_in_tools/code_interpreter_demo
# Upload the CSV into a hosted container, let the model analyse it, save the chart it draws.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA
# Cost: one 1 GB container session ($0.03 per 20 min, as of 2026-11; verify) + tokens
import os
from pathlib import Path

from openai import OpenAI

from export_issue_stats import export_issue_stats  # the sibling script above

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

csv_path = export_issue_stats()
container = client.containers.create(name="atlas-issue-stats", memory_limit="1g")
with csv_path.open("rb") as f:
    client.containers.files.create(container.id, file=f)

resp = client.responses.create(
    model=MODEL,
    tools=[{"type": "code_interpreter", "container": container.id}],
    tool_choice="required",                    # this demo must use the python tool
    include=["code_interpreter_call.outputs"],  # logs and images are opt-in
    instructions="Use the python tool. Keep the final answer to three sentences.",
    input="issues_by_component.csv has one row per bug and component. Which component's bug "
          "count is trending up in 2026? Compare Jan-Jun with Jul-Dec and save a bar chart as PNG.",
)

# script_id: day_038_built_in_tools/diy_aggregation
# The same question with the stdlib: $0, deterministic, and the data never leaves.
import csv
from collections import Counter

from export_issue_stats import export_issue_stats

h1, h2 = Counter(), Counter()
with export_issue_stats().open(encoding="utf-8") as f:
    for row in csv.DictReader(f):
        if row["created"].startswith("2026"):
            (h1 if row["created"][5:7] <= "06" else h2)[row["component"]] += 1

print(f"{'component':<14}{'Jan-Jun':>8}{'Jul-Dec':>8}{'change':>8}")
for comp in sorted(h1 | h2, key=lambda c: (h2[c] - h1[c], c), reverse=True)[:4]:
    print(f"{comp:<14}{h1[comp]:>8}{h2[comp]:>8}{h2[comp] - h1[comp]:>+8}")

wrote 128 rows to issues_by_component.csv
component      Jan-Jun Jul-Dec  change
core                 5      12      +7
scheduler            1       3      +2
auth                 1       3      +2
dashboard            3       4      +1