#!/usr/bin/env python3 """ Grok Direct API Provider - Bypass Cloudflare using x-statsig-id token. This provider makes direct HTTP requests to Grok's internal API without browser automation. Requires: curl_cffi, x-statsig-id token from browser session. Usage: provider = GrokDirectProvider() result = provider.generate_image("a beautiful sunset") print(result) """ import os import json import uuid import time from typing import Optional from dotenv import load_dotenv from curl_cffi import requests load_dotenv() class GrokDirectProvider: """Direct API provider for Grok using curl_cffi with TLS fingerprint impersonation.""" BASE_URL = "https://grok.com" def __init__(self): self.sso_token = os.getenv('GROK_SSO_TOKEN') self.sso_rw_token = os.getenv('GROK_SSO_RW_TOKEN') self.user_id = os.getenv('GROK_USER_ID') self.statsig_id = os.getenv('GROK_STATSIG_ID') if not all([self.sso_token, self.sso_rw_token, self.user_id]): raise ValueError("Missing GROK_SSO_TOKEN, GROK_SSO_RW_TOKEN, or GROK_USER_ID in .env") if not self.statsig_id: raise ValueError( "Missing GROK_STATSIG_ID in .env. " "Capture this from browser DevTools → Network → any POST request headers" ) self.cookies = { 'sso': self.sso_token, 'sso-rw': self.sso_rw_token, 'x-userid': self.user_id, } def _get_headers(self) -> dict: """Build request headers with x-statsig-id.""" return { 'x-xai-request-id': str(uuid.uuid4()), 'x-statsig-id': self.statsig_id, 'sec-ch-ua-platform': '"Linux"', 'referer': 'https://grok.com/', 'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"', 'sec-ch-ua-mobile': '?0', 'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36', 'content-type': 'application/json', 'accept': '*/*', 'origin': 'https://grok.com', 'sec-fetch-site': 'same-origin', 'sec-fetch-mode': 'cors', 'sec-fetch-dest': 'empty', 'accept-language': 'en-US,en;q=0.9', } def generate_image( self, prompt: str, num_images: int = 2, timeout: int = 120 ) -> dict: """ Generate images using Grok's internal API. Args: prompt: Text prompt for image generation num_images: Number of images to generate (default 2) timeout: Request timeout in seconds Returns: dict with conversation_id, response text, and image URLs """ payload = { "temporary": False, "message": prompt, "fileAttachments": [], "imageAttachments": [], "disableSearch": False, "enableImageGeneration": True, "returnImageBytes": False, "returnRawGrokInXaiRequest": False, "enableImageStreaming": True, "imageGenerationCount": num_images, "forceConcise": False, "toolOverrides": {}, "enableSideBySide": True, "sendFinalMetadata": True, "isReasoning": False, "disableTextFollowUps": False, "responseMetadata": {}, "disableMemory": False, "forceSideBySide": False, "modelMode": "MODEL_MODE_AUTO", "isAsync": False } resp = requests.post( f"{self.BASE_URL}/rest/app-chat/conversations/new", headers=self._get_headers(), cookies=self.cookies, json=payload, impersonate='chrome131', timeout=timeout ) if resp.status_code != 200: raise Exception(f"API request failed: {resp.status_code} - {resp.text[:500]}") # Parse streaming response result = { "conversation_id": None, "response_text": "", "image_urls": [], "raw_response": resp.text } for line in resp.text.split('\n'): if not line.strip(): continue try: data = json.loads(line) if 'result' in data: res = data['result'] # Extract conversation ID if 'conversation' in res: result["conversation_id"] = res['conversation'].get('conversationId') # Extract response content if 'response' in res: response_data = res['response'] # Text tokens if 'token' in response_data: result["response_text"] += response_data['token'] # Image attachments if 'imageAttachment' in response_data: img = response_data['imageAttachment'] if 'imageUrl' in img: result["image_urls"].append(img['imageUrl']) elif 'url' in img: result["image_urls"].append(img['url']) # Final model response if 'modelResponse' in response_data: result["response_text"] = response_data['modelResponse'].get('message', '') except json.JSONDecodeError: continue return result def send_message( self, message: str, conversation_id: Optional[str] = None, timeout: int = 60 ) -> dict: """ Send a message to Grok (without image generation). Args: message: Text message to send conversation_id: Optional existing conversation ID timeout: Request timeout in seconds Returns: dict with conversation_id and response text """ payload = { "temporary": False, "message": message, "fileAttachments": [], "imageAttachments": [], "disableSearch": False, "enableImageGeneration": False, "returnImageBytes": False, "returnRawGrokInXaiRequest": False, "enableImageStreaming": False, "imageGenerationCount": 0, "forceConcise": False, "toolOverrides": {}, "enableSideBySide": False, "sendFinalMetadata": True, "isReasoning": False, "disableTextFollowUps": False, "responseMetadata": {}, "disableMemory": False, "forceSideBySide": False, "modelMode": "MODEL_MODE_AUTO", "isAsync": False } if conversation_id: payload["conversationId"] = conversation_id resp = requests.post( f"{self.BASE_URL}/rest/app-chat/conversations/new", headers=self._get_headers(), cookies=self.cookies, json=payload, impersonate='chrome131', timeout=timeout ) if resp.status_code != 200: raise Exception(f"API request failed: {resp.status_code} - {resp.text[:500]}") result = { "conversation_id": None, "response_text": "", } for line in resp.text.split('\n'): if not line.strip(): continue try: data = json.loads(line) if 'result' in data: res = data['result'] if 'conversation' in res: result["conversation_id"] = res['conversation'].get('conversationId') if 'response' in res: response_data = res['response'] if 'token' in response_data: result["response_text"] += response_data['token'] if 'modelResponse' in response_data: result["response_text"] = response_data['modelResponse'].get('message', '') except json.JSONDecodeError: continue return result def test_connection(self) -> bool: """Test if the API connection works.""" try: resp = requests.get( f"{self.BASE_URL}/rest/app-chat/conversations?pageSize=1", headers=self._get_headers(), cookies=self.cookies, impersonate='chrome131', timeout=30 ) return resp.status_code == 200 except Exception: return False if __name__ == '__main__': print("=" * 60) print("Grok Direct Provider Test") print("=" * 60) try: provider = GrokDirectProvider() # Test connection print("\n1. Testing connection...") if provider.test_connection(): print(" ✓ Connection successful!") else: print(" ✗ Connection failed!") exit(1) # Test image generation print("\n2. Testing image generation...") result = provider.generate_image("a cute golden retriever puppy playing in the snow") print(f" Conversation ID: {result['conversation_id']}") print(f" Response: {result['response_text'][:200]}...") print(f" Images found: {len(result['image_urls'])}") for i, url in enumerate(result['image_urls']): print(f" Image {i+1}: {url[:80]}...") print("\n✓ All tests passed!") except Exception as e: print(f"\n✗ Error: {e}") import traceback traceback.print_exc() print("=" * 60)