From a0a3f4b65f326b90c9707622166d8f3cfff3e8cb Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 30 May 2026 15:44:34 +0800 Subject: [PATCH] feat(mcp): integrate netx via netx_mcp and improve MCP install - Gate builtin netx tools; ops docs/skills use mcp__netx__* - Import Cursor mcpServers JSON; pass env_schema defaults to subprocess - Allow source_type=local; UTF-8 MCP stdio on Windows Co-authored-by: Cursor --- README.md | 2 +- _local/system.env.example | 14 +- docs/NETX_MCP_INTEGRATION.md | 149 ++++--- interfaces/admin/routes.py | 4 +- interfaces/admin/static/app.js | 27 +- runtime/tools/expert_registry.py | 21 +- .../tools/experts/network_ops/netx_tools.py | 387 ++---------------- runtime/tools/mcp/adapter.py | 37 +- runtime/tools/mcp/installer.py | 10 + runtime/tools/mcp/runtime.py | 28 +- runtime/workspaces/ops/ROLE_SYSTEM.en.md | 19 +- runtime/workspaces/ops/ROLE_SYSTEM.md | 23 +- .../ops/ops-netx-managed-ne-playbook/SKILL.md | 8 +- .../ops/ops-netx-ume-playbook/SKILL.md | 132 ++---- tests/test_mcp_env_defaults.py | 41 ++ tests/test_mcp_installer.py | 7 + tests/test_netx_expert_registry.py | 35 ++ 17 files changed, 408 insertions(+), 536 deletions(-) create mode 100644 tests/test_mcp_env_defaults.py create mode 100644 tests/test_netx_expert_registry.py diff --git a/README.md b/README.md index eb0d32f3..dbd0ecf9 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ powershell -ExecutionPolicy Bypass -File .\scripts\start_all.ps1 -Background ### 可选：运维专家（network_ops）与 netx -若要在 ops 专家模式下调 **netx** 告警库，需单独启动 netx 服务，并在 `_local/system.env` 中配置 `OCLAW_NETX_BASE_URL`（及可选的 `OCLAW_NETX_API_TOKEN`）。说明见 `docs/NETX_MCP_INTEGRATION.md`。 +若要在 ops 专家模式下调 **netx** 告警库，需单独启动 netx 服务，并在 Admin 安装 **netx MCP**（`server_id=netx`，env `NETX_API_URL`）。说明见 `docs/NETX_MCP_INTEGRATION.md`。迁移期可选 `OCLAW_NETX_BUILTIN_TOOLS=1` 启用旧 inline 工具。 ### 可选：股票分析专家（A股/港股，信号建议） diff --git a/_local/system.env.example b/_local/system.env.example index 93f488ef..79e8c0a6 100644 --- a/_local/system.env.example +++ b/_local/system.env.example @@ -433,16 +433,18 @@ TAVILY_API_KEY= # 【前端】多数上述项在 Admin 设置表单中有对应勾选或输入框。 # ----------------------------------------------------------------------------- -# 二十二、netx（network_ops 内部工具：UME 告警 / 纳管网元 CLI / 诊断） +# 二十二、netx（ops：UME 告警 / 纳管网元 CLI / 诊断） # ----------------------------------------------------------------------------- -# OCLAW_NETX_BASE_URL netx HTTP 根地址（无尾部斜杠）；runtime/tools/experts/network_ops/netx_tools.py +# 推荐路径：Admin 安装 netx MCP（server_id=netx），MCP env 使用 NETX_API_URL / NETX_API_TOKEN。 +# 见 docs/NETX_MCP_INTEGRATION.md。 +# OCLAW_NETX_BUILTIN_TOOLS 默认 0；设为 1 时额外注册旧版 inline netx_* 工具（迁移期双轨，勿与 MCP 同时暴露给模型）。 +# OCLAW_NETX_BASE_URL 仍用于 runtime 锚点 HTTP 探测；与 MCP 的 NETX_API_URL 通常相同。 # 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（类似附件锚点）。 +# OCLAW_NETX_API_TOKEN 可选；netx 若启用 Bearer 鉴权，填与 netx 侧一致的 token（MCP env 亦需一致）。 +# OCLAW_OPS_NETX_CONTEXT_INJECT 默认 1；ops 专家每轮把 netx 最新 UME 同步锚点注入 system。 # 设为 0/false/off 可关闭（压测或 netx 不可达时减少噪音）。 # 【前端】无；仅进程环境变量。 +OCLAW_NETX_BUILTIN_TOOLS=0 OCLAW_NETX_BASE_URL=http://127.0.0.1:8890 OCLAW_NETX_API_TOKEN= OCLAW_OPS_NETX_CONTEXT_INJECT=1 diff --git a/docs/NETX_MCP_INTEGRATION.md b/docs/NETX_MCP_INTEGRATION.md index 6d3cd0cc..aa6c92e7 100644 --- a/docs/NETX_MCP_INTEGRATION.md +++ b/docs/NETX_MCP_INTEGRATION.md @@ -1,10 +1,10 @@ -# netx MCP Integration (same-host) +# netx MCP Integration -This guide wires `oclaw` to the independent ops tool in: +Wire **oclaw** (or any MCP host) to the standard **netx HTTP MCP** in `D:/project/chatgpt/netx`. -- `D:/project/chatgpt/netx` +netx 侧通用安装/更新说明：`D:/project/chatgpt/netx/docs/MCP.md`。 -Assumption: `oclaw` and `netx` run on the same host. +Default path: **stdio MCP → netx REST API** (`NETX_API_URL`). Legacy inline HTTP tools in oclaw are opt-in via `OCLAW_NETX_BUILTIN_TOOLS=1`. ## 1) Start netx API @@ -24,82 +24,127 @@ Health check: curl http://127.0.0.1:8890/health ``` -## 2) Register netx MCP in oclaw Admin +## 2) Install netx MCP in oclaw Admin -Use MCP install payload from: +**推荐：直接粘贴与 Cursor 相同的 `mcpServers` JSON**（与 `netx/mcp.json` 一致），在 Admin → MCP → 安装 JSON 粘贴后安装，无需再手写 `entry_command` / `entry_args`： + +```json +{ + "mcpServers": { + "netx": { + "command": "python", + "args": ["-m", "netx_mcp"], + "env": { + "NETX_API_URL": "http://127.0.0.1:8890", + "NETX_API_TOKEN": "", + "NETX_LANG": "zh" + } + } + } +} +``` + +oclaw 会把 `command` → `entry_command`、`args` → `entry_args`，`env` → 注册表 `env_schema`（含 default，运行时传给 MCP 子进程）；stdio 条目默认 `source_type=local`。与下文「install payload」等价。 + +本机默认 `http://127.0.0.1:8890` 时，`env` 可省略（`netx_mcp` 代码内也有相同默认）；远端 API 或 Token 时在 JSON 的 `env` 里写即可，不必再单独维护 `mcp_install_payload.json`。 + +也可用 oclaw 专用 install payload（字段展开版，便于脚本/文档引用）： - `D:/project/chatgpt/netx/mcp_install_payload.json` -Equivalent manual values: +Equivalent manual values（与上面 `mcpServers` 同义）： -- `source_type`: `local` -- `source_ref`: `netx-local-mcp` -- `server_id`: `netx-local` -- `entry_command`: `python` -- `entry_args`: `["D:/project/chatgpt/netx/netx_api/mcp_server.py"]` -- `timeout_s`: `30` +| Field | Value | +|-------|-------| +| `source_type` | `local` | +| `source_ref` | `netx-mcp-http` | +| `server_id` | **`netx`** | +| `entry_command` | `python` | +| `entry_args` | `["-m", "netx_mcp"]` | +| `timeout_s` | `120` | +| MCP env | `NETX_API_URL`（**可指向远端**）、可选 `NETX_API_TOKEN`、`NETX_LANG` | -Then run: +**两件事情要分开：** -1. `Health` -2. `Sync Tools` +| 组件 | 跑在哪 | 配置 | +|------|--------|------| +| **netx REST API**（告警/网元数据） | 本机或远端服务器 | `NETX_API_URL`，例如 `http://10.0.0.5:8890` | +| **netx MCP 子进程**（stdio，给 oclaw 调工具） | **必须与 oclaw 同机**（或 oclaw 能 `python` 到的环境） | `pip install -e /packages/netx-mcp` 后 `python -m netx_mcp` | -Expected tools: +远端只部署 **netx 服务** 时：把 `NETX_API_URL` 改成远端地址即可；**不需要** `NETX_REPO_ROOT`。 -- `queryAlarms` -- `aggregateAlarms` -- `getImportBatch` -- `runDiagnostics` +本机开发若未 `pip install`，可临时用脚本路径（二选一）： + +```json +"entry_args": ["D:/project/chatgpt/netx/netx_api/mcp_server.py"] +``` + +或在 oclaw 机执行一次（推荐，与远端 API 无关）： + +```powershell +pip install -e D:/project/chatgpt/netx/packages/netx-mcp +``` + +**注意**：`source_type=local` 表示跳过 npm/pypi 的「安装包」步骤，但 oclaw 机上仍须能 `import netx_mcp`（通过上面的 pip 安装）。 + +Then run **Health** → **Sync Tools**. + +### Expected MCP tools (12) + +| MCP tool | oclaw namespaced | Legacy builtin (if enabled) | +|----------|------------------|----------------------------| +| `queryUmeAlarms` | `mcp__netx__queryUmeAlarms` | `netx_query_ume_alarms` | +| `aggregateUmeAlarms` | `mcp__netx__aggregateUmeAlarms` | `netx_aggregate_ume_alarms` | +| `runUmeDiagnostics` | `mcp__netx__runUmeDiagnostics` | `netx_run_ume_diagnostics` | +| `queryUmeNeInventory` | `mcp__netx__queryUmeNeInventory` | `netx_query_ume_ne_inventory` | +| `getUmeNe` | `mcp__netx__getUmeNe` | `netx_get_ume_ne` | +| `queryUmeAlarmsRaw` | `mcp__netx__queryUmeAlarmsRaw` | `netx_query_ume_alarms_raw` | +| `aggregateUmeAlarmsRaw` | `mcp__netx__aggregateUmeAlarmsRaw` | `netx_aggregate_ume_alarms_raw` | +| `listUmeAlarmFields` | `mcp__netx__listUmeAlarmFields` | `netx_list_ume_alarm_fields` | +| `sqlQueryUme` | `mcp__netx__sqlQueryUme` | `netx_sql_query_ume` | +| `listManagedNe` | `mcp__netx__listManagedNe` | `netx_list_managed_ne` | +| `getManagedNe` | `mcp__netx__getManagedNe` | `netx_get_managed_ne` | +| `execManagedNe` | `mcp__netx__execManagedNe` | `netx_exec_managed_ne` | + +**不暴露**（已废弃 Excel 导入批次链路）：`netx_query_alarms`、`netx_list_import_batches`、`netx_sql_query`（带 `batch_id`）等。 ## 3) Bind to ops specialist -In MCP specialist binding, include `netx-local` for your ops specialist/workspace. +In Admin **MCP specialist binding**, include server **`netx`** for the ops workspace/specialist. -## 4) Use from chat +## 4) Dual-track: builtin vs MCP -After binding, model can call namespaced tools like: +| Setting | Effect | +|---------|--------| +| `OCLAW_NETX_BUILTIN_TOOLS=0` (default) | Only MCP tools (`mcp__netx__*`); no duplicate inline `netx_*` in catalog | +| `OCLAW_NETX_BUILTIN_TOOLS=1` | Registers legacy inline HTTP tools **and** MCP if installed — avoid binding both unless testing migration | -- `mcp__netx-local__queryAlarms` -- `mcp__netx-local__aggregateAlarms` -- `mcp__netx-local__getImportBatch` -- `mcp__netx-local__runDiagnostics` +Runtime anchor inject (`OCLAW_OPS_NETX_CONTEXT_INJECT=1`) still works without builtin tools; it only needs netx API reachable at `NETX_API_URL` / `OCLAW_NETX_BASE_URL`. -## 5) External link in Admin +## 5) Cursor / Claude Desktop -`oclaw` admin sidebar includes an external link: +与 §2 相同：直接复制 `D:/project/chatgpt/netx/mcp.json` 到 Cursor 配置即可；oclaw Admin 粘贴同一份 JSON 安装。 -- `Open netx ops tool` -> `http://127.0.0.1:5173/` +## 6) External link in Admin -If your netx host/port differs, update the link in: +Admin sidebar **Open netx ops tool** → `http://127.0.0.1:5173/` (edit in `interfaces/admin/static/index.html` if host/port differs). -- `interfaces/admin/static/index.html` - -## 6) netx -> oclaw AP analyze auth +## 7) netx → oclaw AP analyze auth `netx` can call: - `POST /admin/api/ops-ai/analyze-sync` - `GET /admin/api/ops-ai/health` -Recommended auth: +Shared token: -1. Set shared token in `oclaw` runtime env: - - `OCLAW_OPS_AI_SHARED_TOKEN=` -2. Set same token in `netx`: - - `NETX_OCLAW_ANALYZE_TOKEN=` +1. oclaw: `OCLAW_OPS_AI_SHARED_TOKEN=` +2. netx: `NETX_OCLAW_ANALYZE_TOKEN=` -Then `netx /v1/ap/analyze` can invoke `oclaw` synchronously. +Timeouts: set `NETX_OCLAW_ANALYZE_READ_TIMEOUT_SEC` (default `180`) in netx if analyze-sync is slow. -**Timeouts:** `netx` → `oclaw` uses HTTP; `analyze-sync` often exceeds old ~35s limits. In netx set `NETX_OCLAW_ANALYZE_READ_TIMEOUT_SEC` (default `180` in `netx_api/config.py`) if you still see read timeouts on slow models or multi-tool turns. +Integration status: `GET http://127.0.0.1:8890/v1/integrations/status` -Integration health in netx: +## 8) Observe AP calls in oclaw -- `GET http://127.0.0.1:8890/v1/integrations/status` - -## 7) Observe AP calls in oclaw - -Recent ops-ai analyze calls can be fetched from: - -- `GET /admin/api/ops-ai/logs?limit=50&offset=0` - -Permission: `admin:user:write` (same as admin audit access). +- `GET /admin/api/ops-ai/logs?limit=50&offset=0` (requires `admin:user:write`) diff --git a/interfaces/admin/routes.py b/interfaces/admin/routes.py index 5e780129..df678fb5 100644 --- a/interfaces/admin/routes.py +++ b/interfaces/admin/routes.py @@ -2820,7 +2820,7 @@ def build_admin_router() -> APIRouter: _require_permission(ctx, "admin:tenant:write") source_type = str(payload.get("source_type") or "").strip().lower() source_ref = str(payload.get("source_ref") or "").strip() - if source_type not in {"github", "npm", "pypi"} or not source_ref: + if source_type not in {"github", "npm", "pypi", "local"} or not source_ref: return {"ok": False, "error": "invalid_source"} server_id = _safe_server_id(str(payload.get("server_id") or source_ref)) manifest = McpServerManifest( @@ -3169,7 +3169,7 @@ def build_admin_router() -> APIRouter: _require_permission(ctx, "admin:tenant:write") source_type = str(payload.get("source_type") or "").strip().lower() source_ref = str(payload.get("source_ref") or "").strip() - if source_type not in {"github", "npm", "pypi"} or not source_ref: + if source_type not in {"github", "npm", "pypi", "local"} or not source_ref: return {"ok": False, "error_code": "mcp_invalid_source", "error": "invalid_source"} manifest = McpServerManifest( server_id=_safe_server_id(str(payload.get("server_id") or source_ref)), diff --git a/interfaces/admin/static/app.js b/interfaces/admin/static/app.js index 838ec6f3..eabef9aa 100644 --- a/interfaces/admin/static/app.js +++ b/interfaces/admin/static/app.js @@ -5091,7 +5091,7 @@ async function renderPlugins() { } else if (parsed && typeof parsed === "object" && parsed.mcpServers && typeof parsed.mcpServers === "object") { detectedMcpServersShape = true; installStatus.textContent = - "[json] detected mcpServers shape; streamableHttp/sse entries will be installed via mcp-remote bridge"; + "[json] detected mcpServers shape; stdio uses command/args (local), HTTP/SSE uses mcp-remote bridge"; const envVarRe = /\$\{([A-Za-z_][A-Za-z0-9_]*)\}/g; list = Object.entries(parsed.mcpServers).map(([rawKey, rawServer]) => { const key = String(rawKey || "").trim(); @@ -5149,16 +5149,33 @@ async function renderPlugins() { }; } - // Best-effort passthrough for stdio-like definitions. + // Best-effort passthrough for stdio-like definitions (Cursor command/args/env). const cmd = String(s.command || s.entry_command || "").trim(); const cmdArgs = Array.isArray(s.args) ? s.args.map((x) => String(x)) : []; + const stdioLocal = !!cmd; + const cursorEnv = s.env && typeof s.env === "object" ? s.env : {}; + const envFromCursor = {}; + Object.entries(cursorEnv).forEach(([ek, ev]) => { + const envName = String(ek || "").trim(); + if (!envName) return; + envFromCursor[envName] = { + type: "string", + default: String(ev == null ? "" : ev), + description: "From mcpServers env; copy to mcp_local.env if runtime needs it", + }; + }); + const mergedEnvSchema = { ...envFromCursor, ...envSchema }; + if (s.env_schema && typeof s.env_schema === "object") { + Object.assign(mergedEnvSchema, s.env_schema); + } + const explicitSourceType = String(s.source_type || "").trim(); return { - source_type: String(s.source_type || "npm").trim() || "npm", - source_ref: String(s.source_ref || "mcp-remote").trim() || "mcp-remote", + source_type: explicitSourceType || (stdioLocal ? "local" : "npm"), + source_ref: String(s.source_ref || (stdioLocal ? key : "mcp-remote")).trim() || (stdioLocal ? key : "mcp-remote"), server_id: key || String(s.name || s.server_id || "mcp-server").trim() || "mcp-server", entry_command: cmd, entry_args: cmdArgs, - env_schema: Object.keys(envSchema).length ? envSchema : (s.env_schema && typeof s.env_schema === "object" ? s.env_schema : {}), + env_schema: mergedEnvSchema, required_permissions: Array.isArray(s.required_permissions) ? s.required_permissions.map((x) => String(x)) : [], risk_level: String(s.risk_level || "high"), enabled: Object.prototype.hasOwnProperty.call(s, "isActive") ? !!s.isActive : true, diff --git a/runtime/tools/expert_registry.py b/runtime/tools/expert_registry.py index 30ccf13c..7402a7ca 100644 --- a/runtime/tools/expert_registry.py +++ b/runtime/tools/expert_registry.py @@ -3,6 +3,7 @@ from __future__ import annotations import importlib.util import inspect import logging +import os from collections.abc import Callable from pathlib import Path from typing import Any @@ -21,6 +22,22 @@ _DEPRECATED_TOOL_NAMES: set[str] = { } +def netx_builtin_tools_enabled() -> bool: + """When false, skip ``network_ops/netx_tools.py`` factories (use MCP ``mcp__netx__*`` instead).""" + return str(os.getenv("OCLAW_NETX_BUILTIN_TOOLS") or "0").strip().lower() not in { + "0", + "false", + "no", + "off", + } + + +def _skip_expert_module(module_path: Path) -> bool: + if module_path.stem == "netx_tools" and not netx_builtin_tools_enabled(): + return True + return False + + def _load_module_from_path(module_path: Path, module_name: str) -> Any | None: try: spec = importlib.util.spec_from_file_location(module_name, module_path) @@ -52,12 +69,14 @@ def discover_expert_tool_factories() -> dict[str, list[ToolFactory]]: for module_path in sorted(expert_dir.glob("*.py")): if module_path.name == "__init__.py": continue + if _skip_expert_module(module_path): + continue mod_name = f"runtime.tools.experts.{expert}.{module_path.stem}" module = _load_module_from_path(module_path, mod_name) if module is None: continue exported = getattr(module, "__all__", None) - if isinstance(exported, list) and exported: + if isinstance(exported, list): for name in sorted(exported): value = getattr(module, name, None) if callable(value) and name.endswith("_tool"): diff --git a/runtime/tools/experts/network_ops/netx_tools.py b/runtime/tools/experts/network_ops/netx_tools.py index a0df3873..0223101a 100644 --- a/runtime/tools/experts/network_ops/netx_tools.py +++ b/runtime/tools/experts/network_ops/netx_tools.py @@ -1,7 +1,10 @@ -"""Internal read-only tools: query netx REST API (PostgreSQL backend on netx side). +"""netx ops helpers: runtime context inject + optional legacy inline HTTP tools. + +Default: tools are exposed via stdio MCP (``pip install -e packages/netx-mcp`` then ``python -m netx_mcp``). +Set ``OCLAW_NETX_BUILTIN_TOOLS=1`` to re-register inline ``netx_*`` expert tools. Configure via environment: -- ``OCLAW_NETX_BASE_URL`` (default ``http://127.0.0.1:8890``) +- ``OCLAW_NETX_BASE_URL`` (default ``http://127.0.0.1:8890``) — runtime anchor HTTP probe - ``OCLAW_NETX_API_TOKEN`` (optional) → sent as ``Authorization: Bearer …`` if set. """ @@ -27,6 +30,7 @@ _PROTOCOL_KEY_ZH_TO_EN: dict[str, str] = { "电源": "Power", } + _UME_RAW_GROUP_FIELDS = [ "alarm_alarm_key", "alarm_host_name", @@ -136,24 +140,6 @@ def _http_json(method: str, path: str, *, params: dict[str, Any] | None = None) return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} -def _resolve_latest_import_batch_id() -> dict[str, Any]: - """``GET /v1/batches`` 按 created_at 降序；取第一条为当前最新导入批次。""" - r = _http_json("GET", "/v1/batches", params={"limit": 1}) - if not r.get("ok"): - return {"ok": False, "error": "netx_list_batches_failed", "detail": r.get("detail"), "upstream": r} - data = r.get("data") or {} - items = data.get("items") - if not isinstance(items, list) or not items: - return {"ok": False, "error": "no_import_batches", "detail": "netx 中尚无导入批次，请先导入或显式提供 batch_id"} - first = items[0] - if not isinstance(first, dict): - return {"ok": False, "error": "no_import_batches", "detail": "batch 列表格式异常"} - bid = str(first.get("batch_id") or "").strip() - if not bid: - return {"ok": False, "error": "no_import_batches", "detail": "batch 列表中无 batch_id"} - return {"ok": True, "batch_id": bid, "batch_row": first} - - def _resolve_ume_anchor() -> dict[str, Any]: """Resolve current UME alarm anchor from netx sync status.""" r = _http_json("GET", "/v1/ume/sync/status", params={"page": 1, "page_size": 20}) @@ -219,48 +205,42 @@ 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, " - "netx_query_ume_ne_inventory, netx_get_ume_ne\n" + "- MCP tools (server_id=netx): mcp__netx__queryUmeAlarms, mcp__netx__aggregateUmeAlarms, " + "mcp__netx__runUmeDiagnostics, mcp__netx__queryUmeNeInventory, mcp__netx__getUmeNe, " + "mcp__netx__queryUmeAlarmsRaw, mcp__netx__aggregateUmeAlarmsRaw, mcp__netx__listUmeAlarmFields, " + "mcp__netx__sqlQueryUme, mcp__netx__listManagedNe, mcp__netx__getManagedNe, mcp__netx__execManagedNe\n" "- note: this is only runtime anchor; use tools for alarm/ne evidence.\n" "- English session: user-visible reply must contain NO Chinese/CJK; translate alarm text fields." ) tail_zh = ( - "- 工具: netx_query_ume_alarms、netx_aggregate_ume_alarms、netx_run_ume_diagnostics、" - "netx_query_ume_ne_inventory、netx_get_ume_ne\n" + "- MCP 工具（server_id=netx）：mcp__netx__queryUmeAlarms、mcp__netx__aggregateUmeAlarms、" + "mcp__netx__runUmeDiagnostics、mcp__netx__queryUmeNeInventory、mcp__netx__getUmeNe、" + "mcp__netx__queryUmeAlarmsRaw、mcp__netx__aggregateUmeAlarmsRaw、mcp__netx__listUmeAlarmFields、" + "mcp__netx__sqlQueryUme、mcp__netx__listManagedNe、mcp__netx__getManagedNe、mcp__netx__execManagedNe\n" "- 说明: 此处仅为运行锚点；具体告警/网元信息必须以工具返回为准，勿臆测。" ) return "\n".join(lines_en + [tail_en]) if lang_en else "\n".join(lines_zh + [tail_zh]) err = str(r.get("error") or "") detail = str(r.get("detail") or "")[:240] - if err == "no_import_batches": - if lang_en: - return ( - "[Netx alarm import anchor]\n" - "- batch_id: (none)\n" - "- note: no import batches in netx yet—import alarms or pass batch_id in chat." - ) - return ( - "[当前 netx 告警导入锚点]\n" - "- batch_id: （暂无）\n" - "- 说明: netx 中尚无导入批次；请先导入告警或在对话中提供 batch_id。" - ) if lang_en: return ( "[Netx UME current-alarms anchor]\n" f"- error: {err}\n" f"- detail: {detail}\n" - "- fix: check OCLAW_NETX_BASE_URL and that netx API is reachable." + "- fix: check NETX_API_URL (or OCLAW_NETX_BASE_URL) and that netx API is reachable; " + "ensure MCP server_id=netx is bound and synced." ) return ( "[当前 netx UME告警锚点]\n" f"- 错误: {err}\n" f"- 详情: {detail}\n" - "- 处理: 检查 OCLAW_NETX_BASE_URL 与 netx 服务是否可达。" + "- 处理: 检查 NETX_API_URL（或 OCLAW_NETX_BASE_URL）与 netx 服务是否可达；" + "确认 Admin 已安装并绑定 MCP server_id=netx。" ) def ops_netx_system_context_extension(*, lang: str = "zh") -> str: - """Append to ops specialist system prompt: latest batch_id anchor (direct_loop injection). + """Append to ops specialist system prompt: UME sync anchor (direct_loop injection). Cached briefly to reduce duplicate HTTP calls across tool rounds. """ @@ -281,310 +261,6 @@ def ops_netx_system_context_extension(*, lang: str = "zh") -> str: return text -def netx_query_alarms_tool() -> ToolSpec: - """Paginated alarm rows from netx (same filters as netx UI REST).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - explicit = str(args.get("batch_id") or "").strip() - resolution = "explicit" - batch_id = explicit - if not batch_id: - res = _resolve_latest_import_batch_id() - if not res.get("ok"): - return res - batch_id = str(res.get("batch_id") or "") - resolution = "latest_import" - page = max(1, int(args.get("page") or 1)) - page_size = min(200, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = { - "batch_id": batch_id, - "page": page, - "page_size": page_size, - } - if str(args.get("alarm_code") or "").strip(): - params["alarm_code"] = str(args.get("alarm_code")).strip() - if str(args.get("ne_name") or "").strip(): - params["ne_name"] = str(args.get("ne_name")).strip() - if str(args.get("severity") or "").strip(): - params["severity"] = str(args.get("severity")).strip() - out = _http_json("GET", "/v1/alarms", params=params) - if out.get("ok") and resolution == "latest_import": - out = {**out, "batch_id_used": batch_id, "batch_resolution": resolution} - return out - - return ToolSpec( - name="netx_query_alarms", - description=( - "从独立运维工具 netx 读取告警明细（PostgreSQL 侧由 netx 托管）。" - "batch_id 可选：不传表示使用 netx 当前「最新」导入批次（/v1/batches 第一条）。" - "可选 alarm_code / ne_name / severity（与 netx 告警列表过滤语义一致）；支持分页。" - ), - parameters={ - "type": "object", - "properties": { - "batch_id": {"type": "string", "description": "导入批次 ID；省略则用最新导入批次"}, - "alarm_code": {"type": "string", "description": "告警码包含匹配（可选）"}, - "ne_name": {"type": "string", "description": "网元名包含匹配（可选）"}, - "severity": {"type": "string", "description": "规范化级别 critical/major/minor/warning/..."}, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "alarms", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_aggregate_alarms_tool() -> ToolSpec: - """Aggregate buckets from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - explicit = str(args.get("batch_id") or "").strip() - resolution = "explicit" - batch_id = explicit - if not batch_id: - res = _resolve_latest_import_batch_id() - if not res.get("ok"): - return res - batch_id = str(res.get("batch_id") or "") - resolution = "latest_import" - group_by = str(args.get("group_by") or "severity_norm").strip() - if group_by not in {"severity_norm", "alarm_code", "ne_name"}: - return {"ok": False, "error": "invalid_group_by"} - params: dict[str, Any] = {"group_by": group_by, "batch_id": batch_id} - out = _http_json("GET", "/v1/alarms/aggregate", params=params) - if out.get("ok") and resolution == "latest_import": - out = {**out, "batch_id_used": batch_id, "batch_resolution": resolution} - return out - - return ToolSpec( - name="netx_aggregate_alarms", - description=( - "按 severity_norm / alarm_code / ne_name 对 netx 告警做聚合统计（读 netx API，不落直连 PG）。" - "batch_id 可选：不传则限定为 netx 当前最新导入批次（与告警查询默认语义一致）。" - ), - parameters={ - "type": "object", - "properties": { - "batch_id": {"type": "string"}, - "group_by": { - "type": "string", - "enum": ["severity_norm", "alarm_code", "ne_name"], - "default": "severity_norm", - }, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "alarms", "aggregate", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_run_diagnostics_tool() -> ToolSpec: - """Diagnostics summary (same stats netx uses for dashboard slices).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - explicit = str(args.get("batch_id") or "").strip() - resolution = "explicit" - batch_id = explicit - if not batch_id: - res = _resolve_latest_import_batch_id() - if not res.get("ok"): - return res - batch_id = str(res.get("batch_id") or "") - resolution = "latest_import" - out = _http_json("GET", "/v1/diagnostics", params={"batch_id": batch_id}) - if out.get("ok") and resolution == "latest_import": - out = {**out, "batch_id_used": batch_id, "batch_resolution": resolution} - return out - - return ToolSpec( - name="netx_run_diagnostics", - description=( - "读取 netx /v1/diagnostics 统计摘要（批次维度：级别分布、Top 告警码/网元、协议归类等）。" - "batch_id 可选：不传则使用 netx 当前最新导入批次。" - ), - parameters={ - "type": "object", - "properties": {"batch_id": {"type": "string", "description": "批次 ID；省略则用最新导入批次"}}, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "diagnostics", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_list_import_batches_tool() -> ToolSpec: - """List recent import batches (newest first).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - limit = max(1, min(100, int(args.get("limit") or 20))) - return _http_json("GET", "/v1/batches", params={"limit": limit}) - - return ToolSpec( - name="netx_list_import_batches", - description="列出 netx 最近导入批次（与 UI 一致，按创建时间降序）。用于核对 batch_id 或确认「最新」批次。", - parameters={ - "type": "object", - "properties": { - "limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 20}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "batches", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_list_alarm_fields_tool() -> ToolSpec: - """List alarms_norm columns from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - _ = args - return _http_json("GET", "/v1/alarms/fields", params=None) - - return ToolSpec( - name="netx_list_alarm_fields", - description="列出 netx alarms_norm 表所有字段名（供自由查询时选择字段/确认可用字段）。", - parameters={ - "type": "object", - "properties": {}, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "alarms", "schema", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_query_alarms_raw_tool() -> ToolSpec: - """Power query alarms_norm with all fields.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - explicit = str(args.get("batch_id") or "").strip() - batch_id = explicit - if not batch_id: - res = _resolve_latest_import_batch_id() - if not res.get("ok"): - return res - batch_id = str(res.get("batch_id") or "") - page = max(1, int(args.get("page") or 1)) - page_size = min(200, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = { - "batch_id": batch_id, - "page": page, - "page_size": page_size, - } - if str(args.get("alarm_code") or "").strip(): - params["alarm_code"] = str(args.get("alarm_code")).strip() - if str(args.get("ne_name") or "").strip(): - params["ne_name"] = str(args.get("ne_name")).strip() - if str(args.get("severity") or "").strip(): - params["severity"] = str(args.get("severity")).strip() - if str(args.get("q") or "").strip(): - params["q"] = str(args.get("q")).strip() - if str(args.get("order_by") or "").strip(): - params["order_by"] = str(args.get("order_by")).strip() - if str(args.get("order") or "").strip(): - params["order"] = str(args.get("order")).strip() - return _http_json("GET", "/v1/alarms/raw", params=params) - - return ToolSpec( - name="netx_query_alarms_raw", - description=( - "自由查询 netx alarms_norm：返回所有字段。" - "batch_id 可选（省略则使用最新导入批次）；支持 severity/alarm_code/ne_name/q 过滤与分页；" - "order_by 仅允许 id/alarm_time/severity_norm/ne_name/alarm_code。" - ), - parameters={ - "type": "object", - "properties": { - "batch_id": {"type": "string", "description": "批次 ID；省略则用最新导入批次"}, - "severity": {"type": "string"}, - "alarm_code": {"type": "string"}, - "ne_name": {"type": "string"}, - "q": {"type": "string", "description": "自由文本 contains（alarm_code/ne_name/description/service）"}, - "order_by": {"type": "string", "enum": ["id", "alarm_time", "severity_norm", "ne_name", "alarm_code"]}, - "order": {"type": "string", "enum": ["asc", "desc"]}, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "alarms", "power_query", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_sql_query_tool() -> ToolSpec: - """Execute read-only SQL on netx (server enforced SELECT-only).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - batch_id = str(args.get("batch_id") or "").strip() - if not batch_id: - res = _resolve_latest_import_batch_id() - if not res.get("ok"): - return res - batch_id = str(res.get("batch_id") or "") - sql = str(args.get("sql") or "").strip() - limit = max(1, min(2000, int(args.get("limit") or 200))) - if not sql: - return {"ok": False, "error": "sql_required"} - base = _netx_base_url() - url = f"{base}/v1/sql/query" - try: - # Do not inherit system proxy settings for local netx calls. - with httpx.Client(timeout=60.0, trust_env=False) as client: - resp = client.post(url, json={"sql": sql, "batch_id": batch_id, "limit": limit}, 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 {} - 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]} - - return ToolSpec( - name="netx_sql_query", - description=( - "在 netx 上执行只读 SQL（服务端强制 SELECT-only、单语句、必须包含 :batch_id 参数，并强制 limit）。" - "用于自由组合查询；建议先用 netx_list_alarm_fields 确认可用字段。" - ), - parameters={ - "type": "object", - "properties": { - "batch_id": {"type": "string", "description": "批次 ID；省略则用最新导入批次"}, - "sql": {"type": "string", "description": "必须是 SELECT，且包含 :batch_id 占位符"}, - "limit": {"type": "integer", "minimum": 1, "maximum": 2000, "default": 200}, - }, - "required": ["sql"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "sql", "power_query", "read_only"}), - risk_level="low", - read_only=True, - ) - - def netx_query_ume_alarms_tool() -> ToolSpec: """Paginated UME current alarms from netx.""" @@ -612,7 +288,6 @@ def netx_query_ume_alarms_tool() -> ToolSpec: description=( "读取 netx UME 当前告警明细（实时表）；每条含 host_name（网元主展示键，同步时已写入告警表）。" "支持 severity/ne_id/keyword 与分页。" - "当需要字段级控制或复杂分析时，优先 netx_list_ume_alarm_fields + netx_query_ume_alarms_raw/netx_sql_query_ume。" ), parameters={ "type": "object", @@ -915,7 +590,7 @@ def netx_sql_query_ume_tool() -> ToolSpec: "minimum": 0, "maximum": 30000, "default": 0, - "description": "可选查询超时(ms)，0表示使用数据库默认超时", + "description": "可选查询超时(ms)；0 表示使用数据库默认超时", }, }, "required": ["sql"], @@ -1127,14 +802,9 @@ def netx_exec_managed_ne_tool() -> ToolSpec: ) -__all__ = [ - # "netx_query_alarms_tool", - # "netx_aggregate_alarms_tool", - # "netx_run_diagnostics_tool", - # "netx_list_import_batches_tool", - # "netx_list_alarm_fields_tool", - # "netx_query_alarms_raw_tool", - # "netx_sql_query_tool", +__all__: list[str] = [] + +_LEGACY_EXPORTS = [ "netx_query_ume_alarms_tool", "netx_aggregate_ume_alarms_tool", "netx_run_ume_diagnostics_tool", @@ -1148,3 +818,14 @@ __all__ = [ "netx_get_managed_ne_tool", "netx_exec_managed_ne_tool", ] + + +def _apply_legacy_exports() -> None: + import os as _os + + if str(_os.getenv("OCLAW_NETX_BUILTIN_TOOLS") or "0").strip().lower() in {"0", "false", "no", "off"}: + return + globals()["__all__"] = list(_LEGACY_EXPORTS) + + +_apply_legacy_exports() diff --git a/runtime/tools/mcp/adapter.py b/runtime/tools/mcp/adapter.py index 9e2d353f..7510cb3a 100644 --- a/runtime/tools/mcp/adapter.py +++ b/runtime/tools/mcp/adapter.py @@ -5,7 +5,6 @@ import json import os from typing import Any -from runtime.operations.mcp_env import mcp_env_allowlist_keys from runtime.skills import SkillSpec, materialize_skills_from_tool_specs from runtime.tools.base import ToolSpec from runtime.tools.mcp.filesystem_argv import build_mcp_process_command @@ -13,6 +12,31 @@ from runtime.tools.mcp.runtime import McpProcessRuntime from runtime.tools.public.bailian_webparser_tool import bailian_webparser_tool +def _mcp_row_env_config(row: dict[str, Any]) -> tuple[list[str], dict[str, str]]: + """Per-server env allowlist + defaults from registry ``env_schema`` (e.g. Cursor ``mcpServers.env``).""" + from runtime.operations.mcp_env import mcp_env_allowlist_keys + + schema = row.get("env_schema") if isinstance(row.get("env_schema"), dict) else {} + defaults: dict[str, str] = {} + schema_keys: list[str] = [] + for k, spec in schema.items(): + key = str(k or "").strip() + if not key: + continue + schema_keys.append(key) + if isinstance(spec, dict) and spec.get("default") is not None: + dv = str(spec.get("default") or "").strip() + if dv: + defaults[key] = dv + seen: set[str] = set() + allowlist: list[str] = [] + for k in [*mcp_env_allowlist_keys(), *schema_keys]: + if k and k not in seen: + seen.add(k) + allowlist.append(k) + return allowlist, defaults + + @dataclass class _McpBoundTool: server_id: str @@ -23,9 +47,15 @@ class _McpBoundTool: timeout_s: float = 30.0 required_permissions: frozenset[str] = frozenset() env_allowlist: list[str] | None = None + env_defaults: dict[str, str] | None = None def to_spec(self) -> ToolSpec: - rt = McpProcessRuntime(command=self.command, timeout_s=self.timeout_s, env_allowlist=self.env_allowlist) + rt = McpProcessRuntime( + command=self.command, + timeout_s=self.timeout_s, + env_allowlist=self.env_allowlist, + env_defaults=self.env_defaults, + ) def _handler(args: dict[str, Any]) -> dict[str, Any]: res = rt.call_tool(tool_name=self.tool_name, arguments=args or {}) @@ -112,7 +142,6 @@ def materialize_mcp_tools_for_specialist( return [] out: list[ToolSpec] = [] rows = store.list_mcp_servers(enabled_only=True) if store else [] - env_allowlist = mcp_env_allowlist_keys() for row in rows: server_id = str(row.get("server_id") or "").strip() cmd = str(row.get("entry_command") or "").strip() @@ -120,6 +149,7 @@ def materialize_mcp_tools_for_specialist( continue if binding_server_ids is not None and sp and server_id not in binding_server_ids: continue + env_allowlist, env_defaults = _mcp_row_env_config(row) raw_args = [x for x in (row.get("entry_args") or []) if isinstance(x, str)] command = build_mcp_process_command( cmd, @@ -161,6 +191,7 @@ def materialize_mcp_tools_for_specialist( timeout_s=float(row.get("timeout_s") or 30.0), required_permissions=frozenset(str(x) for x in (row.get("required_permissions") or [])), env_allowlist=env_allowlist, + env_defaults=env_defaults, ).to_spec() out.append(spec) return out diff --git a/runtime/tools/mcp/installer.py b/runtime/tools/mcp/installer.py index 5e54eae3..685ba816 100644 --- a/runtime/tools/mcp/installer.py +++ b/runtime/tools/mcp/installer.py @@ -61,6 +61,8 @@ def _install_command(manifest: McpServerManifest) -> list[str]: if manifest.source_type == "pypi": pkg = manifest.source_ref + (f"=={manifest.version}" if manifest.version else "") return [sys.executable, "-m", "pip", "install", pkg] + if manifest.source_type == "local": + return [] raise ValueError(f"unsupported_source_type:{manifest.source_type}") @@ -85,6 +87,8 @@ def _uninstall_command(manifest: McpServerManifest) -> list[str]: return [sys.executable, "-m", "pip", "uninstall", "-y", str(manifest.source_ref or "").strip()] if manifest.source_type == "github": return [] + if manifest.source_type == "local": + return [] raise ValueError(f"unsupported_source_type:{manifest.source_type}") @@ -93,6 +97,12 @@ def install_mcp_server(manifest: McpServerManifest, *, dry_run: bool = False) -> cmd = _install_command(manifest) except Exception as exc: return McpInstallResult(ok=False, error_code="mcp_invalid_source", error=str(exc)) + if manifest.source_type == "local" and not cmd: + return McpInstallResult( + ok=True, + install_command="", + details={"skipped": True, "reason": "local_source_no_package_install"}, + ) cmd_text = " ".join(cmd) if dry_run: return McpInstallResult(ok=True, install_command=cmd_text, details={"dry_run": True}) diff --git a/runtime/tools/mcp/runtime.py b/runtime/tools/mcp/runtime.py index 0b5a97d3..a9cad06f 100644 --- a/runtime/tools/mcp/runtime.py +++ b/runtime/tools/mcp/runtime.py @@ -16,13 +16,17 @@ class McpProcessRuntime: command: list[str] timeout_s: float = 30.0 env_allowlist: list[str] | None = None + env_defaults: dict[str, str] | None = None _proc: subprocess.Popen[str] | None = None _lock: threading.Lock = field(default_factory=threading.Lock) _initialized: bool = False _request_id: int = 0 @staticmethod - def _build_runtime_env(env_allowlist: list[str] | None) -> dict[str, str] | None: + def _build_runtime_env( + env_allowlist: list[str] | None, + env_defaults: dict[str, str] | None = None, + ) -> dict[str, str] | None: if env_allowlist is None: return None from runtime.operations.mcp_env import mcp_local_env_merged @@ -46,6 +50,16 @@ class McpProcessRuntime: key = str(k or "").strip() if key and key in os.environ and str(os.environ[key] or "").strip(): env[key] = os.environ[key] + for k, v in (env_defaults or {}).items(): + key = str(k or "").strip() + val = str(v or "").strip() + if not key or not val: + continue + if key not in env or not str(env.get(key) or "").strip(): + env[key] = val + # MCP JSON-RPC on stdio is UTF-8; force child Python off Windows GBK console encoding. + env.setdefault("PYTHONIOENCODING", "utf-8") + env.setdefault("PYTHONUTF8", "1") return env @staticmethod @@ -68,11 +82,19 @@ class McpProcessRuntime: def start(self) -> None: if self._proc and self._proc.poll() is None: return - env = self._build_runtime_env(self.env_allowlist) + env = self._build_runtime_env(self.env_allowlist, self.env_defaults) cmd = list(self.command or []) if cmd: cmd[0] = self._resolve_command(str(cmd[0]), env) - popen_kwargs: dict[str, Any] = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, "text": True, "encoding": "utf-8", "env": env} + popen_kwargs: dict[str, Any] = { + "stdin": subprocess.PIPE, + "stdout": subprocess.PIPE, + "stderr": subprocess.PIPE, + "text": True, + "encoding": "utf-8", + "errors": "replace", + "env": env, + } if os.name == "nt": startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index f56a8f89..7e74559a 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -35,19 +35,20 @@ You are the ops specialist (network operations expert). - 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) +## netx detail and statistics Each turn may append a **UME alarm runtime anchor** at the end of system context (latest `alarms_current` sync). Still call tools for alarm/NE evidence when answering. -- Default UME current-alarm path; no import `batch_id`. -- `netx_query_ume_alarms`: current alarm rows (each includes **`host_name`**; filters: `severity` / `ne_id` / `keyword`). -- `netx_aggregate_ume_alarms` / `netx_run_ume_diagnostics`: aggregates and diagnostic summary. -- `netx_query_ume_ne_inventory`: synced NE list (`keyword`). -- `netx_get_ume_ne`: single NE by `ne_id` (includes `raw_json`). +- Default UME current alarms only (no Excel import `batch_id`). +- **MCP (12 tools, `server_id=netx`)**: + - UME alarms: `mcp__netx__queryUmeAlarms`, `mcp__netx__aggregateUmeAlarms`, `mcp__netx__runUmeDiagnostics` + - UME NE inventory: `mcp__netx__queryUmeNeInventory`, `mcp__netx__getUmeNe` + - UME deep query: `mcp__netx__queryUmeAlarmsRaw`, `mcp__netx__aggregateUmeAlarmsRaw`, `mcp__netx__listUmeAlarmFields`, `mcp__netx__sqlQueryUme` + - Managed NE CLI: `mcp__netx__listManagedNe`, `mcp__netx__getManagedNe`, `mcp__netx__execManagedNe` ## 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). +- **MCP**: `mcp__netx__listManagedNe` / `mcp__netx__getManagedNe` / `mcp__netx__execManagedNe`. +- **Legacy builtin** (`OCLAW_NETX_BUILTIN_TOOLS=1`): `netx_list_managed_ne`, `netx_get_managed_ne`, `netx_exec_managed_ne`. -Uses `OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`. Disable anchor inject: `OCLAW_OPS_NETX_CONTEXT_INJECT=0`. +netx API: MCP env `NETX_API_URL` (recommended); anchor probe also uses `OCLAW_NETX_BASE_URL`. Disable anchor inject: `OCLAW_OPS_NETX_CONTEXT_INJECT=0`. diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index b47f6f97..17f4a50d 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -18,8 +18,8 @@ ## 告警与网元展示（强制） - **网元维度一律以 `host_name` 为主键展示**（表格首列、Top 排名键、分组维度、结论中的网元指称）。告警同步后 netx 已把 `host_name` 写入告警表，优先读： - - 列表/分页：`netx_query_ume_alarms` 返回的 **`host_name`** - - Raw/SQL：`alarm_host_name`（与 `ne_host_name` 同值时优先用 `alarm_host_name`） + - 列表/分页：`mcp__netx__queryUmeAlarms`（或 legacy `netx_query_ume_alarms`）返回的 **`host_name`** + - 聚合：`mcp__netx__aggregateUmeAlarms` 的网元维度字段 - **禁止**用 `ne_id` / `alarm_ne_id`（UUID）作为对用户的主展示键；`ne_id` 仅用于工具过滤或内部关联。 - 若 `host_name` 为空，再用 `user_label` / `ne_name` 并标注「host_name 缺失」；仍不得用裸 `ne_id`。 - 按网元统计/聚合：优先 `group_by=alarm_host_name` 或 `group_by=ne_host_name`，勿按 `alarm_ne_id` / `ne_ne_id` 对外展示。 @@ -28,19 +28,20 @@ - 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。 - 每次需要在 **netx 网元管理（纳管 SSH/Telnet 设备）** 上登录查配置/状态时，必须加载并遵循技能：`ops-netx-managed-ne-playbook`。 -## netx 明细与统计（内部工具） +## netx 明细与统计 每轮对话 **system 末尾会自动附带当前 UME 告警运行锚点**（最近一次 `alarms_current` 同步状态），用于快速判断数据新鲜度。涉及告警/统计时仍应用工具拉明细。 -- 默认使用 UME 当前告警链路，不再依赖导入批次 `batch_id`。 -- `netx_query_ume_alarms`：查询 UME 当前告警明细（每条含 **`host_name`**；支持 `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`）。 +- 默认 UME 当前告警；不依赖 Excel 导入 `batch_id`。 +- **MCP（12 个工具，`server_id=netx`）**： + - UME 告警：`mcp__netx__queryUmeAlarms`、`mcp__netx__aggregateUmeAlarms`、`mcp__netx__runUmeDiagnostics` + - UME 网元：`mcp__netx__queryUmeNeInventory`、`mcp__netx__getUmeNe` + - UME 深查：`mcp__netx__queryUmeAlarmsRaw`、`mcp__netx__aggregateUmeAlarmsRaw`、`mcp__netx__listUmeAlarmFields`、`mcp__netx__sqlQueryUme` + - 纳管网元 CLI：`mcp__netx__listManagedNe`、`mcp__netx__getManagedNe`、`mcp__netx__execManagedNe` ## netx 纳管网元（登录设备查 CLI） -- `netx_list_managed_ne` / `netx_get_managed_ne`：网元管理清单与连通详情。 -- `netx_exec_managed_ne`：经 netx 登录设备执行只读 CLI（show/display/ping；禁止改配置）。 +- **MCP**：`mcp__netx__listManagedNe` / `mcp__netx__getManagedNe` / `mcp__netx__execManagedNe`。 +- **旧内置**（`OCLAW_NETX_BUILTIN_TOOLS=1`）：`netx_list_managed_ne`、`netx_get_managed_ne`、`netx_exec_managed_ne`。 -工具走 netx（`OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`）。关闭自动锚点：环境变量 `OCLAW_OPS_NETX_CONTEXT_INJECT=0`。 +netx 服务地址：MCP env `NETX_API_URL`（推荐）；锚点探测亦用 `OCLAW_NETX_BASE_URL`。关闭自动锚点：`OCLAW_OPS_NETX_CONTEXT_INJECT=0`。 diff --git a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md index ec55d3df..864dcdeb 100644 --- a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md @@ -13,11 +13,13 @@ description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手 ## 工具选择顺序 +优先 **MCP**（`mcp__netx__*`）。legacy：`netx_list_managed_ne` 等（`OCLAW_NETX_BUILTIN_TOOLS=1`）。 + 1. **定位设备** - - `netx_list_managed_ne`：`keyword`（名称/IP）、`connect_status=pass` 过滤 - - `netx_get_managed_ne`：单条详情、`connect_detail`（连通失败排障） + - `mcp__netx__listManagedNe`：`keyword`、`connect_status=pass` + - `mcp__netx__getManagedNe`：单条详情、`connect_detail` 2. **登录查信息** - - `netx_exec_managed_ne`：`ne_id` + `commands`（最多 5 条只读命令） + - `mcp__netx__execManagedNe`：`ne_id` + `commands`（最多 5 条只读命令） ## CLI 约束（服务端强制） diff --git a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md index fd0bb931..793713d0 100644 --- a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md @@ -1,97 +1,55 @@ ---- +--- name: ops-netx-ume-playbook -description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查询/聚合/诊断、网元清单与单网元详情、raw 字段过滤与 UME 只读 SQL。 ---- +description: 闈㈠悜 ops 涓撳鐨?netx UME 杩愮淮浣滀笟鎵嬪唽銆傝鐩栧憡璀︽煡璇?鑱氬悎/璇婃柇銆佺綉鍏冩竻鍗曚笌鍗曠綉鍏冭鎯呫€乺aw 瀛楁杩囨护涓?UME 鍙 SQL銆?--- -# Ops Netx UME 作业手册 +# Ops Netx UME 浣滀笟鎵嬪唽 -## 强制使用范围 +## 寮哄埗浣跨敤鑼冨洿 -凡是涉及 netx/UME **告警**或 **网元信息** 的 ops 请求，必须优先加载并遵循本技能。 +鍑℃槸娑夊強 netx/UME **鍛婅**鎴?**缃戝厓淇℃伅** 鐨?ops 璇锋眰锛屽繀椤讳紭鍏堝姞杞藉苟閬靛惊鏈妧鑳姐€? +## 宸ュ叿閫夋嫨椤哄簭 -## 工具选择顺序 - -1. 基础视图（先看整体）： - - `netx_query_ume_alarms` +1. 鍩虹瑙嗗浘锛堝厛鐪嬫暣浣擄級锛? - `netx_query_ume_alarms` - `netx_aggregate_ume_alarms` - `netx_run_ume_diagnostics` -2. 字段感知深查（需要细节）： - - `netx_list_ume_alarm_fields` - - `netx_query_ume_alarms_raw`（优先使用 `select_fields` 控制返回字段） -3. 自定义聚合（非 SQL）： - - `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） +2. 瀛楁鎰熺煡娣辨煡锛堥渶瑕佺粏鑺傦級锛? - `netx_list_ume_alarm_fields` + - `netx_query_ume_alarms_raw`锛堜紭鍏堜娇鐢?`select_fields` 鎺у埗杩斿洖瀛楁锛?3. 鑷畾涔夎仛鍚堬紙闈?SQL锛夛細 + - `netx_aggregate_ume_alarms_raw`锛坄group_by`锛屽彲閫?`group_by2`锛?4. 楂樼骇鍒嗘瀽锛圫QL锛夛細 + - `netx_sql_query_ume`锛堜粎 SELECT銆佷粎 UME 琛紱閲嶆煡璇㈠缓璁缃?`statement_timeout_ms`锛?5. **缃戝厓锛坕nventory锛屼笌 netx銆岀綉鍏冩竻鍗曘€嶅悓婧愶級**锛? - 鍒楄〃/鎼滅储锛歚netx_query_ume_ne_inventory`锛坄keyword` + 鍒嗛〉锛? - 鍗曟潯璇︽儏锛堝惈 `raw_json`锛夛細`netx_get_ume_ne`锛坄ne_id` = UUID锛? +## 蹇€熷喅绛栨爲锛堝己鎺ㄨ崘锛? +- **鍙渶瑕佹暣浣撴€佸娍 / Top 椋庨櫓 / 蹇€熺畝鎶?*锛? - 鍏?`netx_aggregate_ume_alarms` + `netx_run_ume_diagnostics` + - 蹇呰鏃跺啀鐢?`netx_query_ume_alarms` 鐪嬪墠 1 椤靛仛鏍锋湰鏍稿 +- **闇€瑕佲€滃彲寮曠敤璇佹嵁鈥濈殑鍏蜂綋鍛婅鏄庣粏**锛? - 鍏?`netx_list_ume_alarm_fields` + - 鍐?`netx_query_ume_alarms_raw`锛屽苟鐢?`select_fields` 鍙彇蹇呰瀛楁 +- **闇€瑕佹寜浠绘剰瀛楁鍋氱粺璁★紙浣嗕笉鎯冲啓 SQL锛?*锛? - `netx_aggregate_ume_alarms_raw`锛坄group_by` / `group_by2`锛?- **闇€瑕佸鏉傛潯浠?/ 鑷畾涔夎绠?/ 澶氭潯浠跺叧鑱?*锛? - `netx_sql_query_ume`锛堝繀椤昏繃婊?+ `statement_timeout_ms`锛?- **鏌ョ綉鍏冩槸璋併€両P/鏍囩銆佸湪绾跨姸鎬併€佹垨鏍稿鍛婅閲岀殑 ne_id**锛? - 鍏?`netx_query_ume_ne_inventory`锛坄keyword` 鍙～鍚嶇О銆佷富鏈哄悕銆佹爣绛俱€両P 鎴?UUID 鐗囨锛? - 闇€瑕佸畬鏁村瓧娈典笌 `raw_json` 鏃跺啀 `netx_get_ume_ne` -## 快速决策树（强推荐） +## 绾︽潫涓庢姢鏍? +- 浼樺厛浣跨敤闈?SQL 宸ュ叿锛涗粎褰撳伐鍏峰弬鏁版棤娉曡〃杈鹃渶姹傛椂鍐嶇敤 SQL銆?- 榛樿杩囨护浼樺厛绾э紙鍏堟敹鏁涘啀鎵╁睍锛夛細 + - 棣栭€夛細`severity`锛堝厛鎶婇棶棰樼缉灏忓埌 critical/major 绛夛級 + - 鍏舵锛歚keyword`锛堢綉鍏冨悕/鏍囩/IP/瀵硅薄鍚?鍛婅鍏抽敭瀛楋級 + - 鍐嶆锛歚time_from/time_to`锛堟寜 `last_seen_at` 闄愬畾鏃堕棿绐楋級 + - 鏈€鍚庯細`event_type` 鎴?`ne_id`锛堝綋浣犳槑纭煡閬撹閿佸畾浜嬩欢绫诲瀷/缃戝厓鏃讹級 +- 绂佹鈥滀负浜嗗噾鍏ㄩ噺鑰屾棤鑴戠炕椤碘€濓細 + - `netx_query_ume_alarms` 榛樿鍙湅鍓?1 椤碉紙蹇呰鏃舵渶澶?2 椤碉級 + - 濡傞渶鏇村鏁版嵁锛屽繀椤诲厛鏄庣‘杩囨护鏉′欢锛坄severity/ne_id/keyword/time_from/time_to/event_type` 绛夛級鎴栨敼鐢ㄨ仛鍚?SQL +- 鎺у埗鍝嶅簲浣撶Н锛? - 榛樿 `page_size=50`锛堥櫎闈炴槑纭渶瑕佹洿澶氾紝鍚﹀垯涓嶈涓婃潵灏辨媺婊?500锛? - 鍔ㄦ€佽仛鍚堥粯璁?`limit=200` + - 鍚堢悊璁剧疆 `page_size` + - raw 鏌ヨ灏介噺浼?`select_fields`锛涙垨浣跨敤 `field_preset=brief/evidence/ne_debug` + - 鍏堝姞杩囨护鏉′欢锛屽啀澧炲ぇ鍒嗛〉鑼冨洿 +- 鏃堕棿绐楄繃婊ら粯璁ゅ熀浜?`last_seen_at` 璇箟锛岄櫎闈為渶姹傛槑纭姹傚叾瀹冨彛寰勩€?- 鑻ユ暟鎹柊椴滃害涓嶆槑纭紝鍏堟煡鐪?runtime 閿氱偣鐘舵€侊紝鍐嶄笅缁撹銆?- SQL 浣跨敤瑙勫垯锛坄netx_sql_query_ume`锛夛細 + - 寤鸿鎬绘槸璁剧疆 `statement_timeout_ms`锛堜緥濡?3000~10000锛? - 鎺ㄨ崘榛樿浠?`statement_timeout_ms=8000` 寮€濮? - 闄ら潪鍙槸 `count(*)`锛屽惁鍒欏簲鍖呭惈杩囨护鏉′欢锛堣嚦灏戞椂闂寸獥鎴?`ne_id`/涓ラ噸搴﹁繃婊わ級锛岄伩鍏嶅叏琛ㄦ壂鎻? +## 杈撳嚭绾﹀畾 -- **只需要整体态势 / Top 风险 / 快速简报**： - - 先 `netx_aggregate_ume_alarms` + `netx_run_ume_diagnostics` - - 必要时再用 `netx_query_ume_alarms` 看前 1 页做样本核对 -- **需要“可引用证据”的具体告警明细**： - - 先 `netx_list_ume_alarm_fields` - - 再 `netx_query_ume_alarms_raw`，并用 `select_fields` 只取必要字段 -- **需要按任意字段做统计（但不想写 SQL）**： - - `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` 可填名称、主机名、标签、IP 或 UUID 片段） - - 需要完整字段与 `raw_json` 时再 `netx_get_ume_ne` +- 杈撳嚭蹇呴』鍖呭惈锛? - 绠€鏄庣粨璁? - 璇佹嵁渚濇嵁锛堝伐鍏疯緭鍑猴級 + - 鍙墽琛屼笅涓€姝?- 娌℃湁宸ュ叿璇佹嵁鏃讹紝涓嶅緱鑷嗘祴鍛婅浜嬪疄銆?- **鐢ㄦ埛鐢ㄨ嫳鏂囨彁闂椂锛堝己鍒讹級**锛氬洖澶嶄腑**涓嶅緱鍑虹幇浠讳綍姹夊瓧**锛涘伐鍏烽噷鐨勪腑鏂囧憡璀﹀瓧娈碉紙鍘熷洜銆佸璞″悕銆佹弿杩扮瓑锛夊繀椤诲厛**璇戞垚鑻辨枃**鍐嶅啓鍏ヨ〃鏍兼垨姝ｆ枃锛岀姝㈠師鏍风矘璐达紱缃戝厓鍚嶇敤 `host_name`锛屽崗璁被缁村害鐢ㄨ嫳鏂囩被鍒悕锛圤ther/Clock/鈥︼級銆? +### 缃戝厓灞曠ず锛氫互 host_name 涓轰富閿紙寮哄埗锛? +- **鍛婅/缁熻閲屾爣璇嗙綉鍏冩椂锛屼富閿案杩滄槸 `host_name`**锛堜富鏈哄悕锛夛紝涓嶆槸 `ne_id`銆傝〃鏍肩涓€鍒椼€乀op 缃戝厓銆佸垎缁勯敭銆佺粨璁洪噷鐨勭綉鍏冨悕閮界敤瀹冦€?- **浼樺厛鏁版嵁婧?*锛堝悓姝ユ椂宸插啓鍏ュ憡璀﹁〃锛夛細 + - `netx_query_ume_alarms` 鈫?瀛楁 **`host_name`** + - `netx_query_ume_alarms_raw` 鈫?**`alarm_host_name`**锛坄select_fields` / `brief` / `evidence` 棰勮宸插寘鍚級 + - 鑱氬悎 鈫?`group_by=alarm_host_name` 鎴?`group_by=ne_host_name` +- **绂佹**瀵圭敤鎴峰睍绀鸿８ `ne_id` / `alarm_ne_id`锛沗ne_id` 浠呬綔鏌ヨ鍙傛暟銆?- `host_name` 涓虹┖鏃讹細鐢?`user_label` / `ne_name` 骞舵敞鏄庣己澶憋紱浠嶄笉寰楅€€鍥?UUID銆?- 浠呭綋鍒楄〃鎺ュ彛缂?`host_name` 鏃跺啀 `netx_get_ume_ne` / 缃戝厓娓呭崟 / SQL JOIN 琛ュ叏銆? +## 鎺ㄨ崘鍒嗘瀽妯″紡 -## 约束与护栏 - -- 优先使用非 SQL 工具；仅当工具参数无法表达需求时再用 SQL。 -- 默认过滤优先级（先收敛再扩展）： - - 首选：`severity`（先把问题缩小到 critical/major 等） - - 其次：`keyword`（网元名/标签/IP/对象名/告警关键字） - - 再次：`time_from/time_to`（按 `last_seen_at` 限定时间窗） - - 最后：`event_type` 或 `ne_id`（当你明确知道要锁定事件类型/网元时） -- 禁止“为了凑全量而无脑翻页”： - - `netx_query_ume_alarms` 默认只看前 1 页（必要时最多 2 页） - - 如需更多数据，必须先明确过滤条件（`severity/ne_id/keyword/time_from/time_to/event_type` 等）或改用聚合/SQL -- 控制响应体积： - - 默认 `page_size=50`（除非明确需要更多，否则不要上来就拉满 500） - - 动态聚合默认 `limit=200` - - 合理设置 `page_size` - - raw 查询尽量传 `select_fields`；或使用 `field_preset=brief/evidence/ne_debug` - - 先加过滤条件，再增大分页范围 -- 时间窗过滤默认基于 `last_seen_at` 语义，除非需求明确要求其它口径。 -- 若数据新鲜度不明确，先查看 runtime 锚点状态，再下结论。 -- SQL 使用规则（`netx_sql_query_ume`）： - - 建议总是设置 `statement_timeout_ms`（例如 3000~10000） - - 推荐默认从 `statement_timeout_ms=8000` 开始 - - 除非只是 `count(*)`，否则应包含过滤条件（至少时间窗或 `ne_id`/严重度过滤），避免全表扫描 - -## 输出约定 - -- 输出必须包含： - - 简明结论 - - 证据依据（工具输出） - - 可执行下一步 -- 没有工具证据时，不得臆测告警事实。 -- **用户用英文提问时（强制）**：回复中**不得出现任何汉字**；工具里的中文告警字段（原因、对象名、描述等）必须先**译成英文**再写入表格或正文，禁止原样粘贴；网元名用 `host_name`，协议类维度用英文类别名（Other/Clock/…）。 - -### 网元展示：以 host_name 为主键（强制） - -- **告警/统计里标识网元时，主键永远是 `host_name`**（主机名），不是 `ne_id`。表格第一列、Top 网元、分组键、结论里的网元名都用它。 -- **优先数据源**（同步时已写入告警表）： - - `netx_query_ume_alarms` → 字段 **`host_name`** - - `netx_query_ume_alarms_raw` → **`alarm_host_name`**（`select_fields` / `brief` / `evidence` 预设已包含） - - 聚合 → `group_by=alarm_host_name` 或 `group_by=ne_host_name` -- **禁止**对用户展示裸 `ne_id` / `alarm_ne_id`；`ne_id` 仅作查询参数。 -- `host_name` 为空时：用 `user_label` / `ne_name` 并注明缺失；仍不得退回 UUID。 -- 仅当列表接口缺 `host_name` 时再 `netx_get_ume_ne` / 网元清单 / SQL JOIN 补全。 - -## 推荐分析模式 - -- 高风险网元：`netx_aggregate_ume_alarms_raw` + `group_by=alarm_host_name`（首选）或 `ne_host_name` + 严重度过滤；勿按 `alarm_ne_id` 分组对外展示。 -- 严重度分布：`group_by=alarm_perceived_severity`。 -- 事件趋势切片：raw 查询中组合 `time_from/time_to` + `event_type`。 - -## 参考模板 - -- 快速模板见：[reference.md](reference.md) +- 楂橀闄╃綉鍏冿細`netx_aggregate_ume_alarms_raw` + `group_by=alarm_host_name`锛堥閫夛級鎴?`ne_host_name` + 涓ラ噸搴﹁繃婊わ紱鍕挎寜 `alarm_ne_id` 鍒嗙粍瀵瑰灞曠ず銆?- 涓ラ噸搴﹀垎甯冿細`group_by=alarm_perceived_severity`銆?- 浜嬩欢瓒嬪娍鍒囩墖锛歳aw 鏌ヨ涓粍鍚?`time_from/time_to` + `event_type`銆? +## 鍙傝€冩ā鏉? +- 蹇€熸ā鏉胯锛歔reference.md](reference.md) diff --git a/tests/test_mcp_env_defaults.py b/tests/test_mcp_env_defaults.py new file mode 100644 index 00000000..482aa47a --- /dev/null +++ b/tests/test_mcp_env_defaults.py @@ -0,0 +1,41 @@ +"""MCP subprocess env: env_schema defaults from registry (Cursor mcpServers.env import).""" + +from __future__ import annotations + +import os +import unittest +from unittest.mock import patch + +from runtime.tools.mcp.runtime import McpProcessRuntime + + +class McpEnvDefaultsTests(unittest.TestCase): + @patch("runtime.operations.mcp_env.mcp_local_env_merged", return_value={}) + def test_env_schema_default_used_when_not_in_os_environ(self, _mock: object) -> None: + os.environ.pop("NETX_API_URL", None) + try: + env = McpProcessRuntime._build_runtime_env( + ["NETX_API_URL"], + {"NETX_API_URL": "http://10.0.0.5:8890"}, + ) + assert env is not None + self.assertEqual(env.get("NETX_API_URL"), "http://10.0.0.5:8890") + finally: + pass + + @patch("runtime.operations.mcp_env.mcp_local_env_merged", return_value={}) + def test_mcp_local_and_os_environ_override_schema_default(self, _mock: object) -> None: + os.environ["NETX_API_URL"] = "http://from-host:8890" + try: + env = McpProcessRuntime._build_runtime_env( + ["NETX_API_URL"], + {"NETX_API_URL": "http://from-schema:8890"}, + ) + assert env is not None + self.assertEqual(env.get("NETX_API_URL"), "http://from-host:8890") + finally: + os.environ.pop("NETX_API_URL", None) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_installer.py b/tests/test_mcp_installer.py index 3301dd63..5e46cdab 100644 --- a/tests/test_mcp_installer.py +++ b/tests/test_mcp_installer.py @@ -23,6 +23,13 @@ class McpInstallerTests(unittest.TestCase): self.assertIn(" install -g ", res.install_command) self.assertIn("demo-server@1.0.0", res.install_command) + def test_install_local_skips_package_step(self) -> None: + m = McpServerManifest(server_id="netx", source_type="local", source_ref="netx-mcp-http", entry_command="python") + res = install_mcp_server(m, dry_run=False) + self.assertTrue(res.ok) + self.assertEqual(res.install_command, "") + self.assertEqual((res.details or {}).get("reason"), "local_source_no_package_install") + def test_invalid_source(self) -> None: m = McpServerManifest(server_id="x", source_type="invalid", source_ref="x") res = install_mcp_server(m, dry_run=True) diff --git a/tests/test_netx_expert_registry.py b/tests/test_netx_expert_registry.py new file mode 100644 index 00000000..09484eaa --- /dev/null +++ b/tests/test_netx_expert_registry.py @@ -0,0 +1,35 @@ +"""Expert registry: netx builtin tools gated by OCLAW_NETX_BUILTIN_TOOLS.""" + +from __future__ import annotations + +import os + +import pytest + +from runtime.tools import expert_registry + + +@pytest.fixture(autouse=True) +def _clear_expert_cache(): + expert_registry._CACHED_FACTORIES_BY_EXPERT = None + expert_registry._CACHED_SPECS_BY_EXPERT = None + yield + expert_registry._CACHED_FACTORIES_BY_EXPERT = None + expert_registry._CACHED_SPECS_BY_EXPERT = None + + +def test_netx_tools_skipped_when_builtin_disabled(monkeypatch): + monkeypatch.delenv("OCLAW_NETX_BUILTIN_TOOLS", raising=False) + factories = expert_registry.discover_expert_tool_factories() + network_ops = factories.get("network_ops") or [] + names = {f().name for f in network_ops} + assert not any(n.startswith("netx_") for n in names) + + +def test_netx_tools_registered_when_builtin_enabled(monkeypatch): + monkeypatch.setenv("OCLAW_NETX_BUILTIN_TOOLS", "1") + factories = expert_registry.discover_expert_tool_factories() + network_ops = factories.get("network_ops") or [] + names = {f().name for f in network_ops} + assert "netx_query_ume_alarms" in names + assert "netx_exec_managed_ne" in names