From 2485bfec46b4687a69bf3066e3f137ea24b6ad45 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 5 Mar 2026 20:58:14 +0800 Subject: [PATCH] docs: add temperature parameter documentation and code formatting improvements - Added `temperature` parameter to Chat API documentation with implementation notes - Improved code formatting in context_manager.py with consistent string quotes and line breaks - Added section on future runtime LLM parameter override capabilities - Updated API schemas to include temperature parameter (currently unused but reserved for future implementation) --- docs/gns3-copilot/ai-chat-api-design.md | 27 ++ .../gns3_copilot/agent/context_manager.py | 105 ++++---- .../agent/gns3_copilot/agent/gns3_copilot.py | 89 ++++--- .../agent/gns3_copilot/agent/model_factory.py | 14 +- .../agent/gns3_copilot/agent_service.py | 74 +++--- .../gns3_copilot/chat_sessions_repository.py | 62 ++--- .../deprecated/context_manager.py | 107 ++++---- .../gns3_copilot/gns3_client/__init__.py | 38 ++- .../gns3_client/connector_factory.py | 59 ++--- .../gns3_client/context_helpers.py | 8 +- .../gns3_copilot/gns3_client/custom_gns3fy.py | 230 +++++------------- .../gns3_client/gns3_project_info.py | 22 +- .../gns3_client/gns3_topology_reader.py | 34 ++- .../gns3_copilot/project_agent_manager.py | 11 +- .../gns3_copilot/prompts/prompt_loader.py | 1 + .../agent/gns3_copilot/tools_v2/__init__.py | 5 +- .../tools_v2/config_tools_nornir.py | 91 ++----- .../tools_v2/display_tools_nornir.py | 86 ++----- .../gns3_copilot/tools_v2/gns3_create_link.py | 35 +-- .../gns3_copilot/tools_v2/gns3_create_node.py | 13 +- .../tools_v2/gns3_get_node_temp.py | 10 +- .../gns3_copilot/tools_v2/gns3_start_node.py | 59 ++--- .../tools_v2/gns3_update_node_name.py | 21 +- .../tools_v2/vpcs_tools_telnetlib3.py | 37 +-- .../agent/gns3_copilot/utils/__init__.py | 8 +- .../utils/get_gns3_device_port.py | 1 + .../gns3_copilot/utils/gns3_drawing_utils.py | 76 ++---- .../gns3_copilot/utils/llm_config_helper.py | 10 +- .../gns3_copilot/utils/message_converters.py | 117 ++++----- .../gns3_copilot/utils/parse_tool_content.py | 89 +++---- gns3server/api/routes/controller/chat.py | 6 + gns3server/schemas/controller/chat.py | 7 +- 32 files changed, 634 insertions(+), 918 deletions(-) diff --git a/docs/gns3-copilot/ai-chat-api-design.md b/docs/gns3-copilot/ai-chat-api-design.md index ea8151c6e..8a5d33a7a 100644 --- a/docs/gns3-copilot/ai-chat-api-design.md +++ b/docs/gns3-copilot/ai-chat-api-design.md @@ -301,6 +301,7 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 - message: 用户消息内容 - session_id: 会话 ID（可选，不提供则自动创建新会话） - stream: 是否启用流式响应（默认 true） +- temperature: LLM temperature 参数（注意：当前未使用，保留以备将来实现。实际 temperature 从用户的数据库 LLM 配置中读取） - mode: 交互模式（当前仅支持 "text"） **响应**：SSE 流，包含多种类型的消息（见上文消息格式） @@ -366,6 +367,7 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 - message: str - 用户消息内容 - session_id: Optional[str] - 会话 ID（可选） - stream: bool - 是否流式响应（默认 true） +- temperature: Optional[float] - LLM temperature 参数（注意：当前未使用，保留以备将来实现运行时覆盖。当前 temperature 从用户的数据库 LLM 配置中读取） - mode: Literal["text"] - 交互模式 ### ChatSession @@ -635,6 +637,31 @@ agent_manager.remove_agent(project_id) ### 未来可能的扩展 +#### 运行时 LLM 参数覆盖 + +当前 LLM 配置（包括 temperature、max_tokens 等）从用户的数据库配置中读取。将来可以支持在请求时覆盖这些参数： + +**实现方案**： +```python +# 在 chat.py 的 stream_chat 函数中 +if request.temperature is not None: + llm_config["temperature"] = str(request.temperature) +if request.max_tokens is not None: + llm_config["max_tokens"] = str(request.max_tokens) +``` + +**当前状态**： +- `temperature` 参数已添加到 ChatRequest schema，但未实现覆盖逻辑 +- 参数保留在 API 中以保持向后兼容性 +- 代码中已添加 TODO 注释标记实现位置 + +**注意事项**： +- 需要验证参数范围（如 temperature: 0.0-2.0） +- 需要考虑是否记录覆盖值到统计信息 +- 需要在前端 UI 中提供相应的设置选项 + +#### 其他扩展方向 + - 多模态支持（图片、文件） - 语音输入/输出 - 多人协作会话 diff --git a/gns3server/agent/gns3_copilot/agent/context_manager.py b/gns3server/agent/gns3_copilot/agent/context_manager.py index cfbf61564..5f1901a13 100644 --- a/gns3server/agent/gns3_copilot/agent/context_manager.py +++ b/gns3server/agent/gns3_copilot/agent/context_manager.py @@ -44,11 +44,13 @@ Requirements: import json import logging import warnings -from typing import Any, Callable +from typing import Any +from typing import Callable import tiktoken - -from langchain_core.messages import BaseMessage, SystemMessage, trim_messages +from langchain_core.messages import BaseMessage +from langchain_core.messages import SystemMessage +from langchain_core.messages import trim_messages logger = logging.getLogger(__name__) @@ -81,6 +83,7 @@ TOKENS_PER_K = 1000 # Token Counting Functions # ============================================================================ + def count_tokens(text: str) -> int: """ Count tokens in text using tiktoken. @@ -126,11 +129,11 @@ def estimate_tool_tokens(tools: list[Any]) -> int: "function": { "name": tool.name, "description": tool.description, - } + }, } # Add parameters schema if available - if hasattr(tool, 'args_schema') and tool.args_schema: + if hasattr(tool, "args_schema") and tool.args_schema: try: # Try Pydantic v2 method (model_json_schema) tool_schema["function"]["parameters"] = tool.args_schema.model_json_schema() @@ -141,15 +144,15 @@ def estimate_tool_tokens(tools: list[Any]) -> int: except Exception: # Both methods failed, use empty schema logger.debug( - "Failed to get schema for tool %s, using empty parameters", - getattr(tool, 'name', 'unknown') + "Failed to get schema for tool %s, using empty parameters", getattr(tool, "name", "unknown") ) tool_schema["function"]["parameters"] = {} except Exception as e: # model_json_schema() raised an exception logger.debug( "model_json_schema() failed for tool %s: %s, trying v1 fallback", - getattr(tool, 'name', 'unknown'), e + getattr(tool, "name", "unknown"), + e, ) try: tool_schema["function"]["parameters"] = tool.args_schema.schema() @@ -182,7 +185,7 @@ def _count_tokens_for_message(message: BaseMessage) -> int: Estimated token count for the message """ content = "" - if hasattr(message, 'content') and message.content: + if hasattr(message, "content") and message.content: content = str(message.content) return count_tokens(content) @@ -192,6 +195,7 @@ def _count_tokens_for_message(message: BaseMessage) -> int: # Pre-Model Hook Factory # ============================================================================ + def create_pre_model_hook( system_prompt: str, get_topology_func: Callable[[dict], Any] | None = None, @@ -246,6 +250,7 @@ def create_pre_model_hook( - In ReAct loops with multiple LLM calls, topology is re-injected each time - This is acceptable for GNS3-Copilot's usage patterns (low concurrency, short conversations) """ + def pre_model_hook(state: dict) -> dict: """ LangGraph pre_model_hook - called before each LLM invocation. @@ -284,8 +289,7 @@ def create_pre_model_hook( if "context_limit" not in llm_config: logger.error( - "context_limit not found in LLM config. " - "This is a required field. Please configure context_limit." + "context_limit not found in LLM config. " "This is a required field. Please configure context_limit." ) raise ValueError("context_limit is required in LLM config") @@ -313,10 +317,7 @@ def create_pre_model_hook( # Step 2: Inject topology into system prompt messages_with_system = _inject_topology_into_system( - messages=messages, - system_prompt=system_prompt, - state=state, - get_topology_func=get_topology_func + messages=messages, system_prompt=system_prompt, state=state, get_topology_func=get_topology_func ) # Step 3: Calculate token breakdown for logging and validation @@ -324,9 +325,7 @@ def create_pre_model_hook( system_tokens = _count_tokens_for_message(system_message) # Calculate tokens for messages_with_system (including system) - messages_with_system_tokens = sum( - _count_tokens_for_message(m) for m in messages_with_system - ) + messages_with_system_tokens = sum(_count_tokens_for_message(m) for m in messages_with_system) # Calculate available budget model_limit_tokens = context_limit_k * TOKENS_PER_K @@ -345,21 +344,29 @@ def create_pre_model_hook( "This will likely cause LLM call failures. " "Recommendations: 1) Reduce system prompt length, 2) Reduce number of tools, " "3) Use a model with larger context window, or 4) Switch to 'conservative' strategy.", - system_tokens, tool_tokens, max_input_tokens + system_tokens, + tool_tokens, + max_input_tokens, ) elif max_tokens_for_trim < system_tokens * 1.5: # Less than 1.5x system tokens means very little room for history logger.warning( "System prompt (%d tokens) + tools (%d tokens) leave minimal room for conversation history (%d tokens remaining). " "Consider reducing system prompt length or number of tools.", - system_tokens, tool_tokens, max_tokens_for_trim - system_tokens + system_tokens, + tool_tokens, + max_tokens_for_trim - system_tokens, ) # Debug log with clear terminology logger.debug( "Token breakdown: system=%d, all_messages=%d (system+history), tools=%d, trim_budget=%d (limit=%dK, strategy=%s)", - system_tokens, messages_with_system_tokens, tool_tokens, max_tokens_for_trim, - context_limit_k, strategy + system_tokens, + messages_with_system_tokens, + tool_tokens, + max_tokens_for_trim, + context_limit_k, + strategy, ) # Step 4: Trim messages to fit @@ -380,16 +387,25 @@ def create_pre_model_hook( if len(trimmed) < len(messages_with_system): logger.info( "Messages trimmed: %d → %d msgs. Total: ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s", - len(messages_with_system), len(trimmed), - final_total, tool_tokens, final_total + tool_tokens, - context_limit_k, usage_percent, strategy + len(messages_with_system), + len(trimmed), + final_total, + tool_tokens, + final_total + tool_tokens, + context_limit_k, + usage_percent, + strategy, ) else: logger.info( "Context ready: %d msgs, ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s", - len(trimmed), final_total, tool_tokens, - final_total + tool_tokens, context_limit_k, - usage_percent, strategy + len(trimmed), + final_total, + tool_tokens, + final_total + tool_tokens, + context_limit_k, + usage_percent, + strategy, ) return {"messages": trimmed} @@ -433,19 +449,15 @@ def _inject_topology_into_system( if topology_data: topology_str = str(topology_data) - formatted_prompt = system_prompt.replace( - "{{topology_info}}", - f"\n\n## Current Topology\n{topology_str}" - ) - logger.info("✓ Topology injected: %d chars, nodes: %s", - len(topology_str), - list(topology_data.get("nodes", {}).keys())[:5]) # Show first 5 node names + formatted_prompt = system_prompt.replace("{{topology_info}}", f"\n\n## Current Topology\n{topology_str}") + logger.info( + "✓ Topology injected: %d chars, nodes: %s", + len(topology_str), + list(topology_data.get("nodes", {}).keys())[:5], + ) # Show first 5 node names logger.debug("Full topology data: %s", topology_str[:500]) # First 500 chars else: - formatted_prompt = system_prompt.replace( - "{{topology_info}}", - "(No topology information available)" - ) + formatted_prompt = system_prompt.replace("{{topology_info}}", "(No topology information available)") logger.warning("✗ Topology data is None, injecting placeholder") # Filter out existing SystemMessage instances @@ -462,6 +474,7 @@ def _inject_topology_into_system( # Legacy Compatibility (Deprecated) # ============================================================================ + def prepare_context_messages( state_messages: list[Any], system_prompt: str, @@ -482,15 +495,9 @@ def prepare_context_messages( if "{{topology_info}}" not in system_prompt: formatted_prompt = system_prompt elif topology_context: - formatted_prompt = system_prompt.replace( - "{{topology_info}}", - f"\n\n## Current Topology\n{topology_context}" - ) + formatted_prompt = system_prompt.replace("{{topology_info}}", f"\n\n## Current Topology\n{topology_context}") else: - formatted_prompt = system_prompt.replace( - "{{topology_info}}", - "(No topology information available)" - ) + formatted_prompt = system_prompt.replace("{{topology_info}}", "(No topology information available)") return [SystemMessage(content=formatted_prompt)] + state_messages @@ -506,7 +513,7 @@ if __name__ == "__main__": # Test token counting print("Test 1: Token Counting") - print(f" tiktoken encoding: cl100k_base") + print(" tiktoken encoding: cl100k_base") test_text = "Hello world 你好世界" tokens = count_tokens(test_text) @@ -537,7 +544,7 @@ if __name__ == "__main__": print("\nTest 3: Invoke pre_model_hook") test_state = { "messages": [HumanMessage(f"Message {i}: {'x' * 50}") for i in range(5)], - "topology_info": {"project_id": "test123", "nodes": 3} + "topology_info": {"project_id": "test123", "nodes": 3}, } result = hook(test_state) diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index 6d6ae0be6..b0016d61b 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -43,43 +43,40 @@ The agent provides: # Standard library imports import logging import operator -import sys -from pathlib import Path -from typing import Annotated, Literal +from typing import Annotated +from typing import Literal # Third-party imports -from langchain.messages import AnyMessage, SystemMessage, ToolMessage +from langchain.messages import AnyMessage +from langchain.messages import SystemMessage +from langchain.messages import ToolMessage from langchain_core.runnables import RunnableConfig -from langgraph.graph import END, START, StateGraph +from langgraph.graph import END +from langgraph.graph import START +from langgraph.graph import StateGraph from langgraph.managed.is_last_step import RemainingSteps from typing_extensions import TypedDict -# Add backend to path for prompt_manager -sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "backend")) - # Local imports -from gns3server.agent.gns3_copilot.agent.context_manager import ( - create_pre_model_hook, -) +from gns3server.agent.gns3_copilot.agent.context_manager import create_pre_model_hook from gns3server.agent.gns3_copilot.agent.model_factory import ( create_base_model_with_tools, - create_title_model, ) +from gns3server.agent.gns3_copilot.agent.model_factory import create_title_model from gns3server.agent.gns3_copilot.gns3_client import GNS3TopologyTool from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( get_current_llm_config, ) -from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT, load_system_prompt -from gns3server.agent.gns3_copilot.tools_v2 import ( - ExecuteMultipleDeviceConfigCommands, - ExecuteMultipleDeviceCommands, - GNS3CreateNodeTool, - GNS3LinkTool, - GNS3StartNodeTool, - GNS3TemplateTool, - GNS3UpdateNodeNameTool, - VPCSMultiCommands, -) +from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT +from gns3server.agent.gns3_copilot.prompts import load_system_prompt +from gns3server.agent.gns3_copilot.tools_v2 import ExecuteMultipleDeviceCommands +from gns3server.agent.gns3_copilot.tools_v2 import ExecuteMultipleDeviceConfigCommands +from gns3server.agent.gns3_copilot.tools_v2 import GNS3CreateNodeTool +from gns3server.agent.gns3_copilot.tools_v2 import GNS3LinkTool +from gns3server.agent.gns3_copilot.tools_v2 import GNS3StartNodeTool +from gns3server.agent.gns3_copilot.tools_v2 import GNS3TemplateTool +from gns3server.agent.gns3_copilot.tools_v2 import GNS3UpdateNodeNameTool +from gns3server.agent.gns3_copilot.tools_v2 import VPCSMultiCommands # Set up logger for GNS3-Copilot logger = logging.getLogger(__name__) @@ -110,6 +107,7 @@ DEFAULT_CONVERSATION_TITLE = "New Conversation" UNTITLED_SESSION_FALLBACK = "Untitled Session" TITLE_MAX_LENGTH = 40 + # Define state class MessagesState(TypedDict): """ @@ -138,6 +136,7 @@ class MessagesState(TypedDict): # Store GNS3 topology information topology_info: dict | None + # Define llm call node def llm_call(state: dict, config: RunnableConfig | None = None): """ @@ -161,8 +160,9 @@ def llm_call(state: dict, config: RunnableConfig | None = None): "topology_info": None, } - logger.debug("LLM config retrieved from context: provider=%s, model=%s", - llm_config.get("provider"), llm_config.get("model")) + logger.debug( + "LLM config retrieved from context: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model") + ) # Defensive check: skip LLM call if no user messages messages = state.get("messages", []) @@ -189,13 +189,13 @@ def llm_call(state: dict, config: RunnableConfig | None = None): if topology and "error" not in topology: topology_info = topology logger.info( - "Successfully retrieved topology for project_id: %s, name: %s", - project_id, topology.get("name") + "Successfully retrieved topology for project_id: %s, name: %s", project_id, topology.get("name") ) else: logger.warning( "Failed to retrieve topology for project_id %s: %s", - project_id, topology.get("error", "Unknown error") if topology else "No result" + project_id, + topology.get("error", "Unknown error") if topology else "No result", ) except Exception as e: logger.warning("Error retrieving topology for project_id %s: %s", project_id, e) @@ -214,12 +214,10 @@ def llm_call(state: dict, config: RunnableConfig | None = None): ) # Create fresh model with tools for each LLM call - logger.debug("Creating model with tools: provider=%s, model=%s", - llm_config.get("provider"), llm_config.get("model")) - model_with_tools = create_base_model_with_tools( - tools, - llm_config=llm_config + logger.debug( + "Creating model with tools: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model") ) + model_with_tools = create_base_model_with_tools(tools, llm_config=llm_config) # Call pre_hook directly to prepare messages (topology injection + trimming) # Note: LangGraph's pre_model_hook only works with prebuilt agents, not custom StateGraph @@ -231,8 +229,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None): # Invoke model with prepared messages response = model_with_tools.invoke(prepared_messages) - logger.info("LLM call completed: tool_calls=%d", - len(response.tool_calls) if hasattr(response, 'tool_calls') else 0) + logger.info("LLM call completed: tool_calls=%d", len(response.tool_calls) if hasattr(response, "tool_calls") else 0) return { "messages": [response], @@ -240,6 +237,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None): "topology_info": topology_info, } + # Define generate title node def generate_title(state: MessagesState, config: RunnableConfig | None = None) -> dict: """ @@ -270,9 +268,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - # Call the title generation model (create fresh instance for each call) try: title_model = create_title_model(llm_config=llm_config) - response = title_model.invoke( - title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}} - ) + response = title_model.invoke(title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}}) raw_content = response.content new_title = raw_content.strip() @@ -286,7 +282,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - # Safety: truncate long titles and avoid line breaks if len(new_title) > TITLE_MAX_LENGTH: - new_title = new_title[:TITLE_MAX_LENGTH - 2] + "..." + new_title = new_title[: TITLE_MAX_LENGTH - 2] + "..." # Remove unwanted characters new_title = new_title.replace("\n", " ").replace('"', "").replace("'", "") @@ -300,7 +296,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - # Improved fallback: Use user's first message content if messages and len(messages) > 0: first_message = messages[0] - if hasattr(first_message, 'content'): + if hasattr(first_message, "content"): fallback_title = first_message.content[:30].strip() # Remove newlines and extra spaces fallback_title = fallback_title.replace("\n", " ").strip() @@ -323,6 +319,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - # Title already exists → no update needed return {} + # Define tool node def tool_node(state: dict, config: RunnableConfig | None = None): """Performs the tool call""" @@ -337,18 +334,14 @@ def tool_node(state: dict, config: RunnableConfig | None = None): tool = tools_by_name[tool_name] try: observation = tool.invoke(tool_call["args"]) - logger.debug("Tool %s completed: output_length=%d", - tool_name, len(str(observation)) if observation else 0) + logger.debug("Tool %s completed: output_length=%d", tool_name, len(str(observation)) if observation else 0) except Exception as e: logger.error("Tool %s failed: %s", tool_name, e, exc_info=True) observation = f"Error: {str(e)}" - result.append(ToolMessage( - content=observation, - tool_call_id=tool_call["id"], - name=tool_call["name"] - )) + result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"], name=tool_call["name"])) return {"messages": result} + # Routing logic after the LLM node def should_continue( state: MessagesState, @@ -374,6 +367,7 @@ def should_continue( # Normal completion (multi-turn conversation or title already exists) return END + # Routing logic after the tool node, Check remaining_steps def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]: """ @@ -400,6 +394,7 @@ def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]: return END + # Build and compile the agent # Build workflow agent_builder = StateGraph(MessagesState) diff --git a/gns3server/agent/gns3_copilot/agent/model_factory.py b/gns3server/agent/gns3_copilot/agent/model_factory.py index 3b7460775..4b906db6f 100644 --- a/gns3server/agent/gns3_copilot/agent/model_factory.py +++ b/gns3server/agent/gns3_copilot/agent/model_factory.py @@ -32,12 +32,14 @@ Configuration is passed directly from the database. """ import logging -from typing import Any, Optional +from typing import Any +from typing import Optional from langchain.chat_models import init_chat_model logger = logging.getLogger(__name__) + def _load_llm_config( llm_config: Optional[dict[str, Any]] = None, ) -> dict[str, str]: @@ -56,11 +58,7 @@ def _load_llm_config( if not llm_config: raise ValueError("LLM configuration is required") - logger.info( - "Using LLM config: provider=%s, model=%s", - llm_config.get("provider"), - llm_config.get("model") - ) + logger.info("Using LLM config: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model")) return { "model_name": llm_config.get("model", ""), @@ -70,6 +68,7 @@ def _load_llm_config( "temperature": str(llm_config.get("temperature", "0")), } + def create_base_model( llm_config: Optional[dict[str, Any]] = None, ) -> Any: @@ -123,6 +122,7 @@ def create_base_model( logger.error("Failed to create base model: %s", e) raise RuntimeError(f"Failed to create base model: {e}") from e + def create_title_model( llm_config: Optional[dict[str, Any]] = None, ) -> Any: @@ -178,6 +178,7 @@ def create_title_model( logger.error("Failed to create title model: %s", e) raise RuntimeError(f"Failed to create title model: {e}") from e + def create_model_with_tools( model: Any, tools: list[Any], @@ -203,6 +204,7 @@ def create_model_with_tools( logger.error("Failed to bind tools to model: %s", e) raise RuntimeError(f"Failed to bind tools to model: {e}") from e + def create_base_model_with_tools( tools: list[Any], llm_config: Optional[dict[str, Any]] = None, diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index 911b2c537..d6416fe29 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -32,27 +32,37 @@ in the project directory. """ import asyncio -import json import logging import os from datetime import datetime -from typing import AsyncGenerator, Dict, Any, Optional, List +from typing import Any +from typing import AsyncGenerator +from typing import Dict +from typing import List +from typing import Optional from uuid import uuid4 import aiosqlite -from langchain_core.messages import HumanMessage, AIMessage, ToolMessage +from langchain_core.messages import HumanMessage from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver from gns3server.agent.gns3_copilot.agent.gns3_copilot import agent_builder -from gns3server.agent.gns3_copilot.chat_sessions_repository import ChatSessionsRepository +from gns3server.agent.gns3_copilot.chat_sessions_repository import ( + ChatSessionsRepository, +) from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( set_current_jwt_token, +) +from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( set_current_llm_config, ) -from gns3server.agent.gns3_copilot.utils.message_converters import convert_langchain_to_openai +from gns3server.agent.gns3_copilot.utils.message_converters import ( + convert_langchain_to_openai, +) log = logging.getLogger(__name__) + class AgentService: """ Project-level Agent Service with async checkpoint management. @@ -176,7 +186,9 @@ class AgentService: await conn.commit() # Create pinned index (after column is guaranteed to exist) - await conn.execute("CREATE INDEX IF NOT EXISTS idx_pinned_updated ON chat_sessions(pinned DESC, updated_at DESC)") + await conn.execute( + "CREATE INDEX IF NOT EXISTS idx_pinned_updated ON chat_sessions(pinned DESC, updated_at DESC)" + ) await conn.commit() log.debug("chat_sessions table created in checkpoint database") @@ -197,7 +209,7 @@ class AgentService: user_id: Optional[str] = None, jwt_token: Optional[str] = None, mode: str = "text", - llm_config: Optional[Dict[str, Any]] = None + llm_config: Optional[Dict[str, Any]] = None, ) -> AsyncGenerator[Dict[str, Any], None]: """ Stream chat responses from the agent. @@ -234,10 +246,7 @@ class AgentService: if is_new_session: # Create new session session = await repo.create_session( - thread_id=session_id, - user_id=user_id or "", - project_id=project_id or "", - title="New Conversation" + thread_id=session_id, user_id=user_id or "", project_id=project_id or "", title="New Conversation" ) log.debug("Created new chat session: thread_id=%s", session_id) @@ -247,8 +256,9 @@ class AgentService: log.debug("JWT token set in context") if llm_config: set_current_llm_config(llm_config) - log.debug("LLM config set in context: provider=%s, model=%s", - llm_config.get("provider"), llm_config.get("model")) + log.debug( + "LLM config set in context: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model") + ) # Build config - only thread-safe identifiers config = { @@ -258,7 +268,7 @@ class AgentService: }, "metadata": { "user_id": user_id, - } + }, } # Build inputs @@ -353,10 +363,17 @@ class AgentService: input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, - last_message_at=last_message_at + last_message_at=last_message_at, + ) + log.info( + "Session statistics updated: thread_id=%s, messages=%d, llm_calls=%d, tokens=%d+%d=%d", + session_id, + message_count, + llm_calls_count, + input_tokens, + output_tokens, + input_tokens + output_tokens, ) - log.info("Session statistics updated: thread_id=%s, messages=%d, llm_calls=%d, tokens=%d+%d=%d", - session_id, message_count, llm_calls_count, input_tokens, output_tokens, input_tokens + output_tokens) # Sync auto-generated title from checkpoint state final_state = await graph.aget_state(config) @@ -365,8 +382,7 @@ class AgentService: current_session = await repo.get_session_by_thread(session_id) if current_session and current_session.title != generated_title: await repo.update_session(thread_id=session_id, title=generated_title) - log.info("Auto-generated title synced: thread_id=%s, title=%s", - session_id, generated_title) + log.info("Auto-generated title synced: thread_id=%s, title=%s", session_id, generated_title) except Exception as e: log.error("Error in stream_chat: %s", e, exc_info=True) @@ -396,11 +412,7 @@ class AgentService: elif event_type == "on_tool_start": # Tool execution started - return { - "type": "tool_start", - "tool_name": event.get("name", ""), - "session_id": session_id - } + return {"type": "tool_start", "tool_name": event.get("name", ""), "session_id": session_id} elif event_type == "on_tool_end": # Tool execution completed @@ -412,7 +424,7 @@ class AgentService: "type": "tool_end", "tool_name": event.get("name", ""), "tool_output": output, - "session_id": session_id + "session_id": session_id, } return None @@ -441,19 +453,11 @@ class AgentService: title = state.values.get("conversation_title", "New Conversation") - return { - "thread_id": session_id, - "title": title, - "messages": messages - } + return {"thread_id": session_id, "title": title, "messages": messages} except Exception as e: log.error("Error getting history: %s", e, exc_info=True) - return { - "thread_id": session_id, - "title": "New Conversation", - "messages": [] - } + return {"thread_id": session_id, "title": "New Conversation", "messages": []} def _convert_message_to_dict(self, msg) -> Dict[str, Any]: """Convert a LangChain message to OpenAI-compatible dict format.""" diff --git a/gns3server/agent/gns3_copilot/chat_sessions_repository.py b/gns3server/agent/gns3_copilot/chat_sessions_repository.py index 5fca1f0a7..e5908ddcd 100644 --- a/gns3server/agent/gns3_copilot/chat_sessions_repository.py +++ b/gns3server/agent/gns3_copilot/chat_sessions_repository.py @@ -33,13 +33,16 @@ checkpoint database. import json import logging from datetime import datetime -from typing import Any, Dict, List, Optional -from uuid import UUID +from typing import Any +from typing import Dict +from typing import List +from typing import Optional import aiosqlite log = logging.getLogger(__name__) + class ChatSession: """Chat session model.""" @@ -60,7 +63,7 @@ class ChatSession: updated_at: Optional[str] = None, metadata: str = "{}", stats: str = "{}", - pinned: bool = False + pinned: bool = False, ): self.id = id self.thread_id = thread_id @@ -100,6 +103,7 @@ class ChatSession: "pinned": self.pinned, } + class ChatSessionsRepository: """ Repository for managing chat sessions in the checkpoint database. @@ -115,11 +119,7 @@ class ChatSessionsRepository: self.conn = conn async def create_session( - self, - thread_id: str, - user_id: str, - project_id: str, - title: str = "New Conversation" + self, thread_id: str, user_id: str, project_id: str, title: str = "New Conversation" ) -> ChatSession: """ Create a new chat session. @@ -141,7 +141,7 @@ class ChatSessionsRepository: created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?) """, - (thread_id, user_id, project_id, title, now, now) + (thread_id, user_id, project_id, title, now, now), ) await self.conn.commit() @@ -160,10 +160,7 @@ class ChatSessionsRepository: Returns: ChatSession or None """ - cursor = await self.conn.execute( - "SELECT * FROM chat_sessions WHERE id = ?", - (session_id,) - ) + cursor = await self.conn.execute("SELECT * FROM chat_sessions WHERE id = ?", (session_id,)) row = await cursor.fetchone() if row: @@ -180,10 +177,7 @@ class ChatSessionsRepository: Returns: ChatSession or None """ - cursor = await self.conn.execute( - "SELECT * FROM chat_sessions WHERE thread_id = ?", - (thread_id,) - ) + cursor = await self.conn.execute("SELECT * FROM chat_sessions WHERE thread_id = ?", (thread_id,)) row = await cursor.fetchone() if row: @@ -191,10 +185,7 @@ class ChatSessionsRepository: return None async def list_sessions( - self, - user_id: Optional[str] = None, - project_id: Optional[str] = None, - limit: int = 100 + self, user_id: Optional[str] = None, project_id: Optional[str] = None, limit: int = 100 ) -> List[ChatSession]: """ List sessions with optional filters. @@ -239,7 +230,7 @@ class ChatSessionsRepository: input_tokens: Optional[int] = None, output_tokens: Optional[int] = None, total_tokens: Optional[int] = None, - last_message_at: Optional[str] = None + last_message_at: Optional[str] = None, ) -> Optional[ChatSession]: """ Update a session. @@ -316,16 +307,10 @@ class ChatSessionsRepository: True if deleted, False if not found """ # First, delete the checkpoint data - await self.conn.execute( - "DELETE FROM checkpoints WHERE thread_id = ?", - (thread_id,) - ) + await self.conn.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,)) # Then delete the session - cursor = await self.conn.execute( - "DELETE FROM chat_sessions WHERE thread_id = ?", - (thread_id,) - ) + cursor = await self.conn.execute("DELETE FROM chat_sessions WHERE thread_id = ?", (thread_id,)) await self.conn.commit() deleted = cursor.rowcount > 0 @@ -345,24 +330,15 @@ class ChatSessionsRepository: Number of sessions deleted """ # Get all thread_ids for this project - cursor = await self.conn.execute( - "SELECT thread_id FROM chat_sessions WHERE project_id = ?", - (project_id,) - ) + cursor = await self.conn.execute("SELECT thread_id FROM chat_sessions WHERE project_id = ?", (project_id,)) rows = await cursor.fetchall() thread_ids = [row[0] for row in rows] # Delete checkpoints and sessions for thread_id in thread_ids: - await self.conn.execute( - "DELETE FROM checkpoints WHERE thread_id = ?", - (thread_id,) - ) + await self.conn.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,)) - cursor = await self.conn.execute( - "DELETE FROM chat_sessions WHERE project_id = ?", - (project_id,) - ) + cursor = await self.conn.execute("DELETE FROM chat_sessions WHERE project_id = ?", (project_id,)) await self.conn.commit() deleted_count = cursor.rowcount @@ -385,7 +361,7 @@ class ChatSessionsRepository: now = datetime.utcnow().isoformat() await self.conn.execute( "UPDATE chat_sessions SET pinned = ?, updated_at = ? WHERE thread_id = ?", - (1 if pinned else 0, now, thread_id) + (1 if pinned else 0, now, thread_id), ) await self.conn.commit() diff --git a/gns3server/agent/gns3_copilot/deprecated/context_manager.py b/gns3server/agent/gns3_copilot/deprecated/context_manager.py index a284ff184..70c2dcb13 100644 --- a/gns3server/agent/gns3_copilot/deprecated/context_manager.py +++ b/gns3server/agent/gns3_copilot/deprecated/context_manager.py @@ -39,14 +39,13 @@ including: import json import logging -from typing import Any, Literal +from typing import Any +from typing import Literal -from langchain_core.messages import ( - AIMessage, - HumanMessage, - SystemMessage, - ToolMessage, -) +from langchain_core.messages import AIMessage +from langchain_core.messages import HumanMessage +from langchain_core.messages import SystemMessage +from langchain_core.messages import ToolMessage logger = logging.getLogger(__name__) @@ -57,8 +56,8 @@ logger = logging.getLogger(__name__) # Context strategy ratios CONTEXT_STRATEGY_RATIOS = { "conservative": 0.60, # 60% for input, 40% reserved for output - "balanced": 0.75, # 75% for input, 25% reserved for output - "aggressive": 0.85, # 85% for input, 15% reserved for output + "balanced": 0.75, # 75% for input, 25% reserved for output + "aggressive": 0.85, # 85% for input, 15% reserved for output } DEFAULT_CONTEXT_STRATEGY = "balanced" @@ -70,6 +69,7 @@ DEFAULT_CONTEXT_STRATEGY = "balanced" # Global tiktoken encoding cache (lazy loading) _tiktoken_encoding = None + def _get_tiktoken_encoding(): """ Get tiktoken encoding instance (cached). @@ -87,6 +87,7 @@ def _get_tiktoken_encoding(): if _tiktoken_encoding is None: try: import tiktoken + _tiktoken_encoding = tiktoken.get_encoding("cl100k_base") logger.debug("Using tiktoken (cl100k_base) for accurate token counting") except ImportError: @@ -96,6 +97,7 @@ def _get_tiktoken_encoding(): ) return _tiktoken_encoding + def count_tokens_accurately(text: str) -> int: """ Count tokens in text accurately using tiktoken. @@ -119,6 +121,7 @@ def count_tokens_accurately(text: str) -> int: logger.error("tiktoken encoding failed: %s", e) raise + def count_messages_tokens(messages: list[Any]) -> int: """ Count total tokens in a list of messages accurately. @@ -134,16 +137,18 @@ def count_messages_tokens(messages: list[Any]) -> int: """ total = 0 for msg in messages: - if hasattr(msg, 'content') and msg.content: + if hasattr(msg, "content") and msg.content: # Handle both string and complex content content = str(msg.content) total += count_tokens_accurately(content) return total + # ============================================================================ # Tool Definition Token Estimation # ============================================================================ + def estimate_tool_tokens(tools: list[Any]) -> int: """ Estimate the token count of tool definitions. @@ -174,12 +179,12 @@ def estimate_tool_tokens(tools: list[Any]) -> int: "type": "function", "function": { "name": tool.name, - "description": tool.description if hasattr(tool, 'description') else "", - } + "description": tool.description if hasattr(tool, "description") else "", + }, } # Add parameters schema if available - if hasattr(tool, 'args_schema') and tool.args_schema: + if hasattr(tool, "args_schema") and tool.args_schema: try: tool_schema["function"]["parameters"] = tool.args_schema.schema() except Exception: @@ -191,10 +196,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int: tool_tokens = len(encoding.encode(schema_str)) total_tokens += tool_tokens - logger.debug( - "Tool '%s': ~%d tokens (schema size: %d chars)", - tool.name, tool_tokens, len(schema_str) - ) + logger.debug("Tool '%s': ~%d tokens (schema size: %d chars)", tool.name, tool_tokens, len(schema_str)) except Exception as e: logger.error("Failed to estimate tokens for tool '%s': %s", tool.name, e) @@ -203,6 +205,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int: logger.info("Tool definitions estimated at ~%d total tokens (%d tools)", total_tokens, len(tools)) return total_tokens + # ============================================================================ # Model Context Limits # ============================================================================ @@ -221,10 +224,8 @@ def estimate_tool_tokens(tools: list[Any]) -> int: # # Always verify current limits from official provider documentation. -def get_model_context_limit( - model_name: str, - llm_config: dict[str, Any] | None = None -) -> int: + +def get_model_context_limit(model_name: str, llm_config: dict[str, Any] | None = None) -> int: """ Get the context window limit for a given model. @@ -253,7 +254,9 @@ def get_model_context_limit( actual_tokens = db_limit_k * 1000 logger.debug( "Using database config context limit: %dK tokens (%d tokens) for model '%s'", - db_limit_k, actual_tokens, model_name + db_limit_k, + actual_tokens, + model_name, ) return actual_tokens else: @@ -270,9 +273,9 @@ def get_model_context_limit( f"Refer to the model provider's documentation for the current context window size." ) + def calculate_max_tokens( - model_limit: int, - strategy: Literal["conservative", "balanced", "aggressive"] = DEFAULT_CONTEXT_STRATEGY + model_limit: int, strategy: Literal["conservative", "balanced", "aggressive"] = DEFAULT_CONTEXT_STRATEGY ) -> int: """ Calculate the maximum tokens to use, reserving space for output. @@ -290,17 +293,16 @@ def calculate_max_tokens( ratio = CONTEXT_STRATEGY_RATIOS.get(strategy, CONTEXT_STRATEGY_RATIOS[DEFAULT_CONTEXT_STRATEGY]) max_tokens = int(model_limit * ratio) - logger.debug( - "Context limit: model=%d, strategy=%s, usable=%d tokens", - model_limit, strategy, max_tokens - ) + logger.debug("Context limit: model=%d, strategy=%s, usable=%d tokens", model_limit, strategy, max_tokens) return max_tokens + # ============================================================================ # Message Trimming # ============================================================================ + def trim_messages_for_context( messages: list[Any], model_name: str, @@ -353,7 +355,8 @@ def trim_messages_for_context( logger.warning( "Tool definitions (%d tokens) exceed input budget (%d tokens). " "Consider reducing context_limit or using fewer tools.", - tool_tokens, max_tokens + tool_tokens, + max_tokens, ) available_for_messages = 0 @@ -363,13 +366,19 @@ def trim_messages_for_context( if current_tokens <= available_for_messages: logger.debug( "Messages fit in context: %d / %d tokens (available: %d, tools: %d)", - current_tokens, max_tokens, available_for_messages, tool_tokens + current_tokens, + max_tokens, + available_for_messages, + tool_tokens, ) return messages logger.info( "Trimming messages: %d → %d tokens (budget: %d, tools: %d)", - current_tokens, available_for_messages, max_tokens, tool_tokens + current_tokens, + available_for_messages, + max_tokens, + tool_tokens, ) # Manually separate and trim to ensure system messages are preserved @@ -387,7 +396,8 @@ def trim_messages_for_context( # Not enough space for system messages - keep only system messages logger.warning( "System messages (%d tokens) exceed available space (%d tokens), truncating to system only", - system_tokens, available_for_messages + system_tokens, + available_for_messages, ) return system_msgs[:1] if system_msgs else messages[-1:] @@ -399,12 +409,16 @@ def trim_messages_for_context( logger.info( "Trimmed %d → %d messages (system: %d, history: %d → %d)", - len(messages), len(trimmed), - len(system_msgs), len(other_msgs), len(trimmed_other) + len(messages), + len(trimmed), + len(system_msgs), + len(other_msgs), + len(trimmed_other), ) return trimmed + def _trim_to_token_limit(messages: list[Any], max_tokens: int) -> list[Any]: """ Trim messages to fit within token limit using tiktoken. @@ -454,7 +468,7 @@ def _trim_to_token_limit(messages: list[Any], max_tokens: int) -> list[Any]: len(groups) - len(trimmed_groups), len(trimmed), current_tokens, - _count_groups_tokens(trimmed_groups) + _count_groups_tokens(trimmed_groups), ) return trimmed @@ -481,18 +495,18 @@ def _build_message_groups(messages: list[Any]) -> list[list[Any]]: msg = messages[i] # If AIMessage with tool_calls, group it with all following ToolMessages - if isinstance(msg, AIMessage) and hasattr(msg, 'tool_calls') and msg.tool_calls: + if isinstance(msg, AIMessage) and hasattr(msg, "tool_calls") and msg.tool_calls: group = [msg] i += 1 # Collect all following ToolMessages that match these tool_calls - tool_call_ids = {tc['id'] for tc in msg.tool_calls} + tool_call_ids = {tc["id"] for tc in msg.tool_calls} while i < len(messages): next_msg = messages[i] if isinstance(next_msg, ToolMessage): # Check if this ToolMessage belongs to current AIMessage - if hasattr(next_msg, 'tool_call_id') and next_msg.tool_call_id in tool_call_ids: + if hasattr(next_msg, "tool_call_id") and next_msg.tool_call_id in tool_call_ids: group.append(next_msg) i += 1 else: @@ -516,14 +530,16 @@ def _count_groups_tokens(groups: list[list[Any]]) -> int: total = 0 for group in groups: for msg in group: - if hasattr(msg, 'content') and msg.content: + if hasattr(msg, "content") and msg.content: total += count_tokens_accurately(str(msg.content)) return total + # ============================================================================ # Token Usage Summary # ============================================================================ + def get_token_usage_summary( messages: list[Any], model_name: str, @@ -575,10 +591,12 @@ def get_token_usage_summary( "needs_trimming": usage_percentage > 80, } + # ============================================================================ # Main Entry Point - Context Preparation with Template Injection # ============================================================================ + def prepare_context_messages( state_messages: list[Any], system_prompt: str, @@ -685,7 +703,7 @@ def prepare_context_messages( formatted_tokens + trimmed_history_tokens + tool_tokens, model_limit_k, ((formatted_tokens + trimmed_history_tokens) / (model_limit_k * 1000)) * 100, - trim_strategy + trim_strategy, ) else: # No trimming @@ -699,20 +717,19 @@ def prepare_context_messages( formatted_tokens + history_tokens + tool_tokens, model_limit_k, (formatted_tokens / (model_limit_k * 1000)) * 100, - trim_strategy + trim_strategy, ) return trimmed_messages + # ============================================================================ # Module Test # ============================================================================ if __name__ == "__main__": # Simple test - test_messages = [ - HumanMessage(f"Message {i}") for i in range(100) - ] + test_messages = [HumanMessage(f"Message {i}") for i in range(100)] # Test with mock llm_config mock_config = {"context_limit": 8, "context_strategy": "conservative"} @@ -723,7 +740,7 @@ if __name__ == "__main__": topology_context='{"project_id": "test", "nodes": 5}', model_name="gpt-4o", llm_config=mock_config, - tools=None + tools=None, ) print(f"Original: {len(test_messages)} messages") diff --git a/gns3server/agent/gns3_copilot/gns3_client/__init__.py b/gns3server/agent/gns3_copilot/gns3_client/__init__.py index aff51a8ba..a1015ab9e 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/__init__.py +++ b/gns3server/agent/gns3_copilot/gns3_client/__init__.py @@ -47,29 +47,23 @@ Main functions: Upstream gns3fy: https://github.com/davidban77/gns3fy """ -from .connector_factory import ( - get_gns3_connector, - get_gns3_connector_with_llm_config, - get_gns3_server_host, - get_llm_config, -) -from .context_helpers import ( - get_current_jwt_token, - get_current_llm_config, - set_current_jwt_token, - set_current_llm_config, -) -from .custom_gns3fy import ( - CONSOLE_TYPES, - LINK_TYPES, - NODE_TYPES, - Gns3Connector, - Link, - Node, - Project, -) -from .gns3_topology_reader import GNS3TopologyTool +from .connector_factory import get_gns3_connector +from .connector_factory import get_gns3_connector_with_llm_config +from .connector_factory import get_gns3_server_host +from .connector_factory import get_llm_config +from .context_helpers import get_current_jwt_token +from .context_helpers import get_current_llm_config +from .context_helpers import set_current_jwt_token +from .context_helpers import set_current_llm_config +from .custom_gns3fy import CONSOLE_TYPES +from .custom_gns3fy import LINK_TYPES +from .custom_gns3fy import NODE_TYPES +from .custom_gns3fy import Gns3Connector +from .custom_gns3fy import Link +from .custom_gns3fy import Node +from .custom_gns3fy import Project from .gns3_project_info import GNS3ProjectInfoTool +from .gns3_topology_reader import GNS3TopologyTool # Dynamic version management try: diff --git a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py index e18fb89d8..6c127ecbc 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py @@ -41,24 +41,22 @@ Features: import asyncio import concurrent.futures import logging -import threading from typing import Optional from uuid import UUID -# Local imports -from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( get_current_jwt_token, - get_current_llm_config, - set_current_jwt_token, - set_current_llm_config, ) +# Local imports +from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + logger = logging.getLogger(__name__) # Fallback default URL DEFAULT_GNS3_URL = "http://127.0.0.1:3080" + def _get_url_from_controller() -> Optional[str]: """Try to get GNS3 server URL from running Controller instance. @@ -93,6 +91,7 @@ def _get_url_from_controller() -> Optional[str]: logger.warning("Unexpected error getting URL from Controller: %s", str(e)) return None + def _get_url_from_config() -> Optional[str]: """Try to get GNS3 server URL from Config settings. @@ -122,6 +121,7 @@ def _get_url_from_config() -> Optional[str]: logger.warning("Unexpected error getting URL from Config: %s", str(e)) return None + def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = None) -> Optional[Gns3Connector]: """Create and return a Gns3Connector instance with JWT authentication. @@ -184,7 +184,7 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non "Using fallback default URL: %s. " "This may not be correct if your GNS3 server is configured differently. " "Consider providing the URL explicitly or ensuring gns3server is running.", - url + url, ) else: logger.info("Using explicitly provided GNS3 server URL: %s", url) @@ -208,22 +208,20 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non 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]: + +async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optional[str] = None, app=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 + 2. get_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) + app: FastAPI application instance (optional, for direct database access) Returns: Dictionary with keys: @@ -263,21 +261,12 @@ async def get_gns3_connector_with_llm_config( url = _detect_url_for_api() # Step 3: Get LLM config - from gns3server.agent.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 - ) + llm_config = get_llm_config(user_id=user_id, jwt_token=jwt_token, app=app) 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 - } + return {"connector": connector, "llm_config": None} logger.info( f"Successfully initialized GNS3 connector and LLM config for user {user_id}: " @@ -286,15 +275,13 @@ async def get_gns3_connector_with_llm_config( f"llm_model={llm_config.get('model')}" ) - return { - "connector": connector, - "llm_config": llm_config - } + 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. @@ -325,6 +312,7 @@ def _detect_url_for_api() -> Optional[str]: logger.debug("Using fallback URL for API call: %s", DEFAULT_GNS3_URL) return DEFAULT_GNS3_URL + def get_gns3_server_host() -> str: """ Get GNS3 server hostname from Controller or Config. @@ -361,6 +349,7 @@ def get_gns3_server_host() -> str: logger.warning("Failed to extract host from URL %s: %s, using fallback", url, e) return DEFAULT_GNS3_URL.split("://")[1].split(":")[0] + def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: """ Get LLM model configuration for a user. @@ -394,7 +383,9 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: # 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 + from gns3server.agent.gns3_copilot.utils.llm_config_helper import ( + get_user_llm_config_with_app, + ) # Run the async function in sync context try: @@ -402,10 +393,7 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: # We're in an async context with a running loop # This shouldn't happen since this is a sync function with concurrent.futures.ThreadPoolExecutor() as executor: - future = executor.submit( - asyncio.run, - get_user_llm_config_with_app(user_id, app) - ) + future = executor.submit(asyncio.run, get_user_llm_config_with_app(user_id, app)) return future.result(timeout=10) except RuntimeError: # No running event loop - we're in a sync context @@ -417,10 +405,7 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: if loop.is_running(): # Loop is running but not the running loop (edge case) with concurrent.futures.ThreadPoolExecutor() as executor: - future = executor.submit( - asyncio.run, - get_user_llm_config_with_app(user_id, app) - ) + 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 diff --git a/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py b/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py index acae30aaa..33a286a17 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py +++ b/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py @@ -97,8 +97,7 @@ def set_current_llm_config(config: dict) -> None: config: LLM configuration dictionary with provider, model, api_key, etc. """ _llm_config_context.set(config) - logger.debug("LLM config set in context: provider=%s, model=%s", - config.get("provider"), config.get("model")) + logger.debug("LLM config set in context: provider=%s, model=%s", config.get("provider"), config.get("model")) def get_current_llm_config() -> Optional[dict]: @@ -109,8 +108,9 @@ def get_current_llm_config() -> Optional[dict]: """ config = _llm_config_context.get() if config: - logger.debug("LLM config retrieved from context: provider=%s, model=%s", - config.get("provider"), config.get("model")) + logger.debug( + "LLM config retrieved from context: provider=%s, model=%s", config.get("provider"), config.get("model") + ) else: logger.warning("LLM config not found in context") return config diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index 46fb50498..fdf5849bb 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -46,19 +46,20 @@ import time from collections.abc import Callable from dataclasses import field from functools import wraps -from math import cos, pi, sin -from typing import ( - Any, - ParamSpec, - TypeVar, - cast, -) +from math import cos +from math import pi +from math import sin +from typing import Any +from typing import ParamSpec +from typing import TypeVar +from typing import cast from urllib.parse import urlparse import jwt import requests import urllib3 -from pydantic import ConfigDict, field_validator +from pydantic import ConfigDict +from pydantic import field_validator from pydantic.dataclasses import dataclass from requests import HTTPError @@ -98,6 +99,7 @@ CONSOLE_TYPES = [ LINK_TYPES = ["ethernet", "serial"] + class Gns3Connector: """ Connector to be use for interaction against GNS3 server controller API. @@ -149,7 +151,8 @@ class Gns3Connector: if url is None: raise ValueError("URL is required for Gns3Connector") - self.base_url = f"{url.strip('/')}/v{api_version}" + self.url = url.strip('/') # Store original URL for reference + self.base_url = f"{self.url}/v{api_version}" self.user = user self.cred = cred self.headers = {"Content-Type": "application/json"} @@ -174,11 +177,7 @@ class Gns3Connector: self.session.headers["Accept"] = "application/json" # pragma: no cover # Set authentication based on API version - if ( - self.auth_type == "basic" - and self.user is not None - and self.cred is not None - ): + if self.auth_type == "basic" and self.user is not None and self.cred is not None: self.session.auth = (self.user, self.cred) # pragma: no cover elif self.auth_type == "jwt" and self.access_token: @@ -194,9 +193,7 @@ class Gns3Connector: return if not self.user or not self.cred: - raise ValueError( - "Username and password are required for v3 authentication when no JWT token is provided" - ) + raise ValueError("Username and password are required for v3 authentication when no JWT token is provided") # Construct authentication URL (v3 API uses different base URL) auth_url = f"{self.base_url.replace('/v3', '')}/v3/access/users/authenticate" @@ -207,9 +204,7 @@ class Gns3Connector: temp_session.headers["Content-Type"] = "application/json" try: - response = temp_session.post( - auth_url, json=auth_data, verify=self.verify, timeout=10.0 - ) + response = temp_session.post(auth_url, json=auth_data, verify=self.verify, timeout=10.0) if response.status_code == 200: auth_result = response.json() self.access_token = auth_result["access_token"] @@ -217,9 +212,7 @@ class Gns3Connector: self.session.headers["Authorization"] = f"Bearer {self.access_token}" # print(f"Successfully authenticated to v3 API, token obtained") else: - raise HTTPError( - f"v3 API authentication failed: {response.status_code} - {response.text}" - ) + raise HTTPError(f"v3 API authentication failed: {response.status_code} - {response.text}") except Exception as e: raise HTTPError(f"v3 API authentication error: {str(e)}") from e @@ -233,9 +226,7 @@ class Gns3Connector: try: # Decode token without verification to check expiry - decoded: dict[str, Any] = jwt.decode( - token, options={"verify_signature": False} - ) + decoded: dict[str, Any] = jwt.decode(token, options={"verify_signature": False}) exp = decoded.get("exp") if exp is not None: return time.time() > float(exp) @@ -264,12 +255,7 @@ class Gns3Connector: Executes HTTP operations and handles GNS3-specific error logic. """ # Handle JWT authentication - if ( - self.auth_type == "jwt" - and not self.access_token - and self.user - and self.cred - ): + if self.auth_type == "jwt" and not self.access_token and self.user and self.cred: self._authenticate_v3() # Get request function (e.g., session.get, session.post) @@ -337,9 +323,7 @@ class Gns3Connector: response = self.http_call("get", url=f"{self.base_url}/version") return cast(dict[str, Any], response.json()) - def projects_summary( - self, is_print: bool = True - ) -> list[tuple[str, str, int, int, str]] | None: + def projects_summary(self, is_print: bool = True) -> list[tuple[str, str, int, int, str]] | None: """ Returns a summary of the projects in the server. If `is_print` is `False`, it will return a list of tuples like: @@ -349,9 +333,7 @@ class Gns3Connector: _projects_summary = [] for _p in self.get_projects(): # Retrieve the project stats - _stats = self.http_call( - "get", f"{self.base_url}/projects/{_p['project_id']}/stats" - ).json() + _stats = self.http_call("get", f"{self.base_url}/projects/{_p['project_id']}/stats").json() if is_print: print( f"{_p['name']}: {_p['project_id']} -- Nodes: {_stats['nodes']} -- " @@ -376,9 +358,7 @@ class Gns3Connector: response = self.http_call("get", url=f"{self.base_url}/projects").json() return cast(list[dict[str, Any]], response) - def get_project( - self, name: str | None = None, project_id: str | None = None - ) -> dict[str, Any] | None: + def get_project(self, name: str | None = None, project_id: str | None = None) -> dict[str, Any] | None: """ Retrieves a project from either a name or ID @@ -387,9 +367,7 @@ class Gns3Connector: - `name` or `project_id` """ if project_id: - _response = self.http_call( - "get", url=f"{self.base_url}/projects/{project_id}" - ) + _response = self.http_call("get", url=f"{self.base_url}/projects/{project_id}") return cast(dict[str, Any], _response.json()) elif name: try: @@ -400,9 +378,7 @@ class Gns3Connector: else: raise ValueError("Must provide either a name or project_id") - def templates_summary( - self, is_print: bool = True - ) -> list[tuple[str, str, str, bool, str, str]] | None: + def templates_summary(self, is_print: bool = True) -> list[tuple[str, str, str, bool, str, str]] | None: """ Returns a summary of the templates in the server. If `is_print` is `False`, it will return a list of tuples like: @@ -439,9 +415,7 @@ class Gns3Connector: _response_data = self.http_call("get", url=f"{self.base_url}/templates").json() return cast(list[dict[str, Any]], _response_data) - def get_template( - self, name: str | None = None, template_id: str | None = None - ) -> dict[str, Any] | None: + def get_template(self, name: str | None = None, template_id: str | None = None) -> dict[str, Any] | None: """ Retrieves a template from either a name or ID @@ -450,9 +424,7 @@ class Gns3Connector: - `name` or `template_id` """ if template_id: - _response_json = self.http_call( - "get", url=f"{self.base_url}/templates/{template_id}" - ).json() + _response_json = self.http_call("get", url=f"{self.base_url}/templates/{template_id}").json() return cast(dict[str, Any], _response_json) elif name: try: @@ -541,15 +513,11 @@ class Gns3Connector: kwargs["compute_id"] = "local" # Send request - response = self.http_call( - "post", url=f"{self.base_url}/templates", json_data=kwargs - ) + response = self.http_call("post", url=f"{self.base_url}/templates", json_data=kwargs) # Return and convert type return cast(dict[str, Any], response.json()) - def delete_template( - self, name: str | None = None, template_id: str | None = None - ) -> None: + def delete_template(self, name: str | None = None, template_id: str | None = None) -> None: """ Deletes a template by giving its attributes. For more information [API INFO] (http://api.gns3.net/en/2.2/api/v2/controller/template/ @@ -570,9 +538,7 @@ class Gns3Connector: # Final check: ensure template_id has a value at this point if not template_id: - raise ValueError( - "Must provide either a 'name' or 'template_id' to delete a template." - ) + raise ValueError("Must provide either a 'name' or 'template_id' to delete a template.") self.http_call("delete", url=f"{self.base_url}/templates/{template_id}") @@ -584,9 +550,7 @@ class Gns3Connector: - `project_id` """ - _response_data = self.http_call( - "get", url=f"{self.base_url}/projects/{project_id}/nodes" - ).json() + _response_data = self.http_call("get", url=f"{self.base_url}/projects/{project_id}/nodes").json() return cast(list[dict[str, Any]], _response_data) @@ -611,9 +575,7 @@ class Gns3Connector: - `project_id` """ - _response_data = self.http_call( - "get", url=f"{self.base_url}/projects/{project_id}/links" - ).json() + _response_data = self.http_call("get", url=f"{self.base_url}/projects/{project_id}/links").json() return cast(list[dict[str, Any]], _response_data) @@ -688,9 +650,7 @@ class Gns3Connector: return cast(dict[str, Any], _response_data) - def get_compute_images( - self, emulator: str, compute_id: str = "local" - ) -> list[dict[str, Any]]: + def get_compute_images(self, emulator: str, compute_id: str = "local") -> list[dict[str, Any]]: """ Returns a list of images available for a compute. @@ -709,9 +669,7 @@ class Gns3Connector: return cast(list[dict[str, Any]], _response_data) - def upload_compute_image( - self, emulator: str, file_path: str, compute_id: str = "local" - ) -> None: + def upload_compute_image(self, emulator: str, file_path: str, compute_id: str = "local") -> None: """ uploads an image for use by a compute. @@ -748,6 +706,7 @@ class Gns3Connector: return cast(dict[str, Any], _response_data) + def verify_connector_and_id(f: F) -> F: """ Main checker for connector object and respective object's ID for their retrieval @@ -774,13 +733,9 @@ def verify_connector_and_id(f: F) -> F: _url = f"{_conn.base_url}/projects/{_project_id}/nodes" _response = _conn.http_call("get", _url) - extracted = [ - node for node in _response.json() if node["name"] == self.name - ] + extracted = [node for node in _response.json() if node["name"] == self.name] if len(extracted) > 1: # pragma: no cover - raise ValueError( - "Multiple nodes found with same name. Need to submit node_id" - ) + raise ValueError("Multiple nodes found with same name. Need to submit node_id") self.node_id = extracted[0]["node_id"] # Checks for Link if self.__class__.__name__ == "Link": @@ -790,6 +745,7 @@ def verify_connector_and_id(f: F) -> F: return cast(F, wrapper) + @dataclass(config=config) class Link: """ @@ -914,9 +870,7 @@ class Link: if _project_id is None: raise ValueError("Need to submit project_id") if _link_id is None: - raise ValueError( - "Link ID is missing. The link might have already been deleted." - ) + raise ValueError("Link ID is missing. The link might have already been deleted.") _url = f"{_conn.base_url}/projects/{_project_id}/links/{self.link_id}" @@ -942,12 +896,7 @@ class Link: _url = f"{self.connector.base_url}/projects/{self.project_id}/links" - data = { - k: v - for k, v in self.__dict__.items() - if k not in ("connector", "__initialised__") - if v is not None - } + data = {k: v for k, v in self.__dict__.items() if k not in ("connector", "__initialised__") if v is not None} _response = self.connector.http_call("post", _url, json_data=data) @@ -979,9 +928,7 @@ class Link: if not self.project_id: raise ValueError("Need to submit project_id") - _url = ( - f"{self.connector.base_url}/projects/{self.project_id}/links/{self.link_id}" - ) + _url = f"{self.connector.base_url}/projects/{self.project_id}/links/{self.link_id}" # TODO: Verify that the passed kwargs are supported ones _response = self.connector.http_call("put", _url, json_data=kwargs) @@ -989,6 +936,7 @@ class Link: # Update object self._update(_response.json()) + @dataclass(config=config) class Node: """ @@ -1120,9 +1068,7 @@ class Node: if not self.project_id: raise ValueError("Need to submit project_id") - _url = ( - f"{self.connector.base_url}/projects/{self.project_id}/nodes/{self.node_id}" - ) + _url = f"{self.connector.base_url}/projects/{self.project_id}/nodes/{self.node_id}" _response = self.connector.http_call("get", _url) # Update object @@ -1148,10 +1094,7 @@ class Node: if not self.project_id: raise ValueError("Need to submit project_id") - _url = ( - f"{self.connector.base_url}/projects/{self.project_id}/nodes" - f"/{self.node_id}/links" - ) + _url = f"{self.connector.base_url}/projects/{self.project_id}/nodes" f"/{self.node_id}/links" _response = self.connector.http_call("get", _url) # Create the Link array but cleanup cache if there is one @@ -1253,10 +1196,7 @@ class Node: error_detail = _response.json() except Exception: error_detail = _response.text - _msg = ( - f"Failed to stop node: {_response.status_code}, " - f"Detail: {error_detail}" - ) + _msg = f"Failed to stop node: {_response.status_code}, " f"Detail: {error_detail}" raise RuntimeError(_msg) from None @verify_connector_and_id @@ -1305,10 +1245,7 @@ class Node: error_detail = _response.json() except Exception: error_detail = _response.text - _msg = ( - f"Failed to reload node: {_response.status_code}, " - f"Detail: {error_detail}" - ) + _msg = f"Failed to reload node: {_response.status_code}, " f"Detail: {error_detail}" raise RuntimeError(_msg) from None @verify_connector_and_id @@ -1398,9 +1335,7 @@ class Node: _template = self.connector.get_template(name=self.template) if _template is None: raise ValueError(f"Template {self.template} not found") - self.template_id = self.connector.get_template(name=self.template).get( - "template_id" - ) + self.template_id = self.connector.get_template(name=self.template).get("template_id") else: raise ValueError("Need either 'template' of 'template_id'") @@ -1419,14 +1354,9 @@ class Node: if v is not None } - _url = ( - f"{self.connector.base_url}/projects/{self.project_id}/" - f"templates/{self.template_id}" - ) + _url = f"{self.connector.base_url}/projects/{self.project_id}/" f"templates/{self.template_id}" - _response = self.connector.http_call( - "post", _url, json_data={"x": 0, "y": 0, "compute_id": self.compute_id} - ) + _response = self.connector.http_call("post", _url, json_data={"x": 0, "y": 0, "compute_id": self.compute_id}) self._update(_response.json()) @@ -1517,6 +1447,7 @@ class Node: _conn.http_call("post", _url, data=data) + @dataclass(config=config) class Project: """ @@ -1604,9 +1535,7 @@ class Project: if k in self.__dict__: setattr(self, k, v) - def get( - self, get_links: bool = True, get_nodes: bool = True, get_stats: bool = True - ) -> None: + def get(self, get_links: bool = True, get_nodes: bool = True, get_stats: bool = True) -> None: """ Retrieves the projects information. @@ -2045,10 +1974,7 @@ class Project: _nodes_summary = [] for _n in self.nodes: if is_print: - print( - f"{_n.name}: {_n.status} -- Console: {_n.console} -- " - f"ID: {_n.node_id}" - ) + print(f"{_n.name}: {_n.status} -- Console: {_n.console} -- " f"ID: {_n.node_id}") _nodes_summary.append((_n.name, _n.status, _n.console, _n.node_id)) return _nodes_summary if not is_print else None @@ -2084,9 +2010,7 @@ class Project: _nodes_inventory = {} conn = self.connector if not conn: - raise ValueError( - "Gns3Connector not assigned. Please set the connector first." - ) + raise ValueError("Gns3Connector not assigned. Please set the connector first.") _server = urlparse(conn.base_url).hostname @@ -2112,9 +2036,7 @@ class Project: return _nodes_inventory - def links_summary( - self, is_print: bool = True - ) -> list[tuple[str, str, str, str]] | None: + def links_summary(self, is_print: bool = True) -> list[tuple[str, str, str, str]] | None: """ Returns a summary of the links insode the project. If `is_print` is False, it will return a list of tuples like: @@ -2194,9 +2116,7 @@ class Project: except IndexError: return None - def get_node( - self, name: str | None = None, node_id: str | None = None - ) -> Any | None: + def get_node(self, name: str | None = None, node_id: str | None = None) -> Any | None: """ Returns the Node object by searching for the `name` or the `node_id`. @@ -2269,10 +2189,7 @@ class Project: _node.create() self.nodes.append(_node) - print( - f"Created: {_node.name} -- Type: {_node.node_type} -- " - f"Console: {_node.console}" - ) + print(f"Created: {_node.name} -- Type: {_node.node_type} -- " f"Console: {_node.console}") def create_link(self, node_a: str, port_a: str, node_b: str, port_b: str) -> None: """ @@ -2407,9 +2324,7 @@ class Project: ): _matches.append(_l) if not _matches: - raise ValueError( - f"Link not found: {node_a, port_a, node_b, port_b}" - ) # pragma: no cover + raise ValueError(f"Link not found: {node_a, port_a, node_b, port_b}") # pragma: no cover # now to delete the link via GNS3_api _link = _matches[0] @@ -2417,8 +2332,7 @@ class Project: _link_id = _link.link_id _link.delete() print( - f"Deleted Link-ID: {_link_id} From node {node_a}, port: {port_a} <--> " - f"to node {node_b}, port: {port_b}" + f"Deleted Link-ID: {_link_id} From node {node_a}, port: {port_a} <--> " f"to node {node_b}, port: {port_b}" ) @verify_connector_and_id @@ -2451,9 +2365,7 @@ class Project: except StopIteration: return None - def get_snapshot( - self, name: str | None = None, snapshot_id: str | None = None - ) -> dict[str, Any] | None: + def get_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> dict[str, Any] | None: """ Returns the Snapshot by searching for the `name` or the `snapshot_id`. @@ -2510,9 +2422,7 @@ class Project: print(f"Created snapshot: {_snapshot['name']}") @verify_connector_and_id - def delete_snapshot( - self, name: str | None = None, snapshot_id: str | None = None - ) -> None: + def delete_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> None: """ Deletes a snapshot of the project @@ -2536,19 +2446,14 @@ class Project: if not _snapshot: raise ValueError("Snapshot not found") - _url = ( - f"{_conn.base_url}/projects/{_project_id}/snapshots/" - f"{_snapshot['snapshot_id']}" - ) + _url = f"{_conn.base_url}/projects/{_project_id}/snapshots/" f"{_snapshot['snapshot_id']}" _conn.http_call("delete", _url) self.get_snapshots() @verify_connector_and_id - def restore_snapshot( - self, name: str | None = None, snapshot_id: str | None = None - ) -> None: + def restore_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> None: """ Restore a snapshot from disk @@ -2572,10 +2477,7 @@ class Project: if not _snapshot: raise ValueError("Snapshot not found") - _url = ( - f"{_conn.base_url}/projects/{_project_id}/snapshots/" - f"{_snapshot['snapshot_id']}/restore" - ) + _url = f"{_conn.base_url}/projects/{_project_id}/snapshots/" f"{_snapshot['snapshot_id']}/restore" _conn.http_call("post", _url) @@ -2626,11 +2528,7 @@ class Project: self.get_drawings() try: - return next( - _drawing - for _drawing in (self.drawings or []) - if _drawing["drawing_id"] == drawing_id - ) + return next(_drawing for _drawing in (self.drawings or []) if _drawing["drawing_id"] == drawing_id) except (StopIteration, KeyError, TypeError): return None @@ -2793,9 +2691,7 @@ class Project: if not _drawing: raise ValueError("drawing not found") - _url = ( - f"{_conn.base_url}/projects/{_project_id}/drawings/{_drawing['drawing_id']}" - ) + _url = f"{_conn.base_url}/projects/{_project_id}/drawings/{_drawing['drawing_id']}" _conn.http_call("delete", _url) diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py index 8677c2f9d..0729d64f0 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py +++ b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py @@ -38,11 +38,13 @@ from typing import Any from langchain.tools import BaseTool -from gns3server.agent.gns3_copilot.gns3_client import Project, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Project +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + class GNS3ProjectInfoTool(BaseTool): """LangChain tool for retrieving GNS3 project basic information.""" @@ -88,9 +90,7 @@ class GNS3ProjectInfoTool(BaseTool): # Validate project_id parameter if not project_id: logger.error("project_id parameter is required.") - return { - "error": "project_id parameter is required. Please provide a valid project UUID." - } + return {"error": "project_id parameter is required. Please provide a valid project UUID."} # Initialize Gns3Connector using factory function logger.info("Connecting to GNS3 server...") @@ -98,9 +98,7 @@ class GNS3ProjectInfoTool(BaseTool): if server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Use the provided project_id directly logger.info(f"Retrieving project info for project_id: {project_id}") @@ -125,8 +123,13 @@ class GNS3ProjectInfoTool(BaseTool): } # Log result - logger.info("Project info retrieved: name=%s, status=%s, nodes=%d, links=%d", - project.name, project.status, node_count, link_count) + logger.info( + "Project info retrieved: name=%s, status=%s, nodes=%d, links=%d", + project.name, + project.status, + node_count, + link_count, + ) return result @@ -134,6 +137,7 @@ class GNS3ProjectInfoTool(BaseTool): logger.error("Error retrieving GNS3 project info: %s", str(e)) return {"error": f"Failed to retrieve project info: {str(e)}"} + if __name__ == "__main__": # Test the tool tool = GNS3ProjectInfoTool() diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py index 99451e468..2037929cf 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py +++ b/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py @@ -38,11 +38,13 @@ from typing import Any from langchain.tools import BaseTool -from gns3server.agent.gns3_copilot.gns3_client import Project, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Project +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + # Define LangChain tool class class GNS3TopologyTool(BaseTool): """LangChain tool for retrieving GNS3 project topology information.""" @@ -87,9 +89,7 @@ class GNS3TopologyTool(BaseTool): # Validate project_id parameter if not project_id: logger.error("project_id parameter is required.") - return { - "error": "project_id parameter is required. Please provide a valid project UUID." - } + return {"error": "project_id parameter is required. Please provide a valid project UUID."} # Initialize Gns3Connector using factory function logger.info("Connecting to GNS3 server...") @@ -97,9 +97,7 @@ class GNS3TopologyTool(BaseTool): if server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Use the provided project_id directly logger.info(f"Retrieving topology for project_id: {project_id}") @@ -111,18 +109,18 @@ class GNS3TopologyTool(BaseTool): "project_id": project.project_id, "name": project.name, "status": project.status, - "nodes": self._clean_nodes_ports( - copy.deepcopy(project.nodes_inventory()) - ), + "nodes": self._clean_nodes_ports(copy.deepcopy(project.nodes_inventory())), "links": project.links_summary(is_print=False), } # Log topology result - logger.info("Topology retrieved: project_id=%s, name=%s, nodes=%d, links=%d", - topology.get("project_id"), - topology.get("name"), - len(topology.get("nodes", {})), - len(topology.get("links", []))) + logger.info( + "Topology retrieved: project_id=%s, name=%s, nodes=%d, links=%d", + topology.get("project_id"), + topology.get("name"), + len(topology.get("nodes", {})), + len(topology.get("links", [])), + ) logger.debug("Topology details: %s", topology) return topology @@ -138,12 +136,10 @@ class GNS3TopologyTool(BaseTool): """ for node in data.values(): # Iterate through R-1, R-2, R-3, R-4 if "ports" in node and isinstance(node["ports"], list): - node["ports"] = [ - {"name": port["name"], "short_name": port["short_name"]} - for port in node["ports"] - ] + node["ports"] = [{"name": port["name"], "short_name": port["short_name"]} for port in node["ports"]] return data + if __name__ == "__main__": # Test the tool tool = GNS3TopologyTool() diff --git a/gns3server/agent/gns3_copilot/project_agent_manager.py b/gns3server/agent/gns3_copilot/project_agent_manager.py index 551481642..fd5f6f026 100644 --- a/gns3server/agent/gns3_copilot/project_agent_manager.py +++ b/gns3server/agent/gns3_copilot/project_agent_manager.py @@ -32,12 +32,14 @@ Each project has its own AgentService with a dedicated SQLite checkpoint databas import asyncio import logging -from typing import Dict, Optional +from typing import Dict +from typing import Optional from gns3server.agent.gns3_copilot.agent_service import AgentService log = logging.getLogger(__name__) + class ProjectAgentManager: """ Singleton manager for project-level Agent services. @@ -48,12 +50,13 @@ class ProjectAgentManager: """ _instance: Optional["ProjectAgentManager"] = None - _lock = asyncio.Lock() + _lock: asyncio.Lock = asyncio.Lock() + _agents: Dict[str, AgentService] = {} def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) - cls._instance._agents: Dict[str, AgentService] = {} + cls._instance._agents = {} cls._instance._lock = asyncio.Lock() return cls._instance @@ -124,10 +127,12 @@ class ProjectAgentManager: """ return list(self._agents.keys()) + # Global singleton instance _project_agent_manager: Optional[ProjectAgentManager] = None _manager_lock = asyncio.Lock() + async def get_project_agent_manager() -> ProjectAgentManager: """ Get the global ProjectAgentManager singleton instance. diff --git a/gns3server/agent/gns3_copilot/prompts/prompt_loader.py b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py index f93761c55..676fffc16 100644 --- a/gns3server/agent/gns3_copilot/prompts/prompt_loader.py +++ b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py @@ -42,6 +42,7 @@ from .lab_assistant_prompt import LAB_ASSISTANT_PROMPT logger = logging.getLogger(__name__) + def load_system_prompt(llm_config: dict | None = None) -> str: """ Load the system prompt for GNS3-Copilot. diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index 5b5765973..686e28393 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -54,7 +54,8 @@ from .display_tools_nornir import ExecuteMultipleDeviceCommands from .gns3_create_link import GNS3LinkTool from .gns3_create_node import GNS3CreateNodeTool from .gns3_get_node_temp import GNS3TemplateTool -from .gns3_start_node import GNS3StartNodeTool, GNS3StartNodeQuickTool +from .gns3_start_node import GNS3StartNodeQuickTool +from .gns3_start_node import GNS3StartNodeTool from .gns3_update_node_name import GNS3UpdateNodeNameTool from .vpcs_tools_telnetlib3 import VPCSMultiCommands @@ -84,4 +85,4 @@ __all__ = [ ] # Package initialization message -# print(f"GNS3-Copilot Tools package loaded (version {__version__})") \ No newline at end of file +# print(f"GNS3-Copilot Tools package loaded (version {__version__})") diff --git a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py index 8daef1583..e97e65704 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -30,7 +30,6 @@ This module provides a tool to execute configuration commands on multiple device import json import logging -import os import re from typing import Any @@ -39,7 +38,9 @@ from langchain_core.callbacks import CallbackManagerForToolRun from netmiko.exceptions import ReadTimeout from nornir import InitNornir from nornir.core import Nornir -from nornir.core.task import AggregatedResult, Result, Task +from nornir.core.task import AggregatedResult +from nornir.core.task import Result +from nornir.core.task import Task from nornir_netmiko.tasks import netmiko_send_config from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host @@ -48,15 +49,14 @@ from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology # config log logger = logging.getLogger(__name__) + # Local Nornir configuration functions for Cisco IOS Telnet devices def _get_nornir_defaults() -> dict[str, Any]: """Get Nornir default configuration for Cisco IOS.""" return {"data": {"location": "gns3"}} -def _get_nornir_groups_config( - device_type: str = "cisco_ios_telnet", - platform: str = "cisco_ios" -) -> dict[str, Any]: + +def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]: """ Get Nornir group configuration for Cisco IOS Telnet devices. @@ -73,15 +73,12 @@ def _get_nornir_groups_config( "timeout": 120, "username": "", "password": "", - "connection_options": { - "netmiko": {"extras": {"device_type": device_type}} - }, + "connection_options": {"netmiko": {"extras": {"device_type": device_type}}}, } + def _get_nornir_group( - group_name: str = "cisco_IOSv_telnet", - device_type: str = "cisco_ios_telnet", - platform: str = "cisco_ios" + group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios" ) -> dict[str, Any]: """ Get Nornir group configuration for a specific group. @@ -95,10 +92,8 @@ def _get_nornir_group( Dictionary containing group configuration """ # _get_nornir_groups_config now returns the group config directly - return _get_nornir_groups_config( - device_type=device_type, - platform=platform - ) + return _get_nornir_groups_config(device_type=device_type, platform=platform) + class ExecuteMultipleDeviceConfigCommands(BaseTool): """ @@ -174,11 +169,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) - if ( - isinstance(device_configs_list, list) - and len(device_configs_list) > 0 - and "error" in device_configs_list[0] - ): + if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]: return device_configs_list # Create a mapping of device names to their configuration commands @@ -186,9 +177,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Prepare device hosts data try: - hosts_data = self._prepare_device_hosts_data( - device_configs_list, project_id - ) + hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id) except ValueError as e: logger.error("Failed to prepare device hosts data: %s", e) return [{"error": str(e)}] @@ -210,9 +199,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): ) # Process results for all devices - results = self._process_task_results( - device_configs_list, hosts_data, task_result - ) + results = self._process_task_results(device_configs_list, hosts_data, task_result) except Exception as e: # Overall execution failed @@ -226,9 +213,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): return results - def _run_all_device_configs_with_single_retry( - self, task: Task, device_configs_map: dict[str, list[str]] - ) -> Result: + def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result: """Execute configuration commands with single retry mechanism.""" device_name = task.host.name config_commands = device_configs_map.get(device_name, []) @@ -237,9 +222,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): return Result(host=task.host, result="No configuration commands to execute") try: - _result = task.run( - task=netmiko_send_config, config_commands=config_commands - ) + _result = task.run(task=netmiko_send_config, config_commands=config_commands) return Result(host=task.host, result=_result.result) except ReadTimeout as e: @@ -259,9 +242,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Handle prompt detection issues with Cisco IOSv L2 images where the '#' prompt character # may be delayed, causing Netmiko prompt detection failures. Implements retry logic. if "netmiko_send_config (failed)" in str(e): - _result = task.run( - task=netmiko_send_config, config_commands=config_commands - ) + _result = task.run(task=netmiko_send_config, config_commands=config_commands) return Result(host=task.host, result=_result.result) # Log any other exceptions with full details @@ -307,9 +288,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Handle standard models (like GPT/OpenAI) where the framework # has already parsed the JSON into a Python object (dict or list). parsed_input = tool_input - logger.info( - "Using tool input directly as type: %s", type(parsed_input).__name__ - ) + logger.info("Using tool input directly as type: %s", type(parsed_input).__name__) # Handle new format: {"project_id": "...", "device_configs": [...]} if isinstance(parsed_input, dict): @@ -323,9 +302,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): return ([{"error": error_msg}], None) if not self._validate_project_id(project_id): - error_msg = ( - f"Invalid project_id format: {project_id}. Expected UUID format." - ) + error_msg = f"Invalid project_id format: {project_id}. Expected UUID format." logger.error(error_msg) return ([{"error": error_msg}], None) @@ -343,9 +320,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Handle legacy format: [...] elif isinstance(parsed_input, list): - logger.warning( - "Using legacy input format without project_id. Please use new format with project_id." - ) + logger.warning("Using legacy input format without project_id. Please use new format with project_id.") return parsed_input, None else: @@ -369,9 +344,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" return bool(re.match(uuid_pattern, project_id, re.IGNORECASE)) - def _configs_map( - self, device_config_list: list[dict[str, Any]] - ) -> dict[str, list[str]]: + def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]: """Create a mapping of device names to their configuration commands.""" device_configs_map = {} for device_config in device_config_list: @@ -386,9 +359,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): ) -> dict[str, dict[str, Any]]: """Prepare device hosts data from topology information.""" # Extract device names list - device_names = [ - device_config["device_name"] for device_config in device_config_list - ] + device_names = [device_config["device_name"] for device_config in device_config_list] # Get device port information with project_id hosts_data = get_device_ports_from_topology(device_names, project_id) @@ -424,16 +395,11 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): device_type = first_device_data.get("device_type", "cisco_ios_telnet") platform = first_device_data.get("platform", "cisco_ios") - logger.info( - "Extracted from tags: device_type=%s, platform=%s", - device_type, - platform - ) + logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform) # Get latest environment configuration with dynamic device_type and platform groups_data = _get_nornir_groups_config( - device_type=device_type or "cisco_ios_telnet", - platform=platform or "cisco_ios" + device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios" ) defaults = _get_nornir_defaults() @@ -484,9 +450,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): device_result = { "device_name": device_name, "status": "failed", - "error": ( - f"Device '{device_name}' not found in topology or missing console_port" - ), + "error": (f"Device '{device_name}' not found in topology or missing console_port"), } results.append(device_result) continue @@ -508,9 +472,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): if multi_result[0].failed: # Execution failed device_result["status"] = "failed" - device_result["error"] = ( - f"Configuration execution failed: {multi_result[0].result}" - ) + device_result["error"] = f"Configuration execution failed: {multi_result[0].result}" device_result["output"] = multi_result[0].result else: # Execution successful @@ -522,6 +484,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): return results + if __name__ == "__main__": # Example usage with new format # example tool_input with project_id diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index 6d36d6dc7..d7f14c4e9 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -30,34 +30,33 @@ This module provides a tool to execute display commands on multiple devices import json import logging -import os import re from typing import Any -from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host - from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun from netmiko.exceptions import ReadTimeout from nornir import InitNornir from nornir.core import Nornir -from nornir.core.task import AggregatedResult, Result, Task +from nornir.core.task import AggregatedResult +from nornir.core.task import Result +from nornir.core.task import Task from nornir_netmiko.tasks import netmiko_multiline +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology # config log logger = logging.getLogger(__name__) + # Local Nornir configuration functions for Cisco IOS Telnet devices def _get_nornir_defaults() -> dict[str, Any]: """Get Nornir default configuration for Cisco IOS.""" return {"data": {"location": "gns3"}} -def _get_nornir_groups_config( - device_type: str = "cisco_ios_telnet", - platform: str = "cisco_ios" -) -> dict[str, Any]: + +def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]: """ Get Nornir group configuration for Cisco IOS Telnet devices. @@ -74,15 +73,12 @@ def _get_nornir_groups_config( "timeout": 120, "username": "", "password": "", - "connection_options": { - "netmiko": {"extras": {"device_type": device_type}} - }, + "connection_options": {"netmiko": {"extras": {"device_type": device_type}}}, } + def _get_nornir_group( - group_name: str = "cisco_IOSv_telnet", - device_type: str = "cisco_ios_telnet", - platform: str = "cisco_ios" + group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios" ) -> dict[str, Any]: """ Get Nornir group configuration for a specific group. @@ -96,10 +92,8 @@ def _get_nornir_group( Dictionary containing group configuration """ # _get_nornir_groups_config now returns the group config directly - return _get_nornir_groups_config( - device_type=device_type, - platform=platform - ) + return _get_nornir_groups_config(device_type=device_type, platform=platform) + class ExecuteMultipleDeviceCommands(BaseTool): """ @@ -182,11 +176,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) - if ( - isinstance(device_configs_list, list) - and len(device_configs_list) > 0 - and "error" in device_configs_list[0] - ): + if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]: return device_configs_list # Create a mapping of device names to their display commands @@ -194,9 +184,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Prepare device hosts data try: - hosts_data = self._prepare_device_hosts_data( - device_configs_list, project_id - ) + hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id) except ValueError as e: logger.error("Failed to prepare device hosts data: %s", e) return [{"error": str(e)}] @@ -218,9 +206,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): ) # Process results for all devices - results = self._process_task_results( - device_configs_list, hosts_data, task_result - ) + results = self._process_task_results(device_configs_list, hosts_data, task_result) except Exception as e: # Overall execution failed @@ -234,9 +220,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): return results - def _run_all_device_configs_with_single_retry( - self, task: Task, device_configs_map: dict[str, list[str]] - ) -> Result: + def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result: """Execute READ-ONLY diagnostic commands with single retry mechanism.""" device_name = task.host.name diagnostic_commands = device_configs_map.get(device_name, []) @@ -323,9 +307,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Handle standard models (like GPT/OpenAI) where the framework # has already parsed the JSON into a Python object (dict or list). parsed_input = tool_input - logger.info( - "Using tool input directly as type: %s", type(parsed_input).__name__ - ) + logger.info("Using tool input directly as type: %s", type(parsed_input).__name__) # Handle new format: {"project_id": "...", "device_configs": [...]} if isinstance(parsed_input, dict): @@ -339,9 +321,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): return ([{"error": error_msg}], None) if not self._validate_project_id(project_id): - error_msg = ( - f"Invalid project_id format: {project_id}. Expected UUID format." - ) + error_msg = f"Invalid project_id format: {project_id}. Expected UUID format." logger.error(error_msg) return ([{"error": error_msg}], None) @@ -359,9 +339,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Handle legacy format: [...] elif isinstance(parsed_input, list): - logger.warning( - "Using legacy input format without project_id. Please use new format with project_id." - ) + logger.warning("Using legacy input format without project_id. Please use new format with project_id.") return parsed_input, None else: @@ -385,9 +363,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" return bool(re.match(uuid_pattern, project_id, re.IGNORECASE)) - def _configs_map( - self, device_config_list: list[dict[str, Any]] - ) -> dict[str, list[str]]: + def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]: """Create a mapping of device names to their diagnostic commands.""" device_diagnostic_map = {} for device_config in device_config_list: @@ -402,9 +378,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): ) -> dict[str, dict[str, Any]]: """Prepare device hosts data from topology information.""" # Extract device names list - device_names = [ - device_config["device_name"] for device_config in device_config_list - ] + device_names = [device_config["device_name"] for device_config in device_config_list] # Get device port information with project_id hosts_data = get_device_ports_from_topology(device_names, project_id) @@ -440,16 +414,11 @@ class ExecuteMultipleDeviceCommands(BaseTool): device_type = first_device_data.get("device_type", "cisco_ios_telnet") platform = first_device_data.get("platform", "cisco_ios") - logger.info( - "Extracted from tags: device_type=%s, platform=%s", - device_type, - platform - ) + logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform) # Get latest environment configuration with dynamic device_type and platform groups_data = _get_nornir_groups_config( - device_type=device_type or "cisco_ios_telnet", - platform=platform or "cisco_ios" + device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios" ) defaults = _get_nornir_defaults() @@ -500,9 +469,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): device_result = { "device_name": device_name, "status": "failed", - "error": ( - f"Device '{device_name}' not found in topology or missing console_port" - ), + "error": (f"Device '{device_name}' not found in topology or missing console_port"), } results.append(device_result) continue @@ -524,9 +491,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): if multi_result[0].failed: # Execution failed device_result["status"] = "failed" - device_result["error"] = ( - f"Diagnostic command execution failed: {multi_result[0].result}" - ) + device_result["error"] = f"Diagnostic command execution failed: {multi_result[0].result}" device_result["output"] = multi_result[0].result else: # Execution successful @@ -538,6 +503,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): return results + if __name__ == "__main__": # Example usage with new format device_commands = json.dumps( diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py index 707906d2a..e3a328d83 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py @@ -38,11 +38,13 @@ from typing import Any from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun -from gns3server.agent.gns3_copilot.gns3_client import Link, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Link +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + class GNS3LinkTool(BaseTool): """ Tool for creating network links between GNS3 nodes. @@ -96,9 +98,7 @@ class GNS3LinkTool(BaseTool): ] """ - def _run( - self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None - ) -> list[dict[str, Any]]: + def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> list[dict[str, Any]]: """ Creates one or multiple links between nodes in a GNS3 project. @@ -133,11 +133,7 @@ class GNS3LinkTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return [ - { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } - ] + return [{"error": "Failed to connect to GNS3 server. Please check your configuration."}] created_links = [] @@ -160,12 +156,8 @@ class GNS3LinkTool(BaseTool): continue # Get node details - node1 = gns3_server.get_node( - project_id=project_id, node_id=node_id1 - ) - node2 = gns3_server.get_node( - project_id=project_id, node_id=node_id2 - ) + node1 = gns3_server.get_node(project_id=project_id, node_id=node_id1) + node2 = gns3_server.get_node(project_id=project_id, node_id=node_id2) if not node1 or not node2: error_msg = f"Node not found in link {i}" logger.error(error_msg) @@ -174,19 +166,11 @@ class GNS3LinkTool(BaseTool): # Find port information port1_info = next( - ( - port - for port in node1.get("ports", []) - if port.get("name") == port1 - ), + (port for port in node1.get("ports", []) if port.get("name") == port1), None, ) port2_info = next( - ( - port - for port in node2.get("ports", []) - if port.get("name") == port2 - ), + (port for port in node2.get("ports", []) if port.get("name") == port2), None, ) if not port1_info or not port2_info: @@ -249,6 +233,7 @@ class GNS3LinkTool(BaseTool): logger.error("Failed to process link creation: %s", e) return [{"error": f"Failed to process link creation: {str(e)}"}] + if __name__ == "__main__": # Test with single link single_link_input = json.dumps( diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py index b6bfceb4e..1ef1d969c 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py @@ -38,11 +38,13 @@ from typing import Any from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun -from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Node +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + class GNS3CreateNodeTool(BaseTool): """ A LangChain tool to create multiple nodes in a GNS3 project @@ -174,9 +176,7 @@ class GNS3CreateNodeTool(BaseTool): "Invalid input: Node %d missing or invalid template_id, x, or y.", i + 1, ) - return { - "error": f"Node {i + 1} missing or invalid template_id, x, or y." - } + return {"error": f"Node {i + 1} missing or invalid template_id, x, or y."} # Initialize Gns3Connector using factory function logger.info("Connecting to GNS3 server...") @@ -184,9 +184,7 @@ class GNS3CreateNodeTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Create nodes logger.info("Creating %d nodes in project %s...", len(nodes), project_id) @@ -267,6 +265,7 @@ class GNS3CreateNodeTool(BaseTool): logger.error("Failed to process node creation request: %s", e) return {"error": f"Failed to process node creation request: {str(e)}"} + if __name__ == "__main__": # Test the tool locally with multiple nodes test_input = json.dumps( diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py index a4cd48594..ab106edd5 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py @@ -43,6 +43,7 @@ from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + class GNS3TemplateTool(BaseTool): """ A LangChain tool to retrieve all available device templates from a GNS3 server. @@ -102,9 +103,7 @@ class GNS3TemplateTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Retrieve all available templates templates = gns3_server.get_templates() @@ -128,11 +127,10 @@ class GNS3TemplateTool(BaseTool): return result except Exception as e: - logger.error( - "Failed to connect to GNS3 server or retrieve templates: %s", e - ) + logger.error("Failed to connect to GNS3 server or retrieve templates: %s", e) return {"error": f"Failed to retrieve templates: {str(e)}"} + if __name__ == "__main__": # Test's tool locally tool = GNS3TemplateTool() diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py index 1c397c4a8..e6aa21001 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py @@ -39,14 +39,14 @@ from typing import Any from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun -from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Node +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) -def show_progress_bar( - duration: int = 120, interval: int = 1, node_count: int = 1 -) -> None: + +def show_progress_bar(duration: int = 120, interval: int = 1, node_count: int = 1) -> None: """ Display a simple text progress bar for node startup. @@ -63,9 +63,7 @@ def show_progress_bar( # Create progress bar display bar_length = 30 filled_length = int(bar_length * elapsed // duration) - progress_string = ( - "=" * filled_length + ">" + " " * (bar_length - filled_length - 1) - ) + progress_string = "=" * filled_length + ">" + " " * (bar_length - filled_length - 1) # Print progress bar with node count print(f"\r[{progress_string}] {progress:.1f}%", end="", flush=True) @@ -73,6 +71,7 @@ def show_progress_bar( print(f"\n{node_count} node(s) startup completed!") + class GNS3StartNodeTool(BaseTool): """ A LangChain tool to start one or multiple nodes in a GNS3 project. @@ -106,9 +105,7 @@ class GNS3StartNodeTool(BaseTool): Returns: A dictionary with all nodes' details including success/failure status. """ - def _run( - self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None - ) -> dict[str, Any]: + def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]: try: # Parse input JSON input_data = json.loads(tool_input) @@ -130,9 +127,7 @@ class GNS3StartNodeTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # First loop: Send start commands for all nodes logger.info( @@ -142,22 +137,16 @@ class GNS3StartNodeTool(BaseTool): ) for node_id in node_ids: try: - node = Node( - project_id=project_id, node_id=node_id, connector=gns3_server - ) + node = Node(project_id=project_id, node_id=node_id, connector=gns3_server) # Verify node exists node.get() if node.node_id: node.start() logger.info("Start command sent for node %s", node_id) else: - logger.error( - "Node %s not found in project %s", node_id, project_id - ) + logger.error("Node %s not found in project %s", node_id, project_id) except Exception as e: - logger.error( - "Failed to send start command for node %s: %s", node_id, e - ) + logger.error("Failed to send start command for node %s: %s", node_id, e) # Calculate progress bar duration: 140s base + 10s per additional node base_duration = 140 @@ -165,18 +154,14 @@ class GNS3StartNodeTool(BaseTool): total_duration = base_duration + extra_duration # Show progress bar - show_progress_bar( - duration=total_duration, interval=1, node_count=len(node_ids) - ) + show_progress_bar(duration=total_duration, interval=1, node_count=len(node_ids)) # Second loop: Get status for all nodes results = [] logger.info("Retrieving status for %d nodes...", len(node_ids)) for node_id in node_ids: try: - node = Node( - project_id=project_id, node_id=node_id, connector=gns3_server - ) + node = Node(project_id=project_id, node_id=node_id, connector=gns3_server) node.get() # Get latest status node_info = { "node_id": node.node_id, @@ -223,6 +208,7 @@ class GNS3StartNodeTool(BaseTool): logger.error("Failed to start nodes: %s", e) return {"error": f"Failed to start nodes: {str(e)}"} + class GNS3StartNodeQuickTool(BaseTool): """ A LangChain tool to start one or multiple nodes in a GNS3 project WITHOUT waiting. @@ -263,9 +249,7 @@ class GNS3StartNodeQuickTool(BaseTool): NOTE: Nodes will continue booting in background after this tool returns. """ - def _run( - self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None - ) -> dict[str, Any]: + def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]: try: # Parse input JSON input_data = json.loads(tool_input) @@ -287,9 +271,7 @@ class GNS3StartNodeQuickTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Send start commands for all nodes and collect initial status logger.info( @@ -301,9 +283,7 @@ class GNS3StartNodeQuickTool(BaseTool): for node_id in node_ids: try: - node = Node( - project_id=project_id, node_id=node_id, connector=gns3_server - ) + node = Node(project_id=project_id, node_id=node_id, connector=gns3_server) # Verify node exists and get current info node.get() if not node.node_id: @@ -371,15 +351,14 @@ class GNS3StartNodeQuickTool(BaseTool): logger.error("Failed to start nodes: %s", e) return {"error": f"Failed to start nodes: {str(e)}"} + if __name__ == "__main__": # Test with single node print("=== Testing single node startup ===") test_input_single = json.dumps( { "project_id": "", # Replace with actual project UUID - "node_ids": [ - "fbeda109-9a74-4d8c-a749-cc3847911a90" - ], # Replace with actual node UUID + "node_ids": ["fbeda109-9a74-4d8c-a749-cc3847911a90"], # Replace with actual node UUID } ) tool = GNS3StartNodeTool() diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py index 380d4516b..586bbf43b 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py @@ -37,11 +37,13 @@ from typing import Any from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun -from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector +from gns3server.agent.gns3_copilot.gns3_client import Node +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector # Configure logging logger = logging.getLogger(__name__) + class GNS3UpdateNodeNameTool(BaseTool): """ A LangChain tool to update the name of one or multiple nodes in a GNS3 project. @@ -78,9 +80,7 @@ class GNS3UpdateNodeNameTool(BaseTool): Returns: A dictionary with all nodes' update results including success/failure status. """ - def _run( - self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None - ) -> dict[str, Any]: + def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]: try: # Parse input JSON input_data = json.loads(tool_input) @@ -111,9 +111,7 @@ class GNS3UpdateNodeNameTool(BaseTool): if gns3_server is None: logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} # Update node names logger.info("Updating names for %d nodes in project %s...", len(nodes), project_id) @@ -133,9 +131,7 @@ class GNS3UpdateNodeNameTool(BaseTool): ) # Get node to retrieve current name - node = Node( - project_id=project_id, node_id=node_id, connector=gns3_server - ) + node = Node(project_id=project_id, node_id=node_id, connector=gns3_server) node.get() old_name = node.name @@ -152,9 +148,7 @@ class GNS3UpdateNodeNameTool(BaseTool): "status": "success", } results.append(node_info) - logger.info( - "Successfully updated node name: %s -> %s", old_name, new_name - ) + logger.info("Successfully updated node name: %s -> %s", old_name, new_name) else: error_info = { "node_id": node_id, @@ -205,6 +199,7 @@ class GNS3UpdateNodeNameTool(BaseTool): logger.error("Failed to update node names: %s", e) return {"error": f"Failed to update node names: {str(e)}"} + if __name__ == "__main__": # Test with single node print("=== Testing single node name update ===") diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py index c5d36ada7..0b15a59bc 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py @@ -30,7 +30,6 @@ Supports concurrent execution of multiple command groups across multiple VPCS de import json import logging -import os import re import threading from time import sleep @@ -45,6 +44,7 @@ from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology logger = logging.getLogger(__name__) + class VPCSMultiCommands(BaseTool): """ A tool for VPCS (Virtual PC Simulator) devices to view PC configurations and test connectivity. @@ -131,9 +131,7 @@ class VPCSMultiCommands(BaseTool): # Check if device has port information if device_name not in device_ports: - logger.warning( - "Device '%s' not found in topology or missing console port", device_name - ) + logger.warning("Device '%s' not found in topology or missing console port", device_name) results_list[index] = { "device_name": device_name, "status": "error", @@ -199,9 +197,7 @@ class VPCSMultiCommands(BaseTool): ) except Exception as e: - logger.error( - "Error executing commands on device '%s': %s", device_name, str(e) - ) + logger.error("Error executing commands on device '%s': %s", device_name, str(e)) results_list[index] = { "device_name": device_name, "status": "error", @@ -256,9 +252,7 @@ class VPCSMultiCommands(BaseTool): else: # Handle standard models where the framework has already parsed the JSON. parsed_input = tool_input - logger.info( - "Using tool input directly as type: %s", type(parsed_input).__name__ - ) + logger.info("Using tool input directly as type: %s", type(parsed_input).__name__) # Validate input is a dictionary if not isinstance(parsed_input, dict): @@ -278,9 +272,7 @@ class VPCSMultiCommands(BaseTool): # Validate project_id format if not self._validate_project_id(project_id): - error_msg = ( - f"Invalid project_id format: {project_id}. Expected UUID format." - ) + error_msg = f"Invalid project_id format: {project_id}. Expected UUID format." logger.error(error_msg) return ([{"error": error_msg}], "") @@ -305,9 +297,7 @@ class VPCSMultiCommands(BaseTool): # Validate each item in device_configs for i, item in enumerate(device_configs): if not isinstance(item, dict): - error_msg = ( - f"Item at index {i} must be a dictionary, got {type(item).__name__}" - ) + error_msg = f"Item at index {i} must be a dictionary, got {type(item).__name__}" logger.error(error_msg) return ([{"error": error_msg}], "") @@ -324,8 +314,7 @@ class VPCSMultiCommands(BaseTool): if not isinstance(item["commands"], list): error_msg = ( - f"'commands' in item at index {i} must be a list, " - f"but got {type(item['commands']).__name__}" + f"'commands' in item at index {i} must be a list, " f"but got {type(item['commands']).__name__}" ) logger.error(error_msg) return ([{"error": error_msg}], "") @@ -363,20 +352,14 @@ class VPCSMultiCommands(BaseTool): device_configs, project_id = self._validate_tool_input(tool_input) # Check if validation returned an error - if ( - isinstance(device_configs, list) - and len(device_configs) > 0 - and "error" in device_configs[0] - ): + if isinstance(device_configs, list) and len(device_configs) > 0 and "error" in device_configs[0]: return device_configs # Extract all device names from input using set comprehension device_names = {config["device_name"] for config in device_configs} # Get device port mapping with project_id - device_ports = get_device_ports_from_topology( - list(device_names), project_id=project_id - ) + device_ports = get_device_ports_from_topology(list(device_names), project_id=project_id) logger.info( "Retrieved port mappings for %d devices: %s", len(device_ports), @@ -394,7 +377,6 @@ class VPCSMultiCommands(BaseTool): # Create thread for each command group logger.info("Starting parallel execution for %d devices", len(device_configs)) for i, cmd_group in enumerate(device_configs): - device_name = cmd_group["device_name"] thread = threading.Thread( target=self._connect_and_execute_commands, args=( @@ -426,6 +408,7 @@ class VPCSMultiCommands(BaseTool): return results + if __name__ == "__main__": # Example usage command_groups = json.dumps( diff --git a/gns3server/agent/gns3_copilot/utils/__init__.py b/gns3server/agent/gns3_copilot/utils/__init__.py index b11390aae..9cc0371eb 100644 --- a/gns3server/agent/gns3_copilot/utils/__init__.py +++ b/gns3server/agent/gns3_copilot/utils/__init__.py @@ -38,11 +38,9 @@ Author: Guobin Yue # Import main utility functions from .get_gns3_device_port import get_device_ports_from_topology -from .parse_tool_content import ( - format_tool_response, - normalize_tool_response, - parse_tool_content -) +from .parse_tool_content import format_tool_response +from .parse_tool_content import normalize_tool_response +from .parse_tool_content import parse_tool_content # Dynamic version management try: diff --git a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py index ec56908e4..eda3cd19e 100644 --- a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py +++ b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py @@ -32,6 +32,7 @@ from typing import Any logger = logging.getLogger(__name__) + def get_device_ports_from_topology( device_names: list[str], project_id: str | None = None, diff --git a/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py b/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py index d73cd8ca4..c676a1f06 100644 --- a/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py +++ b/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py @@ -31,7 +31,8 @@ Supports ellipse and rectangle shapes for two-node annotations. """ import math -from typing import Any, Literal +from typing import Any +from typing import Literal # Default parameters DEFAULT_DEVICE_WIDTH = 50 @@ -103,6 +104,7 @@ COLOR_SCHEMES = { }, } + def calculate_two_node_shape( node1: dict, node2: dict, @@ -133,13 +135,9 @@ def calculate_two_node_shape( node2_center_x = node2["x"] + (node2_width / 2) node2_center_y = node2["y"] + (node2_height / 2) - distance = math.sqrt( - (node2_center_x - node1_center_x) ** 2 + (node2_center_y - node1_center_y) ** 2 - ) + distance = math.sqrt((node2_center_x - node1_center_x) ** 2 + (node2_center_y - node1_center_y) ** 2) - angle_rad = math.atan2( - node2_center_y - node1_center_y, node2_center_x - node1_center_x - ) + angle_rad = math.atan2(node2_center_y - node1_center_y, node2_center_x - node1_center_x) angle_deg = round(math.degrees(angle_rad)) angle_rad = math.radians(angle_deg) @@ -162,9 +160,7 @@ def calculate_two_node_shape( svg_x = center_x - (rx * math.cos(angle_rad) - ry * math.sin(angle_rad)) svg_y = center_y - (rx * math.sin(angle_rad) + ry * math.cos(angle_rad)) - shape_svg = generate_ellipse_svg( - int(rx), int(ry), color_scheme, int(shape_width), int(shape_height) - ) + shape_svg = generate_ellipse_svg(int(rx), int(ry), color_scheme, int(shape_width), int(shape_height)) offset_distance = ry * text_offset_ratio @@ -183,18 +179,10 @@ def calculate_two_node_shape( shape_width = distance shape_height = max(node1_width, node1_height, node2_width, node2_height) - svg_x = center_x - ( - (shape_width / 2) * math.cos(angle_rad) - - (shape_height / 2) * math.sin(angle_rad) - ) - svg_y = center_y - ( - (shape_width / 2) * math.sin(angle_rad) - + (shape_height / 2) * math.cos(angle_rad) - ) + svg_x = center_x - ((shape_width / 2) * math.cos(angle_rad) - (shape_height / 2) * math.sin(angle_rad)) + svg_y = center_y - ((shape_width / 2) * math.sin(angle_rad) + (shape_height / 2) * math.cos(angle_rad)) - shape_svg = generate_rectangle_svg( - int(shape_width), int(shape_height), color_scheme - ) + shape_svg = generate_rectangle_svg(int(shape_width), int(shape_height), color_scheme) offset_distance = (shape_height / 2) * text_offset_ratio @@ -241,6 +229,7 @@ def calculate_two_node_shape( "metadata": metadata, } + def generate_ellipse_svg( rx: int, ry: int, @@ -249,7 +238,8 @@ def generate_ellipse_svg( svg_height: int, ) -> str: """Generate SVG for ellipse.""" - return f'''''' + return f"""""" + def generate_rectangle_svg( width: int, @@ -257,14 +247,16 @@ def generate_rectangle_svg( color_scheme: dict[str, Any], ) -> str: """Generate SVG for rectangle.""" - return f'''''' + return f"""""" + def generate_text_svg(text: str, color_scheme: dict[str, Any]) -> str: """Generate SVG for text label.""" text_width = len(text) * 8 + 20 text_height = DEFAULT_FONT_SIZE + 16 - return f'''{text}''' + return f"""{text}""" + def _hsv_to_hex(h: int, s: int, v: int) -> str: """Convert HSV to HEX color.""" @@ -299,6 +291,7 @@ def _hsv_to_hex(h: int, s: int, v: int) -> str: return f"#{r:02x}{g:02x}{b:02x}" + def calculate_z_order(area_size: float) -> int: """ Calculate z-order based on area size for proper layering. @@ -316,6 +309,7 @@ def calculate_z_order(area_size: float) -> int: else: return 3 + def _get_color_scheme(area_name: str) -> dict[str, Any]: """ Get color scheme based on area name using keyword inference. @@ -339,33 +333,15 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]: return COLOR_SCHEMES["NORMAL_AREA"] # 3. Logical Isolation - if ( - "VRF" in label - or "VLAN" in label - or "MSTP" in label - or "VXLAN" in label - or "MPLS" in label - ): + if "VRF" in label or "VLAN" in label or "MSTP" in label or "VXLAN" in label or "MPLS" in label: return COLOR_SCHEMES["ISOLATION"] # 4. High Availability - if ( - "VRRP" in label - or "HSRP" in label - or "HA" in label - or "STACK" in label - or "M-LAG" in label - ): + if "VRRP" in label or "HSRP" in label or "HA" in label or "STACK" in label or "M-LAG" in label: return COLOR_SCHEMES["HIGH_AVAILABILITY"] # 5. External/Internet - if ( - "INET" in label - or "OUT" in label - or "EXTERNAL" in label - or "INTERNET" in label - or "DMZ" in label - ): + if "INET" in label or "OUT" in label or "EXTERNAL" in label or "INTERNET" in label or "DMZ" in label: return COLOR_SCHEMES["EXTERNAL"] # 6. Management @@ -404,6 +380,7 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]: return COLOR_SCHEMES["DEFAULT"] + def calculate_two_node_ellipse( node1: dict, node2: dict, @@ -415,15 +392,14 @@ def calculate_two_node_ellipse( Wrapper around calculate_two_node_shape with shape_type="ellipse". """ - result = calculate_two_node_shape( - node1, node2, area_name, "ellipse", text_offset_ratio - ) + result = calculate_two_node_shape(node1, node2, area_name, "ellipse", text_offset_ratio) return { "ellipse": result["shape"], "text": result["text"], "metadata": result["metadata"], } + def calculate_two_node_rectangle( node1: dict, node2: dict, @@ -435,9 +411,7 @@ def calculate_two_node_rectangle( Wrapper around calculate_two_node_shape with shape_type="rectangle". """ - result = calculate_two_node_shape( - node1, node2, area_name, "rectangle", text_offset_ratio - ) + result = calculate_two_node_shape(node1, node2, area_name, "rectangle", text_offset_ratio) metadata = result["metadata"] return { "rectangle": result["shape"], diff --git a/gns3server/agent/gns3_copilot/utils/llm_config_helper.py b/gns3server/agent/gns3_copilot/utils/llm_config_helper.py index 34a6077b1..d0f43be5f 100644 --- a/gns3server/agent/gns3_copilot/utils/llm_config_helper.py +++ b/gns3server/agent/gns3_copilot/utils/llm_config_helper.py @@ -41,17 +41,17 @@ Usage: """ import logging -from typing import Optional, Dict, Any +from typing import Any +from typing import Dict +from typing import Optional from uuid import UUID from fastapi import FastAPI logger = logging.getLogger(__name__) -async def get_user_llm_config_with_app( - user_id: UUID, - app: FastAPI -) -> Optional[Dict[str, Any]]: + +async def get_user_llm_config_with_app(user_id: UUID, app: FastAPI) -> Optional[Dict[str, Any]]: """ Get user's default LLM model configuration with decrypted API key. diff --git a/gns3server/agent/gns3_copilot/utils/message_converters.py b/gns3server/agent/gns3_copilot/utils/message_converters.py index 7c5b3cf1e..87e1d353e 100644 --- a/gns3server/agent/gns3_copilot/utils/message_converters.py +++ b/gns3server/agent/gns3_copilot/utils/message_converters.py @@ -28,12 +28,18 @@ Message format converters for OpenAI-compatible message format. Converts between LangChain messages and OpenAI-compatible format. """ + import json import uuid from datetime import datetime -from typing import Dict, Any +from typing import Any +from typing import Dict + +from langchain_core.messages import AIMessage +from langchain_core.messages import HumanMessage +from langchain_core.messages import SystemMessage +from langchain_core.messages import ToolMessage -from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage def _ensure_string(content: Any) -> str: """Ensure content is a string, converting dicts/lists to JSON if needed.""" @@ -44,6 +50,7 @@ def _ensure_string(content: Any) -> str: else: return str(content) + def convert_langchain_to_openai(lc_message) -> Dict[str, Any]: """ Convert LangChain message to OpenAI-compatible format. @@ -55,53 +62,40 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]: Dictionary in OpenAI-compatible format """ # Generate message ID - msg_id = getattr(lc_message, 'id', None) + msg_id = getattr(lc_message, "id", None) if msg_id is None: msg_id = str(uuid.uuid4()) # Get timestamp - timestamp = getattr(lc_message, 'created_at', None) + timestamp = getattr(lc_message, "created_at", None) if timestamp is None: timestamp = datetime.utcnow().isoformat() - elif hasattr(timestamp, 'isoformat'): + elif hasattr(timestamp, "isoformat"): timestamp = timestamp.isoformat() # Base message structure - base_msg = { - "id": msg_id, - "created_at": timestamp, - "metadata": {} - } + base_msg = {"id": msg_id, "created_at": timestamp, "metadata": {}} # Convert based on message type if isinstance(lc_message, HumanMessage): - return { - **base_msg, - "role": "user", - "content": lc_message.content - } + return {**base_msg, "role": "user", "content": lc_message.content} elif isinstance(lc_message, AIMessage): - msg = { - **base_msg, - "role": "assistant", - "content": lc_message.content - } + msg = {**base_msg, "role": "assistant", "content": lc_message.content} # Handle tool calls - convert to OpenAI format - if hasattr(lc_message, 'tool_calls') and lc_message.tool_calls: + if hasattr(lc_message, "tool_calls") and lc_message.tool_calls: tool_calls = [] for tc in lc_message.tool_calls: # Convert to dict if it's an object tc_dict = tc if isinstance(tc, dict) else tc.model_dump() - tool_calls.append({ - "id": tc_dict.get("id", str(uuid.uuid4())), - "type": "function", - "function": { - "name": tc_dict.get("name", ""), - "arguments": tc_dict.get("args", {}) + tool_calls.append( + { + "id": tc_dict.get("id", str(uuid.uuid4())), + "type": "function", + "function": {"name": tc_dict.get("name", ""), "arguments": tc_dict.get("args", {})}, } - }) + ) msg["tool_calls"] = tool_calls return msg @@ -111,24 +105,17 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]: **base_msg, "role": "tool", "content": _ensure_string(lc_message.content), - "name": getattr(lc_message, 'name', ''), - "tool_call_id": getattr(lc_message, 'tool_call_id', '') + "name": getattr(lc_message, "name", ""), + "tool_call_id": getattr(lc_message, "tool_call_id", ""), } elif isinstance(lc_message, SystemMessage): - return { - **base_msg, - "role": "system", - "content": lc_message.content - } + return {**base_msg, "role": "system", "content": lc_message.content} else: # Fallback for unknown message types - return { - **base_msg, - "role": "unknown", - "content": str(lc_message) - } + return {**base_msg, "role": "unknown", "content": str(lc_message)} + def convert_openai_to_langchain(msg: Dict[str, Any]): """ @@ -153,21 +140,19 @@ def convert_openai_to_langchain(msg: Dict[str, Any]): if "tool_calls" in msg and msg["tool_calls"]: tool_calls = [] for tc in msg["tool_calls"]: - tool_calls.append({ - "id": tc.get("id", str(uuid.uuid4())), - "name": tc.get("function", {}).get("name", ""), - "args": tc.get("function", {}).get("arguments", {}) - }) + tool_calls.append( + { + "id": tc.get("id", str(uuid.uuid4())), + "name": tc.get("function", {}).get("name", ""), + "args": tc.get("function", {}).get("arguments", {}), + } + ) ai_msg.tool_calls = tool_calls return ai_msg elif role == "tool": - return ToolMessage( - content=content, - name=msg.get("name", ""), - tool_call_id=msg.get("tool_call_id", "") - ) + return ToolMessage(content=content, name=msg.get("name", ""), tool_call_id=msg.get("tool_call_id", "")) elif role == "system": return SystemMessage(content=content) @@ -176,6 +161,7 @@ def convert_openai_to_langchain(msg: Dict[str, Any]): # Fallback to HumanMessage for unknown roles return HumanMessage(content=content) + def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]: """ Convert LangGraph streaming event to OpenAI-compatible format. @@ -190,21 +176,17 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]: if event_type == "on_chat_model_stream": chunk = event.get("data", {}).get("chunk", {}) - content = getattr(chunk, 'content', '') + content = getattr(chunk, "content", "") if content: - return { - "type": "content", - "content": content, - "message_id": event.get("metadata", {}).get("msg_id") - } + return {"type": "content", "content": content, "message_id": event.get("metadata", {}).get("msg_id")} # Check for tool call chunks - if hasattr(chunk, 'tool_call_chunks') and chunk.tool_call_chunks: + if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks: for tc_chunk in chunk.tool_call_chunks: - tc_id = getattr(tc_chunk, 'id', None) - tc_name = getattr(tc_chunk, 'name', None) - tc_args = getattr(tc_chunk, 'args', None) + tc_id = getattr(tc_chunk, "id", None) + tc_name = getattr(tc_chunk, "name", None) + tc_args = getattr(tc_chunk, "args", None) if tc_id: return { @@ -212,19 +194,12 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]: "tool_call": { "id": tc_id, "type": "function", - "function": { - "name": tc_name or "", - "arguments": tc_args or "" - } - } + "function": {"name": tc_name or "", "arguments": tc_args or ""}, + }, } elif event_type == "on_tool_start": - return { - "type": "tool_start", - "tool_name": event.get("name", ""), - "metadata": event.get("metadata", {}) - } + return {"type": "tool_start", "tool_name": event.get("name", ""), "metadata": event.get("metadata", {})} elif event_type == "on_tool_end": tool_output = event.get("data", {}).get("output", "") @@ -236,7 +211,7 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]: "type": "tool_end", "tool_output": tool_output, "tool_name": event.get("name", ""), - "metadata": event.get("metadata", {}) + "metadata": event.get("metadata", {}), } # Default empty response diff --git a/gns3server/agent/gns3_copilot/utils/parse_tool_content.py b/gns3server/agent/gns3_copilot/utils/parse_tool_content.py index b13d0db28..6cc32b65d 100644 --- a/gns3server/agent/gns3_copilot/utils/parse_tool_content.py +++ b/gns3server/agent/gns3_copilot/utils/parse_tool_content.py @@ -71,6 +71,7 @@ from typing import Any logger = logging.getLogger(__name__) + def parse_tool_content( content: str | dict | list | int | float | bool | None, fallback_to_raw: bool = True, @@ -172,9 +173,7 @@ def parse_tool_content( # (higher priority as many tools return Python format strings) try: result = ast.literal_eval(s) - logger.info( - "Successfully parsed as Python literal, returning: %s", result - ) + logger.info("Successfully parsed as Python literal, returning: %s", result) return result except (ValueError, SyntaxError): pass @@ -211,8 +210,7 @@ def parse_tool_content( # Handle unsupported types error_msg = ( # type: ignore[unreachable] - "Content must be str, dict, list, int, float, bool, or None, got " - f"{type(content).__name__}" + "Content must be str, dict, list, int, float, bool, or None, got " f"{type(content).__name__}" ) logger.error(error_msg) @@ -227,9 +225,8 @@ def parse_tool_content( logger.info("Returning error: %s", result) return result -def format_tool_response( - content: str | dict | list | int | float | bool | None, indent: int = 2 -) -> str: + +def format_tool_response(content: str | dict | list | int | float | bool | None, indent: int = 2) -> str: """ Format tool response as a beautiful JSON string for UI display. @@ -256,9 +253,7 @@ def format_tool_response( # If the parsed result cannot be serialized, convert to string and wrap logger.error("Cannot serialize parsed result to JSON: %s", e) try: - result = json.dumps( - {"raw": str(content)}, ensure_ascii=False, indent=indent - ) + result = json.dumps({"raw": str(content)}, ensure_ascii=False, indent=indent) logger.info("format_tool_response returning fallback: %s", result) return result except Exception: @@ -276,10 +271,8 @@ def format_tool_response( logger.info("format_tool_response returning error: %s", result) return result -def normalize_tool_response( - response: dict | list | str, - tool_name: str = "unknown" -) -> dict: + +def normalize_tool_response(response: dict | list | str, tool_name: str = "unknown") -> dict: """ Normalize tool response to standard format for consistent frontend display. @@ -310,10 +303,7 @@ def normalize_tool_response( >>> normalize_tool_response([{"device_name": "R1", "status": "success"}]) {'success': True, 'total': 1, 'successful': 1, 'failed': 0, 'data': [...], 'metadata': {}} """ - metadata = { - "tool_name": tool_name, - "normalized_at": datetime.utcnow().isoformat() - } + metadata = {"tool_name": tool_name, "normalized_at": datetime.utcnow().isoformat()} # Handle error responses if isinstance(response, dict) and "error" in response and len(response) == 1: @@ -324,19 +314,12 @@ def normalize_tool_response( "failed": 0, "data": [], "error": str(response["error"]), - "metadata": metadata + "metadata": metadata, } # Handle empty responses if not response: - return { - "success": True, - "total": 0, - "successful": 0, - "failed": 0, - "data": [], - "metadata": metadata - } + return {"success": True, "total": 0, "successful": 0, "failed": 0, "data": [], "metadata": metadata} # Handle list responses (most tools return list of device results) if isinstance(response, list): @@ -358,12 +341,7 @@ def normalize_tool_response( normalized_data.append(normalized_item) else: # Non-dict items in list - normalized_data.append({ - "id": "", - "name": "", - "status": "unknown", - "result": str(item) - }) + normalized_data.append({"id": "", "name": "", "status": "unknown", "result": str(item)}) return { "success": failed == 0, @@ -371,7 +349,7 @@ def normalize_tool_response( "successful": successful, "failed": failed, "data": normalized_data, - "metadata": metadata + "metadata": metadata, } # Handle dict responses (some tools return summary + results) @@ -385,7 +363,7 @@ def normalize_tool_response( "failed": response.get("failed", 0), "data": response.get("data", []), "error": response.get("error"), - "metadata": {**metadata, **response.get("metadata", {})} + "metadata": {**metadata, **response.get("metadata", {})}, } # Legacy format: extract common fields @@ -403,20 +381,24 @@ def normalize_tool_response( data = response["data"] elif "output" in response: # Single device response - data = [{ - "name": response.get("device_name", ""), - "status": response.get("status", "success"), - "result": response["output"] - }] + data = [ + { + "name": response.get("device_name", ""), + "status": response.get("status", "success"), + "result": response["output"], + } + ] # If no data found but have status, create single item if not data and "status" in response: - data = [{ - "name": response.get("device_name") or response.get("name") or "", - "status": response["status"], - "result": response.get("output") or response.get("result") or "", - "error": response.get("error") or "" - }] + data = [ + { + "name": response.get("device_name") or response.get("name") or "", + "status": response["status"], + "result": response.get("output") or response.get("result") or "", + "error": response.get("error") or "", + } + ] # Recursively normalize data items if data and isinstance(data, list): @@ -429,7 +411,7 @@ def normalize_tool_response( "successful": successful, "failed": failed, "data": [], - "metadata": metadata + "metadata": metadata, } # Handle string responses (parse first) @@ -443,15 +425,11 @@ def normalize_tool_response( "total": 1, "successful": 1, "failed": 0, - "data": [{ - "id": "", - "name": "", - "status": "unknown", - "result": str(response) - }], - "metadata": metadata + "data": [{"id": "", "name": "", "status": "unknown", "result": str(response)}], + "metadata": metadata, } + # Test function to verify the implementation def _test_parse_tool_content() -> None: """Test function to verify parse_tool_content works correctly with all input types""" @@ -510,5 +488,6 @@ def _test_parse_tool_content() -> None: valid = "✗" print(f"Format Test {i + 1}: {valid} Input: {repr(input_data)} -> {result}") + if __name__ == "__main__": _test_parse_tool_content() diff --git a/gns3server/api/routes/controller/chat.py b/gns3server/api/routes/controller/chat.py index f559089f1..86b3d4686 100644 --- a/gns3server/api/routes/controller/chat.py +++ b/gns3server/api/routes/controller/chat.py @@ -134,6 +134,12 @@ async def stream_chat( llm_config.get("model"), ) + # TODO: Support runtime temperature override from request.temperature + # Currently, temperature is loaded from the user's LLM config in the database. + # To enable runtime override, uncomment the following: + # if request.temperature is not None: + # llm_config["temperature"] = str(request.temperature) + # 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) diff --git a/gns3server/schemas/controller/chat.py b/gns3server/schemas/controller/chat.py index 64ede4925..ed8950b18 100644 --- a/gns3server/schemas/controller/chat.py +++ b/gns3server/schemas/controller/chat.py @@ -36,7 +36,12 @@ class ChatRequest(BaseModel): message: str = Field(..., description="User message content") session_id: Optional[str] = Field(None, description="Session ID (auto-generated if not provided)") stream: bool = Field(default=True, description="Enable streaming response") - temperature: Optional[float] = Field(None, description="LLM temperature parameter") + temperature: Optional[float] = Field( + None, + description="LLM temperature parameter (NOTE: currently not used. " + "Temperature is loaded from user's LLM config in database. " + "Reserved for future runtime override support.)" + ) mode: Literal["text"] = Field(default="text", description="Interaction mode")