diff --git a/gns3server/agent/gns3_copilot/agent/context_manager.py b/gns3server/agent/gns3_copilot/agent/context_manager.py index fccbb03c9..297f5088e 100644 --- a/gns3server/agent/gns3_copilot/agent/context_manager.py +++ b/gns3server/agent/gns3_copilot/agent/context_manager.py @@ -136,26 +136,37 @@ def estimate_tool_tokens(tools: list[Any]) -> int: 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() + tool_schema["function"]["parameters"] = ( + tool.args_schema.model_json_schema() + ) except AttributeError: # Fallback to Pydantic v1 method (schema) try: - tool_schema["function"]["parameters"] = tool.args_schema.schema() + tool_schema["function"]["parameters"] = ( + tool.args_schema.schema() + ) except Exception: # Both methods failed, use empty schema + tool_name = getattr(tool, "name", "unknown") 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", + tool_name, ) tool_schema["function"]["parameters"] = {} except Exception as e: # model_json_schema() raised an exception + tool_name = getattr(tool, "name", "unknown") logger.debug( - "model_json_schema() failed for tool %s: %s, trying v1 fallback", - getattr(tool, "name", "unknown"), + "model_json_schema() failed for tool %s: %s, " + "trying v1 fallback", + tool_name, e, ) try: - tool_schema["function"]["parameters"] = tool.args_schema.schema() + tool_schema["function"]["parameters"] = ( + tool.args_schema.schema() + ) except Exception: tool_schema["function"]["parameters"] = {} @@ -165,7 +176,12 @@ def estimate_tool_tokens(tools: list[Any]) -> int: total_tokens += tokens except Exception as e: - logger.debug(f"Failed to estimate tokens for tool {getattr(tool, 'name', 'unknown')}: {e}") + tool_name = getattr(tool, "name", "unknown") + logger.debug( + "Failed to estimate tokens for tool %s: %s", + tool_name, + e, + ) # Rough fallback: 1000 tokens per tool total_tokens += 1000 @@ -179,7 +195,8 @@ def _count_tokens_for_message(message: BaseMessage) -> int: This function is called by trim_messages for each message. Args: - message: A single message (HumanMessage, AIMessage, SystemMessage, etc.) + message: A single message (HumanMessage, AIMessage, SystemMessage, + etc.) Returns: Estimated token count for the message @@ -206,13 +223,16 @@ def create_pre_model_hook( Create a pre_model_hook function for LangGraph agent. This hook will be automatically called before each LLM invocation, - handling topology injection, tool token estimation, and message trimming. + handling topology injection, tool token estimation, and message + trimming. IMPORTANT USAGE REQUIREMENTS: - This hook MUST be passed via model.invoke() config, NOT used as a Node return value. + This hook MUST be passed via model.invoke() config, NOT used as + a Node return value. CORRECT Usage: - model.invoke(messages, config={"configurable": {"pre_model_hook": pre_hook}}) + model.invoke(messages, + config={"configurable": {"pre_model_hook": pre_hook}}) INCORRECT Usage: # ❌ Don't use as Node return value @@ -224,56 +244,69 @@ def create_pre_model_hook( messages: Annotated[list, add_messages] # Incompatible! COMPATIBILITY: - - Works with: model.invoke(config={"configurable": {"pre_model_hook": ...}}) + - Works with: model.invoke(config={"configurable": + {"pre_model_hook": ...}}) - Does NOT work as: StateGraph Node return value - Does NOT work with: Annotated[list, add_messages] state reducers - The hook returns a complete message list that overwrites the model input, - which is correct for invoke() but wrong for state updates with add_messages. + The hook returns a complete message list that overwrites the model + input, which is correct for invoke() but wrong for state updates with + add_messages. Args: - system_prompt: System prompt template (must contain {{topology_info}} placeholder) + system_prompt: System prompt template (must contain + {{topology_info}} placeholder) get_topology_func: Function to extract topology from state get_llm_config_func: Function to get LLM config from context - get_tools_func: Optional function to get tools list for token estimation + get_tools_func: Optional function to get tools list for token + estimation Returns: A pre_model_hook function for use with model.invoke(config=...) Thread Safety: - Thread-safe if get_llm_config_func uses request-scoped context (e.g., contextvars) + Thread-safe if get_llm_config_func uses request-scoped context + (e.g., contextvars) Performance Notes: - - SystemMessage is reconstructed on each LLM call (intentional design) + - SystemMessage is reconstructed on each LLM call (intentional + design) - Overhead: ~1-2ms per call (negligible compared to LLM latency) - Trade-off: Simplicity > micro-optimization - - 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) + - 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. - This function is NOT a StateGraph Node. It is a preprocessing hook that - modifies the input to the LLM, not the agent state. + This function is NOT a StateGraph Node. It is a preprocessing hook + that modifies the input to the LLM, not the agent state. Usage Context: Called automatically by LangChain when passed via: - model.invoke(messages, config={"configurable": {"pre_model_hook": this}}) + model.invoke(messages, + config={"configurable": + {"pre_model_hook": this}}) Args: state: Current agent state containing messages and topology_info Returns: dict with 'messages' key containing prepared and trimmed messages - Note: This return value is used by LangChain to replace the model input, - NOT to update the agent state. + Note: This return value is used by LangChain to replace the model + input, NOT to update the agent state. Raises: ValueError: If context_limit is missing or invalid """ - logger.debug("pre_model_hook invoked: messages=%d", len(state.get("messages", []))) + logger.debug( + "pre_model_hook invoked: messages=%d", + len(state.get("messages", [])), + ) messages = state.get("messages", []) if not messages: @@ -285,11 +318,14 @@ def create_pre_model_hook( if not llm_config: logger.error("LLM config not found. context_limit is required.") - raise ValueError("LLM config not found. context_limit is required.") + raise ValueError( + "LLM config not found. context_limit is required." + ) 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") @@ -302,7 +338,11 @@ def create_pre_model_hook( strategy = llm_config.get("context_strategy", DEFAULT_CONTEXT_STRATEGY) if strategy not in CONTEXT_STRATEGY_RATIOS: - logger.warning("Invalid context_strategy '%s', using '%s'", strategy, DEFAULT_CONTEXT_STRATEGY) + logger.warning( + "Invalid context_strategy '%s', using '%s'", + strategy, + DEFAULT_CONTEXT_STRATEGY, + ) strategy = DEFAULT_CONTEXT_STRATEGY # Step 1: Estimate tool tokens @@ -311,13 +351,20 @@ def create_pre_model_hook( try: tools = get_tools_func() tool_tokens = estimate_tool_tokens(tools) - logger.debug("Tool definitions estimated at ~%d tokens (%d tools)", tool_tokens, len(tools)) + logger.debug( + "Tool definitions estimated at ~%d tokens (%d tools)", + tool_tokens, + len(tools), + ) except Exception as e: logger.warning("Failed to estimate tool tokens: %s", e) # 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 @@ -325,7 +372,9 @@ 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 @@ -333,17 +382,19 @@ def create_pre_model_hook( max_input_tokens = int(model_limit_tokens * strategy_ratio) # Calculate max tokens for trim_messages - # Important: trim_messages counts ALL messages (including system), so we only - # subtract tool_tokens here, NOT system_tokens. Let trim_messages handle system. + # Important: trim_messages counts ALL messages (including system), + # so we only subtract tool_tokens here, NOT system_tokens. + # Let trim_messages handle system. max_tokens_for_trim = max_input_tokens - tool_tokens # Validate budget and provide actionable warnings - if system_tokens + tool_tokens > max_input_tokens: + if system_tokens + tool_tokens > max_input_tokens: # noqa: E501 logger.error( - "System prompt (%d tokens) + tools (%d tokens) EXCEED input budget (%d tokens). " - "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 prompt (%d tokens) + tools (%d tokens) EXCEED input " + "budget (%d tokens). This will likely cause LLM call failures. " # noqa: E501 + "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, @@ -351,7 +402,8 @@ def create_pre_model_hook( 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). " + "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, @@ -360,7 +412,8 @@ def create_pre_model_hook( # 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)", + "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, @@ -370,7 +423,8 @@ def create_pre_model_hook( ) # Step 4: Trim messages to fit - # Note: max_tokens_for_trim INCLUDES system message, trim_messages will handle it + # Note: max_tokens_for_trim INCLUDES system message, trim_messages + # will handle it try: trimmed = trim_messages( messages=messages_with_system, @@ -382,11 +436,14 @@ def create_pre_model_hook( # Calculate final token counts final_total = sum(_count_tokens_for_message(m) for m in trimmed) - usage_percent = (final_total + tool_tokens) / model_limit_tokens * 100 + usage_percent = ( + (final_total + tool_tokens) / model_limit_tokens * 100 + ) if len(trimmed) < len(messages_with_system): logger.info( - "Messages trimmed: %d → %d msgs. Total: ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s", + "Messages trimmed: %d → %d msgs. Total: ~%d tokens + %d " + "tools = %d / %dK (%.1f%%), strategy=%s", len(messages_with_system), len(trimmed), final_total, @@ -398,7 +455,8 @@ def create_pre_model_hook( ) else: logger.info( - "Context ready: %d msgs, ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s", + "Context ready: %d msgs, ~%d tokens + %d tools = %d / %dK " + "(%.1f%%), strategy=%s", len(trimmed), final_total, tool_tokens, @@ -449,23 +507,33 @@ 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}") + 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 + 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 - non_system_messages = [m for m in messages if not isinstance(m, SystemMessage)] + non_system_messages = [ + m for m in messages if not isinstance(m, SystemMessage) + ] filtered_count = len(messages) - len(non_system_messages) if filtered_count > 0: - logger.debug("Filtered out %d existing SystemMessage(s)", filtered_count) + logger.debug( + "Filtered out %d existing SystemMessage(s)", filtered_count + ) return [SystemMessage(content=formatted_prompt)] + non_system_messages @@ -487,7 +555,8 @@ def prepare_context_messages( **DEPRECATED**: Use create_pre_model_hook() instead. """ warnings.warn( - "prepare_context_messages() is deprecated. Use create_pre_model_hook() instead.", + "prepare_context_messages() is deprecated. Use " + "create_pre_model_hook() instead.", DeprecationWarning, stacklevel=2, ) @@ -495,9 +564,13 @@ 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 @@ -543,7 +616,9 @@ if __name__ == "__main__": # Test invocation print("\nTest 3: Invoke pre_model_hook") test_state = { - "messages": [HumanMessage(f"Message {i}: {'x' * 50}") for i in range(5)], + "messages": [ + HumanMessage(f"Message {i}: {'x' * 50}") for i in range(5) + ], "topology_info": {"project_id": "test123", "nodes": 3}, } diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index 98a874c8a..4f3f47d5f 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -39,7 +39,8 @@ The agent provides: - Integration with GNS3 topology management Copilot Modes: -- "teaching_assistant" (default): Diagnostic tools only, no configuration changes +- "teaching_assistant" (default): Diagnostic tools only, no configuration + changes - "lab_automation_assistant": Full diagnostic and configuration tools """ @@ -64,19 +65,27 @@ from langgraph.managed.is_last_step import RemainingSteps from typing_extensions import TypedDict # 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, ) -from gns3server.agent.gns3_copilot.agent.model_factory import 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 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 ( + 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 @@ -87,8 +96,9 @@ from gns3server.agent.gns3_copilot.tools_v2 import VPCSMultiCommands # Set up logger for GNS3-Copilot logger = logging.getLogger(__name__) -# Note: LLM model configuration is now managed by the new llm_model_configs system. -# The model_factory module handles model creation with configuration from the database. +# Note: LLM model configuration is now managed by the new llm_model_configs +# system. The model_factory module handles model creation with configuration +# from the database. # Define tools for different copilot modes # Teaching assistant mode: READ-ONLY diagnostic tools only @@ -98,7 +108,8 @@ TEACHING_ASSISTANT_MODE_TOOLS = [ GNS3LinkTool(), # Create links between nodes GNS3StartNodeTool(), # Start GNS3 nodes GNS3UpdateNodeNameTool(), # Update node name - ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands (READ-ONLY) + ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands + # (READ-ONLY) ] # Lab automation assistant mode: Full diagnostic AND configuration tools @@ -108,7 +119,8 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [ GNS3LinkTool(), # Create links between nodes GNS3StartNodeTool(), # Start GNS3 nodes GNS3UpdateNodeNameTool(), # Update node name - ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands (READ-ONLY) + ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands + # (READ-ONLY) ExecuteMultipleDeviceConfigCommands(), # Execute configuration commands VPCSMultiCommands(), # Execute VPCS commands on multiple devices ] @@ -117,7 +129,8 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [ tools = LAB_AUTOMATION_ASSISTANT_MODE_TOOLS # Create combined tool lookup for tool_node (supports both modes) -# tool_node will receive tool calls based on mode-specific tools bound to the model +# tool_node will receive tool calls based on mode-specific tools bound to the +# model ALL_TOOLS = LAB_AUTOMATION_ASSISTANT_MODE_TOOLS tools_by_name = {tool.name: tool for tool in ALL_TOOLS} @@ -135,14 +148,18 @@ class MessagesState(TypedDict): """ GNS3-Copilot conversation state management class. - Maintains the conversation state for the LangGraph workflow, including message history, - call counters, and session titles for comprehensive dialogue management. + Maintains the conversation state for the LangGraph workflow, including + message history, call counters, and session titles for comprehensive + dialogue management. Attributes: - messages: List of conversation messages with cumulative updates using operator.add + messages: List of conversation messages with cumulative updates using + operator.add llm_calls: Counter for tracking the number of LLM invocations - remaining_steps: Is automatically managed by LangGraph's RemainingSteps to track and limit recursion depth. - conversation_title: Optional conversation title for session identification and management + remaining_steps: Is automatically managed by LangGraph's RemainingSteps + to track and limit recursion depth. + conversation_title: Optional conversation title for session + identification and management topology_info: Dictionary containing GNS3 project topology information """ @@ -183,7 +200,9 @@ def llm_call(state: dict, config: RunnableConfig | None = None): } logger.debug( - "LLM config retrieved from context: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model") + "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 @@ -211,16 +230,25 @@ 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", + 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) + logger.warning( + "Error retrieving topology for project_id %s: %s", + project_id, + e, + ) # Store topology_info in state for pre_model_hook to access state["topology_info"] = topology_info @@ -229,10 +257,15 @@ def llm_call(state: dict, config: RunnableConfig | None = None): copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower() if copilot_mode == "lab_automation_assistant": mode_tools = LAB_AUTOMATION_ASSISTANT_MODE_TOOLS - logger.info("Using LAB_AUTOMATION_ASSISTANT mode tools (includes configuration tools)") + logger.info( + "Using LAB_AUTOMATION_ASSISTANT mode tools (includes " + "configuration tools)" + ) else: # teaching_assistant mode (default) mode_tools = TEACHING_ASSISTANT_MODE_TOOLS - logger.info("Using TEACHING_ASSISTANT mode tools (diagnostic tools only)") + logger.info( + "Using TEACHING_ASSISTANT mode tools (diagnostic tools only)" + ) # Create pre_model_hook for automatic topology injection and trimming # Load system prompt based on copilot_mode configuration @@ -241,7 +274,8 @@ def llm_call(state: dict, config: RunnableConfig | None = None): system_prompt=system_prompt, get_topology_func=lambda s: s.get("topology_info"), get_llm_config_func=get_current_llm_config, - get_tools_func=lambda: mode_tools, # Pass mode-specific tools for token estimation + get_tools_func=lambda: mode_tools, # Pass mode-specific tools for + # token estimation ) # Create fresh model with tools for each LLM call @@ -252,14 +286,22 @@ def llm_call(state: dict, config: RunnableConfig | None = None): copilot_mode, len(mode_tools), ) - model_with_tools = create_base_model_with_tools(mode_tools, llm_config=llm_config) + model_with_tools = create_base_model_with_tools( + mode_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 + # Call pre_hook directly to prepare messages (topology injection + + # trimming) + # Note: LangGraph's pre_model_hook only works with prebuilt agents, not + # custom StateGraph logger.info("Calling pre_hook to prepare %d messages", len(messages)) - prepared_state = pre_hook({"messages": messages, "topology_info": topology_info}) + prepared_state = pre_hook( + {"messages": messages, "topology_info": topology_info} + ) prepared_messages = prepared_state["messages"] - logger.info("Messages prepared: %d → %d", len(messages), len(prepared_messages)) + logger.info( + "Messages prepared: %d → %d", len(messages), len(prepared_messages) + ) # Invoke model with prepared messages response = model_with_tools.invoke(prepared_messages) @@ -267,15 +309,22 @@ def llm_call(state: dict, config: RunnableConfig | None = None): # Add metadata with created_at timestamp to AI response if hasattr(response, "metadata"): existing_metadata = response.metadata or {} - response.metadata = {**existing_metadata, "created_at": datetime.utcnow().isoformat()} + response.metadata = { + **existing_metadata, + "created_at": datetime.utcnow().isoformat(), + } else: - # LangChain messages should have metadata attribute, but defensive fallback + # LangChain messages should have metadata attribute, but defensive + # fallback try: response.metadata = {"created_at": datetime.utcnow().isoformat()} except Exception: logger.warning("Could not add metadata to AI response") - 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], @@ -285,10 +334,13 @@ def llm_call(state: dict, config: RunnableConfig | None = None): # Define generate title node -def generate_title(state: MessagesState, config: RunnableConfig | None = None) -> dict: +def generate_title( + state: MessagesState, config: RunnableConfig | None = None +) -> dict: """ - Generate a conversation title using a lightweight assistant LLM (title_model). - This node is only executed when no title has been set yet (first round only). + Generate a conversation title using a lightweight assistant LLM + (title_model). This node is only executed when no title has been set yet + (first round only). """ # Get llm_config from request-scoped context variable @@ -314,24 +366,37 @@ 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() # Validate the generated title if not new_title or len(new_title) < 3: - raise ValueError(f"Generated title too short or empty: '{new_title}'") + raise ValueError( + f"Generated title too short or empty: '{new_title}'" + ) - if new_title in ["New Conversation", "Untitled Session", "GNS3 Session"]: - raise ValueError(f"Generated title is a default value: '{new_title}'") + if new_title in [ + "New Conversation", + "Untitled Session", + "GNS3 Session", + ]: + raise ValueError( + f"Generated title is a default value: '{new_title}'" + ) # Safety: truncate long titles and avoid line breaks if len(new_title) > TITLE_MAX_LENGTH: new_title = new_title[: TITLE_MAX_LENGTH - 2] + "..." # Remove unwanted characters - new_title = new_title.replace("\n", " ").replace('"', "").replace("'", "") + new_title = ( + new_title.replace("\n", " ").replace('"', "").replace("'", "") + ) logger.info("Generated new title: %s", new_title) return {"conversation_title": new_title} @@ -355,11 +420,17 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) - fallback_title = fallback_title[:28] + ".." if fallback_title: - logger.info(f"Using fallback title from user message: '{fallback_title}'") + logger.info( + "Using fallback title from user message: '%s'", + fallback_title, + ) return {"conversation_title": fallback_title} # Final fallback - logger.info(f"Using final fallback title: '{UNTITLED_SESSION_FALLBACK}'") + logger.info( + "Using final fallback title: '%s'", + UNTITLED_SESSION_FALLBACK, + ) return {"conversation_title": UNTITLED_SESSION_FALLBACK} # Title already exists → no update needed @@ -376,17 +447,24 @@ def tool_node(state: dict, config: RunnableConfig | None = None): result = [] for tool_call in tool_calls: tool_name = tool_call["name"] - logger.debug("Executing tool: %s with args: %s", tool_name, tool_call["args"]) + logger.debug( + "Executing tool: %s with args: %s", tool_name, tool_call["args"] + ) 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)}" # Serialize observation to JSON string if it's not already a string - # This ensures ToolMessage.content is always JSON format, not Python str() + # This ensures ToolMessage.content is always JSON format, not Python + # str() if not isinstance(observation, str): observation = json.dumps(observation, ensure_ascii=False, indent=2) @@ -395,7 +473,7 @@ def tool_node(state: dict, config: RunnableConfig | None = None): content=observation, tool_call_id=tool_call["id"], name=tool_call["name"], - metadata={"created_at": datetime.utcnow().isoformat()} + metadata={"created_at": datetime.utcnow().isoformat()}, ) result.append(tool_msg) @@ -410,7 +488,8 @@ def should_continue( Determine the next step after the LLM has produced a response. - If the LLM requested any tool calls → route to tool_node - - If this is the first complete turn (llm_calls == 1) and no title exists → generate a title + - If this is the first complete turn (llm_calls == 1) and no title + exists → generate a title - Otherwise → conversation is complete, go to END """ last_message = state["messages"][-1] @@ -433,8 +512,8 @@ def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]: """ Routing logic after tool execution to prevent infinite recursion. - Determines whether to continue with another LLM call or end the conversation - based on remaining steps and message type. + Determines whether to continue with another LLM call or end the + conversation based on remaining steps and message type. Args: state: Current conversation state with messages and remaining steps @@ -467,13 +546,16 @@ agent_builder.add_node("title_generator_node", generate_title) # Add edges to connect nodes agent_builder.add_edge(START, "llm_call") # Conditional routing after LLM response -# Determines the next step based on whether LLM needs to call tools or generate title +# Determines the next step based on whether LLM needs to call tools or +# generate title agent_builder.add_conditional_edges( "llm_call", should_continue, { - "tool_node": "tool_node", # Route to tool execution if LLM requested tools - "title_generator_node": "title_generator_node", # Generate title on first interaction + "tool_node": "tool_node", # Route to tool execution if LLM requested + # tools + "title_generator_node": "title_generator_node", # Generate title on + # first interaction END: END, # End conversation if no tools needed }, ) @@ -483,7 +565,8 @@ agent_builder.add_conditional_edges( "tool_node", recursion_limit_continue, { - "llm_call": "llm_call", # Continue to LLM if tools executed and steps remain + "llm_call": "llm_call", # Continue to LLM if tools executed and steps + # remain END: END, # End conversation to prevent infinite loops }, ) diff --git a/gns3server/agent/gns3_copilot/agent/model_factory.py b/gns3server/agent/gns3_copilot/agent/model_factory.py index ad5445b72..e638ac734 100644 --- a/gns3server/agent/gns3_copilot/agent/model_factory.py +++ b/gns3server/agent/gns3_copilot/agent/model_factory.py @@ -58,7 +58,11 @@ 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", ""), @@ -90,7 +94,8 @@ def create_base_model( # Log the loaded configuration (mask sensitive data) logger.info( - "Creating base model: name=%s, provider=%s, base_url=%s, temperature=%s", + "Creating base model: name=%s, provider=%s, base_url=%s, " + "temperature=%s", config_vars["model_name"], config_vars["model_provider"], config_vars["base_url"] if config_vars["base_url"] else "default", @@ -130,8 +135,8 @@ def create_title_model( Create a fresh title generation model instance. This creates a model instance suitable for generating conversation titles. - It uses the same configuration as the base model but with a higher temperature - for more creative output. + It uses the same configuration as the base model but with a higher + temperature for more creative output. Args: llm_config: Configuration dictionary from database @@ -147,7 +152,8 @@ def create_title_model( config_vars = _load_llm_config(llm_config) logger.info( - "Creating title model: name=%s, provider=%s, base_url=%s, temperature=1.0", + "Creating title model: name=%s, provider=%s, base_url=%s, " + "temperature=1.0", config_vars["model_name"], config_vars["model_provider"], config_vars["base_url"] if config_vars["base_url"] else "default", diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index 4b03b4a4d..1cdc57c9c 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -108,7 +108,9 @@ class AgentService: return self._checkpointer checkpoint_dir = self._get_checkpoint_dir() - checkpointer_path = os.path.join(checkpoint_dir, "copilot_checkpoints.db") + checkpointer_path = os.path.join( + checkpoint_dir, "copilot_checkpoints.db" + ) log.debug("Creating checkpointer at: %s", checkpointer_path) @@ -118,13 +120,17 @@ class AgentService: await self._checkpointer_conn.close() log.debug("Closed previous checkpointer connection") except Exception as e: - log.warning("Error closing old checkpointer connection: %s", e) + log.warning( + "Error closing old checkpointer connection: %s", e + ) # Create new connection conn = await aiosqlite.connect(checkpointer_path) # Enable WAL mode for better concurrent performance await conn.execute("PRAGMA journal_mode=WAL;") - self._checkpointer_conn = conn # Save connection reference to prevent GC + self._checkpointer_conn = ( + conn # Save connection reference to prevent GC + ) self._checkpointer = AsyncSqliteSaver(conn) # CRITICAL: Initialize database schema @@ -176,22 +182,33 @@ class AgentService: """) # Create indexes - await conn.execute("CREATE INDEX IF NOT EXISTS idx_thread_id ON chat_sessions(thread_id)") - await conn.execute("CREATE INDEX IF NOT EXISTS idx_user_project ON chat_sessions(user_id, project_id)") + await conn.execute( + "CREATE INDEX IF NOT EXISTS idx_thread_id ON " + "chat_sessions(thread_id)" + ) + await conn.execute( + "CREATE INDEX IF NOT EXISTS idx_user_project ON " + "chat_sessions(user_id, project_id)" + ) - # Check if pinned column exists, add it if not (migration for existing databases) + # Check if pinned column exists, add it if not (migration for existing + # databases) cursor = await conn.execute("PRAGMA table_info(chat_sessions)") columns = await cursor.fetchall() column_names = [col[1] for col in columns] if "pinned" not in column_names: log.debug("Adding pinned column to existing chat_sessions table") - await conn.execute("ALTER TABLE chat_sessions ADD COLUMN pinned BOOLEAN DEFAULT FALSE") + await conn.execute( + "ALTER TABLE chat_sessions ADD COLUMN pinned BOOLEAN DEFAULT " + "FALSE" + ) 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)" + "CREATE INDEX IF NOT EXISTS idx_pinned_updated ON " + "chat_sessions(pinned DESC, updated_at DESC)" ) await conn.commit() @@ -202,7 +219,9 @@ class AgentService: if self._graph is None: checkpointer = await self._get_checkpointer() self._graph = agent_builder.compile(checkpointer=checkpointer) - log.info("LangGraph agent compiled for project: %s", self.project_path) + log.info( + "LangGraph agent compiled for project: %s", self.project_path + ) return self._graph async def stream_chat( @@ -225,13 +244,15 @@ class AgentService: user_id: User ID for metadata tracking jwt_token: JWT token for API authentication (optional) mode: Interaction mode (default: "text") - llm_config: LLM configuration dict (provider, model, api_key, etc.) + llm_config: LLM configuration dict (provider, model, api_key, + etc.) Yields: Dict containing SSE-compatible response chunks """ log.info( - "Stream chat started: project_id=%s, user_id=%s, session_id=%s, mode=%s", + "Stream chat started: project_id=%s, user_id=%s, session_id=%s, " + "mode=%s", project_id, user_id, session_id, @@ -250,7 +271,10 @@ 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) @@ -261,7 +285,9 @@ class AgentService: 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") + "LLM config set in context: provider=%s, model=%s", + llm_config.get("provider"), + llm_config.get("model"), ) # Build config - only thread-safe identifiers @@ -304,33 +330,49 @@ class AgentService: ai_response_counted = False tool_messages_counted = 0 - # Initialize tool call stream accumulator for handling progressive tool call arguments + # Initialize tool call stream accumulator for handling progressive + # tool call arguments tool_call_accumulator = ToolCallStreamAccumulator() # Stream events try: - async for event in graph.astream_events(inputs, config=config, version="v2"): + async for event in graph.astream_events( + inputs, config=config, version="v2" + ): event_type = event.get("event", "") data = event.get("data", {}) # Track LLM calls and tokens if event_type == "on_chat_model_start": # Filter out title_generator_node from statistics - langgraph_node = event.get("metadata", {}).get("langgraph_node", "") + langgraph_node = event.get("metadata", {}).get( + "langgraph_node", "" + ) if langgraph_node != "title_generator_node": llm_calls_count += 1 - log.debug("LLM call started, count=%d", llm_calls_count) + log.debug( + "LLM call started, count=%d", llm_calls_count + ) else: - log.debug("Skipping LLM call count for internal node: title_generator_node") + log.debug( + "Skipping LLM call count for internal node: " + "title_generator_node" + ) elif event_type == "on_chat_model_end": # Filter out title_generator_node from token counting - langgraph_node = event.get("metadata", {}).get("langgraph_node", "") + langgraph_node = event.get("metadata", {}).get( + "langgraph_node", "" + ) if langgraph_node == "title_generator_node": - log.debug("Skipping token counting for internal node: title_generator_node") + log.debug( + "Skipping token counting for internal node: " + "title_generator_node" + ) else: # Extract token usage from response metadata - # Try multiple possible locations where token usage might be stored + # Try multiple possible locations where token usage + # might be stored token_info_found = False # Method 1: response.usage_metadata @@ -348,8 +390,12 @@ class AgentService: if hasattr(output_msg, "usage_metadata"): usage = output_msg.usage_metadata if usage: - input_tokens += usage.get("input_tokens", 0) - output_tokens += usage.get("output_tokens", 0) + input_tokens += usage.get( + "input_tokens", 0 + ) + output_tokens += usage.get( + "output_tokens", 0 + ) token_info_found = True # Method 3: Check data directly for token usage fields @@ -358,7 +404,10 @@ class AgentService: input_tokens += data.get("input_tokens", 0) if "output_tokens" in data: output_tokens += data.get("output_tokens", 0) - if "input_tokens" in data or "output_tokens" in data: + if ( + "input_tokens" in data + or "output_tokens" in data + ): token_info_found = True # Count AI response as one message (only once per turn) @@ -370,16 +419,26 @@ class AgentService: elif event_type == "on_tool_end": message_count += 1 # Tool result message tool_messages_counted += 1 - log.debug("Tool message counted, message_count=%d", message_count) + log.debug( + "Tool message counted, message_count=%d", message_count + ) # Convert event to chunk for SSE streaming - # Use accumulator for on_chat_model_stream events to handle progressive tool calls + # Use accumulator for on_chat_model_stream events to handle + # progressive tool calls - # Filter out internal nodes (title_generator_node) from streaming to frontend - langgraph_node = event.get("metadata", {}).get("langgraph_node", "") + # Filter out internal nodes (title_generator_node) from + # streaming to frontend + langgraph_node = event.get("metadata", {}).get( + "langgraph_node", "" + ) if langgraph_node == "title_generator_node": - # Skip all events from the title_generator_node (internal use only) - log.debug("Skipping event from internal node: title_generator_node") + # Skip all events from the title_generator_node (internal + # use only) + log.debug( + "Skipping event from internal node: " + "title_generator_node" + ) continue if event_type == "on_chat_model_stream": @@ -387,7 +446,10 @@ class AgentService: for chunk in chunks: # Add session_id to each chunk chunk["session_id"] = session_id - log.debug("Yielding accumulated chunk: type=%s", chunk.get("type")) + log.debug( + "Yielding accumulated chunk: type=%s", + chunk.get("type"), + ) yield chunk else: # Use stateless converter for other events @@ -407,7 +469,8 @@ class AgentService: last_message_at=last_message_at, ) log.info( - "Session statistics updated: thread_id=%s, messages=%d, llm_calls=%d, tokens=%d+%d=%d", + "Session statistics updated: thread_id=%s, messages=%d, " + "llm_calls=%d, tokens=%d+%d=%d", session_id, message_count, llm_calls_count, @@ -421,15 +484,26 @@ class AgentService: if final_state and "conversation_title" in final_state.values: generated_title = final_state.values["conversation_title"] 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) + 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, + ) except Exception as e: log.error("Error in stream_chat: %s", e, exc_info=True) yield {"type": "error", "error": str(e), "session_id": session_id} - def _convert_event_to_chunk(self, event: Dict[str, Any], session_id: str) -> Optional[Dict[str, Any]]: + def _convert_event_to_chunk( + self, event: Dict[str, Any], session_id: str + ) -> Optional[Dict[str, Any]]: """ Convert LangGraph event to API response chunk. @@ -441,15 +515,17 @@ class AgentService: Dict for SSE response or None if event should be filtered Note: - on_chat_model_stream events are handled by ToolCallStreamAccumulator - before calling this method, so they are not processed here. + on_chat_model_stream events are handled by + ToolCallStreamAccumulator before calling this method, so they are + not processed here. """ event_type = event.get("event", "") data = event.get("data", {}) if event_type == "on_tool_start": # Tool execution started - # Extract tool_call_id from event metadata to associate with tool_call event + # Extract tool_call_id from event metadata to associate with + # tool_call event tool_call_id = event.get("metadata", {}).get("tool_call_id", "") return { "type": "tool_start", @@ -460,9 +536,11 @@ class AgentService: elif event_type == "on_tool_end": # Tool execution completed + # Extract tool output and convert to JSON string output = data.get("output", "") # Convert output to JSON string if it's not already a string - # This ensures dict/list outputs are properly serialized for frontend parsing + # This ensures dict/list outputs are properly serialized for + # frontend parsing if not isinstance(output, str): output = json.dumps(output, ensure_ascii=False, indent=2) return { @@ -474,7 +552,9 @@ class AgentService: return None - async def get_history(self, session_id: str, limit: int = 100) -> Dict[str, Any]: + async def get_history( + self, session_id: str, limit: int = 100 + ) -> Dict[str, Any]: """ Get conversation history for a session. @@ -496,19 +576,31 @@ class AgentService: for msg in state.values["messages"][-limit:]: messages.append(self._convert_message_to_dict(msg)) - title = state.values.get("conversation_title", "New Conversation") + 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.""" return convert_langchain_to_openai(msg) - async def list_sessions(self, user_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: + async def list_sessions( + self, user_id: Optional[str] = None, limit: int = 100 + ) -> List[Dict[str, Any]]: """ List chat sessions for this project. @@ -542,7 +634,9 @@ class AgentService: repo = ChatSessionsRepository(self._checkpointer_conn) return await repo.delete_session(session_id) - async def rename_session(self, session_id: str, new_title: str) -> Optional[Dict[str, Any]]: + async def rename_session( + self, session_id: str, new_title: str + ) -> Optional[Dict[str, Any]]: """ Rename a chat session. @@ -557,10 +651,14 @@ class AgentService: await self._get_checkpointer() repo = ChatSessionsRepository(self._checkpointer_conn) - session = await repo.update_session(thread_id=session_id, title=new_title) + session = await repo.update_session( + thread_id=session_id, title=new_title + ) return session.to_dict() if session else None - async def pin_session(self, session_id: str, pinned: bool = True) -> Optional[Dict[str, Any]]: + async def pin_session( + self, session_id: str, pinned: bool = True + ) -> Optional[Dict[str, Any]]: """ Pin or unpin a chat session. @@ -586,7 +684,10 @@ class AgentService: if self._checkpointer_conn: try: await self._checkpointer_conn.close() - log.debug("Checkpointer connection closed for: %s", self.project_path) + log.debug( + "Checkpointer connection closed for: %s", + self.project_path, + ) except Exception as e: log.warning("Error closing checkpointer connection: %s", e) finally: diff --git a/gns3server/agent/gns3_copilot/chat_sessions_repository.py b/gns3server/agent/gns3_copilot/chat_sessions_repository.py index af07f4db7..6acdd2e17 100644 --- a/gns3server/agent/gns3_copilot/chat_sessions_repository.py +++ b/gns3server/agent/gns3_copilot/chat_sessions_repository.py @@ -119,7 +119,11 @@ 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. @@ -146,11 +150,15 @@ class ChatSessionsRepository: await self.conn.commit() session_id = cursor.lastrowid - log.info("Created chat session: id=%s, thread_id=%s", session_id, thread_id) + log.info( + "Created chat session: id=%s, thread_id=%s", session_id, thread_id + ) return await self.get_session_by_id(session_id) - async def get_session_by_id(self, session_id: int) -> Optional[ChatSession]: + async def get_session_by_id( + self, session_id: int + ) -> Optional[ChatSession]: """ Get a session by its database ID. @@ -160,14 +168,18 @@ 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: return self._row_to_session(row) return None - async def get_session_by_thread(self, thread_id: str) -> Optional[ChatSession]: + async def get_session_by_thread( + self, thread_id: str + ) -> Optional[ChatSession]: """ Get a session by thread_id. @@ -177,7 +189,9 @@ 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: @@ -185,7 +199,10 @@ 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. @@ -288,7 +305,10 @@ class ChatSessionsRepository: params.append(now) params.append(thread_id) - query = f"UPDATE chat_sessions SET {', '.join(updates)} WHERE thread_id = ?" + query = ( + f"UPDATE chat_sessions SET {', '.join(updates)} WHERE thread_id " + f"= ?" + ) await self.conn.execute(query, params) await self.conn.commit() @@ -307,15 +327,21 @@ 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 if deleted: - log.info("Deleted chat session and checkpoints: thread_id=%s", thread_id) + log.info( + "Deleted chat session and checkpoints: thread_id=%s", thread_id + ) return deleted @@ -330,24 +356,37 @@ 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 if deleted_count > 0: - log.info("Deleted %d sessions for project: %s", deleted_count, project_id) + log.info( + "Deleted %d sessions for project: %s", + deleted_count, + project_id, + ) return deleted_count - async def pin_session(self, thread_id: str, pinned: bool = True) -> Optional[ChatSession]: + async def pin_session( + self, thread_id: str, pinned: bool = True + ) -> Optional[ChatSession]: """ Pin or unpin a session. @@ -360,12 +399,17 @@ class ChatSessionsRepository: """ now = datetime.utcnow().isoformat() await self.conn.execute( - "UPDATE chat_sessions SET pinned = ?, updated_at = ? WHERE thread_id = ?", + "UPDATE chat_sessions SET pinned = ?, updated_at = ? WHERE " + "thread_id = ?", (1 if pinned else 0, now, thread_id), ) await self.conn.commit() - log.debug("Session pin status updated: thread_id=%s, pinned=%s", thread_id, pinned) + log.debug( + "Session pin status updated: thread_id=%s, pinned=%s", + thread_id, + pinned, + ) return await self.get_session_by_thread(thread_id) def _row_to_session(self, row) -> ChatSession: diff --git a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py index fb3a0d25b..a2144b2b9 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py @@ -30,7 +30,8 @@ This module provides factory functions for creating Gns3Connector instances with JWT token authentication and context-aware configuration management. Features: -- Context variable based request-scoped data management (JWT tokens, LLM config) +- Context variable based request-scoped data management (JWT tokens, LLM + config) - Auto-detection of GNS3 server URL from Controller/Config - Fallback URL strategy for flexible deployment - LLM configuration retrieval for users @@ -49,7 +50,9 @@ from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( ) # Local imports -from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector +from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import ( + Gns3Connector, +) logger = logging.getLogger(__name__) @@ -69,7 +72,10 @@ def _get_url_from_controller() -> Optional[str]: controller = Controller.instance() local_compute = controller.get_compute("local") - url = f"{local_compute.protocol}://{local_compute.host}:{local_compute.port}" + url = ( + f"{local_compute.protocol}://{local_compute.host}:" + f"{local_compute.port}" + ) logger.debug( "Got GNS3 URL from Controller: %s (protocol=%s, host=%s, port=%s)", url, @@ -88,7 +94,9 @@ def _get_url_from_controller() -> Optional[str]: logger.debug("Local compute not found in Controller: %s", str(e)) return None except Exception as e: - logger.warning("Unexpected error getting URL from Controller: %s", str(e)) + logger.warning( + "Unexpected error getting URL from Controller: %s", str(e) + ) return None @@ -102,7 +110,10 @@ def _get_url_from_config() -> Optional[str]: from gns3server.config import Config server_config = Config.instance().settings.Server - url = f"{server_config.protocol.value}://{server_config.host}:{server_config.port}" + url = ( + f"{server_config.protocol.value}://{server_config.host}:" + f"{server_config.port}" + ) logger.debug( "Got GNS3 URL from Config: %s (protocol=%s, host=%s, port=%s)", url, @@ -122,7 +133,9 @@ def _get_url_from_config() -> Optional[str]: return None -def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = None) -> Optional[Gns3Connector]: +def get_gns3_connector( + jwt_token: Optional[str] = None, url: Optional[str] = None +) -> Optional[Gns3Connector]: """Create and return a Gns3Connector instance with JWT authentication. URL Resolution Strategy (in order): @@ -132,7 +145,8 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non 4. Fallback to DEFAULT_GNS3_URL (http://127.0.0.1:3080) Args: - jwt_token: JWT token for authentication (optional, will be retrieved from context if not provided) + jwt_token: JWT token for authentication (optional, will be retrieved + from context if not provided) url: GNS3 server URL (optional, auto-detected if not provided) Returns: @@ -181,9 +195,9 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non logger.debug("Config not available, using default URL") url = DEFAULT_GNS3_URL logger.warning( - "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.", + "Using fallback default URL: %s. This may not be " + "correct if your GNS3 server is configured " + "differently. Consider providing the URL explicitly.", url, ) else: @@ -205,11 +219,15 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non return connector except Exception as e: - logger.error("Failed to create Gns3Connector: %s", str(e), exc_info=True) + 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, app=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. @@ -221,7 +239,8 @@ async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optio 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) + app: FastAPI application instance (optional, for direct database + access) Returns: Dictionary with keys: @@ -261,7 +280,9 @@ async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optio url = _detect_url_for_api() # Step 3: Get LLM config - llm_config = get_llm_config(user_id=user_id, jwt_token=jwt_token, app=app) + 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}") @@ -269,8 +290,8 @@ async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optio return {"connector": connector, "llm_config": None} logger.info( - f"Successfully initialized GNS3 connector and LLM config for user {user_id}: " - f"connector_url={connector.url}, " + f"Successfully initialized GNS3 connector and LLM config for " + f"user {user_id}: connector_url={connector.url}, " f"llm_provider={llm_config.get('provider')}, " f"llm_model={llm_config.get('model')}" ) @@ -278,7 +299,9 @@ async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optio 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) + logger.error( + f"Failed to get GNS3 connector with LLM config: {e}", exc_info=True + ) return None @@ -317,8 +340,8 @@ def get_gns3_server_host() -> str: """ Get GNS3 server hostname from Controller or Config. - This is a convenience function for extracting the hostname only, - useful for Nornir tools that need the GNS3 server address. + This is a convenience function for extracting the hostname only, useful + for Nornir tools that need the GNS3 server address. Uses the same priority order as get_gns3_connector: 1. Controller.instance().compute("local") @@ -329,7 +352,9 @@ def get_gns3_server_host() -> str: Hostname/IP address string Example: - from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host + from gns3server.agent.gns3_copilot.gns3_client import ( + get_gns3_server_host, + ) host = get_gns3_server_host() print(f"GNS3 server host: {host}") @@ -346,7 +371,9 @@ def get_gns3_server_host() -> str: logger.debug("Extracted GNS3 server host: %s from URL: %s", host, url) return host except Exception as e: - logger.warning("Failed to extract host from URL %s: %s, using fallback", url, e) + logger.warning( + "Failed to extract host from URL %s: %s, using fallback", url, e + ) return DEFAULT_GNS3_URL.split("://")[1].split(":")[0] @@ -355,20 +382,24 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]: Get LLM model configuration for a user. This function retrieves the user's LLM configuration from the database. - When `app` is provided, it directly accesses the database (bypassing API restrictions). - Otherwise, it falls back to API call (which may mask group config API keys). + When `app` is provided, it directly accesses the database (bypassing API + restrictions). Otherwise, it falls back to API call (which may mask + group config API keys). Args: user_id: User UUID (can be string or UUID object) jwt_token: JWT token for authentication - app: FastAPI application instance (optional, for direct database access) + app: FastAPI application instance (optional, for direct database + access) Returns: - Dictionary with LLM configuration keys (provider, model, api_key, etc.), - or None if not found. + Dictionary with LLM configuration keys (provider, model, api_key, + etc.), or None if not found. Example: - from gns3server.agent.gns3_copilot.gns3_client import get_llm_config + from gns3server.agent.gns3_copilot.gns3_client import ( + get_llm_config, + ) config = get_llm_config(user_id, jwt_token) if config: @@ -393,7 +424,9 @@ 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 @@ -405,22 +438,33 @@ 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 - return loop.run_until_complete(get_user_llm_config_with_app(user_id, app)) + return loop.run_until_complete( + get_user_llm_config_with_app(user_id, app) + ) except RuntimeError: # No event loop exists - create a new one loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: - return loop.run_until_complete(get_user_llm_config_with_app(user_id, app)) + return loop.run_until_complete( + get_user_llm_config_with_app(user_id, app) + ) finally: loop.close() - # Fallback: No app provided, try API call (will mask group config API keys) - logger.warning("No app provided for get_llm_config, group config API keys may be masked") + # Fallback: No app provided, try API call (will mask group config API + # keys) + logger.warning( + "No app provided for get_llm_config, group config API keys may be " + "masked" + ) return None except Exception as e: diff --git a/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py b/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py index 424b9e2ae..73e275767 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py +++ b/gns3server/agent/gns3_copilot/gns3_client/context_helpers.py @@ -62,8 +62,12 @@ logger = logging.getLogger(__name__) # Context variables for request-scoped data # Automatically cleaned up when request context ends -_jwt_token_context: ContextVar[Optional[str]] = ContextVar("_jwt_token_context", default=None) -_llm_config_context: ContextVar[Optional[dict]] = ContextVar("_llm_config_context", default=None) +_jwt_token_context: ContextVar[Optional[str]] = ContextVar( + "_jwt_token_context", default=None +) +_llm_config_context: ContextVar[Optional[dict]] = ContextVar( + "_llm_config_context", default=None +) def set_current_jwt_token(token: str) -> None: @@ -94,10 +98,15 @@ def set_current_llm_config(config: dict) -> None: """Set the LLM config for the current request context. Args: - config: LLM configuration dictionary with provider, model, api_key, etc. + 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,7 +118,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") + "LLM config retrieved from context: provider=%s, model=%s", + config.get("provider"), + config.get("model"), ) else: logger.warning("LLM config not found in context") diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index 3540b4cfb..80f4df10d 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -26,12 +26,14 @@ """ Adapted gns3fy module for GNS3-Copilot -This module is based on the upstream gns3fy project (https://github.com/davidban77/gns3fy). +This module is based on the upstream gns3fy project +(https://github.com/davidban77/gns3fy). Modifications made for GNS3-Copilot: -- Adjusted pydantic usages and dataclass configuration to reduce dependency conflicts - with langchain (pydantic version/api differences) -- Kept the original API surface where possible but simplified validators/config +- Adjusted pydantic usages and dataclass configuration to reduce dependency + conflicts with langchain (pydantic version/api differences) +- Kept the original API surface where possible but simplified + validators/config - Added JWT token authentication support - Integrated with context-aware connector factory @@ -124,11 +126,21 @@ class Gns3Connector: ```python >>> # API v2 with basic auth - >>> server = Gns3Connector(url="http://
:3080", user="admin", cred="password", api_version=2) + >>> server = Gns3Connector( + ... url="http://:3080", user="admin", cred="password", + ... api_version=2 + ... ) >>> # API v3 with username/password (auto-fetches JWT token) - >>> server = Gns3Connector(url="http://:3080", user="admin", cred="password", api_version=3) + >>> server = Gns3Connector( + ... url="http://:3080", user="admin", cred="password", + ... api_version=3 + ... ) >>> # API v3 with direct JWT token - >>> server = Gns3Connector(url="http://:3080", jwt_token="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...", api_version=3) + >>> server = Gns3Connector( + ... url="http://:3080", + ... jwt_token="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...", + ... api_version=3 + ... ) >>> print(server.get_version()) {'local': False, 'version': '2.2.0b4'} ``` @@ -151,7 +163,7 @@ class Gns3Connector: if url is None: raise ValueError("URL is required for Gns3Connector") - self.url = url.strip('/') # Store original URL for reference + self.url = url.strip("/") # Store original URL for reference self.base_url = f"{self.url}/v{api_version}" self.user = user self.cred = cred @@ -160,7 +172,8 @@ class Gns3Connector: self.api_calls = 0 # v3 authentication attributes - # If jwt_token is provided directly, use it; otherwise will be fetched via username/password + # If jwt_token is provided directly, use it; otherwise will be + # fetched via username/password self.access_token = jwt_token self.token_expiry = None self.auth_type = "basic" if api_version == 2 else "jwt" @@ -177,11 +190,17 @@ 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: - self.session.headers["Authorization"] = f"Bearer {self.access_token}" + self.session.headers["Authorization"] = ( + f"Bearer {self.access_token}" + ) def _authenticate_v3(self) -> None: """ @@ -193,10 +212,15 @@ 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" + auth_url = ( + f"{self.base_url.replace('/v3', '')}/v3/access/users/authenticate" + ) auth_data = {"username": self.user, "password": self.cred} # Use temporary session for authentication @@ -204,15 +228,22 @@ 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"] # Update session with new token - self.session.headers["Authorization"] = f"Bearer {self.access_token}" + 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} - " + f"{response.text}" + ) except Exception as e: raise HTTPError(f"v3 API authentication error: {str(e)}") from e @@ -226,7 +257,9 @@ 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) @@ -255,7 +288,12 @@ 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) @@ -298,10 +336,15 @@ class Gns3Connector: try: # Only attempt parsing when Content-Type is JSON - if "application/json" in response.headers.get("Content-Type", "").lower(): + if ( + "application/json" + in response.headers.get("Content-Type", "").lower() + ): error_json = response.json() status = error_json.get("status", "Unknown Status") - message = error_json.get("message", "No message provided in JSON.") + message = error_json.get( + "message", "No message provided in JSON." + ) # Construct a more descriptive new error new_err = HTTPError( f"{status}: {message} (Original {response.status_code} Error)", @@ -323,7 +366,9 @@ 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: @@ -333,7 +378,9 @@ 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']} -- " @@ -355,10 +402,14 @@ class Gns3Connector: """ Returns the list of the projects on the server """ - response = self.http_call("get", url=f"{self.base_url}/projects").json() + 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 @@ -367,18 +418,24 @@ 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: - return next(p for p in self.get_projects() if p["name"] == name) + return next( + p for p in self.get_projects() if p["name"] == name + ) except StopIteration: # Project not found return None 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: @@ -412,10 +469,14 @@ class Gns3Connector: """ Returns the templates defined on the server. """ - _response_data = self.http_call("get", url=f"{self.base_url}/templates").json() + _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 @@ -424,11 +485,15 @@ 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: - return next(t for t in self.get_templates() if t["name"] == name) + return next( + t for t in self.get_templates() if t["name"] == name + ) except StopIteration: # Template name not found return None @@ -452,14 +517,17 @@ class Gns3Connector: **Optional Attributes (can be passed via kwargs):** - - `tags` (list): List of tags for the template (e.g., ["device_type:cisco_ios_telnet", "platform:cisco_ios"]) + - `tags` (list): List of tags for the template (e.g., + ["device_type:cisco_ios_telnet", "platform:cisco_ios"]) - Any other template attributes supported by GNS3 API """ # Get existing template _template = self.get_template(name=name, template_id=template_id) # Type check: handle case where get_template might return None if _template is None: - raise ValueError(f"Template not found (name={name}, id={template_id})") + raise ValueError( + f"Template not found (name={name}, id={template_id})" + ) # Update local dictionary and send request _template.update(**kwargs) @@ -485,7 +553,8 @@ class Gns3Connector: **Optional Attributes (can be passed via kwargs):** - - `tags` (list): List of tags for the template (e.g., ["device_type:cisco_ios_telnet", "platform:cisco_ios"]) + - `tags` (list): List of tags for the template (e.g., + ["device_type:cisco_ios_telnet", "platform:cisco_ios"]) - Any other template attributes supported by GNS3 API **Example:** @@ -498,10 +567,13 @@ class Gns3Connector: ... ) ``` """ - # kwargs["name"] might raise KeyError at runtime, for more robust code we can use get first + # kwargs["name"] might raise KeyError at runtime, for more robust + # code we can use get first template_name = kwargs.get("name") if not template_name: - raise ValueError("Attribute 'name' is required to create a template") + raise ValueError( + "Attribute 'name' is required to create a template" + ) # Check if template already exists _template = self.get_template(name=kwargs["name"]) @@ -513,11 +585,15 @@ 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/ @@ -538,9 +614,13 @@ 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}") + self.http_call( + "delete", url=f"{self.base_url}/templates/{template_id}" + ) def get_nodes(self, project_id: str) -> list[dict[str, Any]]: """ @@ -550,7 +630,9 @@ 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) @@ -575,7 +657,9 @@ 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) @@ -650,7 +734,9 @@ 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. @@ -669,7 +755,9 @@ 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. @@ -733,9 +821,15 @@ 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": @@ -814,7 +908,9 @@ class Link: @field_validator("filters") @classmethod - def _valid_filters(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: + def _valid_filters( + cls, value: dict[str, Any] | None + ) -> dict[str, Any] | None: if type(value) is not dict and value is not None: raise ValueError(f"Not a valid filters - {value}") return value @@ -870,7 +966,9 @@ 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}" @@ -896,7 +994,12 @@ 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) @@ -928,7 +1031,10 @@ 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/" + f"{self.link_id}" + ) # TODO: Verify that the passed kwargs are supported ones _response = self.connector.http_call("put", _url, json_data=kwargs) @@ -1042,7 +1148,10 @@ class Node: @field_validator("status") @classmethod def _valid_status(cls, value: Any) -> Any: - if value not in ("stopped", "started", "suspended") and value is not None: + if ( + value not in ("stopped", "started", "suspended") + and value is not None + ): raise ValueError(f"Not a valid status - {value}") return value @@ -1068,7 +1177,10 @@ 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/" + f"{self.node_id}" + ) _response = self.connector.http_call("get", _url) # Update object @@ -1094,7 +1206,10 @@ 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 @@ -1136,7 +1251,9 @@ class Node: else: # api_version 3 - _response = _conn.http_call("post", _url, json_data={"additionalProp1": {}}) + _response = _conn.http_call( + "post", _url, json_data={"additionalProp1": {}} + ) # successful response code 204 if _response.status_code in (204,): self.get() @@ -1145,7 +1262,9 @@ class Node: try: error_detail = _response.json() except Exception: - error_detail = getattr(_response, "text", "No response body") + error_detail = getattr( + _response, "text", "No response body" + ) _msg = ( "Failed to start node: " @@ -1186,7 +1305,9 @@ class Node: return True else: # api_version 3 - _response = _conn.http_call("post", _url, json_data={"additionalProp1": {}}) + _response = _conn.http_call( + "post", _url, json_data={"additionalProp1": {}} + ) # successful response code 204 if _response.status_code in (204,): self.get() @@ -1196,7 +1317,10 @@ 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 @@ -1217,7 +1341,9 @@ class Node: _node_id = self.node_id assert _node_id is not None - _url = f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/reload" + _url = ( + f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/reload" + ) _response = _conn.http_call("post", _url) if "v2" in _url.lower(): # api_version 2 @@ -1235,7 +1361,9 @@ class Node: else: # api_version 3 - _response = _conn.http_call("post", _url, json_data={"additionalProp1": {}}) + _response = _conn.http_call( + "post", _url, json_data={"additionalProp1": {}} + ) # successful response code 204 if _response.status_code in (204,): self.get() @@ -1245,7 +1373,10 @@ 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 @@ -1266,7 +1397,9 @@ class Node: _node_id = self.node_id assert _node_id is not None - _url = f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/suspend" + _url = ( + f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/suspend" + ) _response = _conn.http_call("post", _url) # Update object or perform get if change was not reflected @@ -1335,7 +1468,9 @@ 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'") @@ -1354,9 +1489,16 @@ 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()) @@ -1535,7 +1677,12 @@ 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. @@ -1955,7 +2102,9 @@ class Project: time.sleep(poll_wait_time) self.get_nodes() - def nodes_summary(self, is_print: bool = True) -> list[tuple[Any, ...]] | None: + def nodes_summary( + self, is_print: bool = True + ) -> list[tuple[Any, ...]] | None: """ Returns a summary of the nodes insode the project. If `is_print` is `False`, it will return a list of tuples like: @@ -1974,7 +2123,10 @@ 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 @@ -2010,7 +2162,9 @@ 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 @@ -2036,7 +2190,9 @@ 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: @@ -2067,7 +2223,9 @@ class Project: try: # Add type-safe lookup logic - _node_a = next(x for x in self.nodes if x.node_id == _side_a["node_id"]) + _node_a = next( + x for x in self.nodes if x.node_id == _side_a["node_id"] + ) # Ensure getting str to resolve [return-value] error _port_a = str( next( @@ -2078,7 +2236,9 @@ class Project: ) ) - _node_b = next(x for x in self.nodes if x.node_id == _side_b["node_id"]) + _node_b = next( + x for x in self.nodes if x.node_id == _side_b["node_id"] + ) _port_b = str( next( x["name"] @@ -2116,7 +2276,9 @@ 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`. @@ -2185,13 +2347,20 @@ class Project: if not self.nodes: self.get_nodes() - _node = Node(project_id=self.project_id, connector=self.connector, **kwargs) + _node = Node( + project_id=self.project_id, connector=self.connector, **kwargs + ) _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: + def create_link( + self, node_a: str, port_a: str, node_b: str, port_b: str + ) -> None: """ Creates a link. @@ -2244,7 +2413,9 @@ class Project: ): _matches.append(_l) # pragma: no cover if _matches: - raise ValueError(f"At least one port is used, ID: {_matches[0].link_id}") + raise ValueError( + f"At least one port is used, ID: {_matches[0].link_id}" + ) # Now create the link! _link = Link( @@ -2270,7 +2441,9 @@ class Project: self.links.append(_link) print(f"Created Link-ID: {_link.link_id} -- Type: {_link.link_type}") - def delete_link(self, node_a: str, port_a: str, node_b: str, port_b: str) -> None: + def delete_link( + self, node_a: str, port_a: str, node_b: str, port_b: str + ) -> None: """ Deletes a link. @@ -2324,7 +2497,9 @@ 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] @@ -2332,7 +2507,8 @@ 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 @@ -2361,11 +2537,15 @@ class Project: self.get_snapshots() try: - return next(_p for _p in (self.snapshots or []) if _p[key] == value) + return next( + _p for _p in (self.snapshots or []) if _p[key] == value + ) 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`. @@ -2422,7 +2602,9 @@ 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 @@ -2446,14 +2628,19 @@ 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 @@ -2477,7 +2664,10 @@ 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) @@ -2512,7 +2702,9 @@ class Project: _y = int(radius * (-cos(_angle * index))) n.update(x=_x, y=_y) - def get_drawing(self, drawing_id: str | None = None) -> dict[str, Any] | None: + def get_drawing( + self, drawing_id: str | None = None + ) -> dict[str, Any] | None: """ Returns the drawing by searching for the `svg` or the `drawing_id`. @@ -2528,7 +2720,11 @@ 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 @@ -2633,17 +2829,25 @@ class Project: # Type guard: inform Mypy that self.drawings is now an iterable list # Use or [] with next to find target object current_drawing = next( - (d for d in (self.drawings or []) if d.get("drawing_id") == drawing_id), + ( + d + for d in (self.drawings or []) + if d.get("drawing_id") == drawing_id + ), None, ) if current_drawing is None: - raise ValueError(f"Drawing with ID {drawing_id} not found in project.") + raise ValueError( + f"Drawing with ID {drawing_id} not found in project." + ) # If parameter is None, get original value from current object # This way, Mypy won't report errors for list comprehensions of each field final_svg = svg if svg is not None else current_drawing.get("svg") - final_locked = locked if locked is not None else current_drawing.get("locked") + final_locked = ( + locked if locked is not None else current_drawing.get("locked") + ) final_x = x if x is not None else current_drawing.get("x") final_y = y if y is not None else current_drawing.get("y") final_z = z if z is not None else current_drawing.get("z") @@ -2691,7 +2895,10 @@ 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/" + f"{_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 b0bcf09f4..4422b728d 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py +++ b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py @@ -50,7 +50,8 @@ class GNS3ProjectInfoTool(BaseTool): name: str = "gns3_project_info" description: str = """ - Retrieves basic information of a GNS3 project including name, status, node count and link count. + Retrieves basic information of a GNS3 project including name, status, + node count and link count. Input: `project_id` (str, required): UUID of the GNS3 project. @@ -79,18 +80,26 @@ class GNS3ProjectInfoTool(BaseTool): project_id : The UUID of the specific GNS3 project. Returns: - dict: A dictionary containing project info (name, project_id, node_count, link_count, status), - or an error dictionary if an exception occurs or project_id is not provided. + dict: A dictionary containing project info (name, project_id, node_count, + link_count, status), or an error dictionary if an exception occurs + or project_id is not provided. """ # Log received input - logger.info("Received tool_input: %s, project_id: %s", tool_input, project_id) + logger.info( + "Received tool_input: %s, project_id: %s", tool_input, project_id + ) try: # 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,10 +107,17 @@ 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}") + logger.info( + f"Retrieving project info for project_id: {project_id}" + ) project = Project(project_id=project_id, connector=server) project.get() # Load project details @@ -119,7 +135,13 @@ class GNS3ProjectInfoTool(BaseTool): "status": project.status, "node_count": node_count, "link_count": link_count, - "tuple": (project.name, project.project_id, node_count, link_count, project.status), + "tuple": ( + project.name, + project.project_id, + node_count, + link_count, + project.status, + ), } # Log result 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 7a858cc3e..aa130d129 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py +++ b/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py @@ -27,7 +27,8 @@ GNS3 Topology Reader Tool This module provides a LangChain BaseTool to retrieve the topology of a -specific GNS3 project by project ID. Returns nodes, links, and project metadata. +specific GNS3 project by project ID. Returns nodes, links, and project +metadata. """ @@ -57,7 +58,8 @@ class GNS3TopologyTool(BaseTool): Output: Dictionary with: - `project_id`, `name`, `status`: Project metadata - - `nodes`: Dict of node details (node_id, name, ports, console_port, type, etc.) + - `nodes`: Dict of node details (node_id, name, ports, console_port, + type, etc.) - `links`: List of link connections Use this to understand network structure before making changes. @@ -73,23 +75,31 @@ class GNS3TopologyTool(BaseTool): Synchronous method to retrieve the topology of a specific GNS3 project. Args: - tool_input : Input parameters, typically a dict or Pydantic model containing server_url. - run_manager : Callback manager for tool run. - project_id : The UUID of the specific GNS3 project to retrieve topology from. + tool_input: Input parameters, typically a dict or Pydantic model + containing server_url. + run_manager: Callback manager for tool run. + project_id: The UUID of the specific GNS3 project to retrieve + topology from. Returns: - dict: A dictionary containing the project ID, name, status, nodes, and links, - or an error dictionary if an exception occurs or project_id is not provided. + dict: A dictionary containing the project ID, name, status, nodes, + and links, or an error dictionary if an exception occurs + or project_id is not provided. """ # Log received input - logger.info("Received tool_input: %s, project_id: %s", tool_input, project_id) + logger.info( + "Received tool_input: %s, project_id: %s", tool_input, project_id + ) try: # 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,7 +107,10 @@ 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}") @@ -109,13 +122,16 @@ 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 retrieved: project_id=%s, name=%s, nodes=%d, " + "links=%d", topology.get("project_id"), topology.get("name"), len(topology.get("nodes", {})), @@ -132,11 +148,16 @@ class GNS3TopologyTool(BaseTool): def _clean_nodes_ports(self, data: dict) -> dict: """ Clean and simplify the nodes data structure. - Simplify each node's ports list to only keep name and short_name fields. + + Simplify each node's ports list to only keep name and short_name + fields. """ 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 diff --git a/gns3server/agent/gns3_copilot/project_agent_manager.py b/gns3server/agent/gns3_copilot/project_agent_manager.py index abc6505b2..edc452f32 100644 --- a/gns3server/agent/gns3_copilot/project_agent_manager.py +++ b/gns3server/agent/gns3_copilot/project_agent_manager.py @@ -27,7 +27,8 @@ Project Agent Manager Manages AgentService instances for GNS3 projects using a singleton pattern. -Each project has its own AgentService with a dedicated SQLite checkpoint database. +Each project has its own AgentService with a dedicated SQLite checkpoint +database. """ import asyncio @@ -60,7 +61,9 @@ class ProjectAgentManager: cls._instance._lock = asyncio.Lock() return cls._instance - async def get_agent(self, project_id: str, project_path: str) -> AgentService: + async def get_agent( + self, project_id: str, project_path: str + ) -> AgentService: """ Get or create an AgentService for a project. @@ -73,7 +76,11 @@ class ProjectAgentManager: """ async with self._lock: if project_id not in self._agents: - log.info("Creating new AgentService for project: %s at %s", project_id, project_path) + log.info( + "Creating new AgentService for project: %s at %s", + project_id, + project_path, + ) self._agents[project_id] = AgentService(project_path) return self._agents[project_id] @@ -99,7 +106,10 @@ class ProjectAgentManager: Should be called on server shutdown. """ async with self._lock: - log.info("Closing all AgentService instances (%d projects)", len(self._agents)) + log.info( + "Closing all AgentService instances (%d projects)", + len(self._agents), + ) for project_id, agent in self._agents.items(): log.debug("Closing AgentService for project: %s", project_id) await agent.close() diff --git a/gns3server/agent/gns3_copilot/prompts/__init__.py b/gns3server/agent/gns3_copilot/prompts/__init__.py index 43e810089..4a244d3b1 100644 --- a/gns3server/agent/gns3_copilot/prompts/__init__.py +++ b/gns3server/agent/gns3_copilot/prompts/__init__.py @@ -30,8 +30,8 @@ This package contains system prompts and prompt loading utilities for the GNS3-Copilot AI agent. Available prompts: -- teaching_assistant_prompt: Teaching assistant mode (diagnostics only, no configuration) -- lab_automation_assistant_prompt: Lab automation assistant mode (diagnostics + configuration) +- teaching_assistant_prompt: Teaching assistant mode (diagnostics only) +- lab_automation_assistant_prompt: Lab automation mode (diagnostics + config) """ diff --git a/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py b/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py index 2b8963ad5..84f6ef945 100644 --- a/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py +++ b/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py @@ -52,21 +52,21 @@ You have access to the following tools to help users: | Tool | Purpose | Usage | |------|---------|-------| -| `gns3_template_reader` | Get available node templates | List templates for creating nodes | -| `gns3_create_node` | Create new nodes in topology | Add routers, switches, VPCS, etc. | -| `gns3_link_tool` | Create links between nodes | Connect network topology | -| `gns3_start_node_tool` | Start/stop nodes | Control device power state | +| `gns3_template_reader` | Get available node templates | List templates | +| `gns3_create_node` | Create new nodes in topology | Add routers, switches, VPCS | +| `gns3_link_tool` | Create links between nodes | Connect topology | +| `gns3_start_node_tool` | Start/stop nodes | Control power state | | `gns3_update_node_name_tool` | Update node names | Rename devices | -| `execute_multiple_device_commands` | Execute display commands | Read-only diagnostics (show/display/debug) | -| `execute_multiple_device_config_commands` | Execute configuration commands | Make configuration changes | +| `execute_multiple_device_commands` | Execute display commands | Diagnostics | +| `execute_multiple_device_config_commands` | Execute config commands | Config changes | | `vpcs_multi_commands` | Execute VPCS commands | Configure VPCS devices | --- # TOOL USAGE RULES -1. **Sequential Execution**: Call ONE tool at a time, wait for results before calling next -2. **Topology Awareness**: If topology is already in context, DO NOT call topology reader again +1. **Sequential Execution**: Call ONE tool at a time, wait for results +2. **Topology Awareness**: If topology is in context, DO NOT call reader again 3. **Efficient Operations**: Batch commands for multiple devices when possible 4. **Safety First**: Be cautious with destructive operations (reload, erase, format) @@ -176,7 +176,8 @@ While you have configuration permissions, exercise caution: {{topology_info}} -**Note**: Topology is already retrieved. DO NOT call topology reader again unless needed. +**Note**: Topology is already retrieved. DO NOT call topology reader again unless +needed. --- """ diff --git a/gns3server/agent/gns3_copilot/prompts/prompt_loader.py b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py index b6f5917ab..dcc69df2e 100644 --- a/gns3server/agent/gns3_copilot/prompts/prompt_loader.py +++ b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py @@ -30,8 +30,10 @@ This module provides utilities for loading system prompts. Supports multiple prompt variants based on LLM model configuration. Available Modes (controlled by config.copilot_mode in llm_model_configs): -- "teaching_assistant" (default): Teaching assistant mode - diagnostics only, no configuration -- "lab_automation_assistant": Full lab automation assistant mode - diagnostics and configuration enabled +- "teaching_assistant" (default): Teaching assistant mode - diagnostics only, + no configuration +- "lab_automation_assistant": Full lab automation assistant mode - diagnostics + and configuration enabled """ @@ -47,18 +49,25 @@ def load_system_prompt(llm_config: dict | None = None) -> str: """ Load the system prompt for GNS3-Copilot. - The prompt mode is controlled by the `copilot_mode` field in the LLM model config: - - "teaching_assistant" (default): Teaching assistant mode - diagnostics only, no configuration - - "lab_automation_assistant": Full lab automation assistant mode - diagnostics and configuration enabled + The prompt mode is controlled by the `copilot_mode` field in the LLM + model config: + - "teaching_assistant" (default): Teaching assistant mode - diagnostics + only, no configuration + - "lab_automation_assistant": Full lab automation assistant mode - + diagnostics and configuration enabled Args: - llm_config: LLM model configuration dictionary (flattened structure from get_user_llm_config_full) + llm_config: LLM model configuration dictionary (flattened structure + from get_user_llm_config_full) Returns: str: The system prompt string. """ if not llm_config: - logger.info("No LLM config provided, using default TEACHING_ASSISTANT prompt mode") + logger.info( + "No LLM config provided, using default TEACHING_ASSISTANT " + "prompt mode" + ) return TEACHING_ASSISTANT_PROMPT # llm_config is a flattened dict with copilot_mode at the top level @@ -66,7 +75,10 @@ def load_system_prompt(llm_config: dict | None = None) -> str: mode = llm_config.get("copilot_mode", "teaching_assistant").lower() if mode == "lab_automation_assistant": - logger.info("Using LAB_AUTOMATION_ASSISTANT prompt mode (diagnostics + configuration)") + logger.info( + "Using LAB_AUTOMATION_ASSISTANT prompt mode (diagnostics + " + "configuration)" + ) return LAB_AUTOMATION_ASSISTANT_PROMPT else: logger.info("Using TEACHING_ASSISTANT prompt mode (diagnostics only)") diff --git a/gns3server/agent/gns3_copilot/prompts/teaching_assistant_prompt.py b/gns3server/agent/gns3_copilot/prompts/teaching_assistant_prompt.py index e65b95a68..cb12e9ace 100644 --- a/gns3server/agent/gns3_copilot/prompts/teaching_assistant_prompt.py +++ b/gns3server/agent/gns3_copilot/prompts/teaching_assistant_prompt.py @@ -51,9 +51,11 @@ You are a **GNS3 Lab Teaching Assistant**. 1. ❌ **NEVER** call `execute_multiple_device_config_commands` 2. ❌ **NEVER** say "I've configured..." / "Configuration complete" -3. ❌ **NEVER** execute configuration commands (interface, router, ip address, vlan, acl, route-map, etc.) +3. ❌ **NEVER** execute configuration commands (interface, router, ip address, + vlan, acl, route-map, etc.) -**Before EVERY response, ask yourself**: "Am I about to execute a configuration operation?" +**Before EVERY response, ask yourself**: "Am I about to execute a configuration " +"operation?" → If YES → Stop and provide guidance instead → If NO → Proceed with diagnosis @@ -119,5 +121,6 @@ ip route, ip addr, tcpdump, ping, traceroute {{topology_info}} -**Note**: Topology is already retrieved. DO NOT call topology reader again unless needed. +**Note**: Topology is already retrieved. DO NOT call topology reader again unless +needed. """ diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index a24d42ab5..d81acb873 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -26,7 +26,7 @@ GNS3-Copilot Tools Package -This package provides various tools for interacting with GNS3 network simulator, including: +This package provides various tools for interacting with GNS3 network simulator: - Device configuration command execution - Display command execution - Multiple device command execution using Nornir @@ -36,7 +36,7 @@ This package provides various tools for interacting with GNS3 network simulator, Main modules: - config_tools_nornir: Multiple device configuration command execution tool using Nornir - display_tools_nornir: Multiple device command execution tool using Nornir -- vpcs_tools_telnetlib3: VPCS device configuration tool using telnetlib3 (concurrent execution) +- vpcs_tools_telnetlib3: VPCS device configuration tool using telnetlib3 - gns3_create_node: GNS3 node creation tool - gns3_create_link: GNS3 link creation tool - gns3_start_node: GNS3 node startup tool 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 4d3ecb2cc..cff664baf 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -24,8 +24,8 @@ # """ -This module provides a tool to execute configuration commands on multiple devices - in a GNS3 topology using Nornir. +This module provides a tool to execute configuration commands on multiple + devices in a GNS3 topology using Nornir. """ import json @@ -45,14 +45,17 @@ from nornir_netmiko.tasks import netmiko_send_config from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology -from gns3server.agent.gns3_copilot.utils.command_filter import filter_forbidden_commands +from gns3server.agent.gns3_copilot.utils.command_filter import ( + filter_forbidden_commands, +) # config log logger = logging.getLogger(__name__) # Suppress nornir INFO logs in console (reduce verbosity) -# The logging={"enabled": False} in InitNornir only disables plugin internal logs, -# but nornir.core still logs task execution at INFO level. Set to WARNING to suppress these. +# The logging={"enabled": False} in InitNornir only disables plugin +# internal logs, but nornir.core still logs task execution at INFO level. +# Set to WARNING to suppress these. logging.getLogger("nornir.core").setLevel(logging.WARNING) logging.getLogger("nornir").setLevel(logging.WARNING) @@ -63,7 +66,9 @@ def _get_nornir_defaults() -> dict[str, Any]: 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. @@ -80,12 +85,16 @@ def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: s "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. @@ -99,26 +108,31 @@ 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): """ - A tool to execute configuration commands on multiple devices in a GNS3 topology using Nornir. - This class uses Nornir to manage connections and execute configuration commands - on multiple devices concurrently. + A tool to execute configuration commands on multiple devices in a GNS3 + topology using Nornir. + This class uses Nornir to manage connections and execute configuration + commands on multiple devices concurrently. IMPORTANT SAFETY NOTE: - This tool is intended for configuration changes only. Use extreme caution when - executing configuration commands. + This tool is intended for configuration changes only. Use extreme caution + when executing configuration commands. """ name: str = "execute_multiple_device_config_commands" description: str = """ - Executes CONFIGURATION commands on multiple devices in the current GNS3 topology. - Use this tool ONLY for changing device settings (e.g., 'configure', 'interface', 'ip address', 'router ospf'). + Executes CONFIGURATION commands on multiple devices in the current GNS3 + topology. + Use this tool ONLY for changing device settings (e.g., 'configure', + 'interface', 'ip address', 'router ospf'). For viewing information, use the 'execute_multiple_device_commands' tool. - Input should be a JSON object containing project_id and device configurations. + Input should be a JSON object containing project_id and device configs. Example input: { "project_id": "