feat: let the MCP target a specific design/canvas
Add an optional design_id argument to create_node, create_edge and get_canvas so AI clients can read and populate a canvas other than the first design. Add a list_designs tool wrapping GET /api/v1/designs so those IDs are discoverable. Backward compatible: omitting design_id preserves the existing first-design fallback in the backend (nodes.py / edges.py / canvas.py). Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+21
-2
@@ -43,11 +43,20 @@ _NODE_FIELDS = {
|
||||
},
|
||||
}
|
||||
|
||||
# Optional design/canvas selector. The backend attaches nodes/edges to the first
|
||||
# design when design_id is omitted (see backend nodes.py / edges.py), so these
|
||||
# tools stay backward compatible; pass design_id to target a specific canvas.
|
||||
# Use list_designs to discover the available IDs.
|
||||
_DESIGN_ID_FIELD = {
|
||||
"design_id": {"type": "string", "description": "Target design/canvas ID. Omit to use the default (first) canvas; call list_designs to discover IDs."},
|
||||
}
|
||||
|
||||
|
||||
def _build_tools() -> list[Tool]:
|
||||
create_node_props = {
|
||||
"type": {"type": "string", "enum": NODE_TYPES},
|
||||
**_NODE_FIELDS,
|
||||
**_DESIGN_ID_FIELD,
|
||||
}
|
||||
create_node_props["status"] = {**_NODE_FIELDS["status"], "default": "unknown"}
|
||||
|
||||
@@ -81,6 +90,7 @@ def _build_tools() -> list[Tool]:
|
||||
"target": {"type": "string"},
|
||||
"type": {"type": "string", "enum": ["ethernet", "wifi", "iot", "vlan", "virtual"], "default": "ethernet"},
|
||||
"label": {"type": "string"},
|
||||
**_DESIGN_ID_FIELD,
|
||||
},
|
||||
}),
|
||||
Tool(name="delete_edge", description="Delete a network link", inputSchema={
|
||||
@@ -110,7 +120,7 @@ def _build_tools() -> list[Tool]:
|
||||
}),
|
||||
Tool(name="get_canvas", description="Get the full canvas: all nodes and edges in the homelab topology", inputSchema={
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"properties": {**_DESIGN_ID_FIELD},
|
||||
}),
|
||||
Tool(name="list_nodes", description="List all nodes (devices) in the homelab", inputSchema={
|
||||
"type": "object",
|
||||
@@ -120,6 +130,10 @@ def _build_tools() -> list[Tool]:
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
}),
|
||||
Tool(name="list_designs", description="List all designs (canvases) with their IDs and node/group/text counts", inputSchema={
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
}),
|
||||
]
|
||||
|
||||
|
||||
@@ -192,7 +206,9 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
return await backend.post(f"/api/v1/scan/pending/{args['id']}/hide", {})
|
||||
|
||||
if name == "get_canvas":
|
||||
raw = await backend.get("/api/v1/canvas")
|
||||
design_id = args.get("design_id")
|
||||
path = f"/api/v1/canvas?design_id={design_id}" if design_id else "/api/v1/canvas"
|
||||
raw = await backend.get(path)
|
||||
return _slim_canvas(raw)
|
||||
|
||||
if name == "list_nodes":
|
||||
@@ -201,4 +217,7 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
if name == "list_pending_devices":
|
||||
return await backend.get("/api/v1/scan/pending")
|
||||
|
||||
if name == "list_designs":
|
||||
return await backend.get("/api/v1/designs")
|
||||
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
@@ -77,6 +77,21 @@ async def test_create_edge(mock_backend):
|
||||
mock_backend.post.assert_called_once_with("/api/v1/edges", {"source": "1", "target": "2", "type": "ethernet"})
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_node_with_design_id(mock_backend):
|
||||
# design_id is forwarded to the backend, which attaches the node to that canvas.
|
||||
args = {"type": "server", "label": "Proxmox", "design_id": "design-2"}
|
||||
await _dispatch("create_node", dict(args))
|
||||
mock_backend.post.assert_called_once_with("/api/v1/nodes", args)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_edge_with_design_id(mock_backend):
|
||||
args = {"source": "1", "target": "2", "type": "ethernet", "design_id": "design-2"}
|
||||
await _dispatch("create_edge", dict(args))
|
||||
mock_backend.post.assert_called_once_with("/api/v1/edges", args)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_edge(mock_backend):
|
||||
await _dispatch("delete_edge", {"id": "99"})
|
||||
@@ -162,6 +177,21 @@ async def test_get_canvas_keeps_documentation_fields(mock_backend):
|
||||
assert node["properties"] == [{"name": "rack", "value": "A1"}]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_canvas_with_design_id(mock_backend):
|
||||
mock_backend.get = AsyncMock(return_value={"nodes": [], "edges": []})
|
||||
await _dispatch("get_canvas", {"design_id": "design-2"})
|
||||
mock_backend.get.assert_called_once_with("/api/v1/canvas?design_id=design-2")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_canvas_without_design_id_uses_default(mock_backend):
|
||||
# Backward compatible: no design_id -> unqualified canvas endpoint (first design).
|
||||
mock_backend.get = AsyncMock(return_value={"nodes": [], "edges": []})
|
||||
await _dispatch("get_canvas", {})
|
||||
mock_backend.get.assert_called_once_with("/api/v1/canvas")
|
||||
|
||||
|
||||
def _tool_schema(name: str) -> dict:
|
||||
tool = next(t for t in TOOLS if t.name == name)
|
||||
return tool.inputSchema["properties"]
|
||||
@@ -182,6 +212,19 @@ def test_update_node_schema_exposes_full_node_fields():
|
||||
assert "id" in props
|
||||
|
||||
|
||||
def test_design_id_exposed_on_canvas_targeting_tools():
|
||||
# create_node/create_edge/get_canvas can target a specific canvas.
|
||||
assert "design_id" in _tool_schema("create_node")
|
||||
assert "design_id" in _tool_schema("create_edge")
|
||||
assert "design_id" in _tool_schema("get_canvas")
|
||||
|
||||
|
||||
def test_update_node_schema_has_no_design_id():
|
||||
# The backend NodeUpdate schema can't move a node between designs, so the
|
||||
# update_node tool must not advertise design_id.
|
||||
assert "design_id" not in _tool_schema("update_node")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_nodes(mock_backend):
|
||||
mock_backend.get = AsyncMock(return_value=[{"id": "1", "label": "Freebox"}])
|
||||
@@ -198,6 +241,14 @@ async def test_list_pending_devices(mock_backend):
|
||||
assert result == [{"id": "p1", "ip": "192.168.1.50"}]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_designs(mock_backend):
|
||||
mock_backend.get = AsyncMock(return_value=[{"id": "d1", "name": "Network Topology", "node_count": 12}])
|
||||
result = await _dispatch("list_designs", {})
|
||||
mock_backend.get.assert_called_once_with("/api/v1/designs")
|
||||
assert result == [{"id": "d1", "name": "Network Topology", "node_count": 12}]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_unknown_tool():
|
||||
with pytest.raises(ValueError, match="Unknown tool"):
|
||||
|
||||
Reference in New Issue
Block a user