# A simple CLI tool for testing OpenAI-compatible APIs and measuring streaming performance. It calculates TTFT, generation time, completion tokens, and real-time TPS with optional proxy support.

> Source: <https://gist.github.com/maanimeisam/6f766b5fbf9f3fdfce49b92b68645b02>
> Published: 2026-09-08 08:46:43+00:00

|  | #!/usr/bin/env python3 | 
|  | import json | 
|  | import time | 
|  | import urllib.request | 
|  | import urllib.error | 
|  | def main(): | 
|  | base_url = input("Base URL: ").strip().rstrip("/") | 
|  | model = input("Model: ").strip() | 
|  | api_key = input("API Key: ").strip() | 
|  | proxy = input("Proxy (optional): ").strip() | 
|  | prompt = input("Prompt: ").strip() | 
|  | url = f"{base_url}/chat/completions" | 
|  | payload = { | 
|  | "model": model, | 
|  | "messages": [ | 
|  | { | 
|  | "role": "user", | 
|  | "content": prompt, | 
|  | } | 
|  | ], | 
|  | "stream": True, | 
|  | "stream_options": { | 
|  | "include_usage": True, | 
|  | }, | 
|  | } | 
|  | data = json.dumps(payload).encode() | 
|  | request = urllib.request.Request( | 
|  | url, | 
|  | data=data, | 
|  | headers={ | 
|  | "Authorization": f"Bearer {api_key}", | 
|  | "Content-Type": "application/json", | 
|  | "Accept": "text/event-stream", | 
|  | }, | 
|  | method="POST", | 
|  | ) | 
|  | # Proxy | 
|  | if proxy: | 
|  | proxy_handler = urllib.request.ProxyHandler({ | 
|  | "http": proxy, | 
|  | "https": proxy, | 
|  | }) | 
|  | opener = urllib.request.build_opener(proxy_handler) | 
|  | else: | 
|  | opener = urllib.request.build_opener() | 
|  | print() | 
|  | print("Connecting...") | 
|  | print() | 
|  | request_start = time.perf_counter() | 
|  | first_token_time = None | 
|  | last_token_time = None | 
|  | completion_tokens = None | 
|  | try: | 
|  | with opener.open(request, timeout=600) as response: | 
|  | for raw_line in response: | 
|  | line = raw_line.decode("utf-8", errors="replace").strip() | 
|  | if not line.startswith("data:"): | 
|  | continue | 
|  | data_str = line[5:].strip() | 
|  | if data_str == "[DONE]": | 
|  | break | 
|  | try: | 
|  | chunk = json.loads(data_str) | 
|  | except json.JSONDecodeError: | 
|  | continue | 
|  | # Usage | 
|  | usage = chunk.get("usage") | 
|  | if usage: | 
|  | completion_tokens = usage.get("completion_tokens") | 
|  | choices = chunk.get("choices", []) | 
|  | if not choices: | 
|  | continue | 
|  | delta = choices[0].get("delta", {}) | 
|  | content = delta.get("content") | 
|  | if content: | 
|  | now = time.perf_counter() | 
|  | if first_token_time is None: | 
|  | first_token_time = now | 
|  | ttft = first_token_time - request_start | 
|  | print( | 
|  | f"\n[TTFT: {ttft:.3f}s]\n" | 
|  | ) | 
|  | last_token_time = now | 
|  | print(content, end="", flush=True) | 
|  | except urllib.error.HTTPError as e: | 
|  | print(f"\nHTTP Error {e.code}:") | 
|  | print(e.read().decode(errors="replace")) | 
|  | return | 
|  | except Exception as e: | 
|  | print(f"\nError: {e}") | 
|  | return | 
|  | print("\n") | 
|  | print("=" * 50) | 
|  | if first_token_time is None: | 
|  | print("No tokens received.") | 
|  | return | 
|  | if last_token_time is None: | 
|  | last_token_time = first_token_time | 
|  | generation_time = last_token_time - first_token_time | 
|  | total_time = last_token_time - request_start | 
|  | print(f"Model: {model}") | 
|  | print(f"Completion tokens: {completion_tokens}") | 
|  | print(f"TTFT: {first_token_time - request_start:.3f}s") | 
|  | print(f"Generation time: {generation_time:.3f}s") | 
|  | print(f"Total time: {total_time:.3f}s") | 
|  | if completion_tokens and generation_time > 0: | 
|  | tps = completion_tokens / generation_time | 
|  | print(f"TPS: {tps:.2f} tokens/s") | 
|  | else: | 
|  | print("TPS: N/A") | 
|  | if __name__ == "__main__": | 
|  | main() |