#!/usr/bin/env python3 """ ZenithAI - OpenAI Responses API (/v1/responses): complete, runnable examples. Guide: https://docs.zenithai.si/guides/responses/ ZenithAI speaks OpenAI's Responses protocol at /v1/responses, so the official `openai` Python SDK works unchanged: point `base_url` at your ZenithAI server and use a ZenithAI API key. What this script shows (each one is a small function you can copy): models list the model tiers your key can use (GET /v1/models) text one question, one answer (POST /v1/responses) instructions a system prompt via `instructions` multiturn a conversation: ZenithAI is STATELESS, so you send the history stream streaming text, printed as it arrives (typed events) json structured output that follows a JSON schema (`text.format`) tools function calling: model asks, your code runs it, model answers toolstream function calling while streaming image an image in the input (needs a vision tier and --image FILE) toolimage a function that returns an image to the model (--image FILE) errors what errors look like and how to handle them Setup ----- pip install "openai>=1.66" # any recent openai SDK with client.responses # Linux / macOS export ZENITHAI_BASE_URL="https://dummydomain/v1" # your ZenithAI server + /v1 export ZENITHAI_API_KEY="YOUR_API_KEY" # create one in ZenithAI: Settings > API keys # Windows PowerShell $env:ZENITHAI_BASE_URL = "https://dummydomain/v1" $env:ZENITHAI_API_KEY = "YOUR_API_KEY" Run --- python zenithai_responses_examples.py # every example except `image` and `toolimage` python zenithai_responses_examples.py text stream # only the ones you name python zenithai_responses_examples.py tools --tool-model Genius python zenithai_responses_examples.py image --image photo.jpg --model Smart python zenithai_responses_examples.py toolimage --image bill.png Good to know ------------ * Model names are ZenithAI tiers (Instant, Fast, Smart, Genius); `models` prints the ones available to your key. * Function calling (tools) works on tiers whose /v1/models entry has capabilities.tools = true; a tier without it answers HTTP 400 `unsupported_tools`. The tool examples use --tool-model (default Fast). * Nothing is stored on the server. `store: true` and `previous_response_id` are refused; send the whole conversation in `input` every turn (see `multiturn` and `tools`). Output items can be sent back unchanged. * Function tools only. OpenAI's hosted tools (web_search, file_search, ...) are refused with HTTP 400 `unsupported_tool`. * Limits apply per API key (60 requests a minute by default). When one is hit the answer is HTTP 429 with a `Retry-After` header; the SDK waits and retries for you (see `max_retries` below). * Never print, log or commit your API key. This script only reads it from the environment. """ from __future__ import annotations import argparse import base64 import json import mimetypes import os import sys import time try: import openai from openai import OpenAI except ImportError: # pragma: no cover - friendly message for first-time users sys.exit('This example needs the OpenAI SDK: pip install "openai>=1.66"') # -------------------------------------------------------------------------------------- # Client # -------------------------------------------------------------------------------------- def make_client() -> OpenAI: """One client for every example. base_url your ZenithAI server + /v1 (no trailing slash needed) api_key a ZenithAI API key (sent as `Authorization: Bearer ...`) timeout seconds for one request; tool calls on large tiers can take ~1 minute max_retries the SDK retries 429 (rate limit) and 5xx answers, honouring Retry-After """ base_url = os.environ.get("ZENITHAI_BASE_URL", "").strip().rstrip("/") if not base_url: sys.exit("Set ZENITHAI_BASE_URL first, for example https://your-server/v1 (see the top of this file).") api_key = os.environ.get("ZENITHAI_API_KEY", "").strip() if not api_key: sys.exit("Set ZENITHAI_API_KEY first (see the top of this file).") if not api_key.isascii() or any(ch.isspace() for ch in api_key): sys.exit("ZENITHAI_API_KEY does not look like an API key (it holds spaces or non-ASCII " "characters). Copy only the key itself.") return OpenAI(base_url=base_url, api_key=api_key, timeout=120, max_retries=3) def heading(title: str) -> None: print("\n" + "=" * 78 + f"\n{title}\n" + "=" * 78) def show_usage(response) -> None: """Every response carries token usage; useful for cost and context tracking.""" u = response.usage if u is not None: print(f"[usage] input={u.input_tokens} output={u.output_tokens} total={u.total_tokens} " f"status={response.status} id={response.id}") # -------------------------------------------------------------------------------------- # 1. Models # -------------------------------------------------------------------------------------- def example_models(client: OpenAI, model: str) -> None: """List the tiers your key may use. Use these names as `model`.""" heading("models: the tiers you can use") for m in client.models.list().data: print(f"- {m.id}") # -------------------------------------------------------------------------------------- # 2. Plain text # -------------------------------------------------------------------------------------- def example_text(client: OpenAI, model: str) -> None: """The smallest call: a string in, text out. `response.output_text` joins all text parts of the answer for you. """ heading(f"text: one question, one answer ({model})") response = client.responses.create( model=model, input="In two sentences: why is the sky blue?", max_output_tokens=300, # the output allowance for this answer ) print(response.output_text) show_usage(response) # -------------------------------------------------------------------------------------- # 3. Instructions (system prompt) # -------------------------------------------------------------------------------------- def example_instructions(client: OpenAI, model: str) -> None: """`instructions` is the system prompt: tone, format, rules for this answer.""" heading("instructions: a system prompt") response = client.responses.create( model=model, instructions="You are a concise assistant for an Indian finance team. " "Use Indian number formatting (lakh, crore) and DD-MM-YYYY dates.", input="Write one line stating that revenue was 12500000 rupees on 2026-03-31.", temperature=0.2, # lower = more predictable wording ) print(response.output_text) show_usage(response) # -------------------------------------------------------------------------------------- # 4. Multi-turn conversation (stateless) # -------------------------------------------------------------------------------------- def example_multiturn(client: OpenAI, model: str) -> None: """ZenithAI keeps no conversation on the server. Keep the history yourself: after each turn append the returned output items (`response.output`) to your list, then append the next user message. Output items can be sent back exactly as you received them. """ heading("multiturn: a conversation (you keep the history)") history: list = [{"role": "user", "content": "My name is Asha and I work in Pune. Remember that."}] first = client.responses.create(model=model, input=history, max_output_tokens=200) print("assistant:", first.output_text) # 1) add what the model said, 2) add the next question history += [item.model_dump(exclude_none=True) for item in first.output] history.append({"role": "user", "content": "Which city do I work in, and what is my name?"}) second = client.responses.create(model=model, input=history, max_output_tokens=200) print("assistant:", second.output_text) show_usage(second) # -------------------------------------------------------------------------------------- # 5. Streaming text # -------------------------------------------------------------------------------------- def example_stream(client: OpenAI, model: str) -> None: """With stream=True you get typed events as the answer is written. The ones most apps need: response.output_text.delta a piece of text -> event.delta response.completed the final response object -> event.response response.incomplete stopped early (e.g. max_output_tokens reached) response.failed an error after streaming started -> event.response.error Every event has a `sequence_number`, starting at 0. """ heading("stream: text as it arrives") stream = client.responses.create( model=model, input="List three tips for writing clear emails, one line each.", stream=True, ) final = None for event in stream: if event.type == "response.output_text.delta": print(event.delta, end="", flush=True) elif event.type in ("response.completed", "response.incomplete"): final = event.response elif event.type == "response.failed": print("\n[stream failed]", event.response.error) print() if final is not None: show_usage(final) def example_stream_helper(client: OpenAI, model: str) -> None: """The SDK's helper does the event bookkeeping and gives you the final response.""" heading("stream (helper): client.responses.stream(...)") with client.responses.stream(model=model, input="Say hello in Hindi, Tamil and Marathi.") as stream: for event in stream: if event.type == "response.output_text.delta": print(event.delta, end="", flush=True) final = stream.get_final_response() print() show_usage(final) # -------------------------------------------------------------------------------------- # 6. Structured output (JSON schema) # -------------------------------------------------------------------------------------- INVOICE_SCHEMA = { "type": "object", "properties": { "vendor": {"type": "string"}, "invoice_date": {"type": "string", "description": "DD-MM-YYYY"}, "total_inr": {"type": "number"}, "gst_percent": {"type": "number"}, }, "required": ["vendor", "invoice_date", "total_inr", "gst_percent"], "additionalProperties": False, } def example_json(client: OpenAI, model: str) -> None: """`text.format` = json_schema makes the answer valid JSON in your shape. With "strict": True every property must be listed in "required" and "additionalProperties" must be False. """ heading("json: structured output with a JSON schema") response = client.responses.create( model=model, input="Invoice from Sharma Traders dated 3rd March 2026, total Rs 1,18,000 including 18% GST.", text={"format": {"type": "json_schema", "name": "invoice", "schema": INVOICE_SCHEMA, "strict": True}}, ) data = json.loads(response.output_text) print(json.dumps(data, indent=2, ensure_ascii=False)) show_usage(response) # -------------------------------------------------------------------------------------- # 7. Function calling (tools) # -------------------------------------------------------------------------------------- # Your functions. The model never runs code; it asks you to run these. def get_exchange_rate(currency: str) -> dict: """A stand-in for your own API or database lookup.""" rates = {"USD": 83.2, "EUR": 90.6, "GBP": 105.4, "AED": 22.7} return {"currency": currency.upper(), "inr_per_unit": rates.get(currency.upper())} def convert_amount(amount: float, currency: str) -> dict: rate = get_exchange_rate(currency)["inr_per_unit"] return {"amount": amount, "currency": currency.upper(), "inr": None if rate is None else round(amount * rate, 2)} TOOLS = [ { "type": "function", "name": "get_exchange_rate", "description": "Rupees (INR) for one unit of a foreign currency.", "parameters": { "type": "object", "properties": {"currency": {"type": "string", "description": "ISO code, e.g. USD"}}, "required": ["currency"], "additionalProperties": False, }, "strict": True, }, { "type": "function", "name": "convert_amount", "description": "Convert an amount of a foreign currency to rupees.", "parameters": { "type": "object", "properties": {"amount": {"type": "number"}, "currency": {"type": "string"}}, "required": ["amount", "currency"], "additionalProperties": False, }, "strict": True, }, ] FUNCTIONS = {"get_exchange_rate": get_exchange_rate, "convert_amount": convert_amount} def run_tool_loop(client: OpenAI, model: str, question: str, max_rounds: int = 5) -> None: """The standard loop: 1. send the input with `tools` 2. if the output holds `function_call` items, run each function yourself 3. send again: the earlier input + the returned output items + one {"type": "function_call_output", "call_id": ..., "output": ""} per call 4. repeat until the model answers with text """ items: list = [{"role": "user", "content": question}] for round_no in range(1, max_rounds + 1): response = client.responses.create(model=model, input=items, tools=TOOLS, tool_choice="auto") calls = [item for item in response.output if item.type == "function_call"] if not calls: print("assistant:", response.output_text) show_usage(response) return items += [item.model_dump(exclude_none=True) for item in response.output] for call in calls: args = json.loads(call.arguments) # arguments arrive as a JSON string result = FUNCTIONS[call.name](**args) print(f"[round {round_no}] model called {call.name}({args}) -> {result}") items.append({"type": "function_call_output", "call_id": call.call_id, "output": json.dumps(result)}) print("stopped: too many tool rounds") def example_tools(client: OpenAI, model: str) -> None: """`model` here is the tool model (--tool-model): a tier with capabilities.tools in /v1/models.""" heading(f"tools: function calling ({model})") run_tool_loop(client, model, "How many rupees is 250 US dollars? Use the tools.") def example_toolstream(client: OpenAI, model: str) -> None: """Function calls can be streamed too. The call arrives as response.output_item.added (type function_call) ... response.function_call_arguments.done, and the full call is in the final response's output. Note: with tools, ZenithAI checks the whole call before sending it, so the events may arrive together at the end rather than token by token. """ heading(f"toolstream: function calling while streaming ({model})") with client.responses.stream(model=model, input="What is the rupee rate for 1 euro? Use a tool.", tools=TOOLS) as stream: for event in stream: if event.type == "response.function_call_arguments.done": print(f"[event] arguments complete: {event.arguments}") final = stream.get_final_response() for item in final.output: if item.type == "function_call": print(f"call: {item.name}({item.arguments}) call_id={item.call_id}") show_usage(final) # -------------------------------------------------------------------------------------- # 8. Image input # -------------------------------------------------------------------------------------- def example_image(client: OpenAI, model: str, path: str | None) -> None: """Send an image as a data: URL inside an `input_image` part. Use a tier whose /v1/models entry supports vision. Leave `detail` out (or "auto"). """ heading(f"image: describe a picture ({model})") if not path: print("skipped: pass --image FILE (jpg/png) to run this example") return mime = mimetypes.guess_type(path)[0] or "image/png" with open(path, "rb") as fh: data_url = f"data:{mime};base64," + base64.b64encode(fh.read()).decode("ascii") response = client.responses.create( model=model, input=[{"role": "user", "content": [ {"type": "input_text", "text": "Describe this image in two sentences."}, {"type": "input_image", "image_url": data_url}, ]}], ) print(response.output_text) show_usage(response) def example_toolimage(client: OpenAI, model: str, path: str | None) -> None: """A function can return an image. The model reads it before it answers. function_call_output.output is a string, or a list of input_text and input_image parts. Send the output right after the function_call it answers (the loop below does). Needs a tier with capabilities.tools and capabilities.vision in /v1/models. """ heading("toolimage: a function that returns an image") if not path: print("skipped: pass --image FILE (a photo or scan for the function to return)") return mime = mimetypes.guess_type(path)[0] or "image/png" with open(path, "rb") as fh: data_url = f"data:{mime};base64," + base64.b64encode(fh.read()).decode("ascii") tools = [{ "type": "function", "name": "get_document_scan", "description": "Fetch the scanned image of a document by its id", "parameters": {"type": "object", "properties": {"doc_id": {"type": "string"}}, "required": ["doc_id"], "additionalProperties": False}, "strict": True, }] items: list = [{"role": "user", "content": "Look at document D-17 and describe what it is in two sentences."}] response = client.responses.create(model=model, input=items, tools=tools) for _ in range(5): calls = [item for item in response.output if item.type == "function_call"] if not calls: break items += response.output for call in calls: doc_id = json.loads(call.arguments)["doc_id"] print(f"model asked for get_document_scan({doc_id!r})") items.append({"type": "function_call_output", "call_id": call.call_id, "output": [ {"type": "input_text", "text": f"Scan of document {doc_id}:"}, {"type": "input_image", "image_url": data_url}, ]}) response = client.responses.create(model=model, input=items, tools=tools) print(response.output_text) show_usage(response) # -------------------------------------------------------------------------------------- # 9. Errors # -------------------------------------------------------------------------------------- def example_errors(client: OpenAI, model: str) -> None: """Errors use OpenAI's format: an HTTP status plus {"error": {"message", "type", "param", "code"}}. openai.BadRequestError 400 a field ZenithAI does not support, or bad input openai.AuthenticationError 401 missing or wrong API key openai.NotFoundError 404 unknown model name openai.RateLimitError 429 a limit was hit; the SDK already retried `max_retries` times openai.APIStatusError any other HTTP error (e.g. 503 when the server is busy) """ heading("errors: what they look like") # a) a stored conversation is not supported (ZenithAI is stateless) try: client.responses.create(model=model, input="hello", store=True) except openai.BadRequestError as exc: print(f"400 as expected: code={exc.code!r} param={exc.param!r} message={exc.message!r}") # b) an unknown model name try: client.responses.create(model="no-such-model", input="hello") except (openai.NotFoundError, openai.BadRequestError) as exc: print(f"{exc.status_code} as expected: code={exc.code!r} message={exc.message!r}") # c) a pattern for production code try: response = client.responses.create(model=model, input="Reply with the word OK.", max_output_tokens=20) print("normal call:", response.output_text.strip()) except openai.RateLimitError as exc: wait = exc.response.headers.get("retry-after", "a few") print(f"rate limited: try again in {wait} s") except openai.APIStatusError as exc: print(f"HTTP {exc.status_code}: {exc.message}") except openai.APIConnectionError as exc: print(f"could not reach the server: {exc}") # -------------------------------------------------------------------------------------- # Runner # -------------------------------------------------------------------------------------- EXAMPLES = { "models": example_models, "text": example_text, "instructions": example_instructions, "multiturn": example_multiturn, "stream": example_stream, "streamhelper": example_stream_helper, "json": example_json, "tools": example_tools, "toolstream": example_toolstream, "errors": example_errors, } def main() -> int: parser = argparse.ArgumentParser(description="ZenithAI /v1/responses examples (OpenAI Python SDK).") parser.add_argument("examples", nargs="*", help=f"which to run: {', '.join(list(EXAMPLES) + ['image', 'toolimage'])} (default: all but image and toolimage)") parser.add_argument("--model", default=os.environ.get("ZENITHAI_MODEL", "Fast"), help="tier name (default Fast)") parser.add_argument("--tool-model", default=os.environ.get("ZENITHAI_TOOL_MODEL", "Fast"), help="tier for the tool examples; it must support tools in /v1/models (default Fast)") parser.add_argument("--image", help="image file for the `image` and `toolimage` examples") args = parser.parse_args() client = make_client() chosen = args.examples or list(EXAMPLES) failed = [] for name in chosen: started = time.monotonic() try: if name == "image": example_image(client, args.model, args.image) elif name == "toolimage": example_toolimage(client, args.tool_model, args.image) elif name in ("tools", "toolstream"): EXAMPLES[name](client, args.tool_model) elif name in EXAMPLES: EXAMPLES[name](client, args.model) else: print(f"unknown example {name!r}; choose from {', '.join(list(EXAMPLES) + ['image', 'toolimage'])}") failed.append(name) continue print(f"[{name}] ok in {time.monotonic() - started:.1f} s") except openai.APIError as exc: # keep going with the next example print(f"[{name}] FAILED: {type(exc).__name__}: {exc}") failed.append(name) print("\nall examples passed" if not failed else f"\nfailed: {', '.join(failed)}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())