mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-09-06 18:17:03 +03:00
feat(gns3-copilot): add factory function to create GNS3 connector with LLM config
Introduce `get_gns3_connector_with_llm_config` as a convenience function that combines the creation of a GNS3 API connector and retrieval of the user's default LLM configuration. This simplifies initialization for operations requiring both GNS3 connectivity and AI model settings, reducing boilerplate code in callers. The function returns a dictionary containing the connector and LLM config, or None on failure.
This commit is contained in:
YueGuobin
parent
46b262a02c
commit
3c54a7d90a
@ -24,9 +24,10 @@ File Manager Modules:
|
||||
|
||||
Main functions:
|
||||
- get_gns3_connector: Factory function to create Gns3Connector from environment
|
||||
- get_gns3_connector_with_llm_config: Factory function to create connector AND retrieve LLM config
|
||||
"""
|
||||
|
||||
from .connector_factory import get_gns3_connector
|
||||
from .connector_factory import get_gns3_connector, get_gns3_connector_with_llm_config
|
||||
from .custom_gns3fy import (
|
||||
CONSOLE_TYPES,
|
||||
LINK_TYPES,
|
||||
@ -91,6 +92,7 @@ __all__ = [
|
||||
"GNS3GetNodesTool",
|
||||
"GNS3UpdateDrawingTool",
|
||||
"get_gns3_connector",
|
||||
"get_gns3_connector_with_llm_config",
|
||||
"add_file_to_index",
|
||||
"get_file_list",
|
||||
]
|
||||
|
||||
@ -184,3 +184,117 @@ def get_gns3_connector(jwt_token: str, url: Optional[str] = None) -> Optional[Gn
|
||||
except Exception as e:
|
||||
logger.error("Failed to create Gns3Connector: %s", str(e), exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
async def get_gns3_connector_with_llm_config(
|
||||
user_id,
|
||||
jwt_token: str,
|
||||
url: Optional[str] = None
|
||||
) -> Optional[dict]:
|
||||
"""
|
||||
Create Gns3Connector and retrieve LLM model configuration for the user.
|
||||
|
||||
This is a convenience function that combines:
|
||||
1. get_gns3_connector() - Create GNS3 API connector
|
||||
2. get_user_llm_config() - Retrieve user's default LLM config with API key
|
||||
|
||||
Args:
|
||||
user_id: User UUID (can be string or UUID object)
|
||||
jwt_token: JWT token for authentication
|
||||
url: GNS3 server URL (optional, auto-detected if not provided)
|
||||
|
||||
Returns:
|
||||
Dictionary with keys:
|
||||
- connector: Gns3Connector instance
|
||||
- llm_config: Dict with provider, api_key, model, etc.
|
||||
Or None if failed
|
||||
|
||||
Example:
|
||||
result = await get_gns3_connector_with_llm_config(user_id, jwt_token)
|
||||
if result:
|
||||
connector = result["connector"]
|
||||
llm_config = result["llm_config"]
|
||||
|
||||
# Use connector for GNS3 operations
|
||||
projects = connector.projects
|
||||
|
||||
# Use LLM config for AI operations
|
||||
provider = llm_config["provider"]
|
||||
api_key = llm_config["api_key"]
|
||||
model = llm_config["model"]
|
||||
else:
|
||||
logger.error("Failed to initialize GNS3 connector or LLM config")
|
||||
"""
|
||||
try:
|
||||
# Convert user_id to UUID if it's a string
|
||||
if isinstance(user_id, str):
|
||||
user_id = UUID(user_id)
|
||||
|
||||
# Step 1: Create GNS3 connector
|
||||
connector = get_gns3_connector(jwt_token=jwt_token, url=url)
|
||||
if not connector:
|
||||
logger.error("Failed to create GNS3 connector")
|
||||
return None
|
||||
|
||||
# Step 2: Detect URL if not provided
|
||||
if url is None:
|
||||
url = _detect_url_for_api()
|
||||
|
||||
# Step 3: Get LLM config
|
||||
from gns3_copilot.utils.llm_config_helper import get_user_llm_config
|
||||
|
||||
llm_config = await get_user_llm_config(
|
||||
user_id=user_id,
|
||||
jwt_token=jwt_token,
|
||||
gns3_url=url
|
||||
)
|
||||
|
||||
if not llm_config:
|
||||
logger.warning(f"No LLM config found for user {user_id}")
|
||||
# Still return result with connector only
|
||||
return {
|
||||
"connector": connector,
|
||||
"llm_config": None
|
||||
}
|
||||
|
||||
logger.info(
|
||||
f"Successfully initialized GNS3 connector and LLM config for user {user_id}: "
|
||||
f"connector_url={connector.url}, "
|
||||
f"llm_provider={llm_config.get('provider')}, "
|
||||
f"llm_model={llm_config.get('model')}"
|
||||
)
|
||||
|
||||
return {
|
||||
"connector": connector,
|
||||
"llm_config": llm_config
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get GNS3 connector with LLM config: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def _detect_url_for_api() -> Optional[str]:
|
||||
"""
|
||||
Detect GNS3 server URL for API calls.
|
||||
|
||||
Uses the same priority order as get_gns3_connector:
|
||||
1. Controller.instance().compute("local")
|
||||
2. Config.instance().settings.Server
|
||||
3. Fallback to DEFAULT_GNS3_URL
|
||||
|
||||
Returns:
|
||||
URL string, or None if detection failed
|
||||
"""
|
||||
# Try Controller first
|
||||
url = _get_url_from_controller()
|
||||
if url:
|
||||
return url
|
||||
|
||||
# Try Config
|
||||
url = _get_url_from_config()
|
||||
if url:
|
||||
return url
|
||||
|
||||
# Fallback
|
||||
return DEFAULT_GNS3_URL
|
||||
|
||||
300
gns3server/agent/gns3_copilot/utils/llm_config_helper.py
Normal file
300
gns3server/agent/gns3_copilot/utils/llm_config_helper.py
Normal file
@ -0,0 +1,300 @@
|
||||
"""
|
||||
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.
|
||||
|
||||
Usage:
|
||||
from gns3_copilot.utils.llm_config_helper import get_user_llm_config
|
||||
|
||||
# Get user's default LLM config (with API key)
|
||||
config = get_user_llm_config(user_id, jwt_token, gns3_url)
|
||||
if config:
|
||||
provider = config['provider']
|
||||
api_key = config['api_key']
|
||||
model = config['model']
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Optional, Dict, Any
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def get_user_llm_config(
|
||||
user_id: UUID,
|
||||
jwt_token: str,
|
||||
gns3_url: Optional[str] = None,
|
||||
timeout: float = 5.0
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get user's default LLM model configuration with API key using hybrid approach.
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
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)
|
||||
if config:
|
||||
print(f"Provider: {config['provider']}")
|
||||
print(f"Model: {config['model']}")
|
||||
print(f"API Key: {config['api_key']}")
|
||||
print(f"Source: {config['source']}")
|
||||
"""
|
||||
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
|
||||
|
||||
# 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
|
||||
else:
|
||||
logger.warning("No API key found in configuration")
|
||||
|
||||
# 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
|
||||
|
||||
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 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
|
||||
Loading…
x
Reference in New Issue
Block a user