NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-08-27 12:30:13 +03:00
Code Issues Packages Projects Releases Wiki Activity

docs: add temperature parameter documentation and code formatting improvements

Browse Source 
- Added `temperature` parameter to Chat API documentation with implementation notes
- Improved code formatting in context_manager.py with consistent string quotes and line breaks
- Added section on future runtime LLM parameter override capabilities
- Updated API schemas to include temperature parameter (currently unused but reserved for future implementation)
This commit is contained in:
YueGuobin 2026-03-05 20:58:14 +08:00
parent 7f8364fbf5
commit 2485bfec46
32 changed files with 634 additions and 918 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
docs/gns3-copilot/ai-chat-api-design.md
View File

@ -301,6 +301,7 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。
- message: 用户消息内容
- session_id: 会话 ID（可选，不提供则自动创建新会话）
- stream: 是否启用流式响应（默认 true）
- temperature: LLM temperature 参数（注意：当前未使用，保留以备将来实现。实际 temperature 从用户的数据库 LLM 配置中读取）
- mode: 交互模式（当前仅支持 "text"）
**响应**：SSE 流，包含多种类型的消息（见上文消息格式）
@ -366,6 +367,7 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。
- message: str - 用户消息内容
- session_id: Optional[str] - 会话 ID（可选）
- stream: bool - 是否流式响应（默认 true）
- temperature: Optional[float] - LLM temperature 参数（注意：当前未使用，保留以备将来实现运行时覆盖。当前 temperature 从用户的数据库 LLM 配置中读取）
- mode: Literal["text"] - 交互模式
### ChatSession
@ -635,6 +637,31 @@ agent_manager.remove_agent(project_id)
### 未来可能的扩展
#### 运行时 LLM 参数覆盖
当前 LLM 配置（包括 temperature、max_tokens 等）从用户的数据库配置中读取。将来可以支持在请求时覆盖这些参数：
**实现方案**：
```python
# 在 chat.py 的 stream_chat 函数中
if request.temperature is not None:
llm_config["temperature"] = str(request.temperature)
if request.max_tokens is not None:
llm_config["max_tokens"] = str(request.max_tokens)
```
**当前状态**：
- `temperature` 参数已添加到 ChatRequest schema，但未实现覆盖逻辑
- 参数保留在 API 中以保持向后兼容性
- 代码中已添加 TODO 注释标记实现位置
**注意事项**：
- 需要验证参数范围（如 temperature: 0.0-2.0）
- 需要考虑是否记录覆盖值到统计信息
- 需要在前端 UI 中提供相应的设置选项
#### 其他扩展方向
- 多模态支持（图片、文件）
- 语音输入/输出
- 多人协作会话

105
gns3server/agent/gns3_copilot/agent/context_manager.py
View File

@ -44,11 +44,13 @@ Requirements:
import json
import logging
import warnings
from typing import Any, Callable
from typing import Any
from typing import Callable
import tiktoken
from langchain_core.messages import BaseMessage, SystemMessage, trim_messages
from langchain_core.messages import BaseMessage
from langchain_core.messages import SystemMessage
from langchain_core.messages import trim_messages
logger = logging.getLogger(__name__)
@ -81,6 +83,7 @@ TOKENS_PER_K = 1000
# Token Counting Functions
# ============================================================================
def count_tokens(text: str) -> int:
"""
Count tokens in text using tiktoken.
@ -126,11 +129,11 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
"function": {
"name": tool.name,
"description": tool.description,
}
},
}
# Add parameters schema if available
if hasattr(tool, 'args_schema') and tool.args_schema:
if hasattr(tool, "args_schema") and tool.args_schema:
try:
# Try Pydantic v2 method (model_json_schema)
tool_schema["function"]["parameters"] = tool.args_schema.model_json_schema()
@ -141,15 +144,15 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
except Exception:
# Both methods failed, use empty schema
logger.debug(
"Failed to get schema for tool %s, using empty parameters",
getattr(tool, 'name', 'unknown')
"Failed to get schema for tool %s, using empty parameters", getattr(tool, "name", "unknown")
)
tool_schema["function"]["parameters"] = {}
except Exception as e:
# model_json_schema() raised an exception
logger.debug(
"model_json_schema() failed for tool %s: %s, trying v1 fallback",
getattr(tool, 'name', 'unknown'), e
getattr(tool, "name", "unknown"),
e,
)
try:
tool_schema["function"]["parameters"] = tool.args_schema.schema()
@ -182,7 +185,7 @@ def _count_tokens_for_message(message: BaseMessage) -> int:
Estimated token count for the message
"""
content = ""
if hasattr(message, 'content') and message.content:
if hasattr(message, "content") and message.content:
content = str(message.content)
return count_tokens(content)
@ -192,6 +195,7 @@ def _count_tokens_for_message(message: BaseMessage) -> int:
# Pre-Model Hook Factory
# ============================================================================
def create_pre_model_hook(
system_prompt: str,
get_topology_func: Callable[[dict], Any] | None = None,
@ -246,6 +250,7 @@ def create_pre_model_hook(
- In ReAct loops with multiple LLM calls, topology is re-injected each time
- This is acceptable for GNS3-Copilot's usage patterns (low concurrency, short conversations)
"""
def pre_model_hook(state: dict) -> dict:
"""
LangGraph pre_model_hook - called before each LLM invocation.
@ -284,8 +289,7 @@ def create_pre_model_hook(
if "context_limit" not in llm_config:
logger.error(
"context_limit not found in LLM config. "
"This is a required field. Please configure context_limit."
"context_limit not found in LLM config. " "This is a required field. Please configure context_limit."
)
raise ValueError("context_limit is required in LLM config")
@ -313,10 +317,7 @@ def create_pre_model_hook(
# Step 2: Inject topology into system prompt
messages_with_system = _inject_topology_into_system(
messages=messages,
system_prompt=system_prompt,
state=state,
get_topology_func=get_topology_func
messages=messages, system_prompt=system_prompt, state=state, get_topology_func=get_topology_func
)
# Step 3: Calculate token breakdown for logging and validation
@ -324,9 +325,7 @@ def create_pre_model_hook(
system_tokens = _count_tokens_for_message(system_message)
# Calculate tokens for messages_with_system (including system)
messages_with_system_tokens = sum(
_count_tokens_for_message(m) for m in messages_with_system
)
messages_with_system_tokens = sum(_count_tokens_for_message(m) for m in messages_with_system)
# Calculate available budget
model_limit_tokens = context_limit_k * TOKENS_PER_K
@ -345,21 +344,29 @@ def create_pre_model_hook(
"This will likely cause LLM call failures. "
"Recommendations: 1) Reduce system prompt length, 2) Reduce number of tools, "
"3) Use a model with larger context window, or 4) Switch to 'conservative' strategy.",
system_tokens, tool_tokens, max_input_tokens
system_tokens,
tool_tokens,
max_input_tokens,
)
elif max_tokens_for_trim < system_tokens * 1.5:
# Less than 1.5x system tokens means very little room for history
logger.warning(
"System prompt (%d tokens) + tools (%d tokens) leave minimal room for conversation history (%d tokens remaining). "
"Consider reducing system prompt length or number of tools.",
system_tokens, tool_tokens, max_tokens_for_trim - system_tokens
system_tokens,
tool_tokens,
max_tokens_for_trim - system_tokens,
)
# Debug log with clear terminology
logger.debug(
"Token breakdown: system=%d, all_messages=%d (system+history), tools=%d, trim_budget=%d (limit=%dK, strategy=%s)",
system_tokens, messages_with_system_tokens, tool_tokens, max_tokens_for_trim,
context_limit_k, strategy
system_tokens,
messages_with_system_tokens,
tool_tokens,
max_tokens_for_trim,
context_limit_k,
strategy,
)
# Step 4: Trim messages to fit
@ -380,16 +387,25 @@ def create_pre_model_hook(
if len(trimmed) < len(messages_with_system):
logger.info(
"Messages trimmed: %d → %d msgs. Total: ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s",
len(messages_with_system), len(trimmed),
final_total, tool_tokens, final_total + tool_tokens,
context_limit_k, usage_percent, strategy
len(messages_with_system),
len(trimmed),
final_total,
tool_tokens,
final_total + tool_tokens,
context_limit_k,
usage_percent,
strategy,
)
else:
logger.info(
"Context ready: %d msgs, ~%d tokens + %d tools = %d / %dK (%.1f%%), strategy=%s",
len(trimmed), final_total, tool_tokens,
final_total + tool_tokens, context_limit_k,
usage_percent, strategy
len(trimmed),
final_total,
tool_tokens,
final_total + tool_tokens,
context_limit_k,
usage_percent,
strategy,
)
return {"messages": trimmed}
@ -433,19 +449,15 @@ def _inject_topology_into_system(
if topology_data:
topology_str = str(topology_data)
formatted_prompt = system_prompt.replace(
"{{topology_info}}",
f"\n\n## Current Topology\n{topology_str}"
)
logger.info("✓ Topology injected: %d chars, nodes: %s",
len(topology_str),
list(topology_data.get("nodes", {}).keys())[:5]) # Show first 5 node names
formatted_prompt = system_prompt.replace("{{topology_info}}", f"\n\n## Current Topology\n{topology_str}")
logger.info(
"✓ Topology injected: %d chars, nodes: %s",
len(topology_str),
list(topology_data.get("nodes", {}).keys())[:5],
) # Show first 5 node names
logger.debug("Full topology data: %s", topology_str[:500]) # First 500 chars
else:
formatted_prompt = system_prompt.replace(
"{{topology_info}}",
"(No topology information available)"
)
formatted_prompt = system_prompt.replace("{{topology_info}}", "(No topology information available)")
logger.warning("✗ Topology data is None, injecting placeholder")
# Filter out existing SystemMessage instances
@ -462,6 +474,7 @@ def _inject_topology_into_system(
# Legacy Compatibility (Deprecated)
# ============================================================================
def prepare_context_messages(
state_messages: list[Any],
system_prompt: str,
@ -482,15 +495,9 @@ def prepare_context_messages(
if "{{topology_info}}" not in system_prompt:
formatted_prompt = system_prompt
elif topology_context:
formatted_prompt = system_prompt.replace(
"{{topology_info}}",
f"\n\n## Current Topology\n{topology_context}"
)
formatted_prompt = system_prompt.replace("{{topology_info}}", f"\n\n## Current Topology\n{topology_context}")
else:
formatted_prompt = system_prompt.replace(
"{{topology_info}}",
"(No topology information available)"
)
formatted_prompt = system_prompt.replace("{{topology_info}}", "(No topology information available)")
return [SystemMessage(content=formatted_prompt)] + state_messages
@ -506,7 +513,7 @@ if __name__ == "__main__":
# Test token counting
print("Test 1: Token Counting")
print(f" tiktoken encoding: cl100k_base")
print(" tiktoken encoding: cl100k_base")
test_text = "Hello world 你好世界"
tokens = count_tokens(test_text)
@ -537,7 +544,7 @@ if __name__ == "__main__":
print("\nTest 3: Invoke pre_model_hook")
test_state = {
"messages": [HumanMessage(f"Message {i}: {'x' * 50}") for i in range(5)],
"topology_info": {"project_id": "test123", "nodes": 3}
"topology_info": {"project_id": "test123", "nodes": 3},
}
result = hook(test_state)

89
gns3server/agent/gns3_copilot/agent/gns3_copilot.py
View File

@ -43,43 +43,40 @@ The agent provides:
# Standard library imports
import logging
import operator
import sys
from pathlib import Path
from typing import Annotated, Literal
from typing import Annotated
from typing import Literal
# Third-party imports
from langchain.messages import AnyMessage, SystemMessage, ToolMessage
from langchain.messages import AnyMessage
from langchain.messages import SystemMessage
from langchain.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from langgraph.graph import END
from langgraph.graph import START
from langgraph.graph import StateGraph
from langgraph.managed.is_last_step import RemainingSteps
from typing_extensions import TypedDict
# Add backend to path for prompt_manager
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "backend"))
# Local imports
from gns3server.agent.gns3_copilot.agent.context_manager import (
create_pre_model_hook,
)
from gns3server.agent.gns3_copilot.agent.context_manager import create_pre_model_hook
from gns3server.agent.gns3_copilot.agent.model_factory import (
create_base_model_with_tools,
create_title_model,
)
from gns3server.agent.gns3_copilot.agent.model_factory import create_title_model
from gns3server.agent.gns3_copilot.gns3_client import GNS3TopologyTool
from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
get_current_llm_config,
)
from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT, load_system_prompt
from gns3server.agent.gns3_copilot.tools_v2 import (
ExecuteMultipleDeviceConfigCommands,
ExecuteMultipleDeviceCommands,
GNS3CreateNodeTool,
GNS3LinkTool,
GNS3StartNodeTool,
GNS3TemplateTool,
GNS3UpdateNodeNameTool,
VPCSMultiCommands,
)
from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT
from gns3server.agent.gns3_copilot.prompts import load_system_prompt
from gns3server.agent.gns3_copilot.tools_v2 import ExecuteMultipleDeviceCommands
from gns3server.agent.gns3_copilot.tools_v2 import ExecuteMultipleDeviceConfigCommands
from gns3server.agent.gns3_copilot.tools_v2 import GNS3CreateNodeTool
from gns3server.agent.gns3_copilot.tools_v2 import GNS3LinkTool
from gns3server.agent.gns3_copilot.tools_v2 import GNS3StartNodeTool
from gns3server.agent.gns3_copilot.tools_v2 import GNS3TemplateTool
from gns3server.agent.gns3_copilot.tools_v2 import GNS3UpdateNodeNameTool
from gns3server.agent.gns3_copilot.tools_v2 import VPCSMultiCommands
# Set up logger for GNS3-Copilot
logger = logging.getLogger(__name__)
@ -110,6 +107,7 @@ DEFAULT_CONVERSATION_TITLE = "New Conversation"
UNTITLED_SESSION_FALLBACK = "Untitled Session"
TITLE_MAX_LENGTH = 40
# Define state
class MessagesState(TypedDict):
"""
@ -138,6 +136,7 @@ class MessagesState(TypedDict):
# Store GNS3 topology information
topology_info: dict | None
# Define llm call node
def llm_call(state: dict, config: RunnableConfig | None = None):
"""
@ -161,8 +160,9 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
"topology_info": None,
}
logger.debug("LLM config retrieved from context: provider=%s, model=%s",
llm_config.get("provider"), llm_config.get("model"))
logger.debug(
"LLM config retrieved from context: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model")
)
# Defensive check: skip LLM call if no user messages
messages = state.get("messages", [])
@ -189,13 +189,13 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
if topology and "error" not in topology:
topology_info = topology
logger.info(
"Successfully retrieved topology for project_id: %s, name: %s",
project_id, topology.get("name")
"Successfully retrieved topology for project_id: %s, name: %s", project_id, topology.get("name")
)
else:
logger.warning(
"Failed to retrieve topology for project_id %s: %s",
project_id, topology.get("error", "Unknown error") if topology else "No result"
project_id,
topology.get("error", "Unknown error") if topology else "No result",
)
except Exception as e:
logger.warning("Error retrieving topology for project_id %s: %s", project_id, e)
@ -214,12 +214,10 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
)
# Create fresh model with tools for each LLM call
logger.debug("Creating model with tools: provider=%s, model=%s",
llm_config.get("provider"), llm_config.get("model"))
model_with_tools = create_base_model_with_tools(
tools,
llm_config=llm_config
logger.debug(
"Creating model with tools: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model")
)
model_with_tools = create_base_model_with_tools(tools, llm_config=llm_config)
# Call pre_hook directly to prepare messages (topology injection + trimming)
# Note: LangGraph's pre_model_hook only works with prebuilt agents, not custom StateGraph
@ -231,8 +229,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
# Invoke model with prepared messages
response = model_with_tools.invoke(prepared_messages)
logger.info("LLM call completed: tool_calls=%d",
len(response.tool_calls) if hasattr(response, 'tool_calls') else 0)
logger.info("LLM call completed: tool_calls=%d", len(response.tool_calls) if hasattr(response, "tool_calls") else 0)
return {
"messages": [response],
@ -240,6 +237,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
"topology_info": topology_info,
}
# Define generate title node
def generate_title(state: MessagesState, config: RunnableConfig | None = None) -> dict:
"""
@ -270,9 +268,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) -
# Call the title generation model (create fresh instance for each call)
try:
title_model = create_title_model(llm_config=llm_config)
response = title_model.invoke(
title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}}
)
response = title_model.invoke(title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}})
raw_content = response.content
new_title = raw_content.strip()
@ -286,7 +282,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) -
# Safety: truncate long titles and avoid line breaks
if len(new_title) > TITLE_MAX_LENGTH:
new_title = new_title[:TITLE_MAX_LENGTH - 2] + "..."
new_title = new_title[: TITLE_MAX_LENGTH - 2] + "..."
# Remove unwanted characters
new_title = new_title.replace("\n", " ").replace('"', "").replace("'", "")
@ -300,7 +296,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) -
# Improved fallback: Use user's first message content
if messages and len(messages) > 0:
first_message = messages[0]
if hasattr(first_message, 'content'):
if hasattr(first_message, "content"):
fallback_title = first_message.content[:30].strip()
# Remove newlines and extra spaces
fallback_title = fallback_title.replace("\n", " ").strip()
@ -323,6 +319,7 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) -
# Title already exists → no update needed
return {}
# Define tool node
def tool_node(state: dict, config: RunnableConfig | None = None):
"""Performs the tool call"""
@ -337,18 +334,14 @@ def tool_node(state: dict, config: RunnableConfig | None = None):
tool = tools_by_name[tool_name]
try:
observation = tool.invoke(tool_call["args"])
logger.debug("Tool %s completed: output_length=%d",
tool_name, len(str(observation)) if observation else 0)
logger.debug("Tool %s completed: output_length=%d", tool_name, len(str(observation)) if observation else 0)
except Exception as e:
logger.error("Tool %s failed: %s", tool_name, e, exc_info=True)
observation = f"Error: {str(e)}"
result.append(ToolMessage(
content=observation,
tool_call_id=tool_call["id"],
name=tool_call["name"]
))
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"], name=tool_call["name"]))
return {"messages": result}
# Routing logic after the LLM node
def should_continue(
state: MessagesState,
@ -374,6 +367,7 @@ def should_continue(
# Normal completion (multi-turn conversation or title already exists)
return END
# Routing logic after the tool node, Check remaining_steps
def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]:
"""
@ -400,6 +394,7 @@ def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]:
return END
# Build and compile the agent
# Build workflow
agent_builder = StateGraph(MessagesState)

14
gns3server/agent/gns3_copilot/agent/model_factory.py
View File

@ -32,12 +32,14 @@ Configuration is passed directly from the database.
"""
import logging
from typing import Any, Optional
from typing import Any
from typing import Optional
from langchain.chat_models import init_chat_model
logger = logging.getLogger(__name__)
def _load_llm_config(
llm_config: Optional[dict[str, Any]] = None,
) -> dict[str, str]:
@ -56,11 +58,7 @@ def _load_llm_config(
if not llm_config:
raise ValueError("LLM configuration is required")
logger.info(
"Using LLM config: provider=%s, model=%s",
llm_config.get("provider"),
llm_config.get("model")
)
logger.info("Using LLM config: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model"))
return {
"model_name": llm_config.get("model", ""),
@ -70,6 +68,7 @@ def _load_llm_config(
"temperature": str(llm_config.get("temperature", "0")),
}
def create_base_model(
llm_config: Optional[dict[str, Any]] = None,
) -> Any:
@ -123,6 +122,7 @@ def create_base_model(
logger.error("Failed to create base model: %s", e)
raise RuntimeError(f"Failed to create base model: {e}") from e
def create_title_model(
llm_config: Optional[dict[str, Any]] = None,
) -> Any:
@ -178,6 +178,7 @@ def create_title_model(
logger.error("Failed to create title model: %s", e)
raise RuntimeError(f"Failed to create title model: {e}") from e
def create_model_with_tools(
model: Any,
tools: list[Any],
@ -203,6 +204,7 @@ def create_model_with_tools(
logger.error("Failed to bind tools to model: %s", e)
raise RuntimeError(f"Failed to bind tools to model: {e}") from e
def create_base_model_with_tools(
tools: list[Any],
llm_config: Optional[dict[str, Any]] = None,

74
gns3server/agent/gns3_copilot/agent_service.py
View File

@ -32,27 +32,37 @@ in the project directory.
"""
import asyncio
import json
import logging
import os
from datetime import datetime
from typing import AsyncGenerator, Dict, Any, Optional, List
from typing import Any
from typing import AsyncGenerator
from typing import Dict
from typing import List
from typing import Optional
from uuid import uuid4
import aiosqlite
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from gns3server.agent.gns3_copilot.agent.gns3_copilot import agent_builder
from gns3server.agent.gns3_copilot.chat_sessions_repository import ChatSessionsRepository
from gns3server.agent.gns3_copilot.chat_sessions_repository import (
ChatSessionsRepository,
)
from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
set_current_jwt_token,
)
from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
set_current_llm_config,
)
from gns3server.agent.gns3_copilot.utils.message_converters import convert_langchain_to_openai
from gns3server.agent.gns3_copilot.utils.message_converters import (
convert_langchain_to_openai,
)
log = logging.getLogger(__name__)
class AgentService:
"""
Project-level Agent Service with async checkpoint management.
@ -176,7 +186,9 @@ class AgentService:
await conn.commit()
# Create pinned index (after column is guaranteed to exist)
await conn.execute("CREATE INDEX IF NOT EXISTS idx_pinned_updated ON chat_sessions(pinned DESC, updated_at DESC)")
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_pinned_updated ON chat_sessions(pinned DESC, updated_at DESC)"
)
await conn.commit()
log.debug("chat_sessions table created in checkpoint database")
@ -197,7 +209,7 @@ class AgentService:
user_id: Optional[str] = None,
jwt_token: Optional[str] = None,
mode: str = "text",
llm_config: Optional[Dict[str, Any]] = None
llm_config: Optional[Dict[str, Any]] = None,
) -> AsyncGenerator[Dict[str, Any], None]:
"""
Stream chat responses from the agent.
@ -234,10 +246,7 @@ class AgentService:
if is_new_session:
# Create new session
session = await repo.create_session(
thread_id=session_id,
user_id=user_id or "",
project_id=project_id or "",
title="New Conversation"
thread_id=session_id, user_id=user_id or "", project_id=project_id or "", title="New Conversation"
)
log.debug("Created new chat session: thread_id=%s", session_id)
@ -247,8 +256,9 @@ class AgentService:
log.debug("JWT token set in context")
if llm_config:
set_current_llm_config(llm_config)
log.debug("LLM config set in context: provider=%s, model=%s",
llm_config.get("provider"), llm_config.get("model"))
log.debug(
"LLM config set in context: provider=%s, model=%s", llm_config.get("provider"), llm_config.get("model")
)
# Build config - only thread-safe identifiers
config = {
@ -258,7 +268,7 @@ class AgentService:
},
"metadata": {
"user_id": user_id,
}
},
}
# Build inputs
@ -353,10 +363,17 @@ class AgentService:
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
last_message_at=last_message_at
last_message_at=last_message_at,
)
log.info(
"Session statistics updated: thread_id=%s, messages=%d, llm_calls=%d, tokens=%d+%d=%d",
session_id,
message_count,
llm_calls_count,
input_tokens,
output_tokens,
input_tokens + output_tokens,
)
log.info("Session statistics updated: thread_id=%s, messages=%d, llm_calls=%d, tokens=%d+%d=%d",
session_id, message_count, llm_calls_count, input_tokens, output_tokens, input_tokens + output_tokens)
# Sync auto-generated title from checkpoint state
final_state = await graph.aget_state(config)
@ -365,8 +382,7 @@ class AgentService:
current_session = await repo.get_session_by_thread(session_id)
if current_session and current_session.title != generated_title:
await repo.update_session(thread_id=session_id, title=generated_title)
log.info("Auto-generated title synced: thread_id=%s, title=%s",
session_id, generated_title)
log.info("Auto-generated title synced: thread_id=%s, title=%s", session_id, generated_title)
except Exception as e:
log.error("Error in stream_chat: %s", e, exc_info=True)
@ -396,11 +412,7 @@ class AgentService:
elif event_type == "on_tool_start":
# Tool execution started
return {
"type": "tool_start",
"tool_name": event.get("name", ""),
"session_id": session_id
}
return {"type": "tool_start", "tool_name": event.get("name", ""), "session_id": session_id}
elif event_type == "on_tool_end":
# Tool execution completed
@ -412,7 +424,7 @@ class AgentService:
"type": "tool_end",
"tool_name": event.get("name", ""),
"tool_output": output,
"session_id": session_id
"session_id": session_id,
}
return None
@ -441,19 +453,11 @@ class AgentService:
title = state.values.get("conversation_title", "New Conversation")
return {
"thread_id": session_id,
"title": title,
"messages": messages
}
return {"thread_id": session_id, "title": title, "messages": messages}
except Exception as e:
log.error("Error getting history: %s", e, exc_info=True)
return {
"thread_id": session_id,
"title": "New Conversation",
"messages": []
}
return {"thread_id": session_id, "title": "New Conversation", "messages": []}
def _convert_message_to_dict(self, msg) -> Dict[str, Any]:
"""Convert a LangChain message to OpenAI-compatible dict format."""

62
gns3server/agent/gns3_copilot/chat_sessions_repository.py
View File

@ -33,13 +33,16 @@ checkpoint database.
import json
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional
from uuid import UUID
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
import aiosqlite
log = logging.getLogger(__name__)
class ChatSession:
"""Chat session model."""
@ -60,7 +63,7 @@ class ChatSession:
updated_at: Optional[str] = None,
metadata: str = "{}",
stats: str = "{}",
pinned: bool = False
pinned: bool = False,
):
self.id = id
self.thread_id = thread_id
@ -100,6 +103,7 @@ class ChatSession:
"pinned": self.pinned,
}
class ChatSessionsRepository:
"""
Repository for managing chat sessions in the checkpoint database.
@ -115,11 +119,7 @@ class ChatSessionsRepository:
self.conn = conn
async def create_session(
self,
thread_id: str,
user_id: str,
project_id: str,
title: str = "New Conversation"
self, thread_id: str, user_id: str, project_id: str, title: str = "New Conversation"
) -> ChatSession:
"""
Create a new chat session.
@ -141,7 +141,7 @@ class ChatSessionsRepository:
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?)
""",
(thread_id, user_id, project_id, title, now, now)
(thread_id, user_id, project_id, title, now, now),
)
await self.conn.commit()
@ -160,10 +160,7 @@ class ChatSessionsRepository:
Returns:
ChatSession or None
"""
cursor = await self.conn.execute(
"SELECT * FROM chat_sessions WHERE id = ?",
(session_id,)
)
cursor = await self.conn.execute("SELECT * FROM chat_sessions WHERE id = ?", (session_id,))
row = await cursor.fetchone()
if row:
@ -180,10 +177,7 @@ class ChatSessionsRepository:
Returns:
ChatSession or None
"""
cursor = await self.conn.execute(
"SELECT * FROM chat_sessions WHERE thread_id = ?",
(thread_id,)
)
cursor = await self.conn.execute("SELECT * FROM chat_sessions WHERE thread_id = ?", (thread_id,))
row = await cursor.fetchone()
if row:
@ -191,10 +185,7 @@ class ChatSessionsRepository:
return None
async def list_sessions(
self,
user_id: Optional[str] = None,
project_id: Optional[str] = None,
limit: int = 100
self, user_id: Optional[str] = None, project_id: Optional[str] = None, limit: int = 100
) -> List[ChatSession]:
"""
List sessions with optional filters.
@ -239,7 +230,7 @@ class ChatSessionsRepository:
input_tokens: Optional[int] = None,
output_tokens: Optional[int] = None,
total_tokens: Optional[int] = None,
last_message_at: Optional[str] = None
last_message_at: Optional[str] = None,
) -> Optional[ChatSession]:
"""
Update a session.
@ -316,16 +307,10 @@ class ChatSessionsRepository:
True if deleted, False if not found
"""
# First, delete the checkpoint data
await self.conn.execute(
"DELETE FROM checkpoints WHERE thread_id = ?",
(thread_id,)
)
await self.conn.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,))
# Then delete the session
cursor = await self.conn.execute(
"DELETE FROM chat_sessions WHERE thread_id = ?",
(thread_id,)
)
cursor = await self.conn.execute("DELETE FROM chat_sessions WHERE thread_id = ?", (thread_id,))
await self.conn.commit()
deleted = cursor.rowcount > 0
@ -345,24 +330,15 @@ class ChatSessionsRepository:
Number of sessions deleted
"""
# Get all thread_ids for this project
cursor = await self.conn.execute(
"SELECT thread_id FROM chat_sessions WHERE project_id = ?",
(project_id,)
)
cursor = await self.conn.execute("SELECT thread_id FROM chat_sessions WHERE project_id = ?", (project_id,))
rows = await cursor.fetchall()
thread_ids = [row[0] for row in rows]
# Delete checkpoints and sessions
for thread_id in thread_ids:
await self.conn.execute(
"DELETE FROM checkpoints WHERE thread_id = ?",
(thread_id,)
)
await self.conn.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,))
cursor = await self.conn.execute(
"DELETE FROM chat_sessions WHERE project_id = ?",
(project_id,)
)
cursor = await self.conn.execute("DELETE FROM chat_sessions WHERE project_id = ?", (project_id,))
await self.conn.commit()
deleted_count = cursor.rowcount
@ -385,7 +361,7 @@ class ChatSessionsRepository:
now = datetime.utcnow().isoformat()
await self.conn.execute(
"UPDATE chat_sessions SET pinned = ?, updated_at = ? WHERE thread_id = ?",
(1 if pinned else 0, now, thread_id)
(1 if pinned else 0, now, thread_id),
)
await self.conn.commit()

107
gns3server/agent/gns3_copilot/deprecated/context_manager.py
View File

@ -39,14 +39,13 @@ including:
import json
import logging
from typing import Any, Literal
from typing import Any
from typing import Literal
from langchain_core.messages import (
AIMessage,
HumanMessage,
SystemMessage,
ToolMessage,
)
from langchain_core.messages import AIMessage
from langchain_core.messages import HumanMessage
from langchain_core.messages import SystemMessage
from langchain_core.messages import ToolMessage
logger = logging.getLogger(__name__)
@ -57,8 +56,8 @@ logger = logging.getLogger(__name__)
# Context strategy ratios
CONTEXT_STRATEGY_RATIOS = {
"conservative": 0.60, # 60% for input, 40% reserved for output
"balanced": 0.75, # 75% for input, 25% reserved for output
"aggressive": 0.85, # 85% for input, 15% reserved for output
"balanced": 0.75, # 75% for input, 25% reserved for output
"aggressive": 0.85, # 85% for input, 15% reserved for output
}
DEFAULT_CONTEXT_STRATEGY = "balanced"
@ -70,6 +69,7 @@ DEFAULT_CONTEXT_STRATEGY = "balanced"
# Global tiktoken encoding cache (lazy loading)
_tiktoken_encoding = None
def _get_tiktoken_encoding():
"""
Get tiktoken encoding instance (cached).
@ -87,6 +87,7 @@ def _get_tiktoken_encoding():
if _tiktoken_encoding is None:
try:
import tiktoken
_tiktoken_encoding = tiktoken.get_encoding("cl100k_base")
logger.debug("Using tiktoken (cl100k_base) for accurate token counting")
except ImportError:
@ -96,6 +97,7 @@ def _get_tiktoken_encoding():
)
return _tiktoken_encoding
def count_tokens_accurately(text: str) -> int:
"""
Count tokens in text accurately using tiktoken.
@ -119,6 +121,7 @@ def count_tokens_accurately(text: str) -> int:
logger.error("tiktoken encoding failed: %s", e)
raise
def count_messages_tokens(messages: list[Any]) -> int:
"""
Count total tokens in a list of messages accurately.
@ -134,16 +137,18 @@ def count_messages_tokens(messages: list[Any]) -> int:
"""
total = 0
for msg in messages:
if hasattr(msg, 'content') and msg.content:
if hasattr(msg, "content") and msg.content:
# Handle both string and complex content
content = str(msg.content)
total += count_tokens_accurately(content)
return total
# ============================================================================
# Tool Definition Token Estimation
# ============================================================================
def estimate_tool_tokens(tools: list[Any]) -> int:
"""
Estimate the token count of tool definitions.
@ -174,12 +179,12 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
"type": "function",
"function": {
"name": tool.name,
"description": tool.description if hasattr(tool, 'description') else "",
}
"description": tool.description if hasattr(tool, "description") else "",
},
}
# Add parameters schema if available
if hasattr(tool, 'args_schema') and tool.args_schema:
if hasattr(tool, "args_schema") and tool.args_schema:
try:
tool_schema["function"]["parameters"] = tool.args_schema.schema()
except Exception:
@ -191,10 +196,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
tool_tokens = len(encoding.encode(schema_str))
total_tokens += tool_tokens
logger.debug(
"Tool '%s': ~%d tokens (schema size: %d chars)",
tool.name, tool_tokens, len(schema_str)
)
logger.debug("Tool '%s': ~%d tokens (schema size: %d chars)", tool.name, tool_tokens, len(schema_str))
except Exception as e:
logger.error("Failed to estimate tokens for tool '%s': %s", tool.name, e)
@ -203,6 +205,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
logger.info("Tool definitions estimated at ~%d total tokens (%d tools)", total_tokens, len(tools))
return total_tokens
# ============================================================================
# Model Context Limits
# ============================================================================
@ -221,10 +224,8 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
#
# Always verify current limits from official provider documentation.
def get_model_context_limit(
model_name: str,
llm_config: dict[str, Any] | None = None
) -> int:
def get_model_context_limit(model_name: str, llm_config: dict[str, Any] | None = None) -> int:
"""
Get the context window limit for a given model.
@ -253,7 +254,9 @@ def get_model_context_limit(
actual_tokens = db_limit_k * 1000
logger.debug(
"Using database config context limit: %dK tokens (%d tokens) for model '%s'",
db_limit_k, actual_tokens, model_name
db_limit_k,
actual_tokens,
model_name,
)
return actual_tokens
else:
@ -270,9 +273,9 @@ def get_model_context_limit(
f"Refer to the model provider's documentation for the current context window size."
)
def calculate_max_tokens(
model_limit: int,
strategy: Literal["conservative", "balanced", "aggressive"] = DEFAULT_CONTEXT_STRATEGY
model_limit: int, strategy: Literal["conservative", "balanced", "aggressive"] = DEFAULT_CONTEXT_STRATEGY
) -> int:
"""
Calculate the maximum tokens to use, reserving space for output.
@ -290,17 +293,16 @@ def calculate_max_tokens(
ratio = CONTEXT_STRATEGY_RATIOS.get(strategy, CONTEXT_STRATEGY_RATIOS[DEFAULT_CONTEXT_STRATEGY])
max_tokens = int(model_limit * ratio)
logger.debug(
"Context limit: model=%d, strategy=%s, usable=%d tokens",
model_limit, strategy, max_tokens
)
logger.debug("Context limit: model=%d, strategy=%s, usable=%d tokens", model_limit, strategy, max_tokens)
return max_tokens
# ============================================================================
# Message Trimming
# ============================================================================
def trim_messages_for_context(
messages: list[Any],
model_name: str,
@ -353,7 +355,8 @@ def trim_messages_for_context(
logger.warning(
"Tool definitions (%d tokens) exceed input budget (%d tokens). "
"Consider reducing context_limit or using fewer tools.",
tool_tokens, max_tokens
tool_tokens,
max_tokens,
)
available_for_messages = 0
@ -363,13 +366,19 @@ def trim_messages_for_context(
if current_tokens <= available_for_messages:
logger.debug(
"Messages fit in context: %d / %d tokens (available: %d, tools: %d)",
current_tokens, max_tokens, available_for_messages, tool_tokens
current_tokens,
max_tokens,
available_for_messages,
tool_tokens,
)
return messages
logger.info(
"Trimming messages: %d → %d tokens (budget: %d, tools: %d)",
current_tokens, available_for_messages, max_tokens, tool_tokens
current_tokens,
available_for_messages,
max_tokens,
tool_tokens,
)
# Manually separate and trim to ensure system messages are preserved
@ -387,7 +396,8 @@ def trim_messages_for_context(
# Not enough space for system messages - keep only system messages
logger.warning(
"System messages (%d tokens) exceed available space (%d tokens), truncating to system only",
system_tokens, available_for_messages
system_tokens,
available_for_messages,
)
return system_msgs[:1] if system_msgs else messages[-1:]
@ -399,12 +409,16 @@ def trim_messages_for_context(
logger.info(
"Trimmed %d → %d messages (system: %d, history: %d → %d)",
len(messages), len(trimmed),
len(system_msgs), len(other_msgs), len(trimmed_other)
len(messages),
len(trimmed),
len(system_msgs),
len(other_msgs),
len(trimmed_other),
)
return trimmed
def _trim_to_token_limit(messages: list[Any], max_tokens: int) -> list[Any]:
"""
Trim messages to fit within token limit using tiktoken.
@ -454,7 +468,7 @@ def _trim_to_token_limit(messages: list[Any], max_tokens: int) -> list[Any]:
len(groups) - len(trimmed_groups),
len(trimmed),
current_tokens,
_count_groups_tokens(trimmed_groups)
_count_groups_tokens(trimmed_groups),
)
return trimmed
@ -481,18 +495,18 @@ def _build_message_groups(messages: list[Any]) -> list[list[Any]]:
msg = messages[i]
# If AIMessage with tool_calls, group it with all following ToolMessages
if isinstance(msg, AIMessage) and hasattr(msg, 'tool_calls') and msg.tool_calls:
if isinstance(msg, AIMessage) and hasattr(msg, "tool_calls") and msg.tool_calls:
group = [msg]
i += 1
# Collect all following ToolMessages that match these tool_calls
tool_call_ids = {tc['id'] for tc in msg.tool_calls}
tool_call_ids = {tc["id"] for tc in msg.tool_calls}
while i < len(messages):
next_msg = messages[i]
if isinstance(next_msg, ToolMessage):
# Check if this ToolMessage belongs to current AIMessage
if hasattr(next_msg, 'tool_call_id') and next_msg.tool_call_id in tool_call_ids:
if hasattr(next_msg, "tool_call_id") and next_msg.tool_call_id in tool_call_ids:
group.append(next_msg)
i += 1
else:
@ -516,14 +530,16 @@ def _count_groups_tokens(groups: list[list[Any]]) -> int:
total = 0
for group in groups:
for msg in group:
if hasattr(msg, 'content') and msg.content:
if hasattr(msg, "content") and msg.content:
total += count_tokens_accurately(str(msg.content))
return total
# ============================================================================
# Token Usage Summary
# ============================================================================
def get_token_usage_summary(
messages: list[Any],
model_name: str,
@ -575,10 +591,12 @@ def get_token_usage_summary(
"needs_trimming": usage_percentage > 80,
}
# ============================================================================
# Main Entry Point - Context Preparation with Template Injection
# ============================================================================
def prepare_context_messages(
state_messages: list[Any],
system_prompt: str,
@ -685,7 +703,7 @@ def prepare_context_messages(
formatted_tokens + trimmed_history_tokens + tool_tokens,
model_limit_k,
((formatted_tokens + trimmed_history_tokens) / (model_limit_k * 1000)) * 100,
trim_strategy
trim_strategy,
)
else:
# No trimming
@ -699,20 +717,19 @@ def prepare_context_messages(
formatted_tokens + history_tokens + tool_tokens,
model_limit_k,
(formatted_tokens / (model_limit_k * 1000)) * 100,
trim_strategy
trim_strategy,
)
return trimmed_messages
# ============================================================================
# Module Test
# ============================================================================
if __name__ == "__main__":
# Simple test
test_messages = [
HumanMessage(f"Message {i}") for i in range(100)
]
test_messages = [HumanMessage(f"Message {i}") for i in range(100)]
# Test with mock llm_config
mock_config = {"context_limit": 8, "context_strategy": "conservative"}
@ -723,7 +740,7 @@ if __name__ == "__main__":
topology_context='{"project_id": "test", "nodes": 5}',
model_name="gpt-4o",
llm_config=mock_config,
tools=None
tools=None,
)
print(f"Original: {len(test_messages)} messages")

38
gns3server/agent/gns3_copilot/gns3_client/__init__.py
View File

@ -47,29 +47,23 @@ Main functions:
Upstream gns3fy: https://github.com/davidban77/gns3fy
"""
from .connector_factory import (
get_gns3_connector,
get_gns3_connector_with_llm_config,
get_gns3_server_host,
get_llm_config,
)
from .context_helpers import (
get_current_jwt_token,
get_current_llm_config,
set_current_jwt_token,
set_current_llm_config,
)
from .custom_gns3fy import (
CONSOLE_TYPES,
LINK_TYPES,
NODE_TYPES,
Gns3Connector,
Link,
Node,
Project,
)
from .gns3_topology_reader import GNS3TopologyTool
from .connector_factory import get_gns3_connector
from .connector_factory import get_gns3_connector_with_llm_config
from .connector_factory import get_gns3_server_host
from .connector_factory import get_llm_config
from .context_helpers import get_current_jwt_token
from .context_helpers import get_current_llm_config
from .context_helpers import set_current_jwt_token
from .context_helpers import set_current_llm_config
from .custom_gns3fy import CONSOLE_TYPES
from .custom_gns3fy import LINK_TYPES
from .custom_gns3fy import NODE_TYPES
from .custom_gns3fy import Gns3Connector
from .custom_gns3fy import Link
from .custom_gns3fy import Node
from .custom_gns3fy import Project
from .gns3_project_info import GNS3ProjectInfoTool
from .gns3_topology_reader import GNS3TopologyTool
# Dynamic version management
try:

59
gns3server/agent/gns3_copilot/gns3_client/connector_factory.py
View File

@ -41,24 +41,22 @@ Features:
import asyncio
import concurrent.futures
import logging
import threading
from typing import Optional
from uuid import UUID
# Local imports
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
get_current_jwt_token,
get_current_llm_config,
set_current_jwt_token,
set_current_llm_config,
)
# Local imports
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
logger = logging.getLogger(__name__)
# Fallback default URL
DEFAULT_GNS3_URL = "http://127.0.0.1:3080"
def _get_url_from_controller() -> Optional[str]:
"""Try to get GNS3 server URL from running Controller instance.
@ -93,6 +91,7 @@ def _get_url_from_controller() -> Optional[str]:
logger.warning("Unexpected error getting URL from Controller: %s", str(e))
return None
def _get_url_from_config() -> Optional[str]:
"""Try to get GNS3 server URL from Config settings.
@ -122,6 +121,7 @@ def _get_url_from_config() -> Optional[str]:
logger.warning("Unexpected error getting URL from Config: %s", str(e))
return None
def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = None) -> Optional[Gns3Connector]:
"""Create and return a Gns3Connector instance with JWT authentication.
@ -184,7 +184,7 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non
"Using fallback default URL: %s. "
"This may not be correct if your GNS3 server is configured differently. "
"Consider providing the URL explicitly or ensuring gns3server is running.",
url
url,
)
else:
logger.info("Using explicitly provided GNS3 server URL: %s", url)
@ -208,22 +208,20 @@ def get_gns3_connector(jwt_token: Optional[str] = None, url: Optional[str] = Non
logger.error("Failed to create Gns3Connector: %s", str(e), exc_info=True)
return None
async def get_gns3_connector_with_llm_config(
user_id,
jwt_token: str,
url: Optional[str] = None
) -> Optional[dict]:
async def get_gns3_connector_with_llm_config(user_id, jwt_token: str, url: Optional[str] = None, app=None) -> Optional[dict]:
"""
Create Gns3Connector and retrieve LLM model configuration for the user.
This is a convenience function that combines:
1. get_gns3_connector() - Create GNS3 API connector
2. get_user_llm_config() - Retrieve user's default LLM config with API key
2. get_llm_config() - Retrieve user's default LLM config with API key
Args:
user_id: User UUID (can be string or UUID object)
jwt_token: JWT token for authentication
url: GNS3 server URL (optional, auto-detected if not provided)
app: FastAPI application instance (optional, for direct database access)
Returns:
Dictionary with keys:
@ -263,21 +261,12 @@ async def get_gns3_connector_with_llm_config(
url = _detect_url_for_api()
# Step 3: Get LLM config
from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config
llm_config = await get_user_llm_config(
user_id=user_id,
jwt_token=jwt_token,
gns3_url=url
)
llm_config = get_llm_config(user_id=user_id, jwt_token=jwt_token, app=app)
if not llm_config:
logger.warning(f"No LLM config found for user {user_id}")
# Still return result with connector only
return {
"connector": connector,
"llm_config": None
}
return {"connector": connector, "llm_config": None}
logger.info(
f"Successfully initialized GNS3 connector and LLM config for user {user_id}: "
@ -286,15 +275,13 @@ async def get_gns3_connector_with_llm_config(
f"llm_model={llm_config.get('model')}"
)
return {
"connector": connector,
"llm_config": llm_config
}
return {"connector": connector, "llm_config": llm_config}
except Exception as e:
logger.error(f"Failed to get GNS3 connector with LLM config: {e}", exc_info=True)
return None
def _detect_url_for_api() -> Optional[str]:
"""
Detect GNS3 server URL for API calls.
@ -325,6 +312,7 @@ def _detect_url_for_api() -> Optional[str]:
logger.debug("Using fallback URL for API call: %s", DEFAULT_GNS3_URL)
return DEFAULT_GNS3_URL
def get_gns3_server_host() -> str:
"""
Get GNS3 server hostname from Controller or Config.
@ -361,6 +349,7 @@ def get_gns3_server_host() -> str:
logger.warning("Failed to extract host from URL %s: %s, using fallback", url, e)
return DEFAULT_GNS3_URL.split("://")[1].split(":")[0]
def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
"""
Get LLM model configuration for a user.
@ -394,7 +383,9 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
# If app is provided, use direct database access (preferred)
if app is not None:
from gns3server.agent.gns3_copilot.utils.llm_config_helper import get_user_llm_config_with_app
from gns3server.agent.gns3_copilot.utils.llm_config_helper import (
get_user_llm_config_with_app,
)
# Run the async function in sync context
try:
@ -402,10 +393,7 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
# We're in an async context with a running loop
# This shouldn't happen since this is a sync function
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(
asyncio.run,
get_user_llm_config_with_app(user_id, app)
)
future = executor.submit(asyncio.run, get_user_llm_config_with_app(user_id, app))
return future.result(timeout=10)
except RuntimeError:
# No running event loop - we're in a sync context
@ -417,10 +405,7 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
if loop.is_running():
# Loop is running but not the running loop (edge case)
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(
asyncio.run,
get_user_llm_config_with_app(user_id, app)
)
future = executor.submit(asyncio.run, get_user_llm_config_with_app(user_id, app))
return future.result(timeout=10)
else:
# Loop exists but not running - use it

8
gns3server/agent/gns3_copilot/gns3_client/context_helpers.py
View File

@ -97,8 +97,7 @@ def set_current_llm_config(config: dict) -> None:
config: LLM configuration dictionary with provider, model, api_key, etc.
"""
_llm_config_context.set(config)
logger.debug("LLM config set in context: provider=%s, model=%s",
config.get("provider"), config.get("model"))
logger.debug("LLM config set in context: provider=%s, model=%s", config.get("provider"), config.get("model"))
def get_current_llm_config() -> Optional[dict]:
@ -109,8 +108,9 @@ def get_current_llm_config() -> Optional[dict]:
"""
config = _llm_config_context.get()
if config:
logger.debug("LLM config retrieved from context: provider=%s, model=%s",
config.get("provider"), config.get("model"))
logger.debug(
"LLM config retrieved from context: provider=%s, model=%s", config.get("provider"), config.get("model")
)
else:
logger.warning("LLM config not found in context")
return config

230
gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py
View File

@ -46,19 +46,20 @@ import time
from collections.abc import Callable
from dataclasses import field
from functools import wraps
from math import cos, pi, sin
from typing import (
Any,
ParamSpec,
TypeVar,
cast,
)
from math import cos
from math import pi
from math import sin
from typing import Any
from typing import ParamSpec
from typing import TypeVar
from typing import cast
from urllib.parse import urlparse
import jwt
import requests
import urllib3
from pydantic import ConfigDict, field_validator
from pydantic import ConfigDict
from pydantic import field_validator
from pydantic.dataclasses import dataclass
from requests import HTTPError
@ -98,6 +99,7 @@ CONSOLE_TYPES = [
LINK_TYPES = ["ethernet", "serial"]
class Gns3Connector:
"""
Connector to be use for interaction against GNS3 server controller API.
@ -149,7 +151,8 @@ class Gns3Connector:
if url is None:
raise ValueError("URL is required for Gns3Connector")
self.base_url = f"{url.strip('/')}/v{api_version}"
self.url = url.strip('/') # Store original URL for reference
self.base_url = f"{self.url}/v{api_version}"
self.user = user
self.cred = cred
self.headers = {"Content-Type": "application/json"}
@ -174,11 +177,7 @@ class Gns3Connector:
self.session.headers["Accept"] = "application/json" # pragma: no cover
# Set authentication based on API version
if (
self.auth_type == "basic"
and self.user is not None
and self.cred is not None
):
if self.auth_type == "basic" and self.user is not None and self.cred is not None:
self.session.auth = (self.user, self.cred) # pragma: no cover
elif self.auth_type == "jwt" and self.access_token:
@ -194,9 +193,7 @@ class Gns3Connector:
return
if not self.user or not self.cred:
raise ValueError(
"Username and password are required for v3 authentication when no JWT token is provided"
)
raise ValueError("Username and password are required for v3 authentication when no JWT token is provided")
# Construct authentication URL (v3 API uses different base URL)
auth_url = f"{self.base_url.replace('/v3', '')}/v3/access/users/authenticate"
@ -207,9 +204,7 @@ class Gns3Connector:
temp_session.headers["Content-Type"] = "application/json"
try:
response = temp_session.post(
auth_url, json=auth_data, verify=self.verify, timeout=10.0
)
response = temp_session.post(auth_url, json=auth_data, verify=self.verify, timeout=10.0)
if response.status_code == 200:
auth_result = response.json()
self.access_token = auth_result["access_token"]
@ -217,9 +212,7 @@ class Gns3Connector:
self.session.headers["Authorization"] = f"Bearer {self.access_token}"
# print(f"Successfully authenticated to v3 API, token obtained")
else:
raise HTTPError(
f"v3 API authentication failed: {response.status_code} - {response.text}"
)
raise HTTPError(f"v3 API authentication failed: {response.status_code} - {response.text}")
except Exception as e:
raise HTTPError(f"v3 API authentication error: {str(e)}") from e
@ -233,9 +226,7 @@ class Gns3Connector:
try:
# Decode token without verification to check expiry
decoded: dict[str, Any] = jwt.decode(
token, options={"verify_signature": False}
)
decoded: dict[str, Any] = jwt.decode(token, options={"verify_signature": False})
exp = decoded.get("exp")
if exp is not None:
return time.time() > float(exp)
@ -264,12 +255,7 @@ class Gns3Connector:
Executes HTTP operations and handles GNS3-specific error logic.
"""
# Handle JWT authentication
if (
self.auth_type == "jwt"
and not self.access_token
and self.user
and self.cred
):
if self.auth_type == "jwt" and not self.access_token and self.user and self.cred:
self._authenticate_v3()
# Get request function (e.g., session.get, session.post)
@ -337,9 +323,7 @@ class Gns3Connector:
response = self.http_call("get", url=f"{self.base_url}/version")
return cast(dict[str, Any], response.json())
def projects_summary(
self, is_print: bool = True
) -> list[tuple[str, str, int, int, str]] | None:
def projects_summary(self, is_print: bool = True) -> list[tuple[str, str, int, int, str]] | None:
"""
Returns a summary of the projects in the server. If `is_print` is `False`, it
will return a list of tuples like:
@ -349,9 +333,7 @@ class Gns3Connector:
_projects_summary = []
for _p in self.get_projects():
# Retrieve the project stats
_stats = self.http_call(
"get", f"{self.base_url}/projects/{_p['project_id']}/stats"
).json()
_stats = self.http_call("get", f"{self.base_url}/projects/{_p['project_id']}/stats").json()
if is_print:
print(
f"{_p['name']}: {_p['project_id']} -- Nodes: {_stats['nodes']} -- "
@ -376,9 +358,7 @@ class Gns3Connector:
response = self.http_call("get", url=f"{self.base_url}/projects").json()
return cast(list[dict[str, Any]], response)
def get_project(
self, name: str | None = None, project_id: str | None = None
) -> dict[str, Any] | None:
def get_project(self, name: str | None = None, project_id: str | None = None) -> dict[str, Any] | None:
"""
Retrieves a project from either a name or ID
@ -387,9 +367,7 @@ class Gns3Connector:
- `name` or `project_id`
"""
if project_id:
_response = self.http_call(
"get", url=f"{self.base_url}/projects/{project_id}"
)
_response = self.http_call("get", url=f"{self.base_url}/projects/{project_id}")
return cast(dict[str, Any], _response.json())
elif name:
try:
@ -400,9 +378,7 @@ class Gns3Connector:
else:
raise ValueError("Must provide either a name or project_id")
def templates_summary(
self, is_print: bool = True
) -> list[tuple[str, str, str, bool, str, str]] | None:
def templates_summary(self, is_print: bool = True) -> list[tuple[str, str, str, bool, str, str]] | None:
"""
Returns a summary of the templates in the server. If `is_print` is `False`, it
will return a list of tuples like:
@ -439,9 +415,7 @@ class Gns3Connector:
_response_data = self.http_call("get", url=f"{self.base_url}/templates").json()
return cast(list[dict[str, Any]], _response_data)
def get_template(
self, name: str | None = None, template_id: str | None = None
) -> dict[str, Any] | None:
def get_template(self, name: str | None = None, template_id: str | None = None) -> dict[str, Any] | None:
"""
Retrieves a template from either a name or ID
@ -450,9 +424,7 @@ class Gns3Connector:
- `name` or `template_id`
"""
if template_id:
_response_json = self.http_call(
"get", url=f"{self.base_url}/templates/{template_id}"
).json()
_response_json = self.http_call("get", url=f"{self.base_url}/templates/{template_id}").json()
return cast(dict[str, Any], _response_json)
elif name:
try:
@ -541,15 +513,11 @@ class Gns3Connector:
kwargs["compute_id"] = "local"
# Send request
response = self.http_call(
"post", url=f"{self.base_url}/templates", json_data=kwargs
)
response = self.http_call("post", url=f"{self.base_url}/templates", json_data=kwargs)
# Return and convert type
return cast(dict[str, Any], response.json())
def delete_template(
self, name: str | None = None, template_id: str | None = None
) -> None:
def delete_template(self, name: str | None = None, template_id: str | None = None) -> None:
"""
Deletes a template by giving its attributes. For more information [API INFO]
(http://api.gns3.net/en/2.2/api/v2/controller/template/
@ -570,9 +538,7 @@ class Gns3Connector:
# Final check: ensure template_id has a value at this point
if not template_id:
raise ValueError(
"Must provide either a 'name' or 'template_id' to delete a template."
)
raise ValueError("Must provide either a 'name' or 'template_id' to delete a template.")
self.http_call("delete", url=f"{self.base_url}/templates/{template_id}")
@ -584,9 +550,7 @@ class Gns3Connector:
- `project_id`
"""
_response_data = self.http_call(
"get", url=f"{self.base_url}/projects/{project_id}/nodes"
).json()
_response_data = self.http_call("get", url=f"{self.base_url}/projects/{project_id}/nodes").json()
return cast(list[dict[str, Any]], _response_data)
@ -611,9 +575,7 @@ class Gns3Connector:
- `project_id`
"""
_response_data = self.http_call(
"get", url=f"{self.base_url}/projects/{project_id}/links"
).json()
_response_data = self.http_call("get", url=f"{self.base_url}/projects/{project_id}/links").json()
return cast(list[dict[str, Any]], _response_data)
@ -688,9 +650,7 @@ class Gns3Connector:
return cast(dict[str, Any], _response_data)
def get_compute_images(
self, emulator: str, compute_id: str = "local"
) -> list[dict[str, Any]]:
def get_compute_images(self, emulator: str, compute_id: str = "local") -> list[dict[str, Any]]:
"""
Returns a list of images available for a compute.
@ -709,9 +669,7 @@ class Gns3Connector:
return cast(list[dict[str, Any]], _response_data)
def upload_compute_image(
self, emulator: str, file_path: str, compute_id: str = "local"
) -> None:
def upload_compute_image(self, emulator: str, file_path: str, compute_id: str = "local") -> None:
"""
uploads an image for use by a compute.
@ -748,6 +706,7 @@ class Gns3Connector:
return cast(dict[str, Any], _response_data)
def verify_connector_and_id(f: F) -> F:
"""
Main checker for connector object and respective object's ID for their retrieval
@ -774,13 +733,9 @@ def verify_connector_and_id(f: F) -> F:
_url = f"{_conn.base_url}/projects/{_project_id}/nodes"
_response = _conn.http_call("get", _url)
extracted = [
node for node in _response.json() if node["name"] == self.name
]
extracted = [node for node in _response.json() if node["name"] == self.name]
if len(extracted) > 1: # pragma: no cover
raise ValueError(
"Multiple nodes found with same name. Need to submit node_id"
)
raise ValueError("Multiple nodes found with same name. Need to submit node_id")
self.node_id = extracted[0]["node_id"]
# Checks for Link
if self.__class__.__name__ == "Link":
@ -790,6 +745,7 @@ def verify_connector_and_id(f: F) -> F:
return cast(F, wrapper)
@dataclass(config=config)
class Link:
"""
@ -914,9 +870,7 @@ class Link:
if _project_id is None:
raise ValueError("Need to submit project_id")
if _link_id is None:
raise ValueError(
"Link ID is missing. The link might have already been deleted."
)
raise ValueError("Link ID is missing. The link might have already been deleted.")
_url = f"{_conn.base_url}/projects/{_project_id}/links/{self.link_id}"
@ -942,12 +896,7 @@ class Link:
_url = f"{self.connector.base_url}/projects/{self.project_id}/links"
data = {
k: v
for k, v in self.__dict__.items()
if k not in ("connector", "__initialised__")
if v is not None
}
data = {k: v for k, v in self.__dict__.items() if k not in ("connector", "__initialised__") if v is not None}
_response = self.connector.http_call("post", _url, json_data=data)
@ -979,9 +928,7 @@ class Link:
if not self.project_id:
raise ValueError("Need to submit project_id")
_url = (
f"{self.connector.base_url}/projects/{self.project_id}/links/{self.link_id}"
)
_url = f"{self.connector.base_url}/projects/{self.project_id}/links/{self.link_id}"
# TODO: Verify that the passed kwargs are supported ones
_response = self.connector.http_call("put", _url, json_data=kwargs)
@ -989,6 +936,7 @@ class Link:
# Update object
self._update(_response.json())
@dataclass(config=config)
class Node:
"""
@ -1120,9 +1068,7 @@ class Node:
if not self.project_id:
raise ValueError("Need to submit project_id")
_url = (
f"{self.connector.base_url}/projects/{self.project_id}/nodes/{self.node_id}"
)
_url = f"{self.connector.base_url}/projects/{self.project_id}/nodes/{self.node_id}"
_response = self.connector.http_call("get", _url)
# Update object
@ -1148,10 +1094,7 @@ class Node:
if not self.project_id:
raise ValueError("Need to submit project_id")
_url = (
f"{self.connector.base_url}/projects/{self.project_id}/nodes"
f"/{self.node_id}/links"
)
_url = f"{self.connector.base_url}/projects/{self.project_id}/nodes" f"/{self.node_id}/links"
_response = self.connector.http_call("get", _url)
# Create the Link array but cleanup cache if there is one
@ -1253,10 +1196,7 @@ class Node:
error_detail = _response.json()
except Exception:
error_detail = _response.text
_msg = (
f"Failed to stop node: {_response.status_code}, "
f"Detail: {error_detail}"
)
_msg = f"Failed to stop node: {_response.status_code}, " f"Detail: {error_detail}"
raise RuntimeError(_msg) from None
@verify_connector_and_id
@ -1305,10 +1245,7 @@ class Node:
error_detail = _response.json()
except Exception:
error_detail = _response.text
_msg = (
f"Failed to reload node: {_response.status_code}, "
f"Detail: {error_detail}"
)
_msg = f"Failed to reload node: {_response.status_code}, " f"Detail: {error_detail}"
raise RuntimeError(_msg) from None
@verify_connector_and_id
@ -1398,9 +1335,7 @@ class Node:
_template = self.connector.get_template(name=self.template)
if _template is None:
raise ValueError(f"Template {self.template} not found")
self.template_id = self.connector.get_template(name=self.template).get(
"template_id"
)
self.template_id = self.connector.get_template(name=self.template).get("template_id")
else:
raise ValueError("Need either 'template' of 'template_id'")
@ -1419,14 +1354,9 @@ class Node:
if v is not None
}
_url = (
f"{self.connector.base_url}/projects/{self.project_id}/"
f"templates/{self.template_id}"
)
_url = f"{self.connector.base_url}/projects/{self.project_id}/" f"templates/{self.template_id}"
_response = self.connector.http_call(
"post", _url, json_data={"x": 0, "y": 0, "compute_id": self.compute_id}
)
_response = self.connector.http_call("post", _url, json_data={"x": 0, "y": 0, "compute_id": self.compute_id})
self._update(_response.json())
@ -1517,6 +1447,7 @@ class Node:
_conn.http_call("post", _url, data=data)
@dataclass(config=config)
class Project:
"""
@ -1604,9 +1535,7 @@ class Project:
if k in self.__dict__:
setattr(self, k, v)
def get(
self, get_links: bool = True, get_nodes: bool = True, get_stats: bool = True
) -> None:
def get(self, get_links: bool = True, get_nodes: bool = True, get_stats: bool = True) -> None:
"""
Retrieves the projects information.
@ -2045,10 +1974,7 @@ class Project:
_nodes_summary = []
for _n in self.nodes:
if is_print:
print(
f"{_n.name}: {_n.status} -- Console: {_n.console} -- "
f"ID: {_n.node_id}"
)
print(f"{_n.name}: {_n.status} -- Console: {_n.console} -- " f"ID: {_n.node_id}")
_nodes_summary.append((_n.name, _n.status, _n.console, _n.node_id))
return _nodes_summary if not is_print else None
@ -2084,9 +2010,7 @@ class Project:
_nodes_inventory = {}
conn = self.connector
if not conn:
raise ValueError(
"Gns3Connector not assigned. Please set the connector first."
)
raise ValueError("Gns3Connector not assigned. Please set the connector first.")
_server = urlparse(conn.base_url).hostname
@ -2112,9 +2036,7 @@ class Project:
return _nodes_inventory
def links_summary(
self, is_print: bool = True
) -> list[tuple[str, str, str, str]] | None:
def links_summary(self, is_print: bool = True) -> list[tuple[str, str, str, str]] | None:
"""
Returns a summary of the links insode the project. If `is_print` is False, it
will return a list of tuples like:
@ -2194,9 +2116,7 @@ class Project:
except IndexError:
return None
def get_node(
self, name: str | None = None, node_id: str | None = None
) -> Any | None:
def get_node(self, name: str | None = None, node_id: str | None = None) -> Any | None:
"""
Returns the Node object by searching for the `name` or the `node_id`.
@ -2269,10 +2189,7 @@ class Project:
_node.create()
self.nodes.append(_node)
print(
f"Created: {_node.name} -- Type: {_node.node_type} -- "
f"Console: {_node.console}"
)
print(f"Created: {_node.name} -- Type: {_node.node_type} -- " f"Console: {_node.console}")
def create_link(self, node_a: str, port_a: str, node_b: str, port_b: str) -> None:
"""
@ -2407,9 +2324,7 @@ class Project:
):
_matches.append(_l)
if not _matches:
raise ValueError(
f"Link not found: {node_a, port_a, node_b, port_b}"
) # pragma: no cover
raise ValueError(f"Link not found: {node_a, port_a, node_b, port_b}") # pragma: no cover
# now to delete the link via GNS3_api
_link = _matches[0]
@ -2417,8 +2332,7 @@ class Project:
_link_id = _link.link_id
_link.delete()
print(
f"Deleted Link-ID: {_link_id} From node {node_a}, port: {port_a} <--> "
f"to node {node_b}, port: {port_b}"
f"Deleted Link-ID: {_link_id} From node {node_a}, port: {port_a} <--> " f"to node {node_b}, port: {port_b}"
)
@verify_connector_and_id
@ -2451,9 +2365,7 @@ class Project:
except StopIteration:
return None
def get_snapshot(
self, name: str | None = None, snapshot_id: str | None = None
) -> dict[str, Any] | None:
def get_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> dict[str, Any] | None:
"""
Returns the Snapshot by searching for the `name` or the `snapshot_id`.
@ -2510,9 +2422,7 @@ class Project:
print(f"Created snapshot: {_snapshot['name']}")
@verify_connector_and_id
def delete_snapshot(
self, name: str | None = None, snapshot_id: str | None = None
) -> None:
def delete_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> None:
"""
Deletes a snapshot of the project
@ -2536,19 +2446,14 @@ class Project:
if not _snapshot:
raise ValueError("Snapshot not found")
_url = (
f"{_conn.base_url}/projects/{_project_id}/snapshots/"
f"{_snapshot['snapshot_id']}"
)
_url = f"{_conn.base_url}/projects/{_project_id}/snapshots/" f"{_snapshot['snapshot_id']}"
_conn.http_call("delete", _url)
self.get_snapshots()
@verify_connector_and_id
def restore_snapshot(
self, name: str | None = None, snapshot_id: str | None = None
) -> None:
def restore_snapshot(self, name: str | None = None, snapshot_id: str | None = None) -> None:
"""
Restore a snapshot from disk
@ -2572,10 +2477,7 @@ class Project:
if not _snapshot:
raise ValueError("Snapshot not found")
_url = (
f"{_conn.base_url}/projects/{_project_id}/snapshots/"
f"{_snapshot['snapshot_id']}/restore"
)
_url = f"{_conn.base_url}/projects/{_project_id}/snapshots/" f"{_snapshot['snapshot_id']}/restore"
_conn.http_call("post", _url)
@ -2626,11 +2528,7 @@ class Project:
self.get_drawings()
try:
return next(
_drawing
for _drawing in (self.drawings or [])
if _drawing["drawing_id"] == drawing_id
)
return next(_drawing for _drawing in (self.drawings or []) if _drawing["drawing_id"] == drawing_id)
except (StopIteration, KeyError, TypeError):
return None
@ -2793,9 +2691,7 @@ class Project:
if not _drawing:
raise ValueError("drawing not found")
_url = (
f"{_conn.base_url}/projects/{_project_id}/drawings/{_drawing['drawing_id']}"
)
_url = f"{_conn.base_url}/projects/{_project_id}/drawings/{_drawing['drawing_id']}"
_conn.http_call("delete", _url)

22
gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py
View File

@ -38,11 +38,13 @@ from typing import Any
from langchain.tools import BaseTool
from gns3server.agent.gns3_copilot.gns3_client import Project, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Project
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3ProjectInfoTool(BaseTool):
"""LangChain tool for retrieving GNS3 project basic information."""
@ -88,9 +90,7 @@ class GNS3ProjectInfoTool(BaseTool):
# Validate project_id parameter
if not project_id:
logger.error("project_id parameter is required.")
return {
"error": "project_id parameter is required. Please provide a valid project UUID."
}
return {"error": "project_id parameter is required. Please provide a valid project UUID."}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
@ -98,9 +98,7 @@ class GNS3ProjectInfoTool(BaseTool):
if server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Use the provided project_id directly
logger.info(f"Retrieving project info for project_id: {project_id}")
@ -125,8 +123,13 @@ class GNS3ProjectInfoTool(BaseTool):
}
# Log result
logger.info("Project info retrieved: name=%s, status=%s, nodes=%d, links=%d",
project.name, project.status, node_count, link_count)
logger.info(
"Project info retrieved: name=%s, status=%s, nodes=%d, links=%d",
project.name,
project.status,
node_count,
link_count,
)
return result
@ -134,6 +137,7 @@ class GNS3ProjectInfoTool(BaseTool):
logger.error("Error retrieving GNS3 project info: %s", str(e))
return {"error": f"Failed to retrieve project info: {str(e)}"}
if __name__ == "__main__":
# Test the tool
tool = GNS3ProjectInfoTool()

34
gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py
View File

@ -38,11 +38,13 @@ from typing import Any
from langchain.tools import BaseTool
from gns3server.agent.gns3_copilot.gns3_client import Project, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Project
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
# Define LangChain tool class
class GNS3TopologyTool(BaseTool):
"""LangChain tool for retrieving GNS3 project topology information."""
@ -87,9 +89,7 @@ class GNS3TopologyTool(BaseTool):
# Validate project_id parameter
if not project_id:
logger.error("project_id parameter is required.")
return {
"error": "project_id parameter is required. Please provide a valid project UUID."
}
return {"error": "project_id parameter is required. Please provide a valid project UUID."}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
@ -97,9 +97,7 @@ class GNS3TopologyTool(BaseTool):
if server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Use the provided project_id directly
logger.info(f"Retrieving topology for project_id: {project_id}")
@ -111,18 +109,18 @@ class GNS3TopologyTool(BaseTool):
"project_id": project.project_id,
"name": project.name,
"status": project.status,
"nodes": self._clean_nodes_ports(
copy.deepcopy(project.nodes_inventory())
),
"nodes": self._clean_nodes_ports(copy.deepcopy(project.nodes_inventory())),
"links": project.links_summary(is_print=False),
}
# Log topology result
logger.info("Topology retrieved: project_id=%s, name=%s, nodes=%d, links=%d",
topology.get("project_id"),
topology.get("name"),
len(topology.get("nodes", {})),
len(topology.get("links", [])))
logger.info(
"Topology retrieved: project_id=%s, name=%s, nodes=%d, links=%d",
topology.get("project_id"),
topology.get("name"),
len(topology.get("nodes", {})),
len(topology.get("links", [])),
)
logger.debug("Topology details: %s", topology)
return topology
@ -138,12 +136,10 @@ class GNS3TopologyTool(BaseTool):
"""
for node in data.values(): # Iterate through R-1, R-2, R-3, R-4
if "ports" in node and isinstance(node["ports"], list):
node["ports"] = [
{"name": port["name"], "short_name": port["short_name"]}
for port in node["ports"]
]
node["ports"] = [{"name": port["name"], "short_name": port["short_name"]} for port in node["ports"]]
return data
if __name__ == "__main__":
# Test the tool
tool = GNS3TopologyTool()

11
gns3server/agent/gns3_copilot/project_agent_manager.py
View File

@ -32,12 +32,14 @@ Each project has its own AgentService with a dedicated SQLite checkpoint databas
import asyncio
import logging
from typing import Dict, Optional
from typing import Dict
from typing import Optional
from gns3server.agent.gns3_copilot.agent_service import AgentService
log = logging.getLogger(__name__)
class ProjectAgentManager:
"""
Singleton manager for project-level Agent services.
@ -48,12 +50,13 @@ class ProjectAgentManager:
"""
_instance: Optional["ProjectAgentManager"] = None
_lock = asyncio.Lock()
_lock: asyncio.Lock = asyncio.Lock()
_agents: Dict[str, AgentService] = {}
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._agents: Dict[str, AgentService] = {}
cls._instance._agents = {}
cls._instance._lock = asyncio.Lock()
return cls._instance
@ -124,10 +127,12 @@ class ProjectAgentManager:
"""
return list(self._agents.keys())
# Global singleton instance
_project_agent_manager: Optional[ProjectAgentManager] = None
_manager_lock = asyncio.Lock()
async def get_project_agent_manager() -> ProjectAgentManager:
"""
Get the global ProjectAgentManager singleton instance.

1
gns3server/agent/gns3_copilot/prompts/prompt_loader.py
View File

@ -42,6 +42,7 @@ from .lab_assistant_prompt import LAB_ASSISTANT_PROMPT
logger = logging.getLogger(__name__)
def load_system_prompt(llm_config: dict | None = None) -> str:
"""
Load the system prompt for GNS3-Copilot.

5
gns3server/agent/gns3_copilot/tools_v2/__init__.py
View File

@ -54,7 +54,8 @@ from .display_tools_nornir import ExecuteMultipleDeviceCommands
from .gns3_create_link import GNS3LinkTool
from .gns3_create_node import GNS3CreateNodeTool
from .gns3_get_node_temp import GNS3TemplateTool
from .gns3_start_node import GNS3StartNodeTool, GNS3StartNodeQuickTool
from .gns3_start_node import GNS3StartNodeQuickTool
from .gns3_start_node import GNS3StartNodeTool
from .gns3_update_node_name import GNS3UpdateNodeNameTool
from .vpcs_tools_telnetlib3 import VPCSMultiCommands
@ -84,4 +85,4 @@ __all__ = [
]
# Package initialization message
# print(f"GNS3-Copilot Tools package loaded (version {__version__})")
# print(f"GNS3-Copilot Tools package loaded (version {__version__})")

91
gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py
View File

@ -30,7 +30,6 @@ This module provides a tool to execute configuration commands on multiple device
import json
import logging
import os
import re
from typing import Any
@ -39,7 +38,9 @@ from langchain_core.callbacks import CallbackManagerForToolRun
from netmiko.exceptions import ReadTimeout
from nornir import InitNornir
from nornir.core import Nornir
from nornir.core.task import AggregatedResult, Result, Task
from nornir.core.task import AggregatedResult
from nornir.core.task import Result
from nornir.core.task import Task
from nornir_netmiko.tasks import netmiko_send_config
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
@ -48,15 +49,14 @@ from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
# config log
logger = logging.getLogger(__name__)
# Local Nornir configuration functions for Cisco IOS Telnet devices
def _get_nornir_defaults() -> dict[str, Any]:
"""Get Nornir default configuration for Cisco IOS."""
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config(
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios"
) -> dict[str, Any]:
def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]:
"""
Get Nornir group configuration for Cisco IOS Telnet devices.
@ -73,15 +73,12 @@ def _get_nornir_groups_config(
"timeout": 120,
"username": "",
"password": "",
"connection_options": {
"netmiko": {"extras": {"device_type": device_type}}
},
"connection_options": {"netmiko": {"extras": {"device_type": device_type}}},
}
def _get_nornir_group(
group_name: str = "cisco_IOSv_telnet",
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios"
group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
) -> dict[str, Any]:
"""
Get Nornir group configuration for a specific group.
@ -95,10 +92,8 @@ def _get_nornir_group(
Dictionary containing group configuration
"""
# _get_nornir_groups_config now returns the group config directly
return _get_nornir_groups_config(
device_type=device_type,
platform=platform
)
return _get_nornir_groups_config(device_type=device_type, platform=platform)
class ExecuteMultipleDeviceConfigCommands(BaseTool):
"""
@ -174,11 +169,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Validate input
device_configs_list, project_id = self._validate_tool_input(tool_input)
if (
isinstance(device_configs_list, list)
and len(device_configs_list) > 0
and "error" in device_configs_list[0]
):
if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
return device_configs_list
# Create a mapping of device names to their configuration commands
@ -186,9 +177,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Prepare device hosts data
try:
hosts_data = self._prepare_device_hosts_data(
device_configs_list, project_id
)
hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id)
except ValueError as e:
logger.error("Failed to prepare device hosts data: %s", e)
return [{"error": str(e)}]
@ -210,9 +199,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
)
# Process results for all devices
results = self._process_task_results(
device_configs_list, hosts_data, task_result
)
results = self._process_task_results(device_configs_list, hosts_data, task_result)
except Exception as e:
# Overall execution failed
@ -226,9 +213,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return results
def _run_all_device_configs_with_single_retry(
self, task: Task, device_configs_map: dict[str, list[str]]
) -> Result:
def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result:
"""Execute configuration commands with single retry mechanism."""
device_name = task.host.name
config_commands = device_configs_map.get(device_name, [])
@ -237,9 +222,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return Result(host=task.host, result="No configuration commands to execute")
try:
_result = task.run(
task=netmiko_send_config, config_commands=config_commands
)
_result = task.run(task=netmiko_send_config, config_commands=config_commands)
return Result(host=task.host, result=_result.result)
except ReadTimeout as e:
@ -259,9 +242,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Handle prompt detection issues with Cisco IOSv L2 images where the '#' prompt character
# may be delayed, causing Netmiko prompt detection failures. Implements retry logic.
if "netmiko_send_config (failed)" in str(e):
_result = task.run(
task=netmiko_send_config, config_commands=config_commands
)
_result = task.run(task=netmiko_send_config, config_commands=config_commands)
return Result(host=task.host, result=_result.result)
# Log any other exceptions with full details
@ -307,9 +288,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Handle standard models (like GPT/OpenAI) where the framework
# has already parsed the JSON into a Python object (dict or list).
parsed_input = tool_input
logger.info(
"Using tool input directly as type: %s", type(parsed_input).__name__
)
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
# Handle new format: {"project_id": "...", "device_configs": [...]}
if isinstance(parsed_input, dict):
@ -323,9 +302,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return ([{"error": error_msg}], None)
if not self._validate_project_id(project_id):
error_msg = (
f"Invalid project_id format: {project_id}. Expected UUID format."
)
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
logger.error(error_msg)
return ([{"error": error_msg}], None)
@ -343,9 +320,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Handle legacy format: [...]
elif isinstance(parsed_input, list):
logger.warning(
"Using legacy input format without project_id. Please use new format with project_id."
)
logger.warning("Using legacy input format without project_id. Please use new format with project_id.")
return parsed_input, None
else:
@ -369,9 +344,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
def _configs_map(
self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]:
def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
"""Create a mapping of device names to their configuration commands."""
device_configs_map = {}
for device_config in device_config_list:
@ -386,9 +359,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
) -> dict[str, dict[str, Any]]:
"""Prepare device hosts data from topology information."""
# Extract device names list
device_names = [
device_config["device_name"] for device_config in device_config_list
]
device_names = [device_config["device_name"] for device_config in device_config_list]
# Get device port information with project_id
hosts_data = get_device_ports_from_topology(device_names, project_id)
@ -424,16 +395,11 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
device_type = first_device_data.get("device_type", "cisco_ios_telnet")
platform = first_device_data.get("platform", "cisco_ios")
logger.info(
"Extracted from tags: device_type=%s, platform=%s",
device_type,
platform
)
logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform)
# Get latest environment configuration with dynamic device_type and platform
groups_data = _get_nornir_groups_config(
device_type=device_type or "cisco_ios_telnet",
platform=platform or "cisco_ios"
device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios"
)
defaults = _get_nornir_defaults()
@ -484,9 +450,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (
f"Device '{device_name}' not found in topology or missing console_port"
),
"error": (f"Device '{device_name}' not found in topology or missing console_port"),
}
results.append(device_result)
continue
@ -508,9 +472,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
if multi_result[0].failed:
# Execution failed
device_result["status"] = "failed"
device_result["error"] = (
f"Configuration execution failed: {multi_result[0].result}"
)
device_result["error"] = f"Configuration execution failed: {multi_result[0].result}"
device_result["output"] = multi_result[0].result
else:
# Execution successful
@ -522,6 +484,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return results
if __name__ == "__main__":
# Example usage with new format
# example tool_input with project_id

86
gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py
View File

@ -30,34 +30,33 @@ This module provides a tool to execute display commands on multiple devices
import json
import logging
import os
import re
from typing import Any
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from netmiko.exceptions import ReadTimeout
from nornir import InitNornir
from nornir.core import Nornir
from nornir.core.task import AggregatedResult, Result, Task
from nornir.core.task import AggregatedResult
from nornir.core.task import Result
from nornir.core.task import Task
from nornir_netmiko.tasks import netmiko_multiline
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
# config log
logger = logging.getLogger(__name__)
# Local Nornir configuration functions for Cisco IOS Telnet devices
def _get_nornir_defaults() -> dict[str, Any]:
"""Get Nornir default configuration for Cisco IOS."""
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config(
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios"
) -> dict[str, Any]:
def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]:
"""
Get Nornir group configuration for Cisco IOS Telnet devices.
@ -74,15 +73,12 @@ def _get_nornir_groups_config(
"timeout": 120,
"username": "",
"password": "",
"connection_options": {
"netmiko": {"extras": {"device_type": device_type}}
},
"connection_options": {"netmiko": {"extras": {"device_type": device_type}}},
}
def _get_nornir_group(
group_name: str = "cisco_IOSv_telnet",
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios"
group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
) -> dict[str, Any]:
"""
Get Nornir group configuration for a specific group.
@ -96,10 +92,8 @@ def _get_nornir_group(
Dictionary containing group configuration
"""
# _get_nornir_groups_config now returns the group config directly
return _get_nornir_groups_config(
device_type=device_type,
platform=platform
)
return _get_nornir_groups_config(device_type=device_type, platform=platform)
class ExecuteMultipleDeviceCommands(BaseTool):
"""
@ -182,11 +176,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Validate input
device_configs_list, project_id = self._validate_tool_input(tool_input)
if (
isinstance(device_configs_list, list)
and len(device_configs_list) > 0
and "error" in device_configs_list[0]
):
if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
return device_configs_list
# Create a mapping of device names to their display commands
@ -194,9 +184,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Prepare device hosts data
try:
hosts_data = self._prepare_device_hosts_data(
device_configs_list, project_id
)
hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id)
except ValueError as e:
logger.error("Failed to prepare device hosts data: %s", e)
return [{"error": str(e)}]
@ -218,9 +206,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
)
# Process results for all devices
results = self._process_task_results(
device_configs_list, hosts_data, task_result
)
results = self._process_task_results(device_configs_list, hosts_data, task_result)
except Exception as e:
# Overall execution failed
@ -234,9 +220,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return results
def _run_all_device_configs_with_single_retry(
self, task: Task, device_configs_map: dict[str, list[str]]
) -> Result:
def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result:
"""Execute READ-ONLY diagnostic commands with single retry mechanism."""
device_name = task.host.name
diagnostic_commands = device_configs_map.get(device_name, [])
@ -323,9 +307,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Handle standard models (like GPT/OpenAI) where the framework
# has already parsed the JSON into a Python object (dict or list).
parsed_input = tool_input
logger.info(
"Using tool input directly as type: %s", type(parsed_input).__name__
)
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
# Handle new format: {"project_id": "...", "device_configs": [...]}
if isinstance(parsed_input, dict):
@ -339,9 +321,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return ([{"error": error_msg}], None)
if not self._validate_project_id(project_id):
error_msg = (
f"Invalid project_id format: {project_id}. Expected UUID format."
)
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
logger.error(error_msg)
return ([{"error": error_msg}], None)
@ -359,9 +339,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Handle legacy format: [...]
elif isinstance(parsed_input, list):
logger.warning(
"Using legacy input format without project_id. Please use new format with project_id."
)
logger.warning("Using legacy input format without project_id. Please use new format with project_id.")
return parsed_input, None
else:
@ -385,9 +363,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
def _configs_map(
self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]:
def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
"""Create a mapping of device names to their diagnostic commands."""
device_diagnostic_map = {}
for device_config in device_config_list:
@ -402,9 +378,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
) -> dict[str, dict[str, Any]]:
"""Prepare device hosts data from topology information."""
# Extract device names list
device_names = [
device_config["device_name"] for device_config in device_config_list
]
device_names = [device_config["device_name"] for device_config in device_config_list]
# Get device port information with project_id
hosts_data = get_device_ports_from_topology(device_names, project_id)
@ -440,16 +414,11 @@ class ExecuteMultipleDeviceCommands(BaseTool):
device_type = first_device_data.get("device_type", "cisco_ios_telnet")
platform = first_device_data.get("platform", "cisco_ios")
logger.info(
"Extracted from tags: device_type=%s, platform=%s",
device_type,
platform
)
logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform)
# Get latest environment configuration with dynamic device_type and platform
groups_data = _get_nornir_groups_config(
device_type=device_type or "cisco_ios_telnet",
platform=platform or "cisco_ios"
device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios"
)
defaults = _get_nornir_defaults()
@ -500,9 +469,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (
f"Device '{device_name}' not found in topology or missing console_port"
),
"error": (f"Device '{device_name}' not found in topology or missing console_port"),
}
results.append(device_result)
continue
@ -524,9 +491,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
if multi_result[0].failed:
# Execution failed
device_result["status"] = "failed"
device_result["error"] = (
f"Diagnostic command execution failed: {multi_result[0].result}"
)
device_result["error"] = f"Diagnostic command execution failed: {multi_result[0].result}"
device_result["output"] = multi_result[0].result
else:
# Execution successful
@ -538,6 +503,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return results
if __name__ == "__main__":
# Example usage with new format
device_commands = json.dumps(

35
gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py
View File

@ -38,11 +38,13 @@ from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3server.agent.gns3_copilot.gns3_client import Link, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Link
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3LinkTool(BaseTool):
"""
Tool for creating network links between GNS3 nodes.
@ -96,9 +98,7 @@ class GNS3LinkTool(BaseTool):
]
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> list[dict[str, Any]]:
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> list[dict[str, Any]]:
"""
Creates one or multiple links between nodes in a GNS3 project.
@ -133,11 +133,7 @@ class GNS3LinkTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return [
{
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
]
return [{"error": "Failed to connect to GNS3 server. Please check your configuration."}]
created_links = []
@ -160,12 +156,8 @@ class GNS3LinkTool(BaseTool):
continue
# Get node details
node1 = gns3_server.get_node(
project_id=project_id, node_id=node_id1
)
node2 = gns3_server.get_node(
project_id=project_id, node_id=node_id2
)
node1 = gns3_server.get_node(project_id=project_id, node_id=node_id1)
node2 = gns3_server.get_node(project_id=project_id, node_id=node_id2)
if not node1 or not node2:
error_msg = f"Node not found in link {i}"
logger.error(error_msg)
@ -174,19 +166,11 @@ class GNS3LinkTool(BaseTool):
# Find port information
port1_info = next(
(
port
for port in node1.get("ports", [])
if port.get("name") == port1
),
(port for port in node1.get("ports", []) if port.get("name") == port1),
None,
)
port2_info = next(
(
port
for port in node2.get("ports", [])
if port.get("name") == port2
),
(port for port in node2.get("ports", []) if port.get("name") == port2),
None,
)
if not port1_info or not port2_info:
@ -249,6 +233,7 @@ class GNS3LinkTool(BaseTool):
logger.error("Failed to process link creation: %s", e)
return [{"error": f"Failed to process link creation: {str(e)}"}]
if __name__ == "__main__":
# Test with single link
single_link_input = json.dumps(

13
gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py
View File

@ -38,11 +38,13 @@ from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Node
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3CreateNodeTool(BaseTool):
"""
A LangChain tool to create multiple nodes in a GNS3 project
@ -174,9 +176,7 @@ class GNS3CreateNodeTool(BaseTool):
"Invalid input: Node %d missing or invalid template_id, x, or y.",
i + 1,
)
return {
"error": f"Node {i + 1} missing or invalid template_id, x, or y."
}
return {"error": f"Node {i + 1} missing or invalid template_id, x, or y."}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
@ -184,9 +184,7 @@ class GNS3CreateNodeTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Create nodes
logger.info("Creating %d nodes in project %s...", len(nodes), project_id)
@ -267,6 +265,7 @@ class GNS3CreateNodeTool(BaseTool):
logger.error("Failed to process node creation request: %s", e)
return {"error": f"Failed to process node creation request: {str(e)}"}
if __name__ == "__main__":
# Test the tool locally with multiple nodes
test_input = json.dumps(

10
gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py
View File

@ -43,6 +43,7 @@ from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3TemplateTool(BaseTool):
"""
A LangChain tool to retrieve all available device templates from a GNS3 server.
@ -102,9 +103,7 @@ class GNS3TemplateTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Retrieve all available templates
templates = gns3_server.get_templates()
@ -128,11 +127,10 @@ class GNS3TemplateTool(BaseTool):
return result
except Exception as e:
logger.error(
"Failed to connect to GNS3 server or retrieve templates: %s", e
)
logger.error("Failed to connect to GNS3 server or retrieve templates: %s", e)
return {"error": f"Failed to retrieve templates: {str(e)}"}
if __name__ == "__main__":
# Test's tool locally
tool = GNS3TemplateTool()

59
gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py
View File

@ -39,14 +39,14 @@ from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Node
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
def show_progress_bar(
duration: int = 120, interval: int = 1, node_count: int = 1
) -> None:
def show_progress_bar(duration: int = 120, interval: int = 1, node_count: int = 1) -> None:
"""
Display a simple text progress bar for node startup.
@ -63,9 +63,7 @@ def show_progress_bar(
# Create progress bar display
bar_length = 30
filled_length = int(bar_length * elapsed // duration)
progress_string = (
"=" * filled_length + ">" + " " * (bar_length - filled_length - 1)
)
progress_string = "=" * filled_length + ">" + " " * (bar_length - filled_length - 1)
# Print progress bar with node count
print(f"\r[{progress_string}] {progress:.1f}%", end="", flush=True)
@ -73,6 +71,7 @@ def show_progress_bar(
print(f"\n{node_count} node(s) startup completed!")
class GNS3StartNodeTool(BaseTool):
"""
A LangChain tool to start one or multiple nodes in a GNS3 project.
@ -106,9 +105,7 @@ class GNS3StartNodeTool(BaseTool):
Returns: A dictionary with all nodes' details including success/failure status.
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> dict[str, Any]:
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@ -130,9 +127,7 @@ class GNS3StartNodeTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# First loop: Send start commands for all nodes
logger.info(
@ -142,22 +137,16 @@ class GNS3StartNodeTool(BaseTool):
)
for node_id in node_ids:
try:
node = Node(
project_id=project_id, node_id=node_id, connector=gns3_server
)
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
# Verify node exists
node.get()
if node.node_id:
node.start()
logger.info("Start command sent for node %s", node_id)
else:
logger.error(
"Node %s not found in project %s", node_id, project_id
)
logger.error("Node %s not found in project %s", node_id, project_id)
except Exception as e:
logger.error(
"Failed to send start command for node %s: %s", node_id, e
)
logger.error("Failed to send start command for node %s: %s", node_id, e)
# Calculate progress bar duration: 140s base + 10s per additional node
base_duration = 140
@ -165,18 +154,14 @@ class GNS3StartNodeTool(BaseTool):
total_duration = base_duration + extra_duration
# Show progress bar
show_progress_bar(
duration=total_duration, interval=1, node_count=len(node_ids)
)
show_progress_bar(duration=total_duration, interval=1, node_count=len(node_ids))
# Second loop: Get status for all nodes
results = []
logger.info("Retrieving status for %d nodes...", len(node_ids))
for node_id in node_ids:
try:
node = Node(
project_id=project_id, node_id=node_id, connector=gns3_server
)
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node.get() # Get latest status
node_info = {
"node_id": node.node_id,
@ -223,6 +208,7 @@ class GNS3StartNodeTool(BaseTool):
logger.error("Failed to start nodes: %s", e)
return {"error": f"Failed to start nodes: {str(e)}"}
class GNS3StartNodeQuickTool(BaseTool):
"""
A LangChain tool to start one or multiple nodes in a GNS3 project WITHOUT waiting.
@ -263,9 +249,7 @@ class GNS3StartNodeQuickTool(BaseTool):
NOTE: Nodes will continue booting in background after this tool returns.
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> dict[str, Any]:
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@ -287,9 +271,7 @@ class GNS3StartNodeQuickTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Send start commands for all nodes and collect initial status
logger.info(
@ -301,9 +283,7 @@ class GNS3StartNodeQuickTool(BaseTool):
for node_id in node_ids:
try:
node = Node(
project_id=project_id, node_id=node_id, connector=gns3_server
)
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
# Verify node exists and get current info
node.get()
if not node.node_id:
@ -371,15 +351,14 @@ class GNS3StartNodeQuickTool(BaseTool):
logger.error("Failed to start nodes: %s", e)
return {"error": f"Failed to start nodes: {str(e)}"}
if __name__ == "__main__":
# Test with single node
print("=== Testing single node startup ===")
test_input_single = json.dumps(
{
"project_id": "<PROJECT_UUID>", # Replace with actual project UUID
"node_ids": [
"fbeda109-9a74-4d8c-a749-cc3847911a90"
], # Replace with actual node UUID
"node_ids": ["fbeda109-9a74-4d8c-a749-cc3847911a90"], # Replace with actual node UUID
}
)
tool = GNS3StartNodeTool()

21
gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py
View File

@ -37,11 +37,13 @@ from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3server.agent.gns3_copilot.gns3_client import Node, get_gns3_connector
from gns3server.agent.gns3_copilot.gns3_client import Node
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3UpdateNodeNameTool(BaseTool):
"""
A LangChain tool to update the name of one or multiple nodes in a GNS3 project.
@ -78,9 +80,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
Returns: A dictionary with all nodes' update results including success/failure status.
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> dict[str, Any]:
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@ -111,9 +111,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
# Update node names
logger.info("Updating names for %d nodes in project %s...", len(nodes), project_id)
@ -133,9 +131,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
)
# Get node to retrieve current name
node = Node(
project_id=project_id, node_id=node_id, connector=gns3_server
)
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node.get()
old_name = node.name
@ -152,9 +148,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
"status": "success",
}
results.append(node_info)
logger.info(
"Successfully updated node name: %s -> %s", old_name, new_name
)
logger.info("Successfully updated node name: %s -> %s", old_name, new_name)
else:
error_info = {
"node_id": node_id,
@ -205,6 +199,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
logger.error("Failed to update node names: %s", e)
return {"error": f"Failed to update node names: {str(e)}"}
if __name__ == "__main__":
# Test with single node
print("=== Testing single node name update ===")

37
gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py
View File

@ -30,7 +30,6 @@ Supports concurrent execution of multiple command groups across multiple VPCS de
import json
import logging
import os
import re
import threading
from time import sleep
@ -45,6 +44,7 @@ from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
logger = logging.getLogger(__name__)
class VPCSMultiCommands(BaseTool):
"""
A tool for VPCS (Virtual PC Simulator) devices to view PC configurations and test connectivity.
@ -131,9 +131,7 @@ class VPCSMultiCommands(BaseTool):
# Check if device has port information
if device_name not in device_ports:
logger.warning(
"Device '%s' not found in topology or missing console port", device_name
)
logger.warning("Device '%s' not found in topology or missing console port", device_name)
results_list[index] = {
"device_name": device_name,
"status": "error",
@ -199,9 +197,7 @@ class VPCSMultiCommands(BaseTool):
)
except Exception as e:
logger.error(
"Error executing commands on device '%s': %s", device_name, str(e)
)
logger.error("Error executing commands on device '%s': %s", device_name, str(e))
results_list[index] = {
"device_name": device_name,
"status": "error",
@ -256,9 +252,7 @@ class VPCSMultiCommands(BaseTool):
else:
# Handle standard models where the framework has already parsed the JSON.
parsed_input = tool_input
logger.info(
"Using tool input directly as type: %s", type(parsed_input).__name__
)
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
# Validate input is a dictionary
if not isinstance(parsed_input, dict):
@ -278,9 +272,7 @@ class VPCSMultiCommands(BaseTool):
# Validate project_id format
if not self._validate_project_id(project_id):
error_msg = (
f"Invalid project_id format: {project_id}. Expected UUID format."
)
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
logger.error(error_msg)
return ([{"error": error_msg}], "")
@ -305,9 +297,7 @@ class VPCSMultiCommands(BaseTool):
# Validate each item in device_configs
for i, item in enumerate(device_configs):
if not isinstance(item, dict):
error_msg = (
f"Item at index {i} must be a dictionary, got {type(item).__name__}"
)
error_msg = f"Item at index {i} must be a dictionary, got {type(item).__name__}"
logger.error(error_msg)
return ([{"error": error_msg}], "")
@ -324,8 +314,7 @@ class VPCSMultiCommands(BaseTool):
if not isinstance(item["commands"], list):
error_msg = (
f"'commands' in item at index {i} must be a list, "
f"but got {type(item['commands']).__name__}"
f"'commands' in item at index {i} must be a list, " f"but got {type(item['commands']).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
@ -363,20 +352,14 @@ class VPCSMultiCommands(BaseTool):
device_configs, project_id = self._validate_tool_input(tool_input)
# Check if validation returned an error
if (
isinstance(device_configs, list)
and len(device_configs) > 0
and "error" in device_configs[0]
):
if isinstance(device_configs, list) and len(device_configs) > 0 and "error" in device_configs[0]:
return device_configs
# Extract all device names from input using set comprehension
device_names = {config["device_name"] for config in device_configs}
# Get device port mapping with project_id
device_ports = get_device_ports_from_topology(
list(device_names), project_id=project_id
)
device_ports = get_device_ports_from_topology(list(device_names), project_id=project_id)
logger.info(
"Retrieved port mappings for %d devices: %s",
len(device_ports),
@ -394,7 +377,6 @@ class VPCSMultiCommands(BaseTool):
# Create thread for each command group
logger.info("Starting parallel execution for %d devices", len(device_configs))
for i, cmd_group in enumerate(device_configs):
device_name = cmd_group["device_name"]
thread = threading.Thread(
target=self._connect_and_execute_commands,
args=(
@ -426,6 +408,7 @@ class VPCSMultiCommands(BaseTool):
return results
if __name__ == "__main__":
# Example usage
command_groups = json.dumps(

8
gns3server/agent/gns3_copilot/utils/__init__.py
View File

@ -38,11 +38,9 @@ Author: Guobin Yue
# Import main utility functions
from .get_gns3_device_port import get_device_ports_from_topology
from .parse_tool_content import (
format_tool_response,
normalize_tool_response,
parse_tool_content
)
from .parse_tool_content import format_tool_response
from .parse_tool_content import normalize_tool_response
from .parse_tool_content import parse_tool_content
# Dynamic version management
try:

1
gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py
View File

@ -32,6 +32,7 @@ from typing import Any
logger = logging.getLogger(__name__)
def get_device_ports_from_topology(
device_names: list[str],
project_id: str | None = None,

76
gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py
View File

@ -31,7 +31,8 @@ Supports ellipse and rectangle shapes for two-node annotations.
"""
import math
from typing import Any, Literal
from typing import Any
from typing import Literal
# Default parameters
DEFAULT_DEVICE_WIDTH = 50
@ -103,6 +104,7 @@ COLOR_SCHEMES = {
},
}
def calculate_two_node_shape(
node1: dict,
node2: dict,
@ -133,13 +135,9 @@ def calculate_two_node_shape(
node2_center_x = node2["x"] + (node2_width / 2)
node2_center_y = node2["y"] + (node2_height / 2)
distance = math.sqrt(
(node2_center_x - node1_center_x) ** 2 + (node2_center_y - node1_center_y) ** 2
)
distance = math.sqrt((node2_center_x - node1_center_x) ** 2 + (node2_center_y - node1_center_y) ** 2)
angle_rad = math.atan2(
node2_center_y - node1_center_y, node2_center_x - node1_center_x
)
angle_rad = math.atan2(node2_center_y - node1_center_y, node2_center_x - node1_center_x)
angle_deg = round(math.degrees(angle_rad))
angle_rad = math.radians(angle_deg)
@ -162,9 +160,7 @@ def calculate_two_node_shape(
svg_x = center_x - (rx * math.cos(angle_rad) - ry * math.sin(angle_rad))
svg_y = center_y - (rx * math.sin(angle_rad) + ry * math.cos(angle_rad))
shape_svg = generate_ellipse_svg(
int(rx), int(ry), color_scheme, int(shape_width), int(shape_height)
)
shape_svg = generate_ellipse_svg(int(rx), int(ry), color_scheme, int(shape_width), int(shape_height))
offset_distance = ry * text_offset_ratio
@ -183,18 +179,10 @@ def calculate_two_node_shape(
shape_width = distance
shape_height = max(node1_width, node1_height, node2_width, node2_height)
svg_x = center_x - (
(shape_width / 2) * math.cos(angle_rad)
- (shape_height / 2) * math.sin(angle_rad)
)
svg_y = center_y - (
(shape_width / 2) * math.sin(angle_rad)
+ (shape_height / 2) * math.cos(angle_rad)
)
svg_x = center_x - ((shape_width / 2) * math.cos(angle_rad) - (shape_height / 2) * math.sin(angle_rad))
svg_y = center_y - ((shape_width / 2) * math.sin(angle_rad) + (shape_height / 2) * math.cos(angle_rad))
shape_svg = generate_rectangle_svg(
int(shape_width), int(shape_height), color_scheme
)
shape_svg = generate_rectangle_svg(int(shape_width), int(shape_height), color_scheme)
offset_distance = (shape_height / 2) * text_offset_ratio
@ -241,6 +229,7 @@ def calculate_two_node_shape(
"metadata": metadata,
}
def generate_ellipse_svg(
rx: int,
ry: int,
@ -249,7 +238,8 @@ def generate_ellipse_svg(
svg_height: int,
) -> str:
"""Generate SVG for ellipse."""
return f'''<svg width="{svg_width}" height="{svg_height}"><ellipse cx="{rx}" cy="{ry}" rx="{rx}" ry="{ry}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>'''
return f"""<svg width="{svg_width}" height="{svg_height}"><ellipse cx="{rx}" cy="{ry}" rx="{rx}" ry="{ry}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>"""
def generate_rectangle_svg(
width: int,
@ -257,14 +247,16 @@ def generate_rectangle_svg(
color_scheme: dict[str, Any],
) -> str:
"""Generate SVG for rectangle."""
return f'''<svg width="{width}" height="{height}"><rect x="0" y="0" width="{width}" height="{height}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>'''
return f"""<svg width="{width}" height="{height}"><rect x="0" y="0" width="{width}" height="{height}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>"""
def generate_text_svg(text: str, color_scheme: dict[str, Any]) -> str:
"""Generate SVG for text label."""
text_width = len(text) * 8 + 20
text_height = DEFAULT_FONT_SIZE + 16
return f'''<svg width="{text_width}" height="{text_height}"><text font-family="TypeWriter" font-size="{DEFAULT_FONT_SIZE}.0" font-weight="bold" fill="{color_scheme["stroke"]}" text-anchor="middle" x="{text_width / 2}" y="{text_height / 2 + 4}">{text}</text></svg>'''
return f"""<svg width="{text_width}" height="{text_height}"><text font-family="TypeWriter" font-size="{DEFAULT_FONT_SIZE}.0" font-weight="bold" fill="{color_scheme["stroke"]}" text-anchor="middle" x="{text_width / 2}" y="{text_height / 2 + 4}">{text}</text></svg>"""
def _hsv_to_hex(h: int, s: int, v: int) -> str:
"""Convert HSV to HEX color."""
@ -299,6 +291,7 @@ def _hsv_to_hex(h: int, s: int, v: int) -> str:
return f"#{r:02x}{g:02x}{b:02x}"
def calculate_z_order(area_size: float) -> int:
"""
Calculate z-order based on area size for proper layering.
@ -316,6 +309,7 @@ def calculate_z_order(area_size: float) -> int:
else:
return 3
def _get_color_scheme(area_name: str) -> dict[str, Any]:
"""
Get color scheme based on area name using keyword inference.
@ -339,33 +333,15 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]:
return COLOR_SCHEMES["NORMAL_AREA"]
# 3. Logical Isolation
if (
"VRF" in label
or "VLAN" in label
or "MSTP" in label
or "VXLAN" in label
or "MPLS" in label
):
if "VRF" in label or "VLAN" in label or "MSTP" in label or "VXLAN" in label or "MPLS" in label:
return COLOR_SCHEMES["ISOLATION"]
# 4. High Availability
if (
"VRRP" in label
or "HSRP" in label
or "HA" in label
or "STACK" in label
or "M-LAG" in label
):
if "VRRP" in label or "HSRP" in label or "HA" in label or "STACK" in label or "M-LAG" in label:
return COLOR_SCHEMES["HIGH_AVAILABILITY"]
# 5. External/Internet
if (
"INET" in label
or "OUT" in label
or "EXTERNAL" in label
or "INTERNET" in label
or "DMZ" in label
):
if "INET" in label or "OUT" in label or "EXTERNAL" in label or "INTERNET" in label or "DMZ" in label:
return COLOR_SCHEMES["EXTERNAL"]
# 6. Management
@ -404,6 +380,7 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]:
return COLOR_SCHEMES["DEFAULT"]
def calculate_two_node_ellipse(
node1: dict,
node2: dict,
@ -415,15 +392,14 @@ def calculate_two_node_ellipse(
Wrapper around calculate_two_node_shape with shape_type="ellipse".
"""
result = calculate_two_node_shape(
node1, node2, area_name, "ellipse", text_offset_ratio
)
result = calculate_two_node_shape(node1, node2, area_name, "ellipse", text_offset_ratio)
return {
"ellipse": result["shape"],
"text": result["text"],
"metadata": result["metadata"],
}
def calculate_two_node_rectangle(
node1: dict,
node2: dict,
@ -435,9 +411,7 @@ def calculate_two_node_rectangle(
Wrapper around calculate_two_node_shape with shape_type="rectangle".
"""
result = calculate_two_node_shape(
node1, node2, area_name, "rectangle", text_offset_ratio
)
result = calculate_two_node_shape(node1, node2, area_name, "rectangle", text_offset_ratio)
metadata = result["metadata"]
return {
"rectangle": result["shape"],

10
gns3server/agent/gns3_copilot/utils/llm_config_helper.py
View File

@ -41,17 +41,17 @@ Usage:
"""
import logging
from typing import Optional, Dict, Any
from typing import Any
from typing import Dict
from typing import Optional
from uuid import UUID
from fastapi import FastAPI
logger = logging.getLogger(__name__)
async def get_user_llm_config_with_app(
user_id: UUID,
app: FastAPI
) -> Optional[Dict[str, Any]]:
async def get_user_llm_config_with_app(user_id: UUID, app: FastAPI) -> Optional[Dict[str, Any]]:
"""
Get user's default LLM model configuration with decrypted API key.

117
gns3server/agent/gns3_copilot/utils/message_converters.py
View File

@ -28,12 +28,18 @@
Message format converters for OpenAI-compatible message format.
Converts between LangChain messages and OpenAI-compatible format.
"""
import json
import uuid
from datetime import datetime
from typing import Dict, Any
from typing import Any
from typing import Dict
from langchain_core.messages import AIMessage
from langchain_core.messages import HumanMessage
from langchain_core.messages import SystemMessage
from langchain_core.messages import ToolMessage
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
def _ensure_string(content: Any) -> str:
"""Ensure content is a string, converting dicts/lists to JSON if needed."""
@ -44,6 +50,7 @@ def _ensure_string(content: Any) -> str:
else:
return str(content)
def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
"""
Convert LangChain message to OpenAI-compatible format.
@ -55,53 +62,40 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
Dictionary in OpenAI-compatible format
"""
# Generate message ID
msg_id = getattr(lc_message, 'id', None)
msg_id = getattr(lc_message, "id", None)
if msg_id is None:
msg_id = str(uuid.uuid4())
# Get timestamp
timestamp = getattr(lc_message, 'created_at', None)
timestamp = getattr(lc_message, "created_at", None)
if timestamp is None:
timestamp = datetime.utcnow().isoformat()
elif hasattr(timestamp, 'isoformat'):
elif hasattr(timestamp, "isoformat"):
timestamp = timestamp.isoformat()
# Base message structure
base_msg = {
"id": msg_id,
"created_at": timestamp,
"metadata": {}
}
base_msg = {"id": msg_id, "created_at": timestamp, "metadata": {}}
# Convert based on message type
if isinstance(lc_message, HumanMessage):
return {
**base_msg,
"role": "user",
"content": lc_message.content
}
return {**base_msg, "role": "user", "content": lc_message.content}
elif isinstance(lc_message, AIMessage):
msg = {
**base_msg,
"role": "assistant",
"content": lc_message.content
}
msg = {**base_msg, "role": "assistant", "content": lc_message.content}
# Handle tool calls - convert to OpenAI format
if hasattr(lc_message, 'tool_calls') and lc_message.tool_calls:
if hasattr(lc_message, "tool_calls") and lc_message.tool_calls:
tool_calls = []
for tc in lc_message.tool_calls:
# Convert to dict if it's an object
tc_dict = tc if isinstance(tc, dict) else tc.model_dump()
tool_calls.append({
"id": tc_dict.get("id", str(uuid.uuid4())),
"type": "function",
"function": {
"name": tc_dict.get("name", ""),
"arguments": tc_dict.get("args", {})
tool_calls.append(
{
"id": tc_dict.get("id", str(uuid.uuid4())),
"type": "function",
"function": {"name": tc_dict.get("name", ""), "arguments": tc_dict.get("args", {})},
}
})
)
msg["tool_calls"] = tool_calls
return msg
@ -111,24 +105,17 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
**base_msg,
"role": "tool",
"content": _ensure_string(lc_message.content),
"name": getattr(lc_message, 'name', ''),
"tool_call_id": getattr(lc_message, 'tool_call_id', '')
"name": getattr(lc_message, "name", ""),
"tool_call_id": getattr(lc_message, "tool_call_id", ""),
}
elif isinstance(lc_message, SystemMessage):
return {
**base_msg,
"role": "system",
"content": lc_message.content
}
return {**base_msg, "role": "system", "content": lc_message.content}
else:
# Fallback for unknown message types
return {
**base_msg,
"role": "unknown",
"content": str(lc_message)
}
return {**base_msg, "role": "unknown", "content": str(lc_message)}
def convert_openai_to_langchain(msg: Dict[str, Any]):
"""
@ -153,21 +140,19 @@ def convert_openai_to_langchain(msg: Dict[str, Any]):
if "tool_calls" in msg and msg["tool_calls"]:
tool_calls = []
for tc in msg["tool_calls"]:
tool_calls.append({
"id": tc.get("id", str(uuid.uuid4())),
"name": tc.get("function", {}).get("name", ""),
"args": tc.get("function", {}).get("arguments", {})
})
tool_calls.append(
{
"id": tc.get("id", str(uuid.uuid4())),
"name": tc.get("function", {}).get("name", ""),
"args": tc.get("function", {}).get("arguments", {}),
}
)
ai_msg.tool_calls = tool_calls
return ai_msg
elif role == "tool":
return ToolMessage(
content=content,
name=msg.get("name", ""),
tool_call_id=msg.get("tool_call_id", "")
)
return ToolMessage(content=content, name=msg.get("name", ""), tool_call_id=msg.get("tool_call_id", ""))
elif role == "system":
return SystemMessage(content=content)
@ -176,6 +161,7 @@ def convert_openai_to_langchain(msg: Dict[str, Any]):
# Fallback to HumanMessage for unknown roles
return HumanMessage(content=content)
def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]:
"""
Convert LangGraph streaming event to OpenAI-compatible format.
@ -190,21 +176,17 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]:
if event_type == "on_chat_model_stream":
chunk = event.get("data", {}).get("chunk", {})
content = getattr(chunk, 'content', '')
content = getattr(chunk, "content", "")
if content:
return {
"type": "content",
"content": content,
"message_id": event.get("metadata", {}).get("msg_id")
}
return {"type": "content", "content": content, "message_id": event.get("metadata", {}).get("msg_id")}
# Check for tool call chunks
if hasattr(chunk, 'tool_call_chunks') and chunk.tool_call_chunks:
if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks:
for tc_chunk in chunk.tool_call_chunks:
tc_id = getattr(tc_chunk, 'id', None)
tc_name = getattr(tc_chunk, 'name', None)
tc_args = getattr(tc_chunk, 'args', None)
tc_id = getattr(tc_chunk, "id", None)
tc_name = getattr(tc_chunk, "name", None)
tc_args = getattr(tc_chunk, "args", None)
if tc_id:
return {
@ -212,19 +194,12 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]:
"tool_call": {
"id": tc_id,
"type": "function",
"function": {
"name": tc_name or "",
"arguments": tc_args or ""
}
}
"function": {"name": tc_name or "", "arguments": tc_args or ""},
},
}
elif event_type == "on_tool_start":
return {
"type": "tool_start",
"tool_name": event.get("name", ""),
"metadata": event.get("metadata", {})
}
return {"type": "tool_start", "tool_name": event.get("name", ""), "metadata": event.get("metadata", {})}
elif event_type == "on_tool_end":
tool_output = event.get("data", {}).get("output", "")
@ -236,7 +211,7 @@ def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]:
"type": "tool_end",
"tool_output": tool_output,
"tool_name": event.get("name", ""),
"metadata": event.get("metadata", {})
"metadata": event.get("metadata", {}),
}
# Default empty response

89
gns3server/agent/gns3_copilot/utils/parse_tool_content.py
View File

@ -71,6 +71,7 @@ from typing import Any
logger = logging.getLogger(__name__)
def parse_tool_content(
content: str | dict | list | int | float | bool | None,
fallback_to_raw: bool = True,
@ -172,9 +173,7 @@ def parse_tool_content(
# (higher priority as many tools return Python format strings)
try:
result = ast.literal_eval(s)
logger.info(
"Successfully parsed as Python literal, returning: %s", result
)
logger.info("Successfully parsed as Python literal, returning: %s", result)
return result
except (ValueError, SyntaxError):
pass
@ -211,8 +210,7 @@ def parse_tool_content(
# Handle unsupported types
error_msg = ( # type: ignore[unreachable]
"Content must be str, dict, list, int, float, bool, or None, got "
f"{type(content).__name__}"
"Content must be str, dict, list, int, float, bool, or None, got " f"{type(content).__name__}"
)
logger.error(error_msg)
@ -227,9 +225,8 @@ def parse_tool_content(
logger.info("Returning error: %s", result)
return result
def format_tool_response(
content: str | dict | list | int | float | bool | None, indent: int = 2
) -> str:
def format_tool_response(content: str | dict | list | int | float | bool | None, indent: int = 2) -> str:
"""
Format tool response as a beautiful JSON string for UI display.
@ -256,9 +253,7 @@ def format_tool_response(
# If the parsed result cannot be serialized, convert to string and wrap
logger.error("Cannot serialize parsed result to JSON: %s", e)
try:
result = json.dumps(
{"raw": str(content)}, ensure_ascii=False, indent=indent
)
result = json.dumps({"raw": str(content)}, ensure_ascii=False, indent=indent)
logger.info("format_tool_response returning fallback: %s", result)
return result
except Exception:
@ -276,10 +271,8 @@ def format_tool_response(
logger.info("format_tool_response returning error: %s", result)
return result
def normalize_tool_response(
response: dict | list | str,
tool_name: str = "unknown"
) -> dict:
def normalize_tool_response(response: dict | list | str, tool_name: str = "unknown") -> dict:
"""
Normalize tool response to standard format for consistent frontend display.
@ -310,10 +303,7 @@ def normalize_tool_response(
>>> normalize_tool_response([{"device_name": "R1", "status": "success"}])
{'success': True, 'total': 1, 'successful': 1, 'failed': 0, 'data': [...], 'metadata': {}}
"""
metadata = {
"tool_name": tool_name,
"normalized_at": datetime.utcnow().isoformat()
}
metadata = {"tool_name": tool_name, "normalized_at": datetime.utcnow().isoformat()}
# Handle error responses
if isinstance(response, dict) and "error" in response and len(response) == 1:
@ -324,19 +314,12 @@ def normalize_tool_response(
"failed": 0,
"data": [],
"error": str(response["error"]),
"metadata": metadata
"metadata": metadata,
}
# Handle empty responses
if not response:
return {
"success": True,
"total": 0,
"successful": 0,
"failed": 0,
"data": [],
"metadata": metadata
}
return {"success": True, "total": 0, "successful": 0, "failed": 0, "data": [], "metadata": metadata}
# Handle list responses (most tools return list of device results)
if isinstance(response, list):
@ -358,12 +341,7 @@ def normalize_tool_response(
normalized_data.append(normalized_item)
else:
# Non-dict items in list
normalized_data.append({
"id": "",
"name": "",
"status": "unknown",
"result": str(item)
})
normalized_data.append({"id": "", "name": "", "status": "unknown", "result": str(item)})
return {
"success": failed == 0,
@ -371,7 +349,7 @@ def normalize_tool_response(
"successful": successful,
"failed": failed,
"data": normalized_data,
"metadata": metadata
"metadata": metadata,
}
# Handle dict responses (some tools return summary + results)
@ -385,7 +363,7 @@ def normalize_tool_response(
"failed": response.get("failed", 0),
"data": response.get("data", []),
"error": response.get("error"),
"metadata": {**metadata, **response.get("metadata", {})}
"metadata": {**metadata, **response.get("metadata", {})},
}
# Legacy format: extract common fields
@ -403,20 +381,24 @@ def normalize_tool_response(
data = response["data"]
elif "output" in response:
# Single device response
data = [{
"name": response.get("device_name", ""),
"status": response.get("status", "success"),
"result": response["output"]
}]
data = [
{
"name": response.get("device_name", ""),
"status": response.get("status", "success"),
"result": response["output"],
}
]
# If no data found but have status, create single item
if not data and "status" in response:
data = [{
"name": response.get("device_name") or response.get("name") or "",
"status": response["status"],
"result": response.get("output") or response.get("result") or "",
"error": response.get("error") or ""
}]
data = [
{
"name": response.get("device_name") or response.get("name") or "",
"status": response["status"],
"result": response.get("output") or response.get("result") or "",
"error": response.get("error") or "",
}
]
# Recursively normalize data items
if data and isinstance(data, list):
@ -429,7 +411,7 @@ def normalize_tool_response(
"successful": successful,
"failed": failed,
"data": [],
"metadata": metadata
"metadata": metadata,
}
# Handle string responses (parse first)
@ -443,15 +425,11 @@ def normalize_tool_response(
"total": 1,
"successful": 1,
"failed": 0,
"data": [{
"id": "",
"name": "",
"status": "unknown",
"result": str(response)
}],
"metadata": metadata
"data": [{"id": "", "name": "", "status": "unknown", "result": str(response)}],
"metadata": metadata,
}
# Test function to verify the implementation
def _test_parse_tool_content() -> None:
"""Test function to verify parse_tool_content works correctly with all input types"""
@ -510,5 +488,6 @@ def _test_parse_tool_content() -> None:
valid = "✗"
print(f"Format Test {i + 1}: {valid} Input: {repr(input_data)} -> {result}")
if __name__ == "__main__":
_test_parse_tool_content()

6
gns3server/api/routes/controller/chat.py
View File

@ -134,6 +134,12 @@ async def stream_chat(
llm_config.get("model"),
)
# TODO: Support runtime temperature override from request.temperature
# Currently, temperature is loaded from the user's LLM config in the database.
# To enable runtime override, uncomment the following:
# if request.temperature is not None:
# llm_config["temperature"] = str(request.temperature)
# Get or create AgentService for this project
agent_manager = await get_project_agent_manager()
agent_service = await agent_manager.get_agent(str(project.id), project.path)

7
gns3server/schemas/controller/chat.py
View File

@ -36,7 +36,12 @@ class ChatRequest(BaseModel):
message: str = Field(..., description="User message content")
session_id: Optional[str] = Field(None, description="Session ID (auto-generated if not provided)")
stream: bool = Field(default=True, description="Enable streaming response")
temperature: Optional[float] = Field(None, description="LLM temperature parameter")
temperature: Optional[float] = Field(
None,
description="LLM temperature parameter (NOTE: currently not used. "
"Temperature is loaded from user's LLM config in database. "
"Reserved for future runtime override support.)"
)
mode: Literal["text"] = Field(default="text", description="Interaction mode")