#!/usr/bin/env python3 """Small reproducible, synthetic API evaluation. Never prints or saves API keys. FLONNO_API_KEY=... python3 scripts/evaluate-agent-api.py > report.json This makes billable requests. It is a smoke evaluation, not a general benchmark. """ import concurrent.futures import json import os import time import urllib.error import urllib.request BASE = os.environ.get("FLONNO_BASE_URL", "https://www.flonno.com/v1").rstrip("/") TOOL = {"type": "function", "function": {"name": "lookup_stock", "description": "Look up stock for a product", "parameters": {"type": "object", "properties": {"product": {"type": "object", "properties": {"sku": {"type": "string"}, "warehouse": {"type": "string"}}, "required": ["sku", "warehouse"], "additionalProperties": False}}, "required": ["product"], "additionalProperties": False}}} def request(payload): started = time.monotonic() first = None calls = {} content = "" reasoning = "" usage = None finished = False done = False req = urllib.request.Request(BASE + "/chat/completions", data=json.dumps(payload).encode(), headers={"Authorization": "Bearer " + os.environ["FLONNO_API_KEY"], "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=180) as response: if payload.get("stream"): for line in response: if not line.startswith(b"data:"): continue data = line[5:].strip() if data == b"[DONE]": done = True continue if not data: continue event = json.loads(data) usage = event.get("usage") or usage for choice in event.get("choices", []): delta = choice.get("delta", {}) if any(delta.get(k) for k in ("content", "reasoning_content", "reasoning", "tool_calls")): first = first or time.monotonic() content += delta.get("content") or "" reasoning += delta.get("reasoning_content") or delta.get("reasoning") or "" finished = finished or bool(choice.get("finish_reason")) for call in delta.get("tool_calls", []): item = calls.setdefault(call.get("index", 0), {"id": "", "type": "function", "function": {"name": "", "arguments": ""}}) item["id"] += call.get("id") or "" for k in ("name", "arguments"): item["function"][k] += call.get("function", {}).get(k) or "" else: event = json.load(response) usage = event.get("usage") choice = event["choices"][0] message = choice["message"] content = message.get("content") or "" reasoning = message.get("reasoning_content") or message.get("reasoning") or "" calls = dict(enumerate(message.get("tool_calls") or [])) finished = bool(choice.get("finish_reason")) done = True elapsed = time.monotonic() - started return {"durationMs": round(elapsed * 1000, 2), "firstEventMs": round((first - started) * 1000, 2) if first else None, "content": content, "reasoningCharacters": len(reasoning), "calls": list(calls.values()), "usage": usage, "finished": finished, "done": done} def run_case(kind, index): base = {"model": "qwen3.8-27b", "reasoning_effort": "none", "max_tokens": 256, "temperature": 0} if kind in ("tool", "stream_tool", "parallel_tool"): count = 2 if kind == "parallel_tool" else 1 prompt = "Use lookup_stock for sku ABC-17 in warehouse Zurich." if count == 1 else "Use lookup_stock twice: sku ABC-17 in warehouse Zurich and sku XYZ-42 in warehouse Bern." payload = {**base, "messages": [{"role": "user", "content": prompt}], "tools": [TOOL], "tool_choice": "auto", "parallel_tool_calls": count == 2, "stream": kind != "tool"} else: payload = {**base, "messages": [{"role": "user", "content": "Return only a JSON object with city Zurich and count 3."}], "response_format": {"type": "json_object"}, "stream": True} if kind == "thinking_json": payload.update(reasoning_effort="medium", max_tokens=1024) result = {"case": kind, "index": index} try: response = request(payload) valid_usage = isinstance(response["usage"], dict) and all(isinstance(response["usage"].get(k), int) and response["usage"][k] >= 0 for k in ("prompt_tokens", "completion_tokens", "total_tokens")) if kind in ("tool", "stream_tool", "parallel_tool"): actual = sorted((json.loads(c["function"]["arguments"])["product"]["sku"], json.loads(c["function"]["arguments"])["product"]["warehouse"]) for c in response["calls"] if c["id"] and c["function"]["name"] == "lookup_stock") expected = [("ABC-17", "Zurich"), ("XYZ-42", "Bern")] if kind == "parallel_tool" else [("ABC-17", "Zurich")] valid = actual == expected else: valid = json.loads(response["content"]) == {"city": "Zurich", "count": 3} result.update(response, passed=valid and valid_usage and response["finished"] and response["done"]) except urllib.error.HTTPError as error: result.update(passed=False, error=f"HTTP {error.code}: " + error.read().decode()[:500]) except Exception as error: result.update(passed=False, error=type(error).__name__ + ": " + str(error)[:180]) return result def evaluate(): rows = [run_case(kind, index) for kind in ("tool", "stream_tool", "json", "parallel_tool", "thinking_json") for index in range(3)] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: concurrent_rows = list(executor.map(lambda i: run_case("json", i), range(2))) for row in concurrent_rows: row["case"] = "concurrency_2_json" rows.extend(concurrent_rows) # A real two-turn tool loop, with a synthetic tool result executed by this client. try: initial = request({"model": "qwen3.8-27b", "reasoning_effort": "none", "max_tokens": 256, "messages": [{"role": "user", "content": "Look up stock for sku ABC-17 in warehouse Zurich, then tell me the quantity."}], "tools": [TOOL], "tool_choice": "auto"}) call = initial["calls"][0] valid_call = call["function"]["name"] == "lookup_stock" and json.loads(call["function"]["arguments"]) == {"product": {"sku": "ABC-17", "warehouse": "Zurich"}} follow = request({"model": "qwen3.8-27b", "reasoning_effort": "none", "max_tokens": 128, "stream": True, "messages": [{"role": "user", "content": "Look up stock for sku ABC-17 in warehouse Zurich, then tell me the quantity."}, {"role": "assistant", "content": None, "tool_calls": [call]}, {"role": "tool", "tool_call_id": call["id"], "content": '{"quantity":17}'}], "tools": [TOOL]}) rows.extend([{**initial, "case": "loop_call", "passed": valid_call}, {**follow, "case": "loop_result", "passed": valid_call and "17" in follow["content"] and follow["finished"] and follow["done"]}]) except Exception as error: rows.append({"case": "tool_loop", "passed": False, "error": type(error).__name__}) return {"measuredAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "baseUrl": BASE, "operatorRun": True, "scope": "Synthetic smoke evaluation; 3 trials per main case, concurrency 2. Not independent or statistically representative; no full-load or general accuracy claim.", "requests": rows} if __name__ == "__main__": print(json.dumps(evaluate(), indent=2))