Python Integration

import os
import time
import requests
from typing import Optional, Dict, Any
from dotenv import load_dotenv

load_dotenv()


class LessTokenClient:
    """
    Production-ready LessToken API client with retry logic,
    error handling, and rate limit management.
    """
    
    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: str = 'https://lesstokens.hive-hub.ai',
        max_retries: int = 3,
        retry_delay: int = 1000
    ):
        """
        Initialize LessToken client
        
        Args:
            api_key: API key (defaults to LESSTOKEN_API_KEY env var)
            base_url: API base URL
            max_retries: Maximum retry attempts
            retry_delay: Initial retry delay in milliseconds
        """
        self.api_key = api_key or os.getenv('LESSTOKEN_API_KEY')
        if not self.api_key:
            raise ValueError('API key required')
        
        self.base_url = base_url
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': self.api_key,
            'Content-Type': 'application/json'
        })
    
    def compress(
        self,
        prompt: str,
        max_time_ms: int = 10000
    ) -> Dict[str, Any]:
        """
        Compress text with automatic retry logic
        
        Args:
            prompt: Text to compress
            max_time_ms: Maximum processing time (milliseconds)
            
        Returns:
            dict: Compression response
            
        Raises:
            requests.HTTPError: If all retries fail
        """
        data = {
            'prompt': prompt,
            'options': {
                'maxTimeMs': max_time_ms
            }
        }
        
        return self._retry_request('POST', '/compress', json=data)
    
    def _retry_request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Make HTTP request with exponential backoff retry logic
        
        Args:
            method: HTTP method
            endpoint: API endpoint
            **kwargs: Additional request parameters
            
        Returns:
            dict: API response
            
        Raises:
            requests.HTTPError: If all retries fail
        """
        url = f'{self.base_url}{endpoint}'
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.request(
                    method,
                    url,
                    timeout=30,
                    **kwargs
                )
                response.raise_for_status()
                return response.json()
                
            except requests.HTTPError as e:
                last_exception = e
                status = e.response.status_code
                
                # Don't retry client errors (except 429)
                if 400 <= status < 500 and status != 429:
                    raise
                
                # Last attempt - raise error
                if attempt == self.max_retries - 1:
                    raise
                
                # Calculate delay with exponential backoff
                delay = (2 ** attempt) * self.retry_delay / 1000
                
                # Use Retry-After header if available
                if status == 429:
                    retry_after = e.response.headers.get('Retry-After')
                    if retry_after:
                        delay = int(retry_after)
                
                print(f"Retry {attempt + 1}/{self.max_retries} after {delay}s")
                time.sleep(delay)
        
        raise last_exception
    
    def get_rate_limits(self, response: requests.Response) -> Dict[str, int]:
        """
        Extract rate limit information from response headers
        
        Args:
            response: API response
            
        Returns:
            dict: Rate limit info (limit, remaining, reset)
        """
        return {
            'limit': int(response.headers.get('X-RateLimit-Limit', 0)),
            'remaining': int(response.headers.get('X-RateLimit-Remaining', 0)),
            'reset': int(response.headers.get('X-RateLimit-Reset', 0))
        }
    
    def close(self):
        """Close the HTTP session"""
        self.session.close()
    
    def __enter__(self):
        """Context manager entry"""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit"""
        self.close()


# Example usage
if __name__ == '__main__':
    with LessTokenClient() as client:
        prompt = """
        {
          "user": {
            "name": "John Doe",
            "email": "[email protected]",
            "preferences": {
              "theme": "dark",
              "notifications": true
            }
          }
        }
        """
        
        try:
            result = client.compress(
                prompt,
                max_time_ms=5000  # Optional: faster response
            )
            
            print(f"Compressed: {result['compressedTokens']} tokens")
            print(f"Saved: {result['savingsPercentage']:.2f}%")
            
        except requests.HTTPError as e:
            error_data = e.response.json()
            print(f"Error: {error_data['message']}")
            print(f"Code: {error_data['code']}")

import os
import asyncio
import aiohttp
from typing import Dict, Any
from dotenv import load_dotenv

load_dotenv()


class AsyncLessTokenClient:
    """Async LessToken API client using aiohttp"""
    
    def __init__(
        self,
        api_key: str = None,
        base_url: str = 'https://lesstokens.hive-hub.ai'
    ):
        self.api_key = api_key or os.getenv('LESSTOKEN_API_KEY')
        if not self.api_key:
            raise ValueError('API key required')
        
        self.base_url = base_url
        self.headers = {
            'X-API-Key': self.api_key,
            'Content-Type': 'application/json'
        }
    
    async def compress(
        self,
        prompt: str,
        **options
    ) -> Dict[str, Any]:
        """Compress text asynchronously"""
        async with aiohttp.ClientSession(headers=self.headers) as session:
            url = f'{self.base_url}/compress'
            data = {
                'prompt': prompt,
                'options': options
            }
            
            async with session.post(url, json=data, timeout=30) as response:
                response.raise_for_status()
                return await response.json()
    
    async def compress_batch(
        self,
        prompts: list[str],
        **options
    ) -> list[Dict[str, Any]]:
        """Compress multiple prompts concurrently"""
        tasks = [self.compress(prompt, **options) for prompt in prompts]
        return await asyncio.gather(*tasks)


# Example usage
async def main():
    client = AsyncLessTokenClient()
    
    prompts = [
        "First prompt to compress",
        "Second prompt to compress",
        "Third prompt to compress"
    ]
    
    results = await client.compress_batch(prompts, maxTimeMs=5000)
    
    for i, result in enumerate(results):
        print(f"Prompt {i+1}: {result['savingsPercentage']:.2f}% saved")


if __name__ == '__main__':
    asyncio.run(main())