feat(ops): add netx managed NE query and exec tools

Add ops specialist tools to list/get managed NEs and execute read-only CLI via netx.
Document the new managed-NE workflow in role prompts, env example, and a dedicated ops skill with tests.

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

_local/system.env.example

@@ -433,10 +433,11 @@ TAVILY_API_KEY=
 # 【前端】多数上述项在 Admin 设置表单中有对应勾选或输入框。
 
 # -----------------------------------------------------------------------------
-# 二十二、netx（network_ops 内部工具：UME 告警 / 网元清单 / 诊断）
+# 二十二、netx（network_ops 内部工具：UME 告警 / 纳管网元 CLI / 诊断）
 # -----------------------------------------------------------------------------
 # OCLAW_NETX_BASE_URL    netx HTTP 根地址（无尾部斜杠）；runtime/tools/experts/network_ops/netx_tools.py
-#    调用 /v1/alarms、/v1/alarms/aggregate、/v1/diagnostics。
+#    UME：/v1/ume/alarms、/v1/ume/inventory/ne 等；纳管网元：/v1/managed-ne、/v1/managed-ne/exec。
+#    netx 侧需配置 NETX_CREDENTIAL_SECRET_KEY（设备密码 Fernet 加密）。
 #    默认 http://127.0.0.1:8890（代码内兜底；与本文件一致时可省略）。
 # OCLAW_NETX_API_TOKEN    可选；netx 若启用 Bearer 鉴权，填与 netx 侧一致的 token。
 # OCLAW_OPS_NETX_CONTEXT_INJECT    默认 1；ops 专家每轮把 netx 最新导入 batch_id 注入 system（类似附件锚点）。

runtime/tools/experts/network_ops/netx_tools.py

@@ -98,6 +98,23 @@ def _localize_netx_payload(data: dict[str, Any], *, lang: str) -> dict[str, Any]
     return data
 
 
+def _http_post_json(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]:
+    base = _netx_base_url()
+    url = f"{base}{path}"
+    try:
+        with httpx.Client(timeout=timeout, trust_env=False) as client:
+            resp = client.post(url, json=body, headers=_netx_headers())
+            text = resp.text
+            if not resp.is_success:
+                return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]}
+            data = resp.json() if text else {}
+            if isinstance(data, dict):
+                data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh"))
+            return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}}
+    except Exception as exc:
+        return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]}
+
+
 def _http_json(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
     base = _netx_base_url()
     url = f"{base}{path}"
@@ -970,6 +987,146 @@ def netx_aggregate_ume_alarms_raw_tool() -> ToolSpec:
     )
 
 
