""" API client for Clover - A terminal assistant for AI-powered project management Handles communication with the OpenAI-compatible LLM server """ import json import os 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 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 """ try: # Ensure base_url doesn't have trailing slash and endpoint has leading slash base = self.base_url.rstrip("/") endpoint = endpoint if endpoint.startswith("/") else f"/{endpoint}" url = f"{base}{endpoint}" 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}") response.raise_for_status() return {"success": True, "data": response.json()} 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)}"} 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 Args: messages (list): List of message dictionaries (roles and content) model (str): Model to use **kwargs: Additional parameters for the API Returns: dict: Response from the LLM """ 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") # Convert messages to prompt format expected by Ollama generate endpoint prompt_text = "" for message in messages: role = message.get("role", "user") content = message.get("content", "") # Format messages properly for the model if role == "system": prompt_text += f"System: {content}\n\n" elif role == "assistant": prompt_text += f"Assistant: {content}\n\n" else: # user prompt_text += f"User: {content}\n\n" # Add instruction for assistant response prompt_text += "Assistant:" # Prepare the data for Ollama generate endpoint data = {"model": model, "prompt": prompt_text, "stream": False, **kwargs} result = self._make_request("/api/generate", "POST", data) if not result["success"]: return {"error": result["error"]} # Extract the response from Ollama's generate format try: response_data = result["data"] # In Ollama generate responses, the actual text is in the "response" field if "response" in response_data: return { "choices": [{"message": {"content": response_data["response"]}}] } else: # If we get a different format, return what we found 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"]