""" API client for Clover - A terminal assistant for AI-powered project management Handles communication with the OpenAI-compatible LLM server """ import json import os import time from typing import Any, Dict, Optional import requests from config.settings import load_config class APIClient: """ API client for communicating with the LLM server """ def __init__(self): """Initialize the API client with configuration""" self.config = load_config() self.base_url = self.config.get( "base_url", "error-set-CLOVER_BASE_URL-in-.env-file" ) self.api_key = self.config.get("api_key") self.ssl_verify = self.config.get("ssl_verify", True) # Use requests.Session for connection pooling and consistent SSL settings self.session = requests.Session() self.session.headers.update( { "Content-Type": "application/json", "User-Agent": "clover-cli/1.0", "Accept": "application/json", } ) if self.api_key: self.session.headers["Authorization"] = f"Bearer {self.api_key}" # Warn if using HTTP (non-TLS) connection if self.base_url.startswith("http://") and not self.base_url.startswith( "http://localhost" ): import warnings warnings.warn( f"Using non-TLS connection to {self.base_url}. " "Consider using HTTPS for remote endpoints. " "Set CLOVER_SSL_VERIFY=false to disable SSL verification for self-signed certs.", UserWarning, ) def _make_request( self, endpoint: str, method: str = "GET", data: Optional[Dict] = None ) -> Dict[str, Any]: """ Make a request to the LLM API server with retry and exponential backoff. Retries up to 3 times with exponential backoff (2s base delay) for transient failures (5xx errors, connection errors, timeouts). Uses requests.Session with explicit SSL verification control. Args: endpoint (str): API endpoint method (str): HTTP method (GET, POST) data (dict): Request data Returns: dict: Response from the API """ max_retries = self.config.get("max_retries", 3) base_delay = self.config.get("retry_base_delay", 2) last_exception = None for attempt in range(max_retries + 1): try: url = self.base_url.rstrip("/") + "/" + endpoint.lstrip("/") kwargs = { "timeout": self.config.get("timeout", 300), "verify": self.ssl_verify, } if method == "GET": response = self.session.get(url, **kwargs) elif method == "POST": response = self.session.post(url, json=data, **kwargs) else: raise ValueError(f"Unsupported HTTP method: {method}") # Retry on 5xx server errors if response.status_code >= 500 and attempt < max_retries: delay = base_delay * (2**attempt) print( f"Server error {response.status_code}, " f"retrying in {delay}s (attempt {attempt + 1}/{max_retries})" ) time.sleep(delay) continue response.raise_for_status() return {"success": True, "data": response.json()} except requests.exceptions.Timeout as e: last_exception = e if attempt < max_retries: delay = base_delay * (2**attempt) print( f"Timeout, retrying in {delay}s " f"(attempt {attempt + 1}/{max_retries})" ) time.sleep(delay) continue except requests.exceptions.ConnectionError as e: last_exception = e if attempt < max_retries: delay = base_delay * (2**attempt) print( f"Connection error, retrying in {delay}s " f"(attempt {attempt + 1}/{max_retries})" ) time.sleep(delay) continue except requests.exceptions.RequestException as e: return {"success": False, "error": f"HTTP Request failed: {str(e)}"} except json.JSONDecodeError as e: return {"success": False, "error": f"Invalid JSON response: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} # All retries exhausted if last_exception: return { "success": False, "error": f"Request failed after {max_retries} retries: {str(last_exception)}", } return {"success": False, "error": "Request failed after maximum retries"} def list_models(self) -> Dict[str, Any]: """ Get list of available models from the LLM server Returns: dict: Available models information """ # Use Ollama standard endpoint for model listing result = self._make_request("/api/tags", "GET") if not result["success"]: return {"error": result["error"], "models": []} try: data = result["data"] # Handle Ollama format correctly - the response has a "models" key with array if isinstance(data, dict) and "models" in data: models_list = data["models"] else: # If it's already just a list of models models_list = data if isinstance(data, list) else [] return {"models": models_list} except Exception as e: return {"error": f"Failed to parse models response: {str(e)}", "models": []} def chat_completion( self, messages: list, model: str = None, **kwargs ) -> Dict[str, Any]: """ Get a completion from the LLM using Ollama chat endpoint. Uses /api/chat with proper message list format to preserve multi-turn conversation semantics (system, user, assistant roles). Args: messages (list): List of message dictionaries with 'role' and 'content' model (str): Model to use **kwargs: Additional parameters for the API Returns: dict: Response from the LLM in OpenAI-compatible format """ if model is None: # Reload config to get latest model setting from config.settings import load_config current_config = load_config() model = current_config.get("model", "qwen3-coder:30b") # Filter messages to only include valid roles for chat endpoint chat_messages = [] for message in messages: role = message.get("role", "user") content = message.get("content", "") if role in ("system", "user", "assistant"): chat_messages.append({"role": role, "content": content}) if not chat_messages: return {"error": "No valid messages provided for chat completion"} # Use Ollama chat endpoint with proper message format data = { "model": model, "messages": chat_messages, "stream": False, **kwargs, } result = self._make_request("/api/chat", "POST", data) if not result["success"]: return {"error": result["error"]} # Extract the response from Ollama's chat format try: response_data = result["data"] # Ollama chat endpoint returns message in message.content if "message" in response_data: return { "choices": [ { "message": { "role": response_data["message"].get( "role", "assistant" ), "content": response_data["message"].get( "content", "" ), } } ] } elif "response" in response_data: return { "choices": [{"message": {"content": response_data["response"]}}] } else: return response_data except Exception as e: return {"error": f"Failed to process chat completion result: {str(e)}"} def generate_text(self, prompt: str, model: str = None, **kwargs) -> Dict[str, Any]: """ Generate text using the LLM Args: prompt (str): Prompt to send to the LLM model (str): Model to use **kwargs: Additional parameters for the API Returns: dict: Generated response """ if model is None: model = self.config.get("model", "qwen3-coder:30b") # Format as Ollama generate request data = {"model": model, "prompt": prompt, "stream": False, **kwargs} result = self._make_request("/api/generate", "POST", data) if not result["success"]: return {"error": result["error"]} try: response_data = result["data"] # Extract the actual response text for generate endpoint if "response" in response_data: return { "choices": [{"message": {"content": response_data["response"]}}] } else: # If we got back something else, return it as-is return response_data except Exception as e: return {"error": f"Failed to process generate result: {str(e)}"} def get_model_info(self, model_name: str) -> Dict[str, Any]: """ Get information about a specific model Args: model_name (str): Name of the model Returns: dict: Model information """ # Use Ollama's show endpoint result = self._make_request(f"/api/show/{model_name}", "POST") if not result["success"]: return {"error": result["error"]} return result["data"]