# [DEF:backend/src/api/routes/llm.py:Module] # @TIER: STANDARD # @SEMANTICS: api, routes, llm # @PURPOSE: API routes for LLM provider configuration and management. # @LAYER: UI (API) from fastapi import APIRouter, Depends, HTTPException, status from typing import List, Optional from ...core.logger import logger from ...schemas.auth import User from ...dependencies import get_current_user as get_current_active_user from ...plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType from ...services.llm_provider import LLMProviderService from ...core.database import get_db from sqlalchemy.orm import Session # [DEF:router:Global] # @PURPOSE: APIRouter instance for LLM routes. router = APIRouter(tags=["LLM"]) # [/DEF:router:Global] # [DEF:_is_valid_runtime_api_key:Function] # @PURPOSE: Validate decrypted runtime API key presence/shape. # @PRE: value can be None. # @POST: Returns True only for non-placeholder key. def _is_valid_runtime_api_key(value: Optional[str]) -> bool: key = (value or "").strip() if not key: return False if key in {"********", "EMPTY_OR_NONE"}: return False return len(key) >= 16 # [/DEF:_is_valid_runtime_api_key:Function] # [DEF:get_providers:Function] # @PURPOSE: Retrieve all LLM provider configurations. # @PRE: User is authenticated. # @POST: Returns list of LLMProviderConfig. @router.get("/providers", response_model=List[LLMProviderConfig]) async def get_providers( current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): """ Get all LLM provider configurations. """ logger.info(f"[llm_routes][get_providers][Action] Fetching providers for user: {current_user.username}") service = LLMProviderService(db) providers = service.get_all_providers() return [ LLMProviderConfig( id=p.id, provider_type=LLMProviderType(p.provider_type), name=p.name, base_url=p.base_url, api_key="********", default_model=p.default_model, is_active=p.is_active ) for p in providers ] # [/DEF:get_providers:Function] # [DEF:get_llm_status:Function] # @PURPOSE: Returns whether LLM runtime is configured for dashboard validation. # @PRE: User is authenticated. # @POST: configured=true only when an active provider with valid decrypted key exists. @router.get("/status") async def get_llm_status( current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): service = LLMProviderService(db) providers = service.get_all_providers() active_provider = next((p for p in providers if p.is_active), None) if not active_provider: return {"configured": False, "reason": "no_active_provider"} api_key = service.get_decrypted_api_key(active_provider.id) if not _is_valid_runtime_api_key(api_key): return {"configured": False, "reason": "invalid_api_key"} return { "configured": True, "reason": "ok", "provider_id": active_provider.id, "provider_name": active_provider.name, "provider_type": active_provider.provider_type, "default_model": active_provider.default_model, } # [/DEF:get_llm_status:Function] # [DEF:create_provider:Function] # @PURPOSE: Create a new LLM provider configuration. # @PRE: User is authenticated and has admin permissions. # @POST: Returns the created LLMProviderConfig. @router.post("/providers", response_model=LLMProviderConfig, status_code=status.HTTP_201_CREATED) async def create_provider( config: LLMProviderConfig, current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): """ Create a new LLM provider configuration. """ service = LLMProviderService(db) provider = service.create_provider(config) return LLMProviderConfig( id=provider.id, provider_type=LLMProviderType(provider.provider_type), name=provider.name, base_url=provider.base_url, api_key="********", default_model=provider.default_model, is_active=provider.is_active ) # [/DEF:create_provider:Function] # [DEF:update_provider:Function] # @PURPOSE: Update an existing LLM provider configuration. # @PRE: User is authenticated and has admin permissions. # @POST: Returns the updated LLMProviderConfig. @router.put("/providers/{provider_id}", response_model=LLMProviderConfig) async def update_provider( provider_id: str, config: LLMProviderConfig, current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): """ Update an existing LLM provider configuration. """ service = LLMProviderService(db) provider = service.update_provider(provider_id, config) if not provider: raise HTTPException(status_code=404, detail="Provider not found") return LLMProviderConfig( id=provider.id, provider_type=LLMProviderType(provider.provider_type), name=provider.name, base_url=provider.base_url, api_key="********", default_model=provider.default_model, is_active=provider.is_active ) # [/DEF:update_provider:Function] # [DEF:delete_provider:Function] # @PURPOSE: Delete an LLM provider configuration. # @PRE: User is authenticated and has admin permissions. # @POST: Returns success status. @router.delete("/providers/{provider_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_provider( provider_id: str, current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): """ Delete an LLM provider configuration. """ service = LLMProviderService(db) if not service.delete_provider(provider_id): raise HTTPException(status_code=404, detail="Provider not found") return # [/DEF:delete_provider:Function] # [DEF:test_connection:Function] # @PURPOSE: Test connection to an LLM provider. # @PRE: User is authenticated. # @POST: Returns success status and message. @router.post("/providers/{provider_id}/test") async def test_connection( provider_id: str, current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db) ): logger.info(f"[llm_routes][test_connection][Action] Testing connection for provider_id: {provider_id}") """ Test connection to an LLM provider. """ from ...plugins.llm_analysis.service import LLMClient service = LLMProviderService(db) db_provider = service.get_provider(provider_id) if not db_provider: raise HTTPException(status_code=404, detail="Provider not found") api_key = service.get_decrypted_api_key(provider_id) # Check if API key was successfully decrypted if not api_key: logger.error(f"[llm_routes][test_connection] Failed to decrypt API key for provider {provider_id}") raise HTTPException( status_code=500, detail="Failed to decrypt API key. The provider may have been encrypted with a different encryption key. Please update the provider with a new API key." ) client = LLMClient( provider_type=LLMProviderType(db_provider.provider_type), api_key=api_key, base_url=db_provider.base_url, default_model=db_provider.default_model ) try: # Simple test call await client.client.models.list() return {"success": True, "message": "Connection successful"} except Exception as e: return {"success": False, "error": str(e)} # [/DEF:test_connection:Function] # [DEF:test_provider_config:Function] # @PURPOSE: Test connection with a provided configuration (not yet saved). # @PRE: User is authenticated. # @POST: Returns success status and message. @router.post("/providers/test") async def test_provider_config( config: LLMProviderConfig, current_user: User = Depends(get_current_active_user) ): """ Test connection with a provided configuration. """ from ...plugins.llm_analysis.service import LLMClient logger.info(f"[llm_routes][test_provider_config][Action] Testing config for {config.name}") # Check if API key is provided if not config.api_key or config.api_key == "********": raise HTTPException( status_code=400, detail="API key is required for testing connection" ) client = LLMClient( provider_type=config.provider_type, api_key=config.api_key, base_url=config.base_url, default_model=config.default_model ) try: # Simple test call await client.client.models.list() return {"success": True, "message": "Connection successful"} except Exception as e: return {"success": False, "error": str(e)} # [/DEF:test_provider_config:Function] # [/DEF:backend/src/api/routes/llm.py]