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.

| | #!/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() |