Add netx UME NE inventory tools for ops

Expose netx_query_ume_ne_inventory and netx_get_ume_ne (GET /v1/ume/inventory/ne and /ne/{id}). Update ops playbook, ROLE_SYSTEM, env example comment, and unit tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

_local/system.env.example

@@ -372,7 +372,7 @@ TAVILY_API_KEY=
 # 【前端】多数上述项在 Admin 设置表单中有对应勾选或输入框。
 
 # -----------------------------------------------------------------------------
-# 二十二、netx（network_ops 内部工具：告警明细 / 聚合 / 诊断）
+# 二十二、netx（network_ops 内部工具：UME 告警 / 网元清单 / 诊断）
 # -----------------------------------------------------------------------------
 # OCLAW_NETX_BASE_URL    netx HTTP 根地址（无尾部斜杠）；runtime/tools/experts/network_ops/netx_tools.py
 #    调用 /v1/alarms、/v1/alarms/aggregate、/v1/diagnostics。

runtime/tools/experts/network_ops/netx_tools.py

@@ -164,12 +164,14 @@ def _format_ops_netx_system_extension(r: dict[str, Any], *, lang_en: bool) -> st
                 f"- last_error: {err[:200]}" if lang_en else f"- 最近错误: {err[:200]}"
             )
         tail_en = (
-            "- tools: netx_query_ume_alarms, netx_aggregate_ume_alarms, netx_run_ume_diagnostics\n"
-            "- note: this is only runtime anchor; use tools for alarm evidence."
+            "- tools: netx_query_ume_alarms, netx_aggregate_ume_alarms, netx_run_ume_diagnostics, "
+            "netx_query_ume_ne_inventory, netx_get_ume_ne\n"
+            "- note: this is only runtime anchor; use tools for alarm/ne evidence."
         )
         tail_zh = (
-            "- 工具: netx_query_ume_alarms、netx_aggregate_ume_alarms、netx_run_ume_diagnostics\n"
-            "- 说明: 此处仅为运行锚点；具体告警必须以工具返回为准，勿臆测。"
+            "- 工具: netx_query_ume_alarms、netx_aggregate_ume_alarms、netx_run_ume_diagnostics、"
+            "netx_query_ume_ne_inventory、netx_get_ume_ne\n"
+            "- 说明: 此处仅为运行锚点；具体告警/网元信息必须以工具返回为准，勿臆测。"
         )
         return "\n".join(lines_en + [tail_en]) if lang_en else "\n".join(lines_zh + [tail_zh])
     err = str(r.get("error") or "")
@@ -611,6 +613,74 @@ def netx_run_ume_diagnostics_tool() -> ToolSpec:
     )
 
 
