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

Merge pull request #2858 from yueguobin/refactor/mcp-to-agent-dir

Browse Source 
Move MCP service to agent package + fix auto_close forwarding
This commit is contained in:
Jeremy Grossmann 2026-08-23 18:00:11 +02:00 committed by GitHub
commit d7a1bf9b7b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 191 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
docs/features/mcp-service.md
View File

@ -471,10 +471,10 @@ websocat ws://192.168.1.3:3080/v3/projects/{project_id}/nodes/{node_id}/console/
| File | Purpose |
|------|---------|
| `gns3server/api/routes/mcp/__init__.py` | FastMCP server, tool decorators, SSE transport, JWT auth wrapper |
| `gns3server/api/routes/mcp/projects.py` | Project tool handlers |
| `gns3server/api/routes/mcp/nodes.py` | Node tool handlers |
| `gns3server/api/routes/mcp/links.py` | Link tool handlers |
| `gns3server/api/routes/mcp/templates.py` | Template tool handlers |
| `gns3server/api/routes/mcp/computes.py` | Compute tool handlers |
| `gns3server/agent/mcp/__init__.py` | FastMCP server, tool decorators, SSE transport, JWT auth wrapper |
| `gns3server/agent/mcp/projects.py` | Project tool handlers |
| `gns3server/agent/mcp/nodes.py` | Node tool handlers |
| `gns3server/agent/mcp/links.py` | Link tool handlers |
| `gns3server/agent/mcp/templates.py` | Template tool handlers |
| `gns3server/agent/mcp/computes.py` | Compute tool handlers |
| `gns3server/api/server.py` | Mounts MCP routes via `register_starlette_routes()` |

10
docs/features/project-open-performance.md
View File

@ -113,14 +113,14 @@ if await asyncio.to_thread(bcrypt.checkpw, secret.encode(), db_key.key_hash.enco
|---------|--------|-------|------|
| Node creation Pool | 5 | 100 | `controller/project.py` |
| Link creation Pool | 5 | 100 | `controller/project.py` |
| MCP BATCH_MAX_WORKERS | 10 | 100 | `api/routes/mcp/nodes.py` |
| MCP BATCH_MAX_WORKERS | 10 | 100 | `agent/mcp/nodes.py` |
| MCP HTTP timeout | 10s | 30s | `agent/gns3_copilot/gns3_client/custom_gns3fy.py` |
| HTTP connection pool | 10 (default) | 500/1000 | `agent/gns3_copilot/gns3_client/custom_gns3fy.py` |
| Start nodes Pool | 3 | 3 (unchanged) | `controller/project.py` |
### 8. MCP Auth Returns JWT
**File:** `gns3server/api/routes/mcp/__init__.py`
**File:** `gns3server/agent/mcp/__init__.py`
When an MCP client connects with an API key, the `_resolve_token` function validates the key then returns a fresh short-lived JWT instead of the raw API key. The JWT is stored in a `ContextVar` and reused for all subsequent tool calls within the same SSE session — zero extra bcrypt.
@ -141,8 +141,8 @@ if user:
| `gns3server/controller/udp_link.py` | Pre-allocated port consumption |
| `gns3server/api/routes/controller/api_keys.py` | O(1) key format |
| `gns3server/api/routes/controller/dependencies/authentication.py` | O(1) auth + thread pool bcrypt |
| `gns3server/api/routes/mcp/__init__.py` | Auth returns JWT, tool enhancements |
| `gns3server/api/routes/mcp/nodes.py` | fields filter, inherited template_id, name passthrough |
| `gns3server/api/routes/mcp/links.py` | fields filter, compact array format |
| `gns3server/agent/mcp/__init__.py` | Auth returns JWT, tool enhancements |
| `gns3server/agent/mcp/nodes.py` | fields filter, inherited template_id, name passthrough |
| `gns3server/agent/mcp/links.py` | fields filter, compact array format |
| `gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py` | Timeout 30s, connection pool 500/1000 |
| `gns3server/utils/images.py` | md5sum cache error → warning |

5
gns3server/agent/__init__.py
View File

@ -85,7 +85,10 @@ MCP_AVAILABLE = False
# Try to import MCP dependencies
try:
import mcp.server.fastmcp # noqa: F401 — test import only
# Use importlib so the top-level SDK name "mcp" is not bound in this
# namespace — it would shadow the gns3server.agent.mcp subpackage.
import importlib
importlib.import_module("mcp.server.fastmcp")
MCP_AVAILABLE = True
except ImportError:
# MCP dependencies not installed, disable MCP feature

0
gns3server/api/routes/mcp/__init__.py → gns3server/agent/mcp/__init__.py
View File

0
gns3server/api/routes/mcp/appliances.py → gns3server/agent/mcp/appliances.py
View File

0
gns3server/api/routes/mcp/computes.py → gns3server/agent/mcp/computes.py
View File

0
gns3server/api/routes/mcp/device_config.py → gns3server/agent/mcp/device_config.py
View File

0
gns3server/api/routes/mcp/drawings.py → gns3server/agent/mcp/drawings.py
View File

0
gns3server/api/routes/mcp/images.py → gns3server/agent/mcp/images.py
View File

0
gns3server/api/routes/mcp/links.py → gns3server/agent/mcp/links.py
View File

0
gns3server/api/routes/mcp/nodes.py → gns3server/agent/mcp/nodes.py
View File

5
gns3server/api/routes/mcp/projects.py → gns3server/agent/mcp/projects.py
View File

@ -66,7 +66,10 @@ def create_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
if not name:
return {"error": "name is required"}
conn = _get_connector(gns3_ctx)
return conn.http_call("post", f"{conn.base_url}/projects", json_data={"name": name}).json()
json_data: dict[str, Any] = {"name": name}
if params.get("auto_close") is not None:
json_data["auto_close"] = params["auto_close"]
return conn.http_call("post", f"{conn.base_url}/projects", json_data=json_data).json()
def delete_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:

0
gns3server/api/routes/mcp/server.py → gns3server/agent/mcp/server.py
View File

0
gns3server/api/routes/mcp/snapshots.py → gns3server/agent/mcp/snapshots.py
View File

0
gns3server/api/routes/mcp/symbols.py → gns3server/agent/mcp/symbols.py
View File

0
gns3server/api/routes/mcp/templates.py → gns3server/agent/mcp/templates.py
View File

2
gns3server/api/server.py
View File

@ -51,7 +51,7 @@ from gns3server.core import tasks
from gns3server.agent import MCP_AVAILABLE
if MCP_AVAILABLE:
from gns3server.api.routes import mcp
from gns3server.agent import mcp
_mcp_router = mcp.router
else:
from fastapi import APIRouter

2
gns3server/core/tasks.py
View File

@ -97,7 +97,7 @@ async def startup(app: FastAPI) -> None:
from gns3server.agent import MCP_AVAILABLE
if MCP_AVAILABLE:
from gns3server.api.routes.mcp import set_mcp_server_ready
from gns3server.agent.mcp import set_mcp_server_ready
set_mcp_server_ready(True)
log.info("GNS3 server startup completed")

0
tests/api/routes/mcp/conftest.py → tests/agent/mcp/conftest.py
View File

112
tests/api/routes/mcp/test_handlers.py → tests/agent/mcp/test_handlers.py
View File

@ -15,7 +15,7 @@ def _mock_conn(json_result=None):
return conn
BASE = "gns3server.api.routes.mcp"
BASE = "gns3server.agent.mcp"
@pytest.fixture
@ -31,60 +31,74 @@ class TestProject:
mod = "projects"
def test_list(self, ctx):
from gns3server.api.routes.mcp.projects import list_projects_handler
from gns3server.agent.mcp.projects import list_projects_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([{"project_id": "p1", "name": "Test", "status": "opened"}])
result = list_projects_handler({}, ctx)
assert result["count"] == 1
def test_get(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_handler
from gns3server.agent.mcp.projects import get_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"project_id": "p1"})
result = get_project_handler({"project_id": "p1"}, ctx)
assert result["project_id"] == "p1"
def test_get_missing_id(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_handler
from gns3server.agent.mcp.projects import get_project_handler
assert "error" in get_project_handler({}, ctx)
def test_create(self, ctx):
from gns3server.api.routes.mcp.projects import create_project_handler
from gns3server.agent.mcp.projects import create_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"project_id": "p1"})
result = create_project_handler({"name": "New"}, ctx)
conn = _mock_conn({"project_id": "p1"})
m.return_value = conn
result = create_project_handler({"name": "New", "auto_close": False}, ctx)
assert result["project_id"] == "p1"
conn.http_call.assert_called_once_with(
"post", f"{conn.base_url}/projects", json_data={"name": "New", "auto_close": False}
)
def test_create_without_auto_close(self, ctx):
from gns3server.agent.mcp.projects import create_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"project_id": "p2"})
m.return_value = conn
create_project_handler({"name": "New"}, ctx)
conn.http_call.assert_called_once_with(
"post", f"{conn.base_url}/projects", json_data={"name": "New"}
)
def test_delete(self, ctx):
from gns3server.api.routes.mcp.projects import delete_project_handler
from gns3server.agent.mcp.projects import delete_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_project_handler({"project_id": "p1"}, ctx)
assert "message" in result
def test_open(self, ctx):
from gns3server.api.routes.mcp.projects import open_project_handler
from gns3server.agent.mcp.projects import open_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "opened"})
result = open_project_handler({"project_id": "p1"}, ctx)
assert result["status"] == "opened"
def test_close(self, ctx):
from gns3server.api.routes.mcp.projects import close_project_handler
from gns3server.agent.mcp.projects import close_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "closed"})
result = close_project_handler({"project_id": "p1"}, ctx)
assert "error" not in result
def test_update(self, ctx):
from gns3server.api.routes.mcp.projects import update_project_handler
from gns3server.agent.mcp.projects import update_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"name": "Updated"})
result = update_project_handler({"project_id": "p1", "name": "Updated"}, ctx)
assert result["name"] == "Updated"
def test_stats(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_stats_handler
from gns3server.agent.mcp.projects import get_project_stats_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"nodes": 5, "links": 3})
result = get_project_stats_handler({"project_id": "p1"}, ctx)
@ -99,7 +113,7 @@ class TestNode:
mod = "nodes"
def test_list_fields(self, ctx):
from gns3server.api.routes.mcp.nodes import get_nodes_handler
from gns3server.agent.mcp.nodes import get_nodes_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([
{"node_id": "n1", "name": "R1", "status": "started", "node_type": "qemu", "console": 5000},
@ -108,21 +122,21 @@ class TestNode:
assert result == {"nodes": [{"name": "R1", "status": "started"}], "count": 1}
def test_list_invalid_fields(self, ctx):
from gns3server.api.routes.mcp.nodes import get_nodes_handler
from gns3server.agent.mcp.nodes import get_nodes_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([])
result = get_nodes_handler({"project_id": "p1", "fields": "not-a-list"}, ctx)
assert "error" in result
def test_get(self, ctx):
from gns3server.api.routes.mcp.nodes import get_node_handler
from gns3server.agent.mcp.nodes import get_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1"})
result = get_node_handler({"project_id": "p1", "node_id": "n1"}, ctx)
assert result["name"] == "R1"
def test_create_single_passes_name(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
from gns3server.agent.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"node_id": "n1", "name": "MyRouter"})
m.return_value = conn
@ -137,7 +151,7 @@ class TestNode:
assert result == {"node_id": "n1", "name": "MyRouter"}
def test_create_fields_filter(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
from gns3server.agent.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1", "status": "started"})
result = create_node_handler({
@ -147,7 +161,7 @@ class TestNode:
assert result == {"node_id": "n1", "name": "R1"}
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
from gns3server.agent.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
@ -159,7 +173,7 @@ class TestNode:
conn.http_call.assert_not_called()
def test_create_batch_inherits_template_id(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
from gns3server.agent.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1"})
result = create_node_handler({
@ -169,39 +183,39 @@ class TestNode:
assert result[0]["status"] == "success"
def test_create_missing_project_id(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
from gns3server.agent.mcp.nodes import create_node_handler
assert create_node_handler({}, ctx) == {"error": "project_id is required"}
def test_delete_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import delete_node_handler
from gns3server.agent.mcp.nodes import delete_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_node_handler({"project_id": "p1", "node_ids": ["n1", "n2"]}, ctx)
assert len(result) == 2
def test_start_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import start_node_handler
from gns3server.agent.mcp.nodes import start_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "started"})
result = start_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_stop_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import stop_node_handler
from gns3server.agent.mcp.nodes import stop_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "stopped"})
result = stop_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_suspend_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import suspend_node_handler
from gns3server.agent.mcp.nodes import suspend_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "suspended"})
result = suspend_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_console(self, ctx):
from gns3server.api.routes.mcp.nodes import get_node_console_info_handler
from gns3server.agent.mcp.nodes import get_node_console_info_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"console_url": "ws://host/console"})
result = get_node_console_info_handler({"project_id": "p1", "node_id": "n1"}, ctx)
@ -216,21 +230,21 @@ class TestLink:
mod = "links"
def test_list(self, ctx):
from gns3server.api.routes.mcp.links import get_links_handler
from gns3server.agent.mcp.links import get_links_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([{"link_id": "l1", "link_type": "ethernet"}])
result = get_links_handler({"project_id": "p1", "fields": ["link_id"]}, ctx)
assert result["links"] == [{"link_id": "l1"}]
def test_get(self, ctx):
from gns3server.api.routes.mcp.links import get_link_handler
from gns3server.agent.mcp.links import get_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1", "link_type": "ethernet"})
result = get_link_handler({"project_id": "p1", "link_id": "l1"}, ctx)
assert result["link_id"] == "l1"
def test_create_compact_format(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
from gns3server.agent.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"link_id": "l1", "link_type": "ethernet", "nodes": []})
m.return_value = conn
@ -247,7 +261,7 @@ class TestLink:
)
def test_create_standard_format(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
from gns3server.agent.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1"})
result = create_link_handler({
@ -260,7 +274,7 @@ class TestLink:
assert result["link_id"] == "l1"
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
from gns3server.agent.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
@ -273,14 +287,14 @@ class TestLink:
conn.http_call.assert_not_called()
def test_delete_batch(self, ctx):
from gns3server.api.routes.mcp.links import delete_link_handler
from gns3server.agent.mcp.links import delete_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_link_handler({"project_id": "p1", "link_ids": ["l1", "l2"]}, ctx)
assert len(result) == 2
def test_update(self, ctx):
from gns3server.api.routes.mcp.links import update_link_handler
from gns3server.agent.mcp.links import update_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1", "suspend": True})
result = update_link_handler({
@ -297,14 +311,14 @@ class TestAppliance:
mod = "appliances"
def test_get(self, ctx):
from gns3server.api.routes.mcp.appliances import get_appliance_handler
from gns3server.agent.mcp.appliances import get_appliance_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"appliance_id": "a1", "name": "Cisco ISE"})
result = get_appliance_handler({"appliance_id": "a1"}, ctx)
assert result["name"] == "Cisco ISE"
def test_install_with_version(self, ctx):
from gns3server.api.routes.mcp.appliances import install_appliance_handler
from gns3server.agent.mcp.appliances import install_appliance_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"status": "installed"})
m.return_value = conn
@ -317,7 +331,7 @@ class TestAppliance:
)
def test_install_missing_id(self, ctx):
from gns3server.api.routes.mcp.appliances import install_appliance_handler
from gns3server.agent.mcp.appliances import install_appliance_handler
result = install_appliance_handler({}, ctx)
assert "error" in result
@ -330,7 +344,7 @@ class TestTemplate:
mod = "templates"
def test_list_fields(self, ctx):
from gns3server.api.routes.mcp.templates import list_templates_handler
from gns3server.agent.mcp.templates import list_templates_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([
{"template_id": "t1", "name": "Cisco 7200", "template_type": "dynamips",
@ -340,21 +354,21 @@ class TestTemplate:
assert result["templates"] == [{"template_id": "t1", "name": "Cisco 7200"}]
def test_list_invalid_field(self, ctx):
from gns3server.api.routes.mcp.templates import list_templates_handler
from gns3server.agent.mcp.templates import list_templates_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn()
result = list_templates_handler({"fields": ["does_not_exist"]}, ctx)
assert "error" in result
def test_get(self, ctx):
from gns3server.api.routes.mcp.templates import get_template_handler
from gns3server.agent.mcp.templates import get_template_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"template_id": "t1", "name": "Test"})
result = get_template_handler({"template_id": "t1"}, ctx)
assert result["name"] == "Test"
def test_delete(self, ctx):
from gns3server.api.routes.mcp.templates import delete_template_handler
from gns3server.agent.mcp.templates import delete_template_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_template_handler({"template_id": "t1"}, ctx)
@ -370,7 +384,7 @@ class TestLinkMarker:
mod = "links"
def test_update_direction_both_clears(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
from gns3server.agent.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
@ -384,7 +398,7 @@ class TestLinkMarker:
)
def test_update_direction_tx_sets(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
from gns3server.agent.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
@ -398,7 +412,7 @@ class TestLinkMarker:
)
def test_update_direction_omitted_preserved(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
from gns3server.agent.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
@ -412,7 +426,7 @@ class TestLinkMarker:
)
def test_create_direction_both_omitted(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
from gns3server.agent.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
@ -426,7 +440,7 @@ class TestLinkMarker:
)
def test_create_direction_tx(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
from gns3server.agent.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
@ -451,7 +465,7 @@ class TestMarkerDefinition:
mod = "links"
def test_create_builds_body(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
from gns3server.agent.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
@ -465,7 +479,7 @@ class TestMarkerDefinition:
)
def test_create_ignores_direction(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
from gns3server.agent.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
@ -479,7 +493,7 @@ class TestMarkerDefinition:
)
def test_update_builds_body(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
from gns3server.agent.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
@ -493,7 +507,7 @@ class TestMarkerDefinition:
)
def test_update_ignores_direction(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
from gns3server.agent.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
@ -507,7 +521,7 @@ class TestMarkerDefinition:
)
def test_update_requires_a_field(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
from gns3server.agent.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector"):
result = marker_definition_handler(
{"project_id": "p", "action": "update", "def_name": "arp"}, ctx,

190
tests/api/routes/mcp/test_tool_params.py → tests/agent/mcp/test_tool_params.py
View File

@ -15,7 +15,7 @@ from pathlib import Path
import pytest
MCP_DIR = Path(__file__).resolve().parents[4] / "gns3server" / "api" / "routes" / "mcp"
MCP_DIR = Path(__file__).resolve().parents[3] / "gns3server" / "agent" / "mcp"
TOOL_FILE = MCP_DIR / "__init__.py"
HANDLER_FILES = {
@ -63,6 +63,8 @@ HANDLER_FILES = {
"start_capture_handler": "links.py",
"stop_capture_handler": "links.py",
"download_capture_file_handler": "links.py",
"link_marker_handler": "links.py",
"marker_definition_handler": "links.py",
"list_templates_handler": "templates.py",
"get_template_handler": "templates.py",
"create_template_handler": "templates.py",
@ -115,10 +117,19 @@ def _get_handler_params(handler_name):
params = set()
for node in ast.walk(tree):
if not isinstance(node, ast.FunctionDef):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if node.name != handler_name:
continue
# A handler that forwards params.items() generically (e.g. building the
# request body from all params) accepts any key — return a wildcard and
# let the caller skip static consistency checks for it.
for sub in ast.walk(node):
if (isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute)
and sub.func.attr == "items"
and isinstance(sub.func.value, ast.Name)
and sub.func.value.id in ("params", "params_data")):
return {"*"}
# Found the handler function, search for params.get("xxx")
for sub in ast.walk(node):
if not isinstance(sub, ast.Call):
@ -134,89 +145,83 @@ def _get_handler_params(handler_name):
return params
def _get_tool_params(tool_name, tool_file=TOOL_FILE):
"""Parse __init__.py and extract params passed to _run_handler_sync for a given tool.
Returns the dict literal keys from the _run_handler_sync call.
"""
tree = ast.parse(tool_file.read_text())
for node in ast.walk(tree):
if not isinstance(node, ast.FunctionDef):
continue
if node.name != tool_name:
continue
# Search for _run_handler_sync calls inside this function
for sub in ast.walk(node):
if not isinstance(sub, ast.Call):
continue
if not hasattr(sub.func, "id") or sub.func.id != "_run_handler_sync":
continue
# _run_handler_sync(handler, {dict}) or _run_handler_sync(handler, params)
if len(sub.args) >= 2:
second_arg = sub.args[1]
if isinstance(second_arg, ast.Dict):
keys = set()
for k in second_arg.keys:
if isinstance(k, ast.Constant) and isinstance(k.value, str):
keys.add(k.value)
return keys
elif isinstance(second_arg, ast.Name) and second_arg.id == "params":
return {"*params*"} # special marker for all params passed through
return None
def test_handler_params_all_readable():
"""Every handler registered in __init__.py should have a corresponding file."""
# Extract all handler names from __init__.py by looking for _run_handler_sync calls
# Extract all handler names from __init__.py by looking for dispatch calls
tree = ast.parse(TOOL_FILE.read_text())
handlers_found = set()
for node in ast.walk(tree):
if isinstance(node, ast.Call) and hasattr(node.func, "id") and node.func.id == "_run_handler_sync":
if node.args and isinstance(node.args[0], ast.Name):
handlers_found.add(node.args[0].id)
if not isinstance(node, ast.Call):
continue
handler_name, _ = _dispatch_args(node)
if handler_name:
handlers_found.add(handler_name)
unknown = [h for h in handlers_found if h not in HANDLER_FILES]
assert not unknown, f"Handlers not mapped in HANDLER_FILES: {unknown}"
def _get_tool_fn_name(handler_name):
"""Reverse lookup: find which MCP tool function calls this handler."""
tree = ast.parse(TOOL_FILE.read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Call) and hasattr(node.func, "id") and node.func.id == "_run_handler_sync":
if node.args and isinstance(node.args[0], ast.Name) and node.args[0].id == handler_name:
# Find enclosing function
for parent in ast.walk(tree):
if isinstance(parent, ast.FunctionDef):
for child in ast.walk(parent):
if child is node:
return parent.name
def _dict_literal_keys(d):
"""String keys of an ast.Dict literal (non-constant keys like **spread are skipped)."""
keys = set()
for k in d.keys:
if isinstance(k, ast.Constant) and isinstance(k.value, str):
keys.add(k.value)
return keys
def _initial_params_keys(fn_node):
"""Keys of the initial 'params = {...}' dict literal inside a tool function, or None.
Conditional additions (params["k"] = v) after the literal are not included.
"""
for stmt in ast.walk(fn_node):
if isinstance(stmt, ast.Assign):
if (len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Name)
and stmt.targets[0].id == "params" and isinstance(stmt.value, ast.Dict)):
return _dict_literal_keys(stmt.value)
return None
def _dispatch_args(node):
"""Extract (handler_name, payload_arg) from a handler dispatch call.
Matches both forms used by tools:
- _run_handler_sync(handler, payload)
- asyncio.to_thread(_run_handler_sync, handler, payload)
Returns (None, None) if the call is neither.
"""
if isinstance(node.func, ast.Name) and node.func.id == "_run_handler_sync":
args = node.args
elif (isinstance(node.func, ast.Attribute) and node.func.attr == "to_thread"
and node.args and isinstance(node.args[0], ast.Name) and node.args[0].id == "_run_handler_sync"):
args = node.args[1:]
else:
return None, None
if len(args) < 2 or not isinstance(args[0], ast.Name):
return None, None
return args[0].id, args[1]
def test_tool_handler_param_consistency():
"""For each tool, the params passed to the handler should match what the handler reads."""
tree = ast.parse(TOOL_FILE.read_text())
# Collect all _run_handler_sync calls with dict literals
# Group dispatch calls by (tool, handler): a tool may dispatch the same
# handler from multiple branches (e.g. node_create single/batch modes),
# each passing only its branch's keys — the union covers all of them.
groups = {}
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if not hasattr(node.func, "id") or node.func.id != "_run_handler_sync":
continue
if len(node.args) < 2:
continue
handler_name = node.args[0].id if isinstance(node.args[0], ast.Name) else None
handler_name, second_arg = _dispatch_args(node)
if not handler_name:
continue
# Find the tool function name (enclosing function)
tool_name = None
for parent in ast.walk(tree):
if isinstance(parent, ast.FunctionDef):
if isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef)):
for child in ast.walk(parent):
if child is node:
tool_name = parent.name
@ -224,31 +229,50 @@ def test_tool_handler_param_consistency():
if not tool_name:
continue
second_arg = node.args[1]
exact = False
if isinstance(second_arg, ast.Dict):
passed_keys = set()
for k in second_arg.keys:
if isinstance(k, ast.Constant) and isinstance(k.value, str):
passed_keys.add(k.value)
passed_keys = _dict_literal_keys(second_arg)
exact = True
elif isinstance(second_arg, ast.Name) and second_arg.id == "params":
# Tool builds 'params' as a variable — resolve its initial dict literal.
fn = next((n for n in ast.walk(tree)
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == tool_name), None)
passed_keys = _initial_params_keys(fn) if fn is not None else None
else:
passed_keys = None
if not passed_keys:
continue
handler_params = _get_handler_params(handler_name)
if handler_params is None:
continue
group = groups.setdefault((tool_name, handler_name), {"passed": set(), "exact": False})
group["passed"] |= passed_keys
group["exact"] = group["exact"] or exact
# Check: every passed key is read by the handler
extra_passed = passed_keys - handler_params
assert not extra_passed, (
f"[{tool_name}] Params passed to handler '{handler_name}' but not read: {extra_passed}"
for (tool_name, handler_name), group in groups.items():
handler_params = _get_handler_params(handler_name)
if handler_params is None or "*" in handler_params:
continue
passed_keys = group["passed"]
# Check: every passed key is read by the handler
extra_passed = passed_keys - handler_params
assert not extra_passed, (
f"[{tool_name}] Params passed to handler '{handler_name}' but not read: {extra_passed}"
)
if not group["exact"]:
# The initial literal underestimates what the tool passes (keys may be
# added conditionally), so only the extra-passed direction is checked.
continue
# Check: every handler param is passed (except common/optional ones)
missing = handler_params - passed_keys
# Filter out well-known optional params that handlers check
known_optional = {"fields", "template", "name", "version", "compute_id",
"x", "y", "link_type", "filters", "suspend", "link_style",
"show_filters_icon", "label"}
truly_missing = missing - known_optional
if truly_missing:
pytest.fail(
f"[{tool_name}] Handler '{handler_name}' reads params not passed: {truly_missing}"
)
# Check: every handler param is passed (except common/optional ones)
missing = handler_params - passed_keys
# Filter out well-known optional params that handlers check
known_optional = {"fields", "template", "name", "version", "compute_id",
"x", "y", "link_type", "filters", "suspend", "link_style",
"show_filters_icon", "label"}
truly_missing = missing - known_optional
if truly_missing:
pytest.fail(
f"[{tool_name}] Handler '{handler_name}' reads params not passed: {truly_missing}"
)