"""Small Chat Completions capability probe. Sends at most six requests, never retries. Not a Responses/Anthropic conformance suite, benchmark, or uptime monitor. Install: python -m pip install requests jsonschema Run: python check_api.py --base-url https://api.example.com/v1 --model MODEL --out result.json Set LLM_API_KEY in the environment. Do not put secrets in CLI arguments. """ import argparse, json, os, time from pathlib import Path from urllib.parse import urlparse import requests from jsonschema import validate, ValidationError SCHEMA = {'type': 'object', 'properties': {'sku': {'type': 'string'}, 'quantity': {'type': 'integer'}}, 'required': ['sku', 'quantity'], 'additionalProperties': False} TOOL = {'type': 'function', 'function': {'name': 'lookup_stock', 'description': 'Read stock for a synthetic tutorial SKU.', 'parameters': {'type': 'object', 'properties': {'sku': {'type': 'string'}}, 'required': ['sku'], 'additionalProperties': False}}} class Probe: def __init__(self, base, model, key, token_field='max_completion_tokens', max_output=256): self.url = base.rstrip('/') + '/chat/completions' self.model = model self.key = key self.rows = [] self.requests = 0 self.token_field = token_field self.max_output = max_output def post(self, body, stream=False): self.requests += 1 return requests.post(self.url, headers={'Authorization': 'Bearer ' + self.key}, json={'model': self.model, self.token_field: self.max_output, **body}, stream=stream, timeout=(10, 45), allow_redirects=False) def check(self, name, fn): row = {'check': name} start = time.monotonic() try: detail = fn() row.update(status='PASS', detail=detail) except requests.HTTPError as e: code = e.response.status_code row.update(status='FAIL', detail={'http_status': code, 'reason': 'Request rejected; inspect provider docs and a redacted response. This does not prove the model lacks the feature.'}) except requests.RequestException as e: row.update(status='UNKNOWN', detail={'reason': type(e).__name__}) except (ValueError, KeyError, IndexError, AssertionError, ValidationError) as e: row.update(status='FAIL', detail={'reason': str(e)[:240]}) except Exception as e: row.update(status='UNKNOWN', detail={'reason': 'Unexpected response shape: ' + type(e).__name__}) row['elapsed_seconds'] = round(time.monotonic() - start, 3) self.rows.append(row) def parsed(self, payload): with self.post(payload) as r: r.raise_for_status() if not 200 <= r.status_code < 300: raise ValueError('Redirect/unexpected HTTP status') return r.json() def chat(self): d = self.parsed({'messages': [{'role': 'user', 'content': 'Reply exactly READY.'}]}) msg = d['choices'][0]['message'] if not msg.get('content', '').strip().rstrip('.') == 'READY': raise ValueError('Expected READY text') return {'model_returned': d.get('model'), 'finish_reason': d['choices'][0].get('finish_reason'), 'usage': d.get('usage')} def structured(self): d = self.parsed({'messages': [{'role': 'user', 'content': 'Return sku 000102 as a string and quantity 3 as an integer.'}], 'response_format': {'type': 'json_schema', 'json_schema': {'name': 'order', 'strict': True, 'schema': SCHEMA}}}) value = json.loads(d['choices'][0]['message']['content']) validate(value, SCHEMA) if not value == {'sku': '000102', 'quantity': 3}: raise ValueError('Schema passed, but task values are wrong') return {'schema_and_values_checked': True, 'usage': d.get('usage')} def tool(self): messages = [{'role': 'user', 'content': 'Look up stock for SKU 000102. Use the tool, then reply with only the stock quantity as decimal digits. No other text.'}] d = self.parsed({'messages': messages, 'tools': [TOOL], 'tool_choice': {'type': 'function', 'function': {'name': 'lookup_stock'}}}) message = d['choices'][0]['message'] calls = message.get('tool_calls', []) if not len(calls) == 1: raise ValueError('Expected exactly one tool call') call = calls[0] if not call.get('type') == 'function': raise ValueError('Expected function tool type') if not call['function']['name'] == 'lookup_stock': raise ValueError('Unknown tool') args = json.loads(call['function']['arguments']) validate(args, TOOL['function']['parameters']) if not args == {'sku': '000102'}: raise ValueError('Unexpected SKU') if not call.get('id'): raise ValueError('Missing tool_call_id') messages += [message, {'role': 'tool', 'tool_call_id': call['id'], 'content': json.dumps({'sku': '000102', 'stock': 17})}] final = self.parsed({'messages': messages, 'tools': [TOOL], 'tool_choice': 'none'}) text = final['choices'][0]['message'].get('content', '') if not (isinstance(text, str) and text.strip() == '17' and (not final['choices'][0]['message'].get('tool_calls'))): raise ValueError('Expected a final answer using the tool result') return {'tool_round_trip': True, 'first_usage': d.get('usage'), 'final_usage': final.get('usage'), 'review_note': 'Final answer must equal 17 exactly; fixed synthetic task only.', 'final_answer': text} def stream(self): text = [] finish = None done = False events = 0 usage = None data = [] deadline = time.monotonic() + 60 def event(lines): nonlocal finish, done, events, usage if not lines: return raw = '\n'.join(lines) if raw == '[DONE]': done = True return d = json.loads(raw) events += 1 if d.get('error'): raise ValueError('SSE contained an error event') if d.get('usage') is not None: usage = d['usage'] for c in d.get('choices', []): delta = c.get('delta', {}).get('content') if delta: text.append(delta) if c.get('finish_reason') is not None: finish = c['finish_reason'] with self.post({'messages': [{'role': 'user', 'content': 'Reply exactly READY.'}], 'stream': True}, True) as r: r.raise_for_status() if not 'text/event-stream' in r.headers.get('Content-Type', ''): raise ValueError('Expected SSE Content-Type') r.encoding = 'utf-8' for line in r.iter_lines(decode_unicode=True): if time.monotonic() > deadline: raise ValueError('Streaming deadline exceeded') if line == '': event(data) data = [] if done: break elif line.startswith('data:'): data.append(line[5:].lstrip()) event(data) if not (events and finish == 'stop' and done): raise ValueError('Stream incomplete: need JSON events, stop finish_reason and [DONE] for this Chat Completions profile') if not ''.join(text).strip().rstrip('.') == 'READY': raise ValueError('Unexpected stream text') return {'json_events': events, 'finish_reason': finish, 'done': done, 'usage': usage, 'note': 'No stream_options usage request: absent streamed usage is recorded, not treated as failure.'} def error_shape(self): with self.post({'messages': 'INVALID_EXPECTED_ARRAY'}) as r: if not r.status_code in (400, 422): raise ValueError(f'Expected validation rejection 400/422, got {r.status_code}') d = r.json() if not isinstance(d.get('error'), dict): raise ValueError('Expected JSON error object') return {'http_status': r.status_code, 'error_keys': sorted(d['error'].keys())} def run(self): for (name, fn) in [('chat', self.chat), ('structured-output', self.structured), ('tool-round-trip', self.tool), ('streaming', self.stream), ('error-shape', self.error_shape)]: self.check(name, fn) return {'profile': 'Chat Completions only; six requests maximum; no retries', 'model_requested': self.model, 'endpoint': self.url, 'requests': self.requests, 'results': self.rows, 'limits': 'No Responses, streaming tool arguments, load, fallback, all schemas, pricing or security certification tested. Output cap is requested on every call; provider support must be checked. Set account spend limits; a token parameter is not a guaranteed dollar cap. Socket inactivity timeout 45s; a best-effort stream deadline is checked between complete lines. This is not a wall-clock deadline: a trickling response can run longer. Interrupt a stalled run; do not deploy this starter as an unattended worker.'} if __name__ == '__main__': p = argparse.ArgumentParser() p.add_argument('--base-url', required=True) p.add_argument('--model', required=True) p.add_argument('--out', default='result.json') p.add_argument('--allow-local-http', action='store_true') p.add_argument('--token-field', choices=['max_tokens', 'max_completion_tokens'], default='max_completion_tokens') p.add_argument('--max-output', type=int, default=256) a = p.parse_args() if not 1 <= a.max_output <= 2048: raise SystemExit('--max-output must be 1..2048') u = urlparse(a.base_url) if u.username or u.password or u.query or u.fragment: raise SystemExit('Base URL must not contain credentials, query or fragment') if u.scheme != 'https' and (not (a.allow_local_http and u.scheme == 'http' and (u.hostname in ('127.0.0.1', 'localhost')))): raise SystemExit('Use HTTPS, or explicitly allow loopback HTTP for a local fixture') key = os.environ.get('LLM_API_KEY') if not key: raise SystemExit('Set LLM_API_KEY in your environment') out = Path(a.out) if out.exists(): raise SystemExit('Output exists; choose a new file to preserve previous results') result = Probe(a.base_url, a.model, key, a.token_field, a.max_output).run() out.write_text(json.dumps(result, indent=2), encoding='utf-8') print('\n'.join((f"{r['check']}: {r['status']}" for r in result['results'])))