+def netx_query_ume_ne_inventory_tool() -> ToolSpec:
+    """Paged UME NE inventory synced in netx (PostgreSQL-backed)."""
+
+    def handler(args: dict[str, Any]) -> dict[str, Any]:
+        page = max(1, int(args.get("page") or 1))
+        page_size = min(500, max(1, int(args.get("page_size") or 50)))
+        params: dict[str, Any] = {"page": page, "page_size": page_size}
+        if str(args.get("keyword") or "").strip():
+            params["keyword"] = str(args.get("keyword")).strip()
+        return _http_json("GET", "/v1/ume/inventory/ne", params=params)
+
+    return ToolSpec(
+        name="netx_query_ume_ne_inventory",
+        description=(
+            "查询 netx 已同步的 UME 网元清单（读 /v1/ume/inventory/ne，与 netx Web「网元清单」同源）。"
+            "keyword 可选：匹配 ne_id / ne_name / user_label / ip_address 包含。"
+            "返回 total、page、page_size、items（含在线状态、地址、类型等）。"
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "keyword": {"type": "string", "description": "关键字过滤（可选）"},
+                "page": {"type": "integer", "minimum": 1, "default": 1},
+                "page_size": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50},
+            },
+            "required": [],
+            "additionalProperties": False,
+        },
+        handler=handler,
+        tags=frozenset({"netx", "ops", "ume", "inventory", "ne", "read_only"}),
+        risk_level="low",
+        read_only=True,
+    )
+
+
+def netx_get_ume_ne_tool() -> ToolSpec:
+    """Single UME NE detail by ne_id from netx."""
+
+    def handler(args: dict[str, Any]) -> dict[str, Any]:
+        from urllib.parse import quote
+
+        ne_id = str(args.get("ne_id") or "").strip()
+        if not ne_id:
+            return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"}
+        safe = quote(ne_id, safe="")
+        return _http_json("GET", f"/v1/ume/inventory/ne/{safe}", params=None)
+
+    return ToolSpec(
+        name="netx_get_ume_ne",
+        description=(
+            "按网元 UUID（ne_id）读取 netx 中单条 UME 网元详情（GET /v1/ume/inventory/ne/{ne_id}）。"
+            "含 vendor、source_type、raw_json 等；404 时上游返回 ume_ne_not_found。"
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "ne_id": {"type": "string", "description": "网元 UUID（与清单中 ne_id 一致）"},
+            },
+            "required": ["ne_id"],
+            "additionalProperties": False,
+        },
+        handler=handler,
+        tags=frozenset({"netx", "ops", "ume", "inventory", "ne", "read_only"}),
+        risk_level="low",
+        read_only=True,
+    )
+
+
 def netx_query_ume_alarms_raw_tool() -> ToolSpec:
     """Power query UME current alarms with full alarm+NE fields."""
 
