import json
import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError
history = [{"role": "user", "content": "What is 7 times 8? Use multiply."}]
tools = [{
"type": "function", "name": "multiply", "description": "Multiply two finite numbers.",
"strict": True,
"parameters": {
"type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"], "additionalProperties": False,
},
}]
seen_call_ids = set()
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
try:
for round_index in range(3):
result = client.responses.create(
model=os.environ["API_MODEL"], input=history, tools=tools,
instructions="Use multiply for arithmetic, then answer briefly.",
store=False, include=["reasoning.encrypted_content"],
)
if result.status != "completed":
raise ValueError("Response did not complete.")
calls = []
for item in result.output:
if item.type == "function_call":
if getattr(item, "status", None) not in (None, "completed"):
raise ValueError("Function call did not complete.")
calls.append(item)
elif item.type == "message":
if item.status != "completed" or any(part.type != "output_text" for part in item.content):
raise ValueError("Refusal or incomplete message.")
elif item.type != "reasoning":
raise ValueError("Unsupported output item.")
if not calls:
if not result.output_text.strip():
raise ValueError("No final text.")
print(result.output_text)
break
if round_index == 2 or len(calls) > 4:
raise ValueError("Tool round or call limit reached.")
outputs = []
for call in calls:
if (call.name != "multiply" or not isinstance(call.call_id, str)
or not call.call_id.strip() or call.call_id in seen_call_ids):
raise ValueError("Unknown tool or invalid call ID.")
args = json.loads(call.arguments)
if not isinstance(args, dict) or set(args) != {"a", "b"}:
raise ValueError("Invalid tool arguments.")
if any(type(value) not in (int, float) or not math.isfinite(value) for value in args.values()):
raise ValueError("Expected finite numbers.")
value = args["a"] * args["b"]
if not math.isfinite(value):
raise ValueError("Invalid tool result.")
seen_call_ids.add(call.call_id)
outputs.append({
"type": "function_call_output", "call_id": call.call_id,
"output": json.dumps({"value": value}, allow_nan=False),
})
history.extend(item.model_dump(exclude_none=True) for item in result.output)
history.extend(outputs)
except APIStatusError as error:
raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
except APIConnectionError:
raise SystemExit("Connection failed. No automatic retry.")
except (ValueError, TypeError, AttributeError, OverflowError):
raise SystemExit("Tool workflow failed: check output, call IDs, arguments, and limits.")