diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index 567e48055..c5839b8f2 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -128,10 +128,10 @@ class MessagesState(TypedDict): def llm_call(state: dict, config: RunnableConfig | None = None): """LLM decides whether to call a tool or not""" - # Extract user authentication info from LangGraph config + # Extract config from LangGraph config configurable = config.get("configurable", {}) if config else {} - user_id = configurable.get("user_id") jwt_token = configurable.get("jwt_token") + llm_config = configurable.get("llm_config") # Defensive check: skip LLM call if no user messages messages = state.get("messages", []) @@ -209,12 +209,9 @@ def llm_call(state: dict, config: RunnableConfig | None = None): # print(full_messages) # Create fresh model with tools for each LLM call - # This ensures configuration changes take effect immediately - # Pass user_id and jwt_token for per-user LLM config and API authentication model_with_tools = create_base_model_with_tools( tools, - user_id=user_id, - jwt_token=jwt_token + llm_config=llm_config ) # Store jwt_token in state for Tools to use when calling GNS3 API @@ -233,10 +230,9 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - This node is only executed when no title has been set yet (first round only). """ - # Extract user authentication info from LangGraph config + # Extract config from LangGraph config configurable = config.get("configurable", {}) if config else {} - user_id = configurable.get("user_id") - jwt_token = configurable.get("jwt_token") + llm_config = configurable.get("llm_config") # Only generate a title if it hasn't been set yet current_title = state.get("conversation_title") @@ -253,9 +249,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - # Call the title generation model (create fresh instance for each call) try: - # Create fresh title model instance from current env configuration - # Pass user_id and jwt_token for per-user LLM config - title_model = create_title_model(user_id=user_id, jwt_token=jwt_token) + title_model = create_title_model(llm_config=llm_config) response = title_model.invoke( title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}} ) diff --git a/gns3server/agent/gns3_copilot/agent/model_factory.py b/gns3server/agent/gns3_copilot/agent/model_factory.py index 800238e3f..030800fd0 100644 --- a/gns3server/agent/gns3_copilot/agent/model_factory.py +++ b/gns3server/agent/gns3_copilot/agent/model_factory.py @@ -2,98 +2,58 @@ Model Factory for GNS3-Copilot Agent This module provides factory functions to create fresh LLM model instances. -Configuration is loaded from the llm_model_configs system via connector_factory. +Configuration is passed directly from the API layer. """ import logging from typing import Any, Optional -from uuid import UUID from langchain.chat_models import init_chat_model -from gns3server.agent.gns3_copilot.gns3_client import get_llm_config - logger = logging.getLogger(__name__) def _load_llm_config( - user_id: Optional[UUID] = None, - jwt_token: Optional[str] = None, llm_config: Optional[dict[str, Any]] = None, ) -> dict[str, str]: """ - Load model configuration from llm_config dict or fetch from llm_model_configs system. - - Priority order: - 1. Provided llm_config dictionary (highest priority) - 2. Fetch from llm_model_configs system via connector_factory (requires user_id and jwt_token) + Convert llm_config dict to model factory format. Args: - user_id: User UUID for fetching config from database - jwt_token: JWT token for API authentication - llm_config: Optional configuration dictionary to use directly + llm_config: Configuration dictionary from database Returns: Dictionary containing model configuration. Raises: - ValueError: If no configuration can be found. + ValueError: If configuration is missing or invalid. """ - # Priority 1: Use provided llm_config dictionary - if llm_config: - logger.info("Using provided llm_config dictionary") - return { - "model_name": llm_config.get("model", ""), - "model_provider": llm_config.get("provider", ""), - "api_key": llm_config.get("api_key", ""), - "base_url": llm_config.get("base_url", ""), - "temperature": str(llm_config.get("temperature", "0")), - } - - # Priority 2: Fetch from llm_model_configs system via connector_factory - if user_id and jwt_token: - logger.info(f"Fetching LLM config from database for user {user_id}") - config = get_llm_config(user_id=user_id, jwt_token=jwt_token) - - if config: - logger.info( - f"Successfully loaded LLM config from database: " - f"provider={config.get('provider')}, model={config.get('model')}" - ) - return { - "model_name": config.get("model", ""), - "model_provider": config.get("provider", ""), - "api_key": config.get("api_key", ""), - "base_url": config.get("base_url", ""), - "temperature": str(config.get("temperature", "0")), - } - - # No configuration found - error_msg = "LLM configuration not found" if not llm_config: - if not user_id or not jwt_token: - error_msg += ": user_id and jwt_token are required for fetching LLM configuration" - else: - error_msg += f": no LLM configuration found for user {user_id}" - raise ValueError(error_msg) + raise ValueError("LLM configuration is required") + + logger.info( + "Using LLM config: provider=%s, model=%s", + llm_config.get("provider"), + llm_config.get("model") + ) + + return { + "model_name": llm_config.get("model", ""), + "model_provider": llm_config.get("provider", ""), + "api_key": llm_config.get("api_key", ""), + "base_url": llm_config.get("base_url", ""), + "temperature": str(llm_config.get("temperature", "0")), + } def create_base_model( - user_id: Optional[UUID] = None, - jwt_token: Optional[str] = None, llm_config: Optional[dict[str, Any]] = None, ) -> Any: """ Create a fresh base LLM model instance. - Configuration priority: - 1. llm_config dictionary (if provided) - 2. Fetch from llm_model_configs system via connector_factory (requires user_id and jwt_token) - Args: - user_id: User UUID for fetching config from database - jwt_token: JWT token for API authentication - llm_config: Optional configuration dictionary to use directly + llm_config: Configuration dictionary from database Returns: Any: A new LLM model instance configured with current settings. @@ -103,7 +63,7 @@ def create_base_model( ValueError: If required configuration fields are missing or invalid. RuntimeError: If model creation fails. """ - config_vars = _load_llm_config(user_id, jwt_token, llm_config) + config_vars = _load_llm_config(llm_config) # Log the loaded configuration (mask sensitive data) logger.info( @@ -141,8 +101,6 @@ def create_base_model( def create_title_model( - user_id: Optional[UUID] = None, - jwt_token: Optional[str] = None, llm_config: Optional[dict[str, Any]] = None, ) -> Any: """ @@ -152,14 +110,8 @@ def create_title_model( It uses the same configuration as the base model but with a higher temperature for more creative output. - Configuration priority: - 1. llm_config dictionary (if provided) - 2. Fetch from llm_model_configs system via connector_factory (requires user_id and jwt_token) - Args: - user_id: User UUID for fetching config from database - jwt_token: JWT token for API authentication - llm_config: Optional configuration dictionary to use directly + llm_config: Configuration dictionary from database Returns: Any: A new LLM model instance for title generation. @@ -169,7 +121,7 @@ def create_title_model( ValueError: If required configuration fields are missing or invalid. RuntimeError: If model creation fails. """ - config_vars = _load_llm_config(user_id, jwt_token, llm_config) + config_vars = _load_llm_config(llm_config) logger.info( "Creating title model: name=%s, provider=%s, base_url=%s, temperature=1.0", @@ -232,8 +184,6 @@ def create_model_with_tools( def create_base_model_with_tools( tools: list[Any], - user_id: Optional[UUID] = None, - jwt_token: Optional[str] = None, llm_config: Optional[dict[str, Any]] = None, ) -> Any: """ @@ -242,15 +192,9 @@ def create_base_model_with_tools( This is a convenience function that combines creating the base model and binding tools to it. - Configuration priority: - 1. llm_config dictionary (if provided) - 2. Fetch from llm_model_configs system via connector_factory (requires user_id and jwt_token) - Args: tools: List of tools to bind to the model. - user_id: User UUID for fetching config from database - jwt_token: JWT token for API authentication - llm_config: Optional configuration dictionary to use directly + llm_config: Configuration dictionary from database Returns: Any: A new model instance with tools bound (type varies by provider). @@ -259,5 +203,5 @@ def create_base_model_with_tools( ValueError: If required configuration fields are missing. RuntimeError: If model creation or tool binding fails. """ - base_model = create_base_model(user_id, jwt_token, llm_config) + base_model = create_base_model(llm_config) return create_model_with_tools(base_model, tools) diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index 410a5a00a..4743308f7 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -103,9 +103,9 @@ class AgentService: message: str, session_id: str, project_id: Optional[str] = None, - user_id: Optional[str] = None, jwt_token: Optional[str] = None, - mode: str = "text" + mode: str = "text", + llm_config: Optional[Dict[str, Any]] = None ) -> AsyncGenerator[Dict[str, Any], None]: """ Stream chat responses from the agent. @@ -114,19 +114,19 @@ class AgentService: message: User message session_id: Session/thread ID for conversation continuity project_id: GNS3 project ID (optional, for context) - user_id: User ID for LLM config lookup (optional) jwt_token: JWT token for API authentication (optional) mode: Interaction mode (default: "text") + llm_config: LLM configuration dict (provider, model, api_key, etc.) Yields: Dict containing SSE-compatible response chunks """ - # Build config with user authentication info + # Build config with LLM configuration config = { "configurable": { "thread_id": session_id, - "user_id": user_id, "jwt_token": jwt_token, + "llm_config": llm_config, } } diff --git a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py index cfc0ecff7..70b809c1b 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py @@ -338,16 +338,18 @@ def get_gns3_server_host() -> str: return DEFAULT_GNS3_URL.split("://")[1].split(":")[0] -def get_llm_config(user_id, jwt_token: str) -> Optional[dict]: +def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: """ Get LLM model configuration for a user. - This is a convenience function that retrieves the user's LLM configuration - from the llm_model_configs system. + This function retrieves the user's LLM configuration from the database. + When `app` is provided, it directly accesses the database (bypassing API restrictions). + Otherwise, it falls back to API call (which may mask group config API keys). Args: user_id: User UUID (can be string or UUID object) jwt_token: JWT token for authentication + app: FastAPI application instance (optional, for direct database access) Returns: Dictionary with LLM configuration keys (provider, model, api_key, etc.), @@ -370,55 +372,53 @@ def get_llm_config(user_id, jwt_token: str) -> Optional[dict]: if isinstance(user_id, str): user_id = UUID(user_id) - # Detect GNS3 URL - url = _detect_url_for_api() + # If app is provided, use direct database access (preferred) + if app is not None: + from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config_with_app - # Import the async helper - from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config - - # Check if we're in the main thread with a running event loop - try: - loop = asyncio.get_running_loop() - # We're in an async context with a running loop - # This shouldn't happen since this is a sync function, - # but if it does, we need to run in a separate thread - import concurrent.futures - with concurrent.futures.ThreadPoolExecutor() as executor: - future = executor.submit( - asyncio.run, - get_user_llm_config(user_id, jwt_token, url) - ) - return future.result(timeout=10) - except RuntimeError: - # No running event loop - we're in a sync context (possibly a thread pool) - # This is the expected path for our use case - pass - - # Try to get an existing event loop, or create a new one - try: - loop = asyncio.get_event_loop() - if loop.is_running(): - # Loop is running but not the running loop (edge case) - # Create a new loop in a thread + # Run the async function in sync context + try: + loop = asyncio.get_running_loop() + # We're in an async context with a running loop + # This shouldn't happen since this is a sync function import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit( asyncio.run, - get_user_llm_config(user_id, jwt_token, url) + get_user_llm_config_with_app(user_id, app) ) return future.result(timeout=10) - else: - # Loop exists but not running - use it - return loop.run_until_complete(get_user_llm_config(user_id, jwt_token, url)) - except RuntimeError: - # No event loop exists - create a new one - # This happens when called from a thread pool executor - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + except RuntimeError: + # No running event loop - we're in a sync context + pass + + # Try to get an existing event loop, or create a new one try: - return loop.run_until_complete(get_user_llm_config(user_id, jwt_token, url)) - finally: - loop.close() + loop = asyncio.get_event_loop() + if loop.is_running(): + # Loop is running but not the running loop (edge case) + import concurrent.futures + with concurrent.futures.ThreadPoolExecutor() as executor: + future = executor.submit( + asyncio.run, + get_user_llm_config_with_app(user_id, app) + ) + return future.result(timeout=10) + else: + # Loop exists but not running - use it + return loop.run_until_complete(get_user_llm_config_with_app(user_id, app)) + except RuntimeError: + # No event loop exists - create a new one + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + return loop.run_until_complete(get_user_llm_config_with_app(user_id, app)) + finally: + loop.close() + + # Fallback: No app provided, try API call (will mask group config API keys) + logger.warning("No app provided for get_llm_config, group config API keys may be masked") + return None except Exception as e: logger.error("Failed to get LLM config for user %s: %s", user_id, e) diff --git a/gns3server/agent/gns3_copilot/utils/llm_config_helper.py b/gns3server/agent/gns3_copilot/utils/llm_config_helper.py index 3b5df270a..8db2f07e7 100644 --- a/gns3server/agent/gns3_copilot/utils/llm_config_helper.py +++ b/gns3server/agent/gns3_copilot/utils/llm_config_helper.py @@ -2,19 +2,13 @@ LLM Model Configuration Helper for GNS3 Copilot This module provides utility functions to retrieve LLM model configurations -using a hybrid approach: -1. Get config_id via API (respects priority logic) -2. Get full config via Repository (includes API key) -3. Decrypt API key for LLM service calls - -This allows the copilot module to access API keys that are hidden from API responses -while leveraging the API's default_config selection logic. +with decrypted API keys by directly accessing the database. Usage: - from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config + from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config_with_app # Get user's default LLM config (with API key) - config = get_user_llm_config(user_id, jwt_token, gns3_url) + config = await get_user_llm_config_with_app(user_id, app) if config: provider = config['provider'] api_key = config['api_key'] @@ -25,276 +19,53 @@ import logging from typing import Optional, Dict, Any from uuid import UUID -import httpx +from fastapi import FastAPI logger = logging.getLogger(__name__) -async def get_user_llm_config( +async def get_user_llm_config_with_app( user_id: UUID, - jwt_token: str, - gns3_url: Optional[str] = None, - timeout: float = 5.0 + app: FastAPI ) -> Optional[Dict[str, Any]]: """ - Get user's default LLM model configuration with API key using hybrid approach. + Get user's default LLM model configuration with decrypted API key. - Step 1: Call API to get default_config's config_id (respects priority logic) - Step 2: Query Repository to get full config (includes encrypted API key) - Step 3: Decrypt API key for LLM service usage + This function directly accesses the database through the app reference, + bypassing API security restrictions to get the complete configuration including + decrypted API keys, even for inherited group configurations. Args: user_id: User UUID - jwt_token: JWT token for API authentication - gns3_url: GNS3 server URL (optional, will auto-detect if not provided) - timeout: HTTP request timeout in seconds + app: FastAPI application instance Returns: Configuration dict with provider, api_key, model, etc., or None if not found Example: - config = await get_user_llm_config(user_id, jwt_token) + config = await get_user_llm_config_with_app(user_id, app) if config: print(f"Provider: {config['provider']}") print(f"Model: {config['model']}") print(f"API Key: {config['api_key']}") print(f"Source: {config['source']}") """ + from gns3server.db.tasks import get_user_llm_config_full + try: - # Step 1: Get GNS3 URL if not provided - if gns3_url is None: - gns3_url = _detect_gns3_url() - if not gns3_url: - logger.error("Failed to detect GNS3 server URL") - return None + user_id_str = str(user_id) + config = await get_user_llm_config_full(user_id_str, app) - # Step 2: Call API to get default_config - logger.debug(f"Fetching default config for user {user_id} from API...") - api_response = await _call_llm_configs_api(gns3_url, user_id, jwt_token, timeout) - - if not api_response: - logger.warning(f"No LLM model configurations found for user {user_id}") - return None - - default_config = api_response.get("default_config") - if not default_config: - logger.warning(f"No default LLM model configuration found for user {user_id}") - return None - - config_id = default_config.get("config_id") - source = default_config.get("source") # "user" or "group" - - logger.debug( - f"API returned default_config: config_id={config_id}, source={source}" - ) - - # Step 3: Query Repository for full config - full_config = await _get_full_config_from_db(config_id, source) - if not full_config: - logger.error(f"Failed to retrieve full config from database: config_id={config_id}") - return None - - # Step 4: Decrypt API key - from gns3server.utils.encryption import decrypt, is_encrypted - - config_data = full_config.config.copy() - if "api_key" in config_data and config_data["api_key"]: - try: - if is_encrypted(config_data["api_key"]): - config_data["api_key"] = decrypt(config_data["api_key"]) - logger.debug("Successfully decrypted API key") - except Exception as e: - logger.error(f"Failed to decrypt API key: {e}") - config_data["api_key"] = None + if config: + logger.info( + f"Successfully retrieved LLM config for user {user_id}: " + f"provider={config.get('provider')}, model={config.get('model')}" + ) else: - logger.warning("No API key found in configuration") + logger.warning(f"No LLM configuration found for user {user_id}") - # Step 5: Build simplified config dict - llm_config = { - "config_id": full_config.config_id, - "name": full_config.name, - "model_type": full_config.model_type, - "source": source, - "group_name": default_config.get("group_name"), - "user_id": full_config.user_id, - "group_id": full_config.group_id, - **config_data # provider, api_key, model, temperature, etc. - } - - logger.info( - f"Successfully retrieved LLM config for user {user_id}: " - f"provider={llm_config.get('provider')}, " - f"model={llm_config.get('model')}, " - f"source={llm_config.get('source')}" - ) - - return llm_config + return config except Exception as e: logger.error(f"Failed to retrieve LLM config for user {user_id}: {e}", exc_info=True) return None - - -async def _call_llm_configs_api( - gns3_url: str, - user_id: UUID, - jwt_token: str, - timeout: float -) -> Optional[Dict[str, Any]]: - """ - Call GNS3 API to get user's LLM model configurations. - - Args: - gns3_url: GNS3 server URL - user_id: User UUID - jwt_token: JWT token for authentication - timeout: Request timeout in seconds - - Returns: - API response dict, or None if failed - """ - try: - url = f"{gns3_url}/v3/access/users/{user_id}/llm-model-configs" - headers = { - "Authorization": f"Bearer {jwt_token}", - "Content-Type": "application/json" - } - - logger.debug(f"Calling API: GET {url}") - - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.get(url, headers=headers) - - if response.status_code == 200: - return response.json() - elif response.status_code == 404: - logger.warning(f"LLM configs endpoint returned 404 for user {user_id}") - return None - else: - logger.error( - f"API returned error {response.status_code}: {response.text}" - ) - return None - - except httpx.TimeoutException: - logger.error(f"Timeout while calling LLM configs API") - return None - except Exception as e: - logger.error(f"Failed to call LLM configs API: {e}", exc_info=True) - return None - - -async def _get_full_config_from_db(config_id: UUID, source: str) -> Optional[Any]: - """ - Get full configuration from database by config_id. - - Args: - config_id: Configuration UUID - source: Configuration source ("user" or "group") - - Returns: - Full config object from database, or None if not found - """ - try: - from gns3server.db.repositories.llm_model_configs import LLMModelConfigsRepository - from gns3server.db import get_session - - with get_session() as db_session: - repo = LLMModelConfigsRepository(db_session) - - if source == "user": - logger.debug(f"Fetching user config from DB: config_id={config_id}") - config = await repo.get_user_config(config_id) - else: - logger.debug(f"Fetching group config from DB: config_id={config_id}") - config = await repo.get_group_config(config_id) - - if config: - logger.debug(f"Successfully retrieved config from DB: config_id={config_id}") - else: - logger.warning(f"Config not found in DB: config_id={config_id}, source={source}") - - return config - - except ImportError as e: - logger.error(f"Failed to import required modules: {e}") - return None - except Exception as e: - logger.error(f"Failed to retrieve config from DB: {e}", exc_info=True) - return None - - -def _detect_gns3_url() -> Optional[str]: - """ - Auto-detect GNS3 server URL using existing connector_factory logic. - - Returns: - GNS3 server URL, or None if detection failed - """ - try: - from gns3server.agent.gns3_copilot.gns3_client.connector_factory import ( - _get_url_from_controller, - _get_url_from_config, - DEFAULT_GNS3_URL - ) - - # Try Controller first - url = _get_url_from_controller() - if url: - logger.debug(f"Auto-detected GNS3 URL from Controller: {url}") - return url - - # Try Config - url = _get_url_from_config() - if url: - logger.debug(f"Auto-detected GNS3 URL from Config: {url}") - return url - - # Fallback - logger.warning(f"Using fallback GNS3 URL: {DEFAULT_GNS3_URL}") - return DEFAULT_GNS3_URL - - except ImportError as e: - logger.error(f"Failed to import connector_factory: {e}") - return None - except Exception as e: - logger.error(f"Failed to auto-detect GNS3 URL: {e}", exc_info=True) - return None - - -# Synchronous wrapper for backward compatibility -def get_user_llm_config_sync(user_id: UUID, jwt_token: str, gns3_url: Optional[str] = None) -> Optional[Dict[str, Any]]: - """ - Synchronous wrapper for get_user_llm_config. - - This is a convenience function for code that cannot use async/await. - It runs the async function in a new event loop. - - Args: - user_id: User UUID - jwt_token: JWT token for API authentication - gns3_url: GNS3 server URL (optional) - - Returns: - Configuration dict, or None if not found - """ - import asyncio - - try: - # Try to get running event loop - loop = asyncio.get_event_loop() - if loop.is_running(): - # If loop is already running, we need to run in a separate thread - import concurrent.futures - with concurrent.futures.ThreadPoolExecutor() as executor: - future = executor.submit( - asyncio.run, - get_user_llm_config(user_id, jwt_token, gns3_url) - ) - return future.result(timeout=10) - else: - # No loop running, use run() directly - return asyncio.run(get_user_llm_config(user_id, jwt_token, gns3_url)) - except Exception as e: - logger.error(f"Failed to run async get_user_llm_config: {e}", exc_info=True) - return None diff --git a/gns3server/api/routes/controller/chat.py b/gns3server/api/routes/controller/chat.py index 95d130af4..56a329f0e 100644 --- a/gns3server/api/routes/controller/chat.py +++ b/gns3server/api/routes/controller/chat.py @@ -91,6 +91,18 @@ async def stream_chat( auth_header = http_request.headers.get("Authorization", "") jwt_token = auth_header.replace("Bearer ", "") if auth_header else None + # Get FastAPI app reference (for database access) + app = http_request.app + + # Get user's LLM config (with decrypted API key) + from gns3server.db.tasks import get_user_llm_config_full + llm_config = await get_user_llm_config_full(user_id, app) + if not llm_config: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="LLM configuration not found. Please configure your LLM settings first." + ) + # Get or create AgentService for this project agent_manager = await get_project_agent_manager() agent_service = await agent_manager.get_agent(str(project.id), project.path) @@ -105,9 +117,9 @@ async def stream_chat( message=request.message, session_id=session_id, project_id=str(project.id), - user_id=user_id, jwt_token=jwt_token, - mode=request.mode + mode=request.mode, + llm_config=llm_config ): try: # Validate and serialize chunk diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index b2458a219..887ff9bf1 100644 --- a/gns3server/db/tasks.py +++ b/gns3server/db/tasks.py @@ -21,7 +21,7 @@ import os from fastapi import FastAPI from pydantic import ValidationError -from typing import List +from typing import List, Optional from sqlalchemy import event from sqlalchemy.engine import Engine from sqlalchemy.exc import SQLAlchemyError @@ -309,3 +309,99 @@ async def monitor_images_on_filesystem(app: FastAPI): log.info(f"Discovered image '{image_path}' has been added to the database") except SQLAlchemyError as e: log.warning(f"Error while adding image '{image_path}' to the database: {e}") + + +async def get_user_llm_config_full(user_id: str, app: FastAPI) -> Optional[dict]: + """ + Get user's full LLM configuration with decrypted API key for Copilot. + + This is a system-level function that bypasses API security restrictions. + It retrieves the complete configuration including decrypted API keys, + even for inherited group configurations. + + Args: + user_id: User UUID + app: FastAPI application instance + + Returns: + Dictionary with LLM configuration (provider, model, api_key, etc.) + or None if not found. + """ + from uuid import UUID + from gns3server.db.repositories.llm_model_configs import LLMModelConfigsRepository + from gns3server.utils.encryption import decrypt, is_encrypted + + try: + user_uuid = UUID(user_id) if isinstance(user_id, str) else user_id + + async with AsyncSession(app.state._db_engine, expire_on_commit=False) as session: + repo = LLMModelConfigsRepository(session) + + # Get effective configs (own + inherited from groups) + result = await repo.get_user_effective_configs( + user_uuid, + current_user_id=user_uuid, # Viewing own config + current_user_is_superadmin=False + ) + + if not result or not result.get("default_config"): + log.warning(f"No default LLM configuration found for user {user_id}") + return None + + default_config = result["default_config"] + config_id = default_config["config_id"] + source = default_config["source"] # "user" or "group" + + # Get full config from database + if source == "user": + full_config = await repo.get_user_config(config_id) + else: + full_config = await repo.get_group_config(config_id) + + if not full_config: + log.error(f"Failed to retrieve full config from database: config_id={config_id}") + return None + + # Decrypt API key + config_data = full_config.config.copy() + if "api_key" in config_data and config_data["api_key"]: + try: + if is_encrypted(config_data["api_key"]): + config_data["api_key"] = decrypt(config_data["api_key"]) + log.debug(f"Successfully decrypted API key for user {user_id}") + except Exception as e: + log.error(f"Failed to decrypt API key: {e}") + config_data["api_key"] = None + + # Build configuration dict + llm_config = { + "config_id": str(full_config.config_id), + "name": full_config.name, + "model_type": str(full_config.model_type), + "source": source, + "group_name": default_config.get("group_name"), + "user_id": str(full_config.user_id) if full_config.user_id else None, + "group_id": str(full_config.group_id) if full_config.group_id else None, + **config_data # provider, api_key, model, temperature, etc. + } + + # Validate required fields + if not llm_config.get("provider"): + log.error(f"LLM config missing 'provider' field: {config_id}") + return None + + if not llm_config.get("model"): + log.error(f"LLM config missing 'model' field: {config_id}") + return None + + log.info( + f"Retrieved LLM config for user {user_id}: " + f"provider={llm_config.get('provider')}, model={llm_config.get('model')}, source={source}" + ) + + return llm_config + + except Exception as e: + log.error(f"Failed to retrieve LLM config for user {user_id}: {e}", exc_info=True) + return None +