mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-05 14:51:13 +02:00
Merge branch '3.1' into mypy-agent
This commit is contained in:
38 files changed
+149
-180
No files matched your search
@@ -140,7 +140,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
|
||||
for tool in tools:
|
||||
try:
|
||||
# Build tool schema in OpenAI format
|
||||
tool_schema = {
|
||||
tool_schema: dict[str, Any] = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": tool.name,
|
||||
|
||||
@@ -110,7 +110,7 @@ def create_base_model(
|
||||
|
||||
try:
|
||||
# Prepare parameters for init_chat_model
|
||||
init_params = {
|
||||
init_params: dict[str, Any] = {
|
||||
"model": config_vars["model_name"],
|
||||
"model_provider": config_vars["model_provider"],
|
||||
"api_key": config_vars["api_key"],
|
||||
@@ -177,7 +177,7 @@ def create_title_model(
|
||||
|
||||
try:
|
||||
# Prepare parameters for init_chat_model
|
||||
init_params = {
|
||||
init_params: dict[str, Any] = {
|
||||
"model": config_vars["model_name"],
|
||||
"model_provider": config_vars["model_provider"],
|
||||
"api_key": config_vars["api_key"],
|
||||
|
||||
@@ -347,7 +347,7 @@ def get_gns3_server_host() -> str:
|
||||
host = get_gns3_server_host()
|
||||
print(f"GNS3 server host: {host}")
|
||||
"""
|
||||
url = _detect_url_for_api()
|
||||
url = _detect_url_for_api() or DEFAULT_GNS3_URL
|
||||
|
||||
# Extract host from URL
|
||||
# URL format: protocol://host:port
|
||||
|
||||
@@ -49,7 +49,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None)
|
||||
"""
|
||||
inventory: dict[str, Any] = {}
|
||||
for n in nodes:
|
||||
inventory[n.get("name")] = {
|
||||
inventory[n.get("name") or ""] = {
|
||||
"server": server_host,
|
||||
"name": n.get("name"),
|
||||
"node_id": n.get("node_id"),
|
||||
@@ -68,7 +68,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None)
|
||||
return inventory
|
||||
|
||||
|
||||
def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||||
def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Build a human/LLM-friendly link list resolving node and port names.
|
||||
|
||||
@@ -76,7 +76,7 @@ def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]
|
||||
Links whose endpoints cannot be resolved are skipped, mirroring the
|
||||
original Project.links_summary behavior.
|
||||
"""
|
||||
summary: list[dict[str, str]] = []
|
||||
summary: list[dict[str, Any]] = []
|
||||
for link in links:
|
||||
if not link.get("nodes"):
|
||||
continue
|
||||
|
||||
@@ -248,7 +248,7 @@ class SkillsLoader:
|
||||
continue
|
||||
# Use device_type from YAML content as the key
|
||||
# Fallback to filename stem if device_type not present
|
||||
skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None
|
||||
skill_key = skill_data.get("device_type")
|
||||
if not skill_key:
|
||||
skill_key = yaml_file.stem
|
||||
logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key")
|
||||
|
||||
@@ -37,8 +37,10 @@ from typing import Optional, Dict, Any
|
||||
|
||||
try:
|
||||
import git
|
||||
|
||||
GIT_AVAILABLE = True
|
||||
except ImportError:
|
||||
git = None
|
||||
GIT_AVAILABLE = False
|
||||
|
||||
from gns3server.config import Config
|
||||
from .loader import SkillsLoader
|
||||
@@ -66,7 +68,7 @@ class SkillsManager:
|
||||
- Version tracking
|
||||
"""
|
||||
|
||||
def __init__(self, repo_url: str = None, branch: str = "main", auto_update: bool = False):
|
||||
def __init__(self, repo_url: Optional[str] = None, branch: str = "main", auto_update: bool = False):
|
||||
"""
|
||||
Initialize the skills manager.
|
||||
|
||||
@@ -90,7 +92,7 @@ class SkillsManager:
|
||||
self._repo: Optional["git.Repo"] = None
|
||||
self._prompt_count = 0
|
||||
|
||||
if git is None:
|
||||
if not GIT_AVAILABLE:
|
||||
logger.warning("GitPython is not installed. Skills management features will be limited.")
|
||||
|
||||
def initialize(self) -> bool:
|
||||
@@ -134,7 +136,7 @@ class SkillsManager:
|
||||
- Up to date → nothing
|
||||
- Network error → use existing files, log warning
|
||||
"""
|
||||
if git is None or self._repo is None:
|
||||
if not GIT_AVAILABLE or self._repo is None:
|
||||
return
|
||||
|
||||
# Check for uncommitted changes
|
||||
@@ -173,7 +175,7 @@ class SkillsManager:
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if git is None:
|
||||
if not GIT_AVAILABLE:
|
||||
logger.error("GitPython is not installed. Cannot clone repository.")
|
||||
return False
|
||||
|
||||
@@ -347,7 +349,7 @@ class SkillsManager:
|
||||
Returns:
|
||||
Commit hash as string, or empty string if not available
|
||||
"""
|
||||
if git is None or self._repo is None:
|
||||
if not GIT_AVAILABLE or self._repo is None:
|
||||
try:
|
||||
self._repo = git.Repo(self.local_path)
|
||||
except Exception:
|
||||
@@ -409,7 +411,7 @@ class SkillsManager:
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if git is None:
|
||||
if not GIT_AVAILABLE:
|
||||
logger.error("GitPython is not installed. Cannot rollback.")
|
||||
return False
|
||||
|
||||
@@ -436,7 +438,7 @@ class SkillsManager:
|
||||
Returns:
|
||||
List of commit information dictionaries
|
||||
"""
|
||||
if git is None:
|
||||
if not GIT_AVAILABLE:
|
||||
return []
|
||||
|
||||
try:
|
||||
|
||||
@@ -78,7 +78,7 @@ try:
|
||||
# so we use importlib to get the actual module
|
||||
import importlib
|
||||
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
sd: Any = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
# Recalculate platforms lists to include custom device types
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
@@ -196,7 +196,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 len(device_configs_list) > 0 and "error" in device_configs_list[0]:
|
||||
return device_configs_list
|
||||
|
||||
# Filter forbidden commands and store blocked commands info
|
||||
|
||||
@@ -78,7 +78,7 @@ try:
|
||||
# so we use importlib to get the actual module
|
||||
import importlib
|
||||
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
sd: Any = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
# Recalculate platforms lists to include custom device types
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
@@ -199,7 +199,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 len(device_configs_list) > 0 and "error" in device_configs_list[0]:
|
||||
return device_configs_list
|
||||
|
||||
# Filter forbidden commands and store blocked commands info
|
||||
|
||||
@@ -286,7 +286,7 @@ class GNS3PacketFilterTool(BaseTool):
|
||||
|
||||
# Check if output contains "Invalid" indicating syntax error
|
||||
if "Invalid" in result.stdout or "Invalid" in result.stderr:
|
||||
error_lines = []
|
||||
error_lines: list[str] = []
|
||||
if "Invalid" in result.stderr:
|
||||
error_lines.extend(line for line in result.stderr.split("\n") if "Invalid" in line)
|
||||
if "Invalid" in result.stdout:
|
||||
|
||||
@@ -143,7 +143,8 @@ class PacketAnalysisTool(BaseTool):
|
||||
stderr=subprocess.DEVNULL,
|
||||
text=True,
|
||||
)
|
||||
tshark.stdout.close()
|
||||
if tshark.stdout is not None:
|
||||
tshark.stdout.close()
|
||||
stdout, _ = grep.communicate(timeout=30)
|
||||
tshark.wait(timeout=5)
|
||||
except Exception as e:
|
||||
|
||||
@@ -68,7 +68,7 @@ try:
|
||||
# CRITICAL: Update netmiko.ssh_dispatcher platforms lists
|
||||
import importlib
|
||||
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
sd: Any = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
# Recalculate platforms lists to include custom device types
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
@@ -179,7 +179,7 @@ class VPCSCommands(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 len(device_configs_list) > 0 and "error" in device_configs_list[0]:
|
||||
return device_configs_list
|
||||
|
||||
# Create a mapping of device names to their commands
|
||||
@@ -386,7 +386,7 @@ class VPCSCommands(BaseTool):
|
||||
def _prepare_device_hosts_data(
|
||||
self,
|
||||
device_configs_list: list[dict[str, Any]],
|
||||
project_id: str,
|
||||
project_id: str | None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
@@ -415,7 +415,7 @@ class VPCSCommands(BaseTool):
|
||||
device_ports = get_device_ports_from_topology(device_names, project_id=project_id, jwt_token=jwt_token, url=url)
|
||||
|
||||
# Build Nornir inventory hosts data
|
||||
hosts_data = {}
|
||||
hosts_data: dict[str, Any] = {}
|
||||
for device_name in device_names:
|
||||
if device_name not in device_ports:
|
||||
logger.error("Device '%s' not found in topology", device_name)
|
||||
|
||||
@@ -37,6 +37,7 @@ 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 ToolCall
|
||||
from langchain_core.messages import ToolMessage
|
||||
|
||||
|
||||
@@ -86,7 +87,7 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
|
||||
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()
|
||||
tc_dict = tc
|
||||
tool_calls.append(
|
||||
{
|
||||
"id": tc_dict.get("id", str(uuid.uuid4())),
|
||||
@@ -139,7 +140,7 @@ def convert_openai_to_langchain(msg: Dict[str, Any]):
|
||||
|
||||
# Restore tool calls if present
|
||||
if "tool_calls" in msg and msg["tool_calls"]:
|
||||
tool_calls = []
|
||||
tool_calls: list[ToolCall] = []
|
||||
for tc in msg["tool_calls"]:
|
||||
tool_calls.append(
|
||||
{
|
||||
|
||||
@@ -214,9 +214,7 @@ def parse_tool_content(
|
||||
return content
|
||||
|
||||
# Handle unsupported types
|
||||
error_msg = ( # type: ignore[unreachable]
|
||||
f"Content must be str, dict, list, int, float, bool, or None, got {type(content).__name__}"
|
||||
)
|
||||
error_msg = f"Content must be str, dict, list, int, float, bool, or None, got {type(content).__name__}"
|
||||
logger.error(error_msg)
|
||||
|
||||
if strict_mode:
|
||||
|
||||
Reference in new issue
Block a user