haeaeae/homelable
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mcp): expose zones to the MCP server

Browse Source 
An AI client could create nodes but had no way to create the areas that
organize them, or to move a node into one (#365).

Four tools, all on the existing /api/v1/nodes endpoints: create_zone
(a groupRect node, with size and border/text colours), list_zones (each
zone plus the ids it contains), add_to_zone and remove_from_zone, both
taking a list of node ids.

The canvas stores a child's position relative to its parent, so
add_to_zone rebases each node on the zone and remove_from_zone restores
absolute coordinates — otherwise a node jumps by the zone's offset the
next time the canvas loads. add_to_zone skips what it cannot move (the
zone itself, a node already inside it, an unknown id, and any node the
zone descends from, which would build a cycle) and reports them back
rather than failing the batch.

groupRect stays out of NODE_TYPES: create_node remains a device tool,
and the enum sync tests keep passing.

ha-relevant: no
This commit is contained in:
Pouzor
2026-09-01 08:52:31 +02:00
committed by Pouzor - Rémy Jardient
parent 3301f00e57
commit 1ec7ea7e8e
2 changed files with 238 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp/app/tools.py
+136
View File
@@ -64,6 +64,18 @@ _NODE_FIELDS = {
},
}
# A zone is a groupRect node: canvas furniture that visually groups the nodes
# parented to it (a VLAN, a site, a rack room). It is deliberately kept out of
# NODE_TYPES — create_node makes devices, these tools make and fill zones.
ZONE_TYPE = "groupRect"
_ZONE_COLOR_FIELDS = {
"border": {"type": "string", "description": "Border/label colour, e.g. '#00d4ff'."},
"border_style": {"type": "string", "enum": ["solid", "dashed", "dotted"]},
"border_width": {"type": "number"},
"text_color": {"type": "string", "description": "Label colour, e.g. '#e6edf3'."},
}
# 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.
@@ -178,6 +190,38 @@ def _build_tools() -> list[Tool]:
"required": ["id"],
"properties": {"id": {"type": "string"}},
}),
Tool(name="create_zone", description="Create a zone (a groupRect area) on the canvas. A zone visually groups the nodes parented to it — a VLAN, a site, a room. Use add_to_zone to fill it.", inputSchema={
"type": "object",
"required": ["label"],
"properties": {
"label": {"type": "string", "description": "Zone name, shown in its top-left corner."},
"pos_x": {"type": "number", "description": "X position on the canvas. Omit to auto-place."},
"pos_y": {"type": "number", "description": "Y position on the canvas. Omit to auto-place."},
"width": {"type": "number", "description": "Zone width in pixels (default 360)."},
"height": {"type": "number", "description": "Zone height in pixels (default 240)."},
**_ZONE_COLOR_FIELDS,
**_DESIGN_ID_FIELD,
},
}),
Tool(name="list_zones", description="List the zones (groupRect areas) on the canvas with their id, label, position, size and the nodes each one contains.", inputSchema={
"type": "object",
"properties": {},
}),
Tool(name="add_to_zone", description="Move one or more nodes into a zone. Positions are rebased on the zone so the nodes keep their place on screen. A node already in the zone, the zone itself, and any node the zone is nested in are skipped.", inputSchema={
"type": "object",
"required": ["zone_id", "node_ids"],
"properties": {
"zone_id": {"type": "string", "description": "Target zone id — call list_zones to discover it."},
"node_ids": {"type": "array", "items": {"type": "string"}, "description": "Ids of the nodes to move in."},
},
}),
Tool(name="remove_from_zone", description="Take nodes back out of the zone they are in, restoring their absolute canvas position. Nodes that are not in a zone are left alone.", inputSchema={
"type": "object",
"required": ["node_ids"],
"properties": {
"node_ids": {"type": "array", "items": {"type": "string"}, "description": "Ids of the nodes to detach."},
},
}),
Tool(name="list_designs", description="List all designs (canvases) with their IDs and node/group/text counts", inputSchema={
"type": "object",
"properties": {},
@@ -258,6 +302,34 @@ def _slim_node_summary(n: dict) -> dict:
return {"id": n.get("id"), "label": n.get("label"), "type": n.get("type"), "status": n.get("status")}
def _slim_zone(z: dict, nodes: list[dict]) -> dict:
"""A zone plus the ids of the nodes parented to it."""
return {
"id": z.get("id"),
"label": z.get("label"),
"pos_x": z.get("pos_x"),
"pos_y": z.get("pos_y"),
"width": z.get("width"),
"height": z.get("height"),
"node_ids": [n["id"] for n in nodes if n.get("parent_id") == z.get("id")],
}
def _is_ancestor(nodes_by_id: dict[str, dict], maybe_ancestor_id: str, node_id: str) -> bool:
"""True when `maybe_ancestor_id` sits on `node_id`'s parent chain. Cycle-safe."""
seen = {node_id}
current = nodes_by_id.get(node_id)
while current:
pid = current.get("parent_id")
if not pid or pid in seen:
return False
if pid == maybe_ancestor_id:
return True
seen.add(pid)
current = nodes_by_id.get(pid)
return False
async def _dispatch(name: str, args: dict) -> dict:
if name == "create_node":
return await backend.post("/api/v1/nodes", args)
@@ -331,6 +403,70 @@ async def _dispatch(name: str, args: dict) -> dict:
if name == "restore_device":
return await backend.post(f"/api/v1/scan/pending/{args['id']}/restore", {})
if name == "create_zone":
colors = {k: args.pop(k) for k in list(_ZONE_COLOR_FIELDS) if k in args}
body = {
"type": ZONE_TYPE,
"status": "unknown",
"width": args.pop("width", 360),
"height": args.pop("height", 240),
**args,
}
if colors:
body["custom_colors"] = colors
return await backend.post("/api/v1/nodes", body)
if name == "list_zones":
nodes = await backend.get("/api/v1/nodes")
return [_slim_zone(n, nodes) for n in nodes if n.get("type") == ZONE_TYPE]
if name == "add_to_zone":
zone_id = args["zone_id"]
nodes = await backend.get("/api/v1/nodes")
by_id = {n["id"]: n for n in nodes}
zone = by_id.get(zone_id)
if zone is None or zone.get("type") != ZONE_TYPE:
raise ValueError(f"add_to_zone: {zone_id} is not a zone")
moved, skipped = [], []
for node_id in args["node_ids"]:
node = by_id.get(node_id)
if (
node is None
or node_id == zone_id
or node.get("parent_id") == zone_id
or _is_ancestor(by_id, node_id, zone_id)
):
skipped.append(node_id)
continue
# The canvas stores a child's position relative to its parent, so
# rebase or the node jumps by the zone's offset on the next load.
await backend.patch(f"/api/v1/nodes/{node_id}", {
"parent_id": zone_id,
"pos_x": (node.get("pos_x") or 0) - (zone.get("pos_x") or 0),
"pos_y": (node.get("pos_y") or 0) - (zone.get("pos_y") or 0),
})
moved.append(node_id)
return {"zone_id": zone_id, "moved": moved, "skipped": skipped}
if name == "remove_from_zone":
nodes = await backend.get("/api/v1/nodes")
by_id = {n["id"]: n for n in nodes}
detached, skipped = [], []
for node_id in args["node_ids"]:
node = by_id.get(node_id)
zone = by_id.get(node.get("parent_id") or "") if node else None
if node is None or zone is None or zone.get("type") != ZONE_TYPE:
skipped.append(node_id)
continue
# Zone-relative → absolute, so the node stays where it looks.
await backend.patch(f"/api/v1/nodes/{node_id}", {
"parent_id": None,
"pos_x": (node.get("pos_x") or 0) + (zone.get("pos_x") or 0),
"pos_y": (node.get("pos_y") or 0) + (zone.get("pos_y") or 0),
})
detached.append(node_id)
return {"detached": detached, "skipped": skipped}
if name == "list_designs":
return await backend.get("/api/v1/designs")
mcp/tests/test_tools.py
+102
View File
@@ -530,3 +530,105 @@ async def test_list_hidden_devices(mock_backend):
async def test_restore_device(mock_backend):
await _dispatch("restore_device", {"id": "5"})
mock_backend.post.assert_called_once_with("/api/v1/scan/pending/5/restore", {})
# ── Zones (#365) ─────────────────────────────────────────────────────────────
_ZONE_NODES = [
{"id": "z1", "type": "groupRect", "label": "DMZ", "pos_x": 100, "pos_y": 100, "width": 400, "height": 300},
{"id": "n1", "type": "router", "label": "Router", "pos_x": 160, "pos_y": 220},
{"id": "n2", "type": "server", "label": "NAS", "pos_x": 200, "pos_y": 260, "parent_id": "z1"},
]
@pytest.mark.anyio
async def test_create_zone_posts_a_grouprect_node(mock_backend):
await _dispatch("create_zone", {"label": "DMZ", "pos_x": 10, "pos_y": 20})
mock_backend.post.assert_called_once_with("/api/v1/nodes", {
"type": "groupRect", "status": "unknown", "width": 360, "height": 240,
"label": "DMZ", "pos_x": 10, "pos_y": 20,
})
@pytest.mark.anyio
async def test_create_zone_folds_colors_into_custom_colors(mock_backend):
await _dispatch("create_zone", {"label": "VLAN 10", "border": "#39d353", "border_style": "dashed", "width": 500})
body = mock_backend.post.call_args[0][1]
assert body["custom_colors"] == {"border": "#39d353", "border_style": "dashed"}
assert body["width"] == 500
assert "border" not in body
def test_create_zone_schema_requires_label():
tool = next(t for t in TOOLS if t.name == "create_zone")
assert tool.inputSchema["required"] == ["label"]
def test_zone_type_is_not_offered_by_create_node():
"""A zone is canvas furniture — create_node stays a device tool."""
tool = next(t for t in TOOLS if t.name == "create_node")
assert "groupRect" not in tool.inputSchema["properties"]["type"]["enum"]
@pytest.mark.anyio
async def test_list_zones_returns_only_zones_with_their_children(mock_backend):
mock_backend.get = AsyncMock(return_value=list(_ZONE_NODES))
result = await _dispatch("list_zones", {})
assert result == [{
"id": "z1", "label": "DMZ", "pos_x": 100, "pos_y": 100,
"width": 400, "height": 300, "node_ids": ["n2"],
}]
@pytest.mark.anyio
async def test_add_to_zone_rebases_positions_on_the_zone(mock_backend):
mock_backend.get = AsyncMock(return_value=list(_ZONE_NODES))
result = await _dispatch("add_to_zone", {"zone_id": "z1", "node_ids": ["n1"]})
mock_backend.patch.assert_called_once_with(
"/api/v1/nodes/n1", {"parent_id": "z1", "pos_x": 60, "pos_y": 120}
)
assert result == {"zone_id": "z1", "moved": ["n1"], "skipped": []}
@pytest.mark.anyio
async def test_add_to_zone_skips_the_zone_itself_and_nodes_already_in_it(mock_backend):
mock_backend.get = AsyncMock(return_value=list(_ZONE_NODES))
result = await _dispatch("add_to_zone", {"zone_id": "z1", "node_ids": ["z1", "n2", "ghost"]})
mock_backend.patch.assert_not_called()
assert result["moved"] == []
assert result["skipped"] == ["z1", "n2", "ghost"]
@pytest.mark.anyio
async def test_add_to_zone_refuses_a_node_the_zone_descends_from(mock_backend):
"""Parenting a zone's own ancestor under it would build a cycle."""
mock_backend.get = AsyncMock(return_value=[
{"id": "outer", "type": "groupRect", "pos_x": 0, "pos_y": 0},
{"id": "z1", "type": "groupRect", "pos_x": 100, "pos_y": 100, "parent_id": "outer"},
])
result = await _dispatch("add_to_zone", {"zone_id": "z1", "node_ids": ["outer"]})
mock_backend.patch.assert_not_called()
assert result["skipped"] == ["outer"]
@pytest.mark.anyio
async def test_add_to_zone_rejects_a_target_that_is_not_a_zone(mock_backend):
mock_backend.get = AsyncMock(return_value=list(_ZONE_NODES))
with pytest.raises(ValueError, match="is not a zone"):
await _dispatch("add_to_zone", {"zone_id": "n1", "node_ids": ["n2"]})
@pytest.mark.anyio
async def test_remove_from_zone_restores_absolute_positions(mock_backend):
mock_backend.get = AsyncMock(return_value=list(_ZONE_NODES))
result = await _dispatch("remove_from_zone", {"node_ids": ["n2", "n1"]})
mock_backend.patch.assert_called_once_with(
"/api/v1/nodes/n2", {"parent_id": None, "pos_x": 300, "pos_y": 360}
)
# n1 has no parent — nothing to detach.
assert result == {"detached": ["n2"], "skipped": ["n1"]}
def test_zone_tools_are_registered():
names = {t.name for t in TOOLS}
assert {"create_zone", "list_zones", "add_to_zone", "remove_from_zone"} <= names