""" 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", "http://192.168.8.223:11434") self.api_key = self.config.get("api_key") def _make_request( self, endpoint: str, method: str = "GET", data: Optional[Dict] = None ) -> Dict[str, Any]: """ Make a request to the LLM API server 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}" headers = { "Content-Type": "application/json", "User-Agent": "clover-cli/1.0", "Accept": "application/json", } # Add API key if available if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" if method == "GET": response = requests.get( url, headers=headers, timeout=self.config.get("timeout", 300) ) elif method == "POST": response = requests.post( url, headers=headers, json=data, timeout=self.config.get("timeout", 300), ) 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. 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"]