@@ -868,6 +938,8 @@ __all__ = [
     "netx_query_ume_alarms_tool",
     "netx_aggregate_ume_alarms_tool",
     "netx_run_ume_diagnostics_tool",
+    "netx_query_ume_ne_inventory_tool",
+    "netx_get_ume_ne_tool",
     "netx_query_ume_alarms_raw_tool",
     "netx_aggregate_ume_alarms_raw_tool",
     "netx_list_ume_alarm_fields_tool",

runtime/workspaces/ops/ROLE_SYSTEM.md

@@ -16,7 +16,7 @@
 - 先结论，再给证据与最小修复步骤。
 
 ## 必须加载技能
-- 每次处理 netx/UME 告警问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。
+- 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。
 
 ## netx 明细与统计（内部工具）
 
@@ -25,5 +25,7 @@
 - 默认使用 UME 当前告警链路，不再依赖导入批次 `batch_id`。
 - `netx_query_ume_alarms`：查询 UME 当前告警明细（支持 `severity/ne_id/keyword`）。
 - `netx_aggregate_ume_alarms` / `netx_run_ume_diagnostics`：查询 UME 聚合与诊断摘要。
+- `netx_query_ume_ne_inventory`：分页查询已同步的 UME 网元清单（可选 `keyword`）。
+- `netx_get_ume_ne`：按 `ne_id`（UUID）取单网元详情（含 `raw_json`）。
 
 工具走 netx（`OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`）。关闭自动锚点：环境变量 `OCLAW_OPS_NETX_CONTEXT_INJECT=0`。

skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md

@@ -1,13 +1,13 @@
 ---
 name: ops-netx-ume-playbook
-description: 面向 ops 专家的 netx UME 告警分析标准作业手册。用于任何告警查询、聚合统计、诊断摘要、raw 字段过滤与 UME SQL 分析场景。
+description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查询/聚合/诊断、网元清单与单网元详情、raw 字段过滤与 UME 只读 SQL。
 ---
 
 # Ops Netx UME 作业手册
 
 ## 强制使用范围
 
-凡是涉及 netx/UME 告警的 ops 请求，必须优先加载并遵循本技能。
+凡是涉及 netx/UME **告警**或 **网元信息** 的 ops 请求，必须优先加载并遵循本技能。
 
 ## 工具选择顺序
 
@@ -22,6 +22,9 @@ description: 面向 ops 专家的 netx UME 告警分析标准作业手册。用
    - `netx_aggregate_ume_alarms_raw`（`group_by`，可选 `group_by2`）
 4. 高级分析（SQL）：
    - `netx_sql_query_ume`（仅 SELECT、仅 UME 表；重查询建议设置 `statement_timeout_ms`）
+5. **网元（inventory，与 netx「网元清单」同源）**：
+   - 列表/搜索：`netx_query_ume_ne_inventory`（`keyword` + 分页）
+   - 单条详情（含 `raw_json`）：`netx_get_ume_ne`（`ne_id` = UUID）
 
 ## 快速决策树（强推荐）
 
@@ -35,6 +38,9 @@ description: 面向 ops 专家的 netx UME 告警分析标准作业手册。用
   - `netx_aggregate_ume_alarms_raw`（`group_by` / `group_by2`）
 - **需要复杂条件 / 自定义计算 / 多条件关联**：
   - `netx_sql_query_ume`（必须过滤 + `statement_timeout_ms`）
+- **查网元是谁、IP/标签、在线状态、或核对告警里的 ne_id**：
+  - 先 `netx_query_ume_ne_inventory`（`keyword` 可填名称片段或 UUID 片段）
+  - 需要完整字段与 `raw_json` 时再 `netx_get_ume_ne`
 
 ## 约束与护栏
 

skills/_workspace/ops/ops-netx-ume-playbook/reference.md

@@ -43,6 +43,14 @@
 - 工具：`netx_run_ume_diagnostics`
 - 用途：在深挖前先快速形成“概览简报”（严重度、Top 网元、Top 事件类型等）
 
+## 4b) 网元清单与详情
+
+- 工具：`netx_query_ume_ne_inventory` — 对应 netx `GET /v1/ume/inventory/ne`
+  - 参数：`keyword`（可选，匹配 ne_id/ne_name/user_label/ip）、`page`、`page_size`（最大 500）
+- 工具：`netx_get_ume_ne` — `GET /v1/ume/inventory/ne/{ne_id}`
+  - 参数：`ne_id`（UUID，与清单字段一致）
+- 与告警 raw 中的 `ne_*` 字段互补：需要 UME 原始 JSON 时用 `netx_get_ume_ne` 的 `raw_json`。
+
 ## 5) SQL 深度分析（受限）
 
 - 工具：`netx_sql_query_ume`

tests/test_netx_ume_ne_tools.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+
+def test_netx_query_ume_ne_inventory_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    import oclaw.runtime.tools.experts.network_ops.netx_tools as nt
+
+    calls: list[tuple[str, str, dict[str, Any] | None]] = []
+
+    def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        calls.append((method, path, params))
+        return {"ok": True, "data": {"total": 0, "page": 1, "page_size": 50, "items": []}}
+
+    monkeypatch.setattr(nt, "_http_json", fake)
+    spec = nt.netx_query_ume_ne_inventory_tool()
+    out = spec.handler({"keyword": "10.0.0", "page": 2, "page_size": 100})
+    assert out.get("ok") is True
+    assert len(calls) == 1
+    assert calls[0][0] == "GET"
+    assert calls[0][1] == "/v1/ume/inventory/ne"
+    assert calls[0][2] == {"page": 2, "page_size": 100, "keyword": "10.0.0"}
+
+
+def test_netx_get_ume_ne_requires_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    import oclaw.runtime.tools.experts.network_ops.netx_tools as nt
+
+    monkeypatch.setattr(nt, "_http_json", lambda *a, **k: {"ok": True, "data": {}})
+    spec = nt.netx_get_ume_ne_tool()
+    out = spec.handler({})
+    assert out.get("ok") is False
+    assert out.get("error_code") == "ne_id_required"
+
+
+def test_netx_get_ume_ne_quotes_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    import oclaw.runtime.tools.experts.network_ops.netx_tools as nt
+
+    paths: list[str] = []
+
+    def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        paths.append(path)
+        return {"ok": True, "data": {"ne_id": "x"}}
+
+    monkeypatch.setattr(nt, "_http_json", fake)
+    spec = nt.netx_get_ume_ne_tool()
+    nid = "550e8400-e29b-41d4-a716-446655440000"
+    spec.handler({"ne_id": nid})
+    assert paths == [f"/v1/ume/inventory/ne/{nid}"]