#!/usr/bin/env python3 """Live probes for the OpenAI tools domain (function calling, built-in tools, containers, skills). Raw sanitized responses -> tmp-live/tools/.json ; every call logged to reports/live-requests.jsonl. Usage: .venv/bin/python tmp-live/tools_probe.py [probe ...] """ from __future__ import annotations import base64, io, json, sys, time, zipfile from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from scripts.live import openai_request, save_sanitized, mask # noqa: E402 OUT = ROOT / "tmp-live" / "tools" NANO = "gpt-5.4-nano" MINI = "gpt-5.4-mini" # rough per-token prices (USD / 1M) used only for est_cost_usd logging PRICE = {NANO: (0.20, 1.25), MINI: (0.75, 4.50), "gpt-4.1-nano": (0.10, 0.40), "computer-use-preview": (3.0, 12.0)} def est(model: str, body) -> float: u = (body or {}).get("usage") if isinstance(body, dict) else None if not u: return 0.0 i, o = PRICE.get(model, (1.0, 5.0)) return (u.get("input_tokens", u.get("prompt_tokens", 0)) * i + u.get("output_tokens", u.get("completion_tokens", 0)) * o) / 1e6 def save(name: str, obj) -> None: save_sanitized(obj, OUT / f"{name}.json") def kinds(body) -> list[str]: if not isinstance(body, dict): return [] return [f"{it.get('type')}:{it.get('status', '')}" for it in body.get("output", [])] def resp(name: str, payload: dict, extra_cost: float = 0.0, model: str | None = None, note: str = ""): model = model or payload.get("model", NANO) st, body, hdrs = openai_request("POST", "/v1/responses", payload, est_cost_usd=extra_cost, note=note or name) save(name, {"status": st, "request": payload, "response": body}) c = est(model, body) + extra_cost print(f"[{name}] HTTP {st} model={model} cost~${c:.4f} output={kinds(body)}" + (f" ERROR={json.dumps(body.get('error'))[:300]}" if isinstance(body, dict) and body.get("error") else "")) return st, body def stream(name: str, payload: dict): payload = {**payload, "stream": True} st, lines, hdrs = openai_request("POST", "/v1/responses", payload, stream=True, note=name + " (stream)") events, samples = [], {} if st != 200: body = b"".join(l.encode() for l in lines) if not isinstance(lines, (dict, bytes)) else lines print(f"[{name}] HTTP {st} {str(body)[:300]}") return st, [] for line in lines: if line.startswith("data: "): try: ev = json.loads(line[6:]) except Exception: continue t = ev.get("type") events.append(t) samples.setdefault(t, ev) save(name, {"status": st, "request": payload, "event_sequence": events, "samples": samples}) print(f"[{name}] HTTP {st} events={len(events)} distinct={sorted(set(events))}") return st, events WEATHER = {"type": "function", "name": "get_weather", "description": "Get current weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": False}, "strict": True} def p_function(): st, body = resp("function_call_forced", {"model": NANO, "input": "What is the weather in Paris?", "tools": [WEATHER], "tool_choice": {"type": "function", "name": "get_weather"}, "max_output_tokens": 64, "reasoning": {"effort": "low"}}) call = next((it for it in body.get("output", []) if it.get("type") == "function_call"), None) if call: print(" function_call item keys:", sorted(call.keys()), "args=", call.get("arguments")) resp("function_call_output_roundtrip", {"model": NANO, "previous_response_id": body["id"], "input": [{"type": "function_call_output", "call_id": call["call_id"], "output": json.dumps({"temp_c": 18, "sky": "clear"})}], "tools": [WEATHER], "max_output_tokens": 32, "reasoning": {"effort": "low"}}) stream("function_call_stream", {"model": NANO, "input": "Weather in Rome?", "tools": [WEATHER], "tool_choice": "required", "max_output_tokens": 64, "reasoning": {"effort": "low"}}) # strict schema violation: additionalProperties missing -> expect 400 bad = {**WEATHER, "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}} resp("function_strict_invalid_schema", {"model": NANO, "input": "hi", "tools": [bad], "max_output_tokens": 16}) # parallel_tool_calls false + tool_choice none + allowed_tools resp("function_allowed_tools", {"model": NANO, "input": "Weather in Oslo?", "tools": [WEATHER, {**WEATHER, "name": "get_time"}], "tool_choice": {"type": "allowed_tools", "mode": "required", "tools": [{"type": "function", "name": "get_weather"}]}, "parallel_tool_calls": False, "max_output_tokens": 64, "reasoning": {"effort": "low"}}) # namespace resp("function_namespace", {"model": NANO, "input": "Weather in Lima?", "tool_choice": "required", "tools": [{"type": "namespace", "name": "weather", "description": "Weather tools", "tools": [WEATHER]}], "max_output_tokens": 64, "reasoning": {"effort": "low"}}) # chat completions function call (gpt-4.1-nano) st, body, _ = openai_request("POST", "/v1/chat/completions", { "model": "gpt-4.1-nano", "messages": [{"role": "user", "content": "Weather in Paris?"}], "tools": [{"type": "function", "function": {"name": "get_weather", "parameters": WEATHER["parameters"], "strict": True}}], "tool_choice": {"type": "function", "function": {"name": "get_weather"}}, "max_tokens": 32}, note="chat completions function call") save("chat_function_call", {"status": st, "response": body}) print(f"[chat_function_call] HTTP {st}", json.dumps(body.get("choices", [{}])[0].get("message", {}).get("tool_calls"))[:300] if isinstance(body, dict) else body) def p_custom(): tool = {"type": "custom", "name": "answer_yes_no", "description": "Reply with exactly yes or no.", "format": {"type": "grammar", "syntax": "regex", "definition": r"^(yes|no)$"}} for m in (NANO, MINI): st, body = resp(f"custom_grammar_{m}", {"model": m, "input": "Is 2+2 equal to 4? Use the tool.", "tools": [tool], "tool_choice": {"type": "custom", "name": "answer_yes_no"}, "max_output_tokens": 32, "reasoning": {"effort": "low"}}) if st == 200: break resp("custom_text_format", {"model": NANO, "input": "Say OK via the tool.", "tool_choice": "required", "tools": [{"type": "custom", "name": "echo", "description": "Echo free text."}], "max_output_tokens": 32, "reasoning": {"effort": "low"}}) def p_web_search(): resp("web_search", {"model": NANO, "input": "What is today's date in Toronto? Reply in 5 words.", "tools": [{"type": "web_search", "search_context_size": "low", "user_location": {"type": "approximate", "country": "CA", "city": "Toronto"}}], "include": ["web_search_call.action.sources"], "max_output_tokens": 64, "reasoning": {"effort": "low"}}, extra_cost=0.01) def p_code_interpreter(): st, body = resp("code_interpreter", {"model": NANO, "input": "Run print(2+2) in python and reply with just the number.", "tools": [{"type": "code_interpreter", "container": {"type": "auto"}}], "include": ["code_interpreter_call.outputs"], "max_output_tokens": 64, "reasoning": {"effort": "low"}}, extra_cost=0.03) ci = next((it for it in body.get("output", []) if it.get("type") == "code_interpreter_call"), None) if st == 200 else None if ci: cid = ci["container_id"] print(" container_id", cid, "code=", repr(ci.get("code"))[:80], "outputs=", json.dumps(ci.get("outputs"))[:200]) st, c, _ = openai_request("GET", f"/v1/containers/{cid}", note="retrieve container"); save("container_retrieve", c); print(" GET container", st, json.dumps(c)[:300]) st, f, _ = openai_request("GET", f"/v1/containers/{cid}/files", note="list container files"); save("container_files_list", f); print(" GET files", st, json.dumps(f)[:300]) st, l, _ = openai_request("GET", "/v1/containers?limit=2", note="list containers"); save("containers_list", l); print(" GET containers", st, [x.get("id") for x in l.get("data", [])] if isinstance(l, dict) else l) st, d, _ = openai_request("DELETE", f"/v1/containers/{cid}", note="delete container"); save("container_delete", d); print(" DELETE container", st, json.dumps(d)[:200]) # explicit container create/delete (no model call) to record create shape st, c, _ = openai_request("POST", "/v1/containers", {"name": "atlas-tools-agent", "memory_limit": "1g", "expires_after": {"anchor": "last_active_at", "minutes": 5}}, note="create container") save("container_create", c); print(" POST containers", st, json.dumps(c)[:300]) if st == 200: cid = c["id"] st, cf, _ = openai_request("POST", f"/v1/containers/{cid}/files", data=( b"--b\r\nContent-Disposition: form-data; name=\"file\"; filename=\"hello.txt\"\r\nContent-Type: text/plain\r\n\r\nhello atlas\r\n--b--\r\n"), content_type="multipart/form-data; boundary=b", note="create container file") save("container_file_create", cf); print(" POST container file", st, json.dumps(cf)[:300]) if st == 200: fid = cf["id"] st, r, _ = openai_request("GET", f"/v1/containers/{cid}/files/{fid}", note="retrieve container file"); save("container_file_retrieve", r); print(" GET file", st) st, r, _ = openai_request("GET", f"/v1/containers/{cid}/files/{fid}/content", note="container file content"); print(" GET content", st, r[:40] if isinstance(r, bytes) else r) st, r, _ = openai_request("DELETE", f"/v1/containers/{cid}/files/{fid}", note="delete container file"); save("container_file_delete", r); print(" DELETE file", st, json.dumps(r)[:200]) st, d, _ = openai_request("DELETE", f"/v1/containers/{cid}", note="delete container"); print(" DELETE container", st) def p_image_invalid(): resp("image_generation_invalid_size", {"model": NANO, "input": "Draw a red dot.", "tools": [{"type": "image_generation", "size": "1x1"}], "max_output_tokens": 16}) def p_mcp(): resp("mcp_deepwiki", {"model": NANO, "input": "Use the deepwiki tool read_wiki_structure for repo openai/openai-python and reply with the first topic name only.", "tools": [{"type": "mcp", "server_label": "deepwiki", "server_url": "https://mcp.deepwiki.com/mcp", "require_approval": "never", "allowed_tools": ["read_wiki_structure"]}], "max_output_tokens": 64, "reasoning": {"effort": "low"}}) # approval flow: require_approval always -> expect mcp_approval_request st, body = resp("mcp_approval_request", {"model": NANO, "input": "Call read_wiki_structure for openai/openai-python.", "tools": [{"type": "mcp", "server_label": "deepwiki", "server_url": "https://mcp.deepwiki.com/mcp", "require_approval": "always", "allowed_tools": ["read_wiki_structure"]}], "tool_choice": {"type": "mcp", "server_label": "deepwiki", "name": "read_wiki_structure"}, "max_output_tokens": 64, "reasoning": {"effort": "low"}}) ar = next((it for it in body.get("output", []) if it.get("type") == "mcp_approval_request"), None) if st == 200 else None if ar: resp("mcp_approval_denied", {"model": NANO, "previous_response_id": body["id"], "input": [{"type": "mcp_approval_response", "approval_request_id": ar["id"], "approve": False, "reason": "atlas test"}], "tools": [{"type": "mcp", "server_label": "deepwiki", "server_url": "https://mcp.deepwiki.com/mcp", "require_approval": "always", "allowed_tools": ["read_wiki_structure"]}], "max_output_tokens": 32, "reasoning": {"effort": "low"}}) PNG1 = base64.b64encode(bytes.fromhex( "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4890000000d4944415478da63f8ffff3f0300" "0500020a2c2d3e850000000049454e44ae426082")).decode() def p_computer(): resp("computer_use_preview", {"model": "computer-use-preview", "input": [{"role": "user", "content": [{"type": "input_text", "text": "Do nothing. Reply with OK."}, {"type": "input_image", "image_url": f"data:image/png;base64,{PNG1}"}]}], "tools": [{"type": "computer_use_preview", "display_width": 1, "display_height": 1, "environment": "browser"}], "truncation": "auto", "max_output_tokens": 32}, model="computer-use-preview") def p_skills(): st, body, _ = openai_request("GET", "/v1/skills?limit=5", note="list skills"); save("skills_list", body); print("[skills_list]", st, json.dumps(body)[:300]) buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as z: z.writestr("atlas-hello/SKILL.md", "---\nname: atlas-hello\ndescription: Atlas test skill. Prints OK when asked.\n---\n\n# Atlas hello\n\nWhen asked, run `echo OK`.\n") zip_bytes = buf.getvalue() mp = (b"--b\r\nContent-Disposition: form-data; name=\"files\"; filename=\"atlas-hello.zip\"\r\nContent-Type: application/zip\r\n\r\n" + zip_bytes + b"\r\n--b--\r\n") st, sk, _ = openai_request("POST", "/v1/skills", data=mp, content_type="multipart/form-data; boundary=b", note="create skill (zip)") save("skill_create", sk); print("[skill_create]", st, json.dumps(sk)[:400]) if st != 200: return sid = sk["id"] st, r, _ = openai_request("GET", f"/v1/skills/{sid}", note="get skill"); save("skill_get", r); print("[skill_get]", st) st, r, _ = openai_request("GET", f"/v1/skills/{sid}/versions", note="list skill versions"); save("skill_versions", r); print("[skill_versions]", st, json.dumps(r)[:300]) st, r, _ = openai_request("GET", f"/v1/skills/{sid}/versions/1", note="get skill version"); save("skill_version_get", r); print("[skill_version_get]", st) st, r, _ = openai_request("GET", f"/v1/skills/{sid}/content", note="skill content"); print("[skill_content]", st, (r[:4] if isinstance(r, bytes) else r)) st, r, _ = openai_request("POST", f"/v1/skills/{sid}/versions", data=mp, content_type="multipart/form-data; boundary=b", note="create skill version"); save("skill_version_create", r); print("[skill_version_create]", st, json.dumps(r)[:300]) st, r, _ = openai_request("POST", f"/v1/skills/{sid}", {"default_version": "2"}, note="set default version"); save("skill_set_default", r); print("[skill_set_default]", st, json.dumps(r)[:200]) st, r, _ = openai_request("DELETE", f"/v1/skills/{sid}/versions/1", note="delete skill version"); save("skill_version_delete", r); print("[skill_version_delete]", st, json.dumps(r)[:200]) # use the skill in hosted shell (nano supports hosted_shell + skills per model page) resp("shell_hosted_skill", {"model": NANO, "input": "Use the atlas-hello skill: run echo OK and reply with the output only.", "tools": [{"type": "shell", "environment": {"type": "container_auto", "skills": [{"type": "skill_reference", "skill_id": sid}]}}], "max_output_tokens": 128, "reasoning": {"effort": "low"}}, extra_cost=0.03) st, r, _ = openai_request("DELETE", f"/v1/skills/{sid}", note="delete skill"); save("skill_delete", r); print("[skill_delete]", st, json.dumps(r)[:200]) def p_shell_patch(): resp("shell_local", {"model": NANO, "input": "Propose the shell command `echo OK`. Do not explain.", "tools": [{"type": "shell", "environment": {"type": "local"}}], "tool_choice": {"type": "shell"}, "max_output_tokens": 64, "reasoning": {"effort": "low"}}) resp("apply_patch", {"model": NANO, "input": "Create a new file hello.txt containing the single line OK using apply_patch.", "tools": [{"type": "apply_patch"}], "tool_choice": {"type": "apply_patch"}, "max_output_tokens": 128, "reasoning": {"effort": "low"}}) resp("local_shell_nano", {"model": NANO, "input": "Propose `echo OK`.", "tools": [{"type": "local_shell"}], "max_output_tokens": 32}) def p_tool_search(): deferred = {**WEATHER, "defer_loading": True} resp("tool_search_mini", {"model": MINI, "input": "Weather in Paris? Find and use a tool.", "tools": [{"type": "tool_search"}, {"type": "namespace", "name": "weather", "description": "Weather lookup tools", "tools": [deferred]}], "max_output_tokens": 96, "reasoning": {"effort": "low"}}) def p_file_search(): st, vs, _ = openai_request("POST", "/v1/vector_stores", {"name": "atlas-tools-agent"}, note="create vector store (tools agent)") save("vs_create", vs); print("[vs_create]", st, vs.get("id") if isinstance(vs, dict) else vs) if st != 200: return vsid = vs["id"] content = b"Atlas fact sheet. The secret codeword for the API Atlas tools agent is PELICAN-42. Nothing else matters here. Filler text to reach two hundred bytes of content for the test file ok." mp = (b"--b\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nassistants\r\n--b\r\nContent-Disposition: form-data; name=\"file\"; filename=\"atlas.txt\"\r\nContent-Type: text/plain\r\n\r\n" + content + b"\r\n--b--\r\n") st, f, _ = openai_request("POST", "/v1/files", data=mp, content_type="multipart/form-data; boundary=b", note="upload file for file_search") fid = f.get("id"); print("[file_upload]", st, fid) st, vf, _ = openai_request("POST", f"/v1/vector_stores/{vsid}/files", {"file_id": fid}, note="attach file to vector store") for _ in range(20): st, vf, _ = openai_request("GET", f"/v1/vector_stores/{vsid}/files/{fid}", note="poll vs file status") if vf.get("status") in ("completed", "failed"): break time.sleep(1.5) print("[vs_file]", vf.get("status")) resp("file_search", {"model": NANO, "input": "What is the secret codeword? Reply with the codeword only.", "tools": [{"type": "file_search", "vector_store_ids": [vsid], "max_num_results": 2}], "include": ["file_search_call.results"], "max_output_tokens": 32, "reasoning": {"effort": "low"}}, extra_cost=0.0025) st, d, _ = openai_request("DELETE", f"/v1/vector_stores/{vsid}", note="delete vector store"); print("[vs_delete]", st) st, d, _ = openai_request("DELETE", f"/v1/files/{fid}", note="delete file"); print("[file_delete]", st) def p_retry_streams(): # web search with enough output budget, streamed to capture events + final response stream("web_search_stream", {"model": NANO, "input": "What is today's date in Toronto? Reply in 5 words.", "tools": [{"type": "web_search", "search_context_size": "low", "user_location": {"type": "approximate", "country": "CA", "city": "Toronto"}}], "include": ["web_search_call.action.sources"], "max_output_tokens": 600, "reasoning": {"effort": "low"}}) from scripts.live import log_request log_request("openai", "POST", "/v1/responses", 200, 0.01, "web search per-call fee ($10/1k)") # file search retry st, vs, _ = openai_request("POST", "/v1/vector_stores", {"name": "atlas-tools-agent"}, note="create vector store (tools agent, retry)") vsid = vs["id"] content = b"Atlas fact sheet. The secret codeword for the API Atlas tools agent is PELICAN-42. Nothing else matters here. Filler text to reach two hundred bytes of content for the test file ok." mp = (b"--b\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nassistants\r\n--b\r\nContent-Disposition: form-data; name=\"file\"; filename=\"atlas.txt\"\r\nContent-Type: text/plain\r\n\r\n" + content + b"\r\n--b--\r\n") st, f, _ = openai_request("POST", "/v1/files", data=mp, content_type="multipart/form-data; boundary=b", note="upload file for file_search") fid = f["id"] openai_request("POST", f"/v1/vector_stores/{vsid}/files", {"file_id": fid}, note="attach file") for _ in range(20): st, vf, _ = openai_request("GET", f"/v1/vector_stores/{vsid}/files/{fid}", note="poll") if vf.get("status") in ("completed", "failed"): break time.sleep(1.5) stream("file_search_stream", {"model": NANO, "input": "What is the secret codeword? Reply with the codeword only.", "tools": [{"type": "file_search", "vector_store_ids": [vsid], "max_num_results": 2}], "include": ["file_search_call.results"], "max_output_tokens": 300, "reasoning": {"effort": "low"}}) log_request("openai", "POST", "/v1/responses", 200, 0.0025, "file search per-call fee ($2.50/1k)") openai_request("DELETE", f"/v1/vector_stores/{vsid}", note="delete vector store") openai_request("DELETE", f"/v1/files/{fid}", note="delete file") # cheap streams for event capture stream("mcp_stream", {"model": NANO, "input": "Use read_wiki_structure for openai/openai-python; reply with the first topic only.", "tools": [{"type": "mcp", "server_label": "deepwiki", "server_url": "https://mcp.deepwiki.com/mcp", "require_approval": "never", "allowed_tools": ["read_wiki_structure"]}], "max_output_tokens": 64, "reasoning": {"effort": "low"}}) stream("custom_stream", {"model": NANO, "input": "Is 2+2 equal to 4? Use the tool.", "tool_choice": {"type": "custom", "name": "answer_yes_no"}, "tools": [{"type": "custom", "name": "answer_yes_no", "format": {"type": "grammar", "syntax": "regex", "definition": r"^(yes|no)$"}}], "max_output_tokens": 32, "reasoning": {"effort": "low"}}) stream("shell_local_stream", {"model": NANO, "input": "Propose the shell command `echo OK`.", "tool_choice": {"type": "shell"}, "tools": [{"type": "shell", "environment": {"type": "local"}}], "max_output_tokens": 64, "reasoning": {"effort": "low"}}) stream("apply_patch_stream", {"model": NANO, "input": "Create hello.txt containing OK using apply_patch.", "tool_choice": {"type": "apply_patch"}, "tools": [{"type": "apply_patch"}], "max_output_tokens": 128, "reasoning": {"effort": "low"}}) PROBES = {"retry": p_retry_streams, "function": p_function, "custom": p_custom, "web_search": p_web_search, "code_interpreter": p_code_interpreter, "image": p_image_invalid, "mcp": p_mcp, "computer": p_computer, "skills": p_skills, "shell_patch": p_shell_patch, "tool_search": p_tool_search, "file_search": p_file_search} if __name__ == "__main__": OUT.mkdir(parents=True, exist_ok=True) for name in (sys.argv[1:] or PROBES): print(f"===== {name}") try: PROBES[name]() except Exception as e: # noqa: BLE001 print(f"[{name}] EXCEPTION {mask(repr(e))[:300]}")