+def netx_list_managed_ne_tool() -> ToolSpec:
+    """List netx managed NEs (inventory for CLI login targets)."""
+
+    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()
+        if str(args.get("vendor") or "").strip():
+            params["vendor"] = str(args.get("vendor")).strip()
+        if str(args.get("connect_status") or "").strip():
+            params["connect_status"] = str(args.get("connect_status")).strip()
+        return _http_json("GET", "/v1/managed-ne", params=params)
+
+    return ToolSpec(
+        name="netx_list_managed_ne",
+        description=(
+            "列出 netx「网元管理」中已纳管的设备（GET /v1/managed-ne）。"
+            "返回 id、name、ip、vendor、device_type、connect_status 等（不含密码）。"
+            "keyword 可匹配名称/IP/用户名/标签；connect_status 可选 unknown/testing/pass/fail。"
+            "登录查配置前先用本工具定位 ne_id，再 netx_get_managed_ne / netx_exec_managed_ne。"
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "keyword": {"type": "string", "description": "名称/IP/用户名/标签包含（可选）"},
+                "vendor": {"type": "string", "description": "厂商过滤（可选）"},
+                "connect_status": {
+                    "type": "string",
+                    "enum": ["unknown", "testing", "pass", "fail"],
+                    "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", "managed_ne", "inventory", "read_only"}),
+        risk_level="low",
+        read_only=True,
+    )
+
+
+def netx_get_managed_ne_tool() -> ToolSpec:
+    """Single managed NE metadata from netx."""
+
+    def handler(args: dict[str, Any]) -> dict[str, Any]:
+        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"}
+        return _http_json("GET", f"/v1/managed-ne/{ne_id}", params=None)
+
+    return ToolSpec(
+        name="netx_get_managed_ne",
+        description=(
+            "读取 netx 单条纳管网元详情（GET /v1/managed-ne/{ne_id}）。"
+            "含 connect_status、connect_message、connect_detail（连通测试日志）、跳板 hop_* 配置摘要。"
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "ne_id": {"type": "string", "description": "网元 UUID（列表 items[].id）"},
+            },
+            "required": ["ne_id"],
+            "additionalProperties": False,
+        },
+        handler=handler,
+        tags=frozenset({"netx", "ops", "managed_ne", "read_only"}),
+        risk_level="low",
+        read_only=True,
+    )
+
+
+def netx_exec_managed_ne_tool() -> ToolSpec:
+    """Run read-only CLI on a managed NE via netx."""
+
+    def handler(args: dict[str, Any]) -> dict[str, Any]:
+        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"}
+        raw_cmds = args.get("commands")
+        if not isinstance(raw_cmds, list) or not raw_cmds:
+            return {"ok": False, "error": "commands_required", "error_code": "commands_required"}
+        commands = [str(c).strip() for c in raw_cmds if str(c).strip()]
+        if not commands:
+            return {"ok": False, "error": "commands_required", "error_code": "commands_required"}
+        if len(commands) > 5:
+            return {"ok": False, "error": "too_many_commands", "error_code": "too_many_commands"}
+        body: dict[str, Any] = {"ne_id": ne_id, "commands": commands}
+        rts = args.get("read_timeout_sec")
+        if rts is not None:
+            body["read_timeout_sec"] = int(rts)
+        out = _http_post_json("/v1/managed-ne/exec", body, timeout=300.0)
+        if not out.get("ok"):
+            return out
+        data = out.get("data") or {}
+        if isinstance(data, dict) and data.get("ok") is False:
+            return {"ok": False, "data": data, "error": str(data.get("error") or "exec_failed")}
+        return {"ok": True, "data": data}
+
+    return ToolSpec(
+        name="netx_exec_managed_ne",
+        description=(
+            "经 netx 登录「网元管理」中的设备并执行只读 CLI（POST /v1/managed-ne/exec）。"
+            "每条命令须以 show / display / get / ping / traceroute / terminal length / ? 开头；"
+            "禁止管道符、分号及改配置类命令；单次最多 5 条；默认读超时 60s。"
+            "返回合并输出（含命令回显）；失败时含 error/detail。"
+            "先 netx_list_managed_ne 解析 ne_id；若 connect_status 非 pass 可先 netx_get_managed_ne 看 connect_detail。"
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "ne_id": {"type": "string", "description": "纳管网元 UUID"},
+                "commands": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "minItems": 1,
+                    "maxItems": 5,
+                    "description": "只读 CLI 列表，如 show version、display interface brief",
+                },
+                "read_timeout_sec": {
+                    "type": "integer",
+                    "minimum": 10,
+                    "maximum": 120,
+                    "description": "单条命令 Netmiko 读超时（秒），默认 60",
+                },
+            },
+            "required": ["ne_id", "commands"],
+            "additionalProperties": False,
+        },
+        handler=handler,
+        tags=frozenset({"netx", "ops", "managed_ne", "cli", "exec"}),
+        risk_level="medium",
+        read_only=False,
+    )
+
+
 __all__ = [
     # "netx_query_alarms_tool",
     # "netx_aggregate_alarms_tool",
@@ -987,4 +1144,7 @@ __all__ = [
     "netx_aggregate_ume_alarms_raw_tool",
     "netx_list_ume_alarm_fields_tool",
     "netx_sql_query_ume_tool",
+    "netx_list_managed_ne_tool",
+    "netx_get_managed_ne_tool",
+    "netx_exec_managed_ne_tool",
 ]

runtime/workspaces/ops/ROLE_SYSTEM.en.md

@@ -31,8 +31,9 @@ You are the ops specialist (network operations expert).
 - If `host_name` is empty, fall back to `user_label` / `ne_name` with a "host_name missing" note — never bare `ne_id`.
 - NE stats/aggregates: prefer `group_by=alarm_host_name` or `group_by=ne_host_name`; do not group by `alarm_ne_id` / `ne_ne_id` for user output.
 
-## Required skill
+## Required skills
 - For every netx/UME **alarm or NE** request, load and follow skill: `ops-netx-ume-playbook` (skill text may be Chinese; **user-facing output must still match the user's language**).
+- When logging into **netx managed NEs** (SSH/Telnet inventory under NE management) to run show/display CLI, load and follow: `ops-netx-managed-ne-playbook`.
 
 ## netx detail and statistics (internal tools)
 
@@ -44,4 +45,9 @@ Each turn may append a **UME alarm runtime anchor** at the end of system context
 - `netx_query_ume_ne_inventory`: synced NE list (`keyword`).
 - `netx_get_ume_ne`: single NE by `ne_id` (includes `raw_json`).
 
+## netx managed NE (device CLI)
+
+- `netx_list_managed_ne` / `netx_get_managed_ne`: managed inventory and connect-test detail.
+- `netx_exec_managed_ne`: read-only CLI via netx login (show/display/ping; no config changes).
+
 Uses `OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`. Disable anchor inject: `OCLAW_OPS_NETX_CONTEXT_INJECT=0`.

runtime/workspaces/ops/ROLE_SYSTEM.md

@@ -26,6 +26,7 @@
 
 ## 必须加载技能
 - 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。
+- 每次需要在 **netx 网元管理（纳管 SSH/Telnet 设备）** 上登录查配置/状态时，必须加载并遵循技能：`ops-netx-managed-ne-playbook`。
 
 ## netx 明细与统计（内部工具）
 
@@ -37,4 +38,9 @@
 - `netx_query_ume_ne_inventory`：分页查询已同步的 UME 网元清单（可选 `keyword`）。
 - `netx_get_ume_ne`：按 `ne_id`（UUID）取单网元详情（含 `raw_json`）。
 
+## netx 纳管网元（登录设备查 CLI）
+
+- `netx_list_managed_ne` / `netx_get_managed_ne`：网元管理清单与连通详情。
+- `netx_exec_managed_ne`：经 netx 登录设备执行只读 CLI（show/display/ping；禁止改配置）。
+
 工具走 netx（`OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`）。关闭自动锚点：环境变量 `OCLAW_OPS_NETX_CONTEXT_INJECT=0`。

skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md

@@ -0,0 +1,41 @@
+---
+name: ops-netx-managed-ne-playbook
+description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手册。覆盖设备清单、连通状态、经 netx 登录设备执行只读 CLI。
+---
+
+# Ops Netx 纳管网元作业手册
+
+## 强制使用范围
+
+凡是需要在 **netx 网元管理** 中已录入的设备上**登录并查询**（show/display/ping 等）时，必须加载并遵循本技能。
+
+与 UME 网元清单（`ops-netx-ume-playbook`）不同：本手册针对 **SSH/Telnet 纳管设备**（含 ZTE/华为/思科跳板），不是 UME REST 同步清单。
+
+## 工具选择顺序
+
+1. **定位设备**
+   - `netx_list_managed_ne`：`keyword`（名称/IP）、`connect_status=pass` 过滤
+   - `netx_get_managed_ne`：单条详情、`connect_detail`（连通失败排障）
+2. **登录查信息**
+   - `netx_exec_managed_ne`：`ne_id` + `commands`（最多 5 条只读命令）
+
+## CLI 约束（服务端强制）
+
+- 允许前缀：`show `、`display `、`get `、`ping `、`traceroute `、`terminal length `、`?`
+- 禁止：`|`、`;`、换行拼接、改配置类（configure/write/copy/reload/delete 等）
+- 示例：
+  - 思科：`show version`、`show configuration | include hostname` **不可**（含 `|`）→ 改用 `show configuration` 或连通测试已解析的 hostname
+  - 华为：`display version`、`display interface brief`
+  - ZTE：`show version`、`show interface`
+
+## 排障流程
+
+1. `connect_status` 为 `fail`：先 `netx_get_managed_ne` 阅读 `connect_detail`，勿反复盲 exec
+2. 设备经跳板：详情中确认 `hop_enabled`、`hop_vendor`、模板是否正确
+3. 超时：对慢命令提高 `read_timeout_sec`（最大 120），或减少单次命令条数
+
+## 输出约定
+
+- 结论 + **工具返回 output 摘录**（勿编造 CLI 结果）
+- 标明 `ip_address`、`name`、`ne_id`（对用户展示优先 name/IP，ne_id 作关联键）
+- 英文会话：用户可见回复不得含汉字（CLI 原文可摘录但需说明为设备原文）

tests/test_netx_managed_ne_tools.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+
+def test_netx_list_managed_ne_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    import 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, "items": []}}
+
+    monkeypatch.setattr(nt, "_http_json", fake)
+    spec = nt.netx_list_managed_ne_tool()
+    out = spec.handler({"keyword": "192.168", "connect_status": "pass", "page": 1, "page_size": 20})
+    assert out.get("ok") is True
+    assert calls[0] == ("GET", "/v1/managed-ne", {"page": 1, "page_size": 20, "keyword": "192.168", "connect_status": "pass"})
+
+
+def test_netx_exec_managed_ne_posts_body(monkeypatch: pytest.MonkeyPatch) -> None:
+    import runtime.tools.experts.network_ops.netx_tools as nt
+
+    bodies: list[dict[str, Any]] = []
+
+    def fake_post(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]:
+        bodies.append(body)
+        return {"ok": True, "data": {"ok": True, "output": "R2#show version\n..."}}
+
+    monkeypatch.setattr(nt, "_http_post_json", fake_post)
+    spec = nt.netx_exec_managed_ne_tool()
+    out = spec.handler({"ne_id": "abc", "commands": ["show version"], "read_timeout_sec": 90})
+    assert out.get("ok") is True
+    assert bodies[0]["ne_id"] == "abc"
+    assert bodies[0]["commands"] == ["show version"]
+    assert bodies[0]["read_timeout_sec"] == 90
+
+
+def test_netx_exec_requires_commands(monkeypatch: pytest.MonkeyPatch) -> None:
+    import runtime.tools.experts.network_ops.netx_tools as nt
+
+    monkeypatch.setattr(nt, "_http_post_json", lambda *a, **k: {"ok": True, "data": {}})
+    spec = nt.netx_exec_managed_ne_tool()
+    out = spec.handler({"ne_id": "abc"})
+    assert out.get("ok") is False
+    assert out.get("error_code") == "commands_required"