From 5ac1521217f2d00992002b3d210f5a7d2e652f99 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Wed, 30 Sep 2026 14:47:45 +0800 Subject: [PATCH] feat(mcp): accept project/node names as tool locators MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LLM agents transcribe UUIDs unreliably — a 2-character slip invents an id that never existed and is invisible to the eye. Tool parameters now accept names wherever the server guarantees uniqueness: project names among live projects, node names within a project. Links, drawings, appliances and snapshots have no names and stay UUID-only. Resolution sits in the single dispatch choke point (_run_handler_sync): UUID-shaped values pass through untouched, names resolve with one list call, and an unknown name fails fast listing the known names instead of a confusing downstream 404. As a backstop for residual UUID typos, error replies carrying " doesn't exist" are matched against the in-scope ids; a single candidate within edit distance 2 earns a "Did you mean ...?" hint. --- gns3server/agent/mcp/__init__.py | 157 ++++++++++++---------- gns3server/agent/mcp/locators.py | 223 +++++++++++++++++++++++++++++++ tests/agent/mcp/test_locators.py | 170 +++++++++++++++++++++++ 3 files changed, 478 insertions(+), 72 deletions(-) create mode 100644 gns3server/agent/mcp/locators.py create mode 100644 tests/agent/mcp/test_locators.py diff --git a/gns3server/agent/mcp/__init__.py b/gns3server/agent/mcp/__init__.py index 0f57ac7ed..180087905 100644 --- a/gns3server/agent/mcp/__init__.py +++ b/gns3server/agent/mcp/__init__.py @@ -122,6 +122,7 @@ from .images import ( install_images_handler, prune_images_handler, ) +from .locators import add_did_you_mean, resolve_locators from .projects import ( close_project_handler, create_project_handler, @@ -339,7 +340,19 @@ def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]: "jwt_username": _jwt_username_var.get(), "jwt_token_version": _jwt_token_version_var.get(), } - result = handler(params, ctx) + + def run(h, p): + return h(p, ctx) + + # Name-shaped project_id/node_id values become UUIDs before the handler + # runs; a name that matches nothing fails fast with the known names. + params, resolution_error = resolve_locators(params, run) + if resolution_error is not None: + result = resolution_error + else: + result = handler(params, ctx) + # backstop for UUID transcription slips: "Did you mean ...?" + result = add_did_you_mean(result, params, run) return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}] @@ -351,7 +364,7 @@ async def project_list() -> list[dict[str, Any]]: @mcp.tool() async def project_get( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Get detailed information about a specific project.""" return await asyncio.to_thread(_run_handler_sync, get_project_handler, {"project_id": project_id}) @@ -368,7 +381,7 @@ async def project_create( @mcp.tool() async def project_delete( - project_id: Annotated[str, Field(description="UUID of the project to delete")], + project_id: Annotated[str, Field(description="Project name or UUID to delete")], ) -> list[dict[str, Any]]: """Delete a GNS3 project permanently.""" return await asyncio.to_thread(_run_handler_sync, delete_project_handler, {"project_id": project_id}) @@ -376,7 +389,7 @@ async def project_delete( @mcp.tool() async def project_open( - project_id: Annotated[str, Field(description="UUID of the project to open")], + project_id: Annotated[str, Field(description="Project name or UUID to open")], ) -> list[dict[str, Any]]: """Open a closed GNS3 project.""" return await asyncio.to_thread(_run_handler_sync, open_project_handler, {"project_id": project_id}) @@ -384,7 +397,7 @@ async def project_open( @mcp.tool() async def project_close( - project_id: Annotated[str, Field(description="UUID of the project to close")], + project_id: Annotated[str, Field(description="Project name or UUID to close")], ) -> list[dict[str, Any]]: """Close an open GNS3 project.""" return await asyncio.to_thread(_run_handler_sync, close_project_handler, {"project_id": project_id}) @@ -392,7 +405,7 @@ async def project_close( @mcp.tool() async def project_stats( - project_id: Annotated[str, Field(description="UUID of the project to get statistics for")], + project_id: Annotated[str, Field(description="Project name or UUID to get statistics for")], ) -> list[dict[str, Any]]: """Get statistics (nodes, links, snapshots, drawings) for a project.""" return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) @@ -400,7 +413,7 @@ async def project_stats( @mcp.tool() async def project_update( - project_id: Annotated[str, Field(description="UUID of the project to update")], + project_id: Annotated[str, Field(description="Project name or UUID to update")], name: Annotated[str | None, Field(description="New project name")] = None, auto_close: Annotated[bool | None, Field(description="Close project when last client leaves")] = None, auto_open: Annotated[bool | None, Field(description="Project opens when GNS3 starts")] = None, @@ -442,7 +455,7 @@ async def project_update( @mcp.tool() async def project_duplicate( - project_id: Annotated[str, Field(description="UUID of the project to duplicate")], + project_id: Annotated[str, Field(description="Project name or UUID to duplicate")], name: Annotated[str, Field(description="New project name")], reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False, ) -> list[dict[str, Any]]: @@ -455,7 +468,7 @@ async def project_duplicate( @mcp.tool() async def project_readme_get( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Get the content of a project's README.md file — the project documentation (Markdown format).""" return await asyncio.to_thread(_run_handler_sync, get_project_readme_handler, {"project_id": project_id}) @@ -463,7 +476,7 @@ async def project_readme_get( @mcp.tool() async def project_readme_update( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], content: Annotated[str, Field(description="Content to write to README.md (Markdown format)")], ) -> list[dict[str, Any]]: """Update or create a project's README.md file — the project documentation (Markdown format).""" @@ -477,7 +490,7 @@ async def project_readme_update( @mcp.tool() async def node_list( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], fields: Annotated[ list[str] | None, Field( @@ -491,8 +504,8 @@ async def node_list( @mcp.tool() async def node_get( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], fields: Annotated[ list[str] | None, Field( @@ -514,8 +527,8 @@ async def node_get( @mcp.tool() async def node_start( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None, + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None, node_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — start multiple nodes in parallel') ] = None, @@ -531,8 +544,8 @@ async def node_start( @mcp.tool() async def node_stop( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None, + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None, node_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — stop multiple nodes in parallel') ] = None, @@ -548,8 +561,8 @@ async def node_stop( @mcp.tool() async def node_suspend( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None, + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None, node_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — suspend multiple nodes in parallel') ] = None, @@ -565,7 +578,7 @@ async def node_suspend( @mcp.tool() async def node_create( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], template_id: Annotated[ str | None, Field(description="Template UUID (required for single mode; used as default in batch mode)") ] = None, @@ -626,8 +639,8 @@ async def node_create( @mcp.tool() async def node_delete( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None, + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None, node_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — delete multiple nodes in parallel') ] = None, @@ -643,8 +656,8 @@ async def node_delete( @mcp.tool() async def node_update( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node to update")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID to update")], **kwargs: Any, ) -> list[dict[str, Any]]: """Update a node's properties (name, position, etc.).""" @@ -654,8 +667,8 @@ async def node_update( @mcp.tool() async def node_console( - project_id: Annotated[str, Field(description="UUID of the project containing the node")], - node_id: Annotated[str, Field(description="UUID of the node to get console info for")], + project_id: Annotated[str, Field(description="Project name or UUID containing the node")], + node_id: Annotated[str, Field(description="Node name or UUID to get console info for")], ) -> list[dict[str, Any]]: """Get WebSocket console connection info for a node. @@ -693,7 +706,7 @@ async def node_console( @mcp.tool() async def link_list( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], fields: Annotated[ list[str] | None, Field( @@ -707,7 +720,7 @@ async def link_list( @mcp.tool() async def link_get( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str, Field(description="UUID of the link")], ) -> list[dict[str, Any]]: """Get detailed information about a specific link.""" @@ -716,7 +729,7 @@ async def link_get( @mcp.tool() async def link_create( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], nodes: Annotated[ list | None, Field(description="Single mode: [{node_id, adapter_number, port_number}] or compact [id, ad, pt, id, ad, pt]"), @@ -762,7 +775,7 @@ async def link_create( @mcp.tool() async def link_delete( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None, link_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — delete multiple links in parallel') @@ -779,7 +792,7 @@ async def link_delete( @mcp.tool() async def link_update( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str, Field(description="UUID of the link to update")], **kwargs: Any, ) -> list[dict[str, Any]]: @@ -841,7 +854,7 @@ async def link_update( @mcp.tool() async def link_available_filters( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str, Field(description="UUID of the link")], ) -> list[dict[str, Any]]: """List the packet filter types available for a link with their parameters @@ -1002,8 +1015,8 @@ async def compute_images( @mcp.tool() async def node_file_list( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], path: Annotated[str, Field(description="Subdirectory path within node directory (optional)")] = "", recursive: Annotated[bool, Field(description="Recursively list all files (optional, default: false)")] = False, ) -> list[dict[str, Any]]: @@ -1026,8 +1039,8 @@ async def node_file_list( @mcp.tool() async def node_file_get( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], file_path: Annotated[str, Field(description="Path to the file within the node directory")], offset: Annotated[int, Field(description="Line offset to start reading from (optional, default: 0)")] = 0, limit: Annotated[int, Field(description="Maximum number of lines to return (optional, default: 200)")] = 200, @@ -1056,8 +1069,8 @@ async def node_file_get( @mcp.tool() async def node_file_write( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], file_path: Annotated[str, Field(description="Path to the file within the node directory")], content: Annotated[str, Field(description="Content to write to the file")], ) -> list[dict[str, Any]]: @@ -1076,8 +1089,8 @@ async def node_file_write( @mcp.tool() async def node_file_delete( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], file_path: Annotated[str, Field(description="Path to the file within the node directory")], ) -> list[dict[str, Any]]: """Delete a file from a node directory. Cannot be undone.""" @@ -1097,7 +1110,7 @@ async def node_file_delete( @mcp.tool() async def node_start_all( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Start all nodes in a project.""" return await asyncio.to_thread( @@ -1111,7 +1124,7 @@ async def node_start_all( @mcp.tool() async def node_stop_all( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Stop all nodes in a project.""" return await asyncio.to_thread( @@ -1125,7 +1138,7 @@ async def node_stop_all( @mcp.tool() async def node_suspend_all( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Suspend all nodes in a project.""" return await asyncio.to_thread( @@ -1139,8 +1152,8 @@ async def node_suspend_all( @mcp.tool() async def node_duplicate( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node to duplicate")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID to duplicate")], x: Annotated[int, Field(description="X coordinate for the new node")] = 0, y: Annotated[int, Field(description="Y coordinate for the new node")] = 0, z: Annotated[int, Field(description="Z layer for the new node")] = 0, @@ -1161,8 +1174,8 @@ async def node_duplicate( @mcp.tool() async def node_isolate( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node to isolate")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID to isolate")], ) -> list[dict[str, Any]]: """Isolate a node by suspending all its attached links (network isolation).""" return await asyncio.to_thread( @@ -1177,8 +1190,8 @@ async def node_isolate( @mcp.tool() async def node_unisolate( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node to unisolate")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID to unisolate")], ) -> list[dict[str, Any]]: """Un-isolate a node by resuming all its suspended links.""" return await asyncio.to_thread( @@ -1193,8 +1206,8 @@ async def node_unisolate( @mcp.tool() async def node_links( - project_id: Annotated[str, Field(description="UUID of the project")], - node_id: Annotated[str, Field(description="UUID of the node")], + project_id: Annotated[str, Field(description="Project name or UUID")], + node_id: Annotated[str, Field(description="Node name or UUID")], ) -> list[dict[str, Any]]: """List all links connected to a specific node.""" return await asyncio.to_thread( @@ -1212,7 +1225,7 @@ async def node_links( @mcp.tool() async def link_reset( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None, link_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — reset multiple links in parallel') @@ -1239,7 +1252,7 @@ async def link_reset( @mcp.tool() async def link_capture_start( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None, data_link_type: Annotated[str, Field(description="Data link type (default: DLT_EN10MB)")] = "DLT_EN10MB", capture_file_name: Annotated[str | None, Field(description="Capture file name (optional)")] = None, @@ -1265,7 +1278,7 @@ async def link_capture_start( @mcp.tool() async def link_capture_stop( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None, link_ids: Annotated[ list[str] | None, @@ -1283,7 +1296,7 @@ async def link_capture_stop( @mcp.tool() async def link_capture_download( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None, link_ids: Annotated[ list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — get download URLs for multiple captures') @@ -1314,7 +1327,7 @@ async def link_capture_download( @mcp.tool() async def link_marker( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], link_id: Annotated[str, Field(description="UUID of the link")], action: Annotated[str, Field(description="Action: create, update, or delete")], bpf: Annotated[ @@ -1382,7 +1395,7 @@ async def link_marker( @mcp.tool() async def marker_definition( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], action: Annotated[str, Field(description="Action: create, update, delete, or list")], bpf: Annotated[ str | None, Field(description="BPF expression, e.g. 'arp', 'ospf', 'tcp port 22' (required for create)") @@ -1433,7 +1446,7 @@ async def marker_definition( @mcp.tool() async def snapshot_list( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """List all snapshots of a project.""" return await asyncio.to_thread( @@ -1447,7 +1460,7 @@ async def snapshot_list( @mcp.tool() async def snapshot_create( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], name: Annotated[str, Field(description="Name for the new snapshot")], ) -> list[dict[str, Any]]: """Create a new snapshot of a project. @@ -1468,7 +1481,7 @@ async def snapshot_create( @mcp.tool() async def snapshot_delete( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], snapshot_id: Annotated[str, Field(description="UUID of the snapshot to delete")], ) -> list[dict[str, Any]]: """Delete a snapshot from a project. Cannot be undone.""" @@ -1484,7 +1497,7 @@ async def snapshot_delete( @mcp.tool() async def snapshot_restore( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], snapshot_id: Annotated[str, Field(description="UUID of the snapshot to restore")], ) -> list[dict[str, Any]]: """Restore a project to a previous snapshot state. The project may be closed and reopened.""" @@ -1503,7 +1516,7 @@ async def snapshot_restore( @mcp.tool() async def drawing_list( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """List all drawings (labels, shapes, images) on a project canvas.""" return await asyncio.to_thread( @@ -1517,7 +1530,7 @@ async def drawing_list( @mcp.tool() async def drawing_create( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], svg: Annotated[str, Field(description="SVG content for the drawing")], x: Annotated[int, Field(description="X coordinate (default: 0)")] = 0, y: Annotated[int, Field(description="Y coordinate (default: 0)")] = 0, @@ -1557,7 +1570,7 @@ async def drawing_create( @mcp.tool() async def drawing_get( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], drawing_id: Annotated[str, Field(description="UUID of the drawing")], ) -> list[dict[str, Any]]: """Get detailed information about a specific drawing.""" @@ -1573,7 +1586,7 @@ async def drawing_get( @mcp.tool() async def drawing_update( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], drawing_id: Annotated[str, Field(description="UUID of the drawing")], svg: Annotated[str | None, Field(description="New SVG content")] = None, locked: Annotated[bool | None, Field(description="Lock or unlock the drawing")] = None, @@ -1593,7 +1606,7 @@ async def drawing_update( @mcp.tool() async def drawing_delete( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], drawing_id: Annotated[str, Field(description="UUID of the drawing to delete")], ) -> list[dict[str, Any]]: """Delete a drawing from a project canvas. Cannot be undone.""" @@ -1612,7 +1625,7 @@ async def drawing_delete( @mcp.tool() async def project_lock( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Lock all drawings and nodes in a project to prevent accidental changes.""" return await asyncio.to_thread( @@ -1626,7 +1639,7 @@ async def project_lock( @mcp.tool() async def project_unlock( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Unlock a project to allow editing of drawings and nodes.""" return await asyncio.to_thread( @@ -1640,7 +1653,7 @@ async def project_unlock( @mcp.tool() async def project_locked( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], ) -> list[dict[str, Any]]: """Check whether a project is locked (preventing edits to drawings and nodes).""" return await asyncio.to_thread( @@ -1868,7 +1881,7 @@ async def image_install() -> list[dict[str, Any]]: @mcp.tool() async def device_config_send( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], device_configs: Annotated[ list, Field( @@ -1905,7 +1918,7 @@ async def device_config_send( @mcp.tool() async def device_show_run( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], device_configs: Annotated[ list, Field( @@ -1947,7 +1960,7 @@ async def device_show_run( @mcp.tool() async def vpcs_config_set( - project_id: Annotated[str, Field(description="UUID of the project")], + project_id: Annotated[str, Field(description="Project name or UUID")], device_configs: Annotated[ list, Field( diff --git a/gns3server/agent/mcp/locators.py b/gns3server/agent/mcp/locators.py new file mode 100644 index 000000000..4bffb8014 --- /dev/null +++ b/gns3server/agent/mcp/locators.py @@ -0,0 +1,223 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# Author: Yue Guobin +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +Name locators for MCP tools. + +LLM agents transcribe UUIDs unreliably (a 2-character slip invents an id +that never existed and is invisible to the eye), so MCP tool parameters +accept names wherever the server guarantees uniqueness: project names are +unique among live projects (the controller rejects duplicates) and node +names are unique within a project (auto-suffixed on collision). Links, +drawings, appliances and snapshots have no names and stay UUID-only. + +Resolution happens in the single dispatch choke point (_run_handler_sync): +UUID-shaped values pass through untouched, names are resolved with one +list call, and a name that matches nothing fails fast with the known +names instead of a confusing downstream 404. As a backstop for residual +UUID typos, error replies carrying " doesn't exist" are matched +against the ids visible in scope; a single candidate within edit +distance 2 earns a "Did you mean ..." hint. +""" + +import re +from typing import Any, Callable + +from .projects import list_projects_handler +from gns3server.agent.gns3_copilot.gns3_client.api_handlers import get_links_handler, get_nodes_handler + +_UUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") +_ID_IN_TEXT_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") + +# how many names to list when a locator matches nothing +_MAX_HINT_NAMES = 15 + + +def looks_like_uuid(value: Any) -> bool: + """ + Whether *value* is a UUID-shaped string (anything else is treated as a name). + """ + + return isinstance(value, str) and bool(_UUID_RE.match(value)) + + +def resolve_locators( + params: dict[str, Any], run: Callable[[Callable, dict], Any] +) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + """ + Rewrite name-shaped ``project_id``/``node_id`` values into UUIDs. + + :param params: tool params (top-level keys only; nested structures are + left alone — link_create's node tuples require ids) + :param run: callback running (handler, params) through the same context + :returns: (resolved params, None) on success, (None, error dict) on failure + """ + + params = dict(params) + + project_ref = params.get("project_id") + if project_ref is not None and not looks_like_uuid(project_ref): + result = run(list_projects_handler, {}) + projects = _as_items(result, "projects") + if projects is None: + return None, {"error": f"Could not resolve project '{project_ref}': {_as_error(result)}"} + match = _match_by_name(projects, project_ref) + if match is None: + return None, {"error": f"Project '{project_ref}' not found. Known projects: {_known_names(projects)}"} + params["project_id"] = match["project_id"] + + node_ref = params.get("node_id") + if node_ref is not None and not looks_like_uuid(node_ref) and params.get("project_id"): + result = run(get_nodes_handler, {"project_id": params["project_id"]}) + nodes = _as_items(result, "nodes") + if nodes is None: + return None, {"error": f"Could not resolve node '{node_ref}': {_as_error(result)}"} + match = _match_by_name(nodes, node_ref) + if match is None: + return None, {"error": f"Node '{node_ref}' not found. Known nodes: {_known_names(nodes)}"} + params["node_id"] = match["node_id"] + + return params, None + + +def add_did_you_mean( + result: Any, params: dict[str, Any], run: Callable[[Callable, dict], Any] +) -> Any: + """ + Append a "Did you mean ...?" hint to an error reply that names a + non-existent UUID, when exactly one in-scope id is within edit + distance 2 of it (the signature of a transcription slip). + """ + + if not isinstance(result, dict): + return result + error = result.get("error") + if not isinstance(error, str) or ("doesn't exist" not in error and "not found" not in error.lower()): + return result + bad_ids = set(_ID_IN_TEXT_RE.findall(error)) + if not bad_ids: + return result + + candidates = _collect_candidates(params, run) + if not candidates: + return result + + hints = [] + for bad in bad_ids: + close = [(cid, label) for cid, label in candidates if bad != cid and _edit_distance_at_most(bad.lower(), cid.lower(), 2)] + if len(close) == 1: + cid, label = close[0] + hints.append(f"Did you mean {label} ({cid})?") + if not hints: + return result + result = dict(result) + result["error"] = error.rstrip() + " " + " ".join(hints) + return result + + +def _as_items(result: Any, key: str) -> list | None: + """ + Normalize a handler reply into a list of dicts (None = failure). + """ + + if isinstance(result, list): + return result + if isinstance(result, dict): + items = result.get(key) + if isinstance(items, list): + return items + return None + + +def _as_error(result: Any) -> Any: + return result.get("error", result) if isinstance(result, dict) else result + + +def _match_by_name(items: list, name: str) -> dict | None: + """ + The unique item whose name equals *name* (exact first, then a unique + case-insensitive match); None when absent or ambiguous. + """ + + exact = [item for item in items if item.get("name") == name] + if len(exact) == 1: + return exact[0] + if not exact: + folded = [item for item in items if isinstance(item.get("name"), str) and item["name"].lower() == name.lower()] + if len(folded) == 1: + return folded[0] + return None + + +def _known_names(items: list) -> str: + names = sorted(str(item.get("name", "?")) for item in items) + if len(names) > _MAX_HINT_NAMES: + names = [*names[:_MAX_HINT_NAMES], f"... ({len(names) - _MAX_HINT_NAMES} more)"] + return ", ".join(names) + + +def _collect_candidates(params: dict[str, Any], run: Callable[[Callable, dict], Any]) -> list[tuple[str, str]]: + """ + (id, label) pairs in scope for a hint: every project, plus the nodes + and links of the project the call targeted. Best-effort — any failure + just yields what was gathered so far. + """ + + candidates = [] + try: + projects = _as_items(run(list_projects_handler, {}), "projects") or [] + candidates.extend((p["project_id"], f"project '{p.get('name', '?')}'") for p in projects if p.get("project_id")) + except Exception: # best-effort hint enrichment + pass + project_id = params.get("project_id") + if looks_like_uuid(project_id): + try: + nodes = _as_items(run(get_nodes_handler, {"project_id": project_id}), "nodes") or [] + candidates.extend((n["node_id"], f"node '{n.get('name', '?')}'") for n in nodes if n.get("node_id")) + except Exception: # best-effort hint enrichment + pass + try: + links = _as_items(run(get_links_handler, {"project_id": project_id}), "links") or [] + candidates.extend((link["link_id"], "link") for link in links if link.get("link_id")) + except Exception: # best-effort hint enrichment + pass + return candidates + + +def _edit_distance_at_most(a: str, b: str, cap: int) -> bool: + """ + Levenshtein distance with an early bail above *cap* (ids are short; + the cap keeps the DP band trivial). + """ + + if abs(len(a) - len(b)) > cap: + return False + if a == b: + return True + previous = list(range(len(b) + 1)) + for i, char_a in enumerate(a, 1): + current = [i] + best = i + for j, char_b in enumerate(b, 1): + substitution = previous[j - 1] + (char_a != char_b) + value = min(previous[j] + 1, current[j - 1] + 1, substitution) + current.append(value) + best = min(best, value) + if best > cap: + return False + previous = current + return previous[-1] <= cap diff --git a/tests/agent/mcp/test_locators.py b/tests/agent/mcp/test_locators.py new file mode 100644 index 000000000..ee4bd1cee --- /dev/null +++ b/tests/agent/mcp/test_locators.py @@ -0,0 +1,170 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +Name locators: name->UUID resolution and the UUID-typo "Did you mean" hint. +""" + +from gns3server.agent.mcp import locators +from gns3server.agent.mcp.locators import add_did_you_mean, looks_like_uuid, resolve_locators + +PROJECT_ID = "d1ab2349-2e53-4e41-a9ae-b5ffcd912408" +NODE_ID = "b78a2630-2c9e-4b43-9397-c855c8053d1a" +NODE_ID_TYPO = "b78a2630-3b9e-4b43-9397-c855c8053d1a" # 2c9e -> 3b9e (distance 2) + + +def _run_factory(projects=None, nodes=None): + """A run() callback answering the two list handlers from canned data.""" + + calls = [] + + def run(handler, params): + calls.append((handler, dict(params))) + if handler is locators.list_projects_handler: + if isinstance(projects, Exception): + raise projects + return {"projects": projects if projects is not None else []} + if handler is locators.get_nodes_handler: + if isinstance(nodes, Exception): + raise nodes + return nodes if nodes is not None else [] + raise AssertionError(f"unexpected handler {handler}") + + return run, calls + + +def test_looks_like_uuid(): + + assert looks_like_uuid(PROJECT_ID) is True + assert looks_like_uuid(PROJECT_ID.upper()) is True + assert looks_like_uuid("FRRDocker-4") is False + assert looks_like_uuid("") is False + assert looks_like_uuid(None) is False + assert looks_like_uuid(1234) is False + + +def test_uuid_params_pass_through_without_calls(): + """UUID-shaped values never trigger resolution lookups.""" + + run, calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "lab"}]) + params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": NODE_ID}, run) + assert error is None + assert params == {"project_id": PROJECT_ID, "node_id": NODE_ID} + assert calls == [] + + +def test_project_name_resolved(): + """A name-shaped project_id becomes the project's UUID.""" + + run, calls = _run_factory(projects=[{"project_id": "11111111-1111-1111-1111-111111111111", "name": "other"}, + {"project_id": PROJECT_ID, "name": "my-lab"}]) + params, error = resolve_locators({"project_id": "my-lab"}, run) + assert error is None + assert params["project_id"] == PROJECT_ID + assert len(calls) == 1 # only the project list + + +def test_project_name_not_found_lists_known_names(): + + run, _calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}]) + params, error = resolve_locators({"project_id": "no-such-lab"}, run) + assert params is None + assert "not found" in error["error"] + assert "my-lab" in error["error"] + + +def test_node_name_resolved_within_project(): + + run, calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}], + nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}]) + params, error = resolve_locators({"project_id": "my-lab", "node_id": "FRRDocker-4"}, run) + assert error is None + assert params["project_id"] == PROJECT_ID + assert params["node_id"] == NODE_ID + # node lookup went to the already-resolved project + assert calls[1] == (locators.get_nodes_handler, {"project_id": PROJECT_ID}) + + +def test_node_name_case_insensitive_unique_match(): + + run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}]) + params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "frrdocker-4"}, run) + assert error is None + assert params["node_id"] == NODE_ID + + +def test_node_name_not_found_lists_known_nodes(): + + run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}]) + params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "vpcs-1"}, run) + assert params is None + assert "FRRDocker-4" in error["error"] + + +def test_handler_failure_during_resolution_is_reported(): + + run, _calls = _run_factory(nodes={"error": "boom"}) + params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "FRRDocker-4"}, run) + assert params is None + assert "Could not resolve node" in error["error"] + + +def test_did_you_mean_appended_for_transcription_slip(): + """The 2c9e -> 3b9e class of typo earns exactly one hint.""" + + run, _calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}], + nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}]) + result = add_did_you_mean( + {"error": f"Node ID {NODE_ID_TYPO} doesn't exist"}, {"project_id": PROJECT_ID}, run + ) + assert f"Did you mean node 'FRRDocker-4' ({NODE_ID})" in result["error"] + # original message preserved + assert NODE_ID_TYPO in result["error"] + + +def test_no_hint_when_too_far_or_ambiguous(): + + far = "b78a2630-9999-4b43-9397-c855c8053d1a" # distance 3 + run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "n1"}]) + result = add_did_you_mean({"error": f"Node ID {far} doesn't exist"}, {"project_id": PROJECT_ID}, run) + assert "Did you mean" not in result["error"] + + two = [NODE_ID, "b78a2630-2b9e-4b43-9397-c855c8053d1a"] # both within distance 2 of the typo + run2, _calls2 = _run_factory(nodes=[{"node_id": i, "name": f"n{i}"} for i in two]) + result2 = add_did_you_mean( + {"error": f"Node ID {NODE_ID_TYPO} doesn't exist"}, {"project_id": PROJECT_ID}, run2 + ) + assert "Did you mean" not in result2["error"] + + +def test_non_error_results_untouched(): + + run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "n1"}]) + payload = [{"node_id": NODE_ID}] + assert add_did_you_mean(payload, {"project_id": PROJECT_ID}, run) is payload + assert add_did_you_mean({"filters": {}}, {"project_id": PROJECT_ID}, run) == {"filters": {}} + # errors without an id carry nothing to hint against + result = add_did_you_mean({"error": "Node not found: no id here"}, {"project_id": PROJECT_ID}, run) + assert result == {"error": "Node not found: no id here"} + + +def test_edit_distance_at_most(): + + assert locators._edit_distance_at_most("abcd", "abcd", 2) + assert locators._edit_distance_at_most("abcd", "abce", 2) + assert locators._edit_distance_at_most("2c9e", "3b9e", 2) + assert not locators._edit_distance_at_most("abcd", "dcba", 2) + assert not locators._edit_distance_at_most("abc", "abcdef", 2)