Add createTopologyFolder MCP tool for topology regions.

This commit is contained in:
oliver 2026-08-04 17:30:56 +08:00
parent 8991bb31d1
commit 4abb037dce
5 changed files with 78 additions and 13 deletions

View file

@ -28,11 +28,11 @@ pip install "git+https://github.com/hansjone/netx.git#subdirectory=packages/netx
复制 [`mcp.json`](./mcp.json) 到 Cursor / oclaw(`server_id=netx-topology`),可与 `netx` 同时存在。
## 工具(13)
## 工具(14)
| 类别 | 工具 |
|------|------|
| 树/画布 | `getTopologyTree`, `listTopologyViews`, `getTopologyView`, `createTopologyView` |
| 树/区域/画布 | `getTopologyTree`, `createTopologyFolder`, `listTopologyViews`, `getTopologyView`, `createTopologyView` |
| 画图 | `addTopologyViewNodes` / `remove…` / `update…Positions`(**优先筛选**,API 自选 id;也可 id 列表), `projectTopologyNeighbors` |
| Fabric 只读 | `getTopologyFabricSummary`, `listTopologyFabricNodes`, `searchTopologyFabricNodes`, `queryTopologyNeighborhood`, `queryTopologyEdges` |

View file

@ -57,6 +57,22 @@ def _create_topology_view(args: dict[str, Any]) -> dict[str, Any]:
return _data(http_json("POST", "/v1/topology/views", body=body))
def _create_topology_folder(args: dict[str, Any]) -> dict[str, Any]:
"""Create a region folder under the topology root (API kind=region only)."""
name = str(args.get("name") or "").strip()
if not name:
return {"ok": False, "error": "name_required"}
body: dict[str, Any] = {
"name": name,
"kind": "region",
"sort_order": int(args.get("sort_order") or 0),
}
parent_id = str(args.get("parent_id") or "").strip()
if parent_id:
body["parent_id"] = parent_id
return _data(http_json("POST", "/v1/topology/folders", body=body))
_CHUNK = 500
@ -343,7 +359,7 @@ def _query_topology_edges(args: dict[str, Any]) -> dict[str, Any]:
HTTP_MCP_TOOLS: list[dict[str, Any]] = [
{
"name": "getTopologyTree",
"description": "Get topology folder tree (sites/regions) with nested views — start here before createTopologyView.",
"description": "Get topology folder tree (sites/regions) with nested views — start here; create region with createTopologyFolder if missing.",
"inputSchema": {"type": "object", "properties": {}, "required": [], "additionalProperties": False},
},
{
@ -361,9 +377,29 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [
"additionalProperties": False,
},
},
{
"name": "createTopologyFolder",
"description": (
"Create a region folder under the topology root. Returns folder id for createTopologyView. "
"Also auto-creates a default physical map under the region. Requires ne:write."
),
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Region display name"},
"parent_id": {
"type": "string",
"description": "Optional; must be topology root id (default: root)",
},
"sort_order": {"type": "integer", "default": 0},
},
"required": ["name"],
"additionalProperties": False,
},
},
{
"name": "createTopologyView",
"description": "Create a topology canvas under a folder (folder_id from getTopologyTree).",
"description": "Create a topology canvas under a folder (folder_id from getTopologyTree or createTopologyFolder).",
"inputSchema": {
"type": "object",
"properties": {
@ -569,6 +605,7 @@ _HANDLERS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
"getTopologyTree": _get_topology_tree,
"listTopologyViews": _list_topology_views,
"getTopologyView": _get_topology_view,
"createTopologyFolder": _create_topology_folder,
"createTopologyView": _create_topology_view,
"addTopologyViewNodes": _add_topology_view_nodes,
"removeTopologyViewNodes": _remove_topology_view_nodes,
@ -585,6 +622,7 @@ TOOL_REQUIRED_SCOPE: dict[str, str] = {
"getTopologyTree": "ne:read",
"listTopologyViews": "ne:read",
"getTopologyView": "ne:read",
"createTopologyFolder": "ne:write",
"createTopologyView": "ne:write",
"addTopologyViewNodes": "ne:write",
"removeTopologyViewNodes": "ne:write",

View file

@ -13,7 +13,8 @@ from netx_topology_mcp.server import _fetch_scopes
def test_tool_list_has_draw_and_query_tools() -> None:
names = {str(t.get("name") or "") for t in HTTP_MCP_TOOLS}
assert len(names) == 13
assert len(names) == 14
assert "createTopologyFolder" in names
assert "createTopologyView" in names
assert "addTopologyViewNodes" in names
assert "updateTopologyViewPositions" in names
@ -23,6 +24,29 @@ def test_tool_list_has_draw_and_query_tools() -> None:
assert "populateTopologyView" not in names
def test_create_folder_requires_name() -> None:
out = call_http_tool("createTopologyFolder", {})
assert out.get("isError") is True
payload = json.loads(out["content"][0]["text"])
assert payload["error"] == "name_required"
def test_create_folder_posts_region() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"id": "f1", "name": "华北", "kind": "region"}}
out = call_http_tool("createTopologyFolder", {"name": "华北", "sort_order": 1})
mock_http.assert_called_once()
assert mock_http.call_args[0][0] == "POST"
assert mock_http.call_args[0][1] == "/v1/topology/folders"
body = mock_http.call_args[1]["body"]
assert body["name"] == "华北"
assert body["kind"] == "region"
assert body["sort_order"] == 1
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
assert payload["id"] == "f1"
def test_add_nodes_rejects_managed_ume_ids() -> None:
out = call_http_tool(
"addTopologyViewNodes",
@ -137,8 +161,10 @@ def test_tools_for_scopes_filters_write() -> None:
read_only = {str(t.get("name") or "") for t in tools_for_scopes(["ne:read"])}
assert "queryTopologyEdges" in read_only
assert "createTopologyView" not in read_only
assert "createTopologyFolder" not in read_only
write = {str(t.get("name") or "") for t in tools_for_scopes(["ne:read", "ne:write"])}
assert "createTopologyView" in write
assert "createTopologyFolder" in write
def test_fetch_scopes_unwraps_envelope() -> None:
@ -175,7 +201,7 @@ def test_stdio_initialize_and_tools_list() -> None:
proc.stdin.flush()
list_resp = json.loads(proc.stdout.readline())
tools = list_resp["result"]["tools"]
assert len(tools) == 13
assert len(tools) == 14
finally:
proc.terminate()
proc.wait(timeout=5)