From c141a372634f08d6c93e4c1f0b970b7b964a8dcc Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 14 Oct 2026 01:46:38 +0800 Subject: [PATCH] netx-mcp 0.3.1: fix alarm paging, tool-level arg errors, leaner outputs. - queryNmsAlarms no longer silently clamps page to 2; add is_cleared. - Bad integer args return invalid_argument tool errors instead of JSON-RPC crashes. - Nested {"ok": true, "data": {"ok": false}} results are flagged isError. - stdio: answer ping, never respond to notifications, version from metadata. - getNmsNe resolves exact host_name/IP; severity lowercased. - Outputs use nms_ne_id/source=nms; drop ME{uuid} names, echoed field lists, idle hop_* fields. - Fabric paths: parallel links count as one hop (edges[].parallel_links). Co-Authored-By: Claude Opus 5.5 --- netx_api/topology_fabric_nodes.py | 17 +- packages/netx-mcp/README.md | 4 +- packages/netx-mcp/pyproject.toml | 2 +- packages/netx-mcp/src/netx_mcp/http_client.py | 4 +- packages/netx-mcp/src/netx_mcp/http_tools.py | 250 ++++++++++++++---- packages/netx-mcp/src/netx_mcp/server.py | 18 +- packages/netx-mcp/tests/test_mcp_http.py | 99 ++++++- tests/test_mcp_http.py | 6 +- tests/test_topology.py | 21 ++ 9 files changed, 358 insertions(+), 63 deletions(-) diff --git a/netx_api/topology_fabric_nodes.py b/netx_api/topology_fabric_nodes.py index ba37251..5164ff0 100644 --- a/netx_api/topology_fabric_nodes.py +++ b/netx_api/topology_fabric_nodes.py @@ -620,9 +620,13 @@ def find_fabric_paths( if len(edge_path) >= max_hops: continue _ensure_adj({node}) + # Parallel links (LAG members / dual fibers) between the same two NEs are one hop: + # expanding each would return near-identical paths that differ only by port. + seen_nbrs: set[str] = set() for nbr, eid in adj_cache.get(node, []): - if nbr in visited: + if nbr in visited or nbr in seen_nbrs: continue + seen_nbrs.add(nbr) explored += 1 new_path = edge_path + [eid] if nbr == to_id: @@ -650,10 +654,15 @@ def find_fabric_paths( "vendor": n.vendor or "", } + def _parallel_count(e: TopoFabricEdge) -> int: + # Only expanded nodes have adjacency loaded; the path's last node never is. + src, dst = (e.a_node_id, e.b_node_id) if e.a_node_id in adj_loaded else (e.b_node_id, e.a_node_id) + return sum(1 for nbr, _ in adj_cache.get(src, []) if nbr == dst) or 1 + def _edge_summary(e: TopoFabricEdge) -> dict[str, Any]: a_node = node_map.get(e.a_node_id) b_node = node_map.get(e.b_node_id) - return { + out = { "a_name": (a_node.name if a_node else "") or "", "b_name": (b_node.name if b_node else "") or "", "a_ip": (a_node.ip if a_node else "") or "", @@ -662,6 +671,10 @@ def find_fabric_paths( "b_port": e.b_port or "", "status": _normalize_edge_status(e.status or "active"), } + parallel = _parallel_count(e) + if parallel > 1: + out["parallel_links"] = parallel + return out def _path_node_ids(edge_ids: list[str]) -> list[str]: ids = [from_id] diff --git a/packages/netx-mcp/README.md b/packages/netx-mcp/README.md index 72ce048..009c1ec 100644 --- a/packages/netx-mcp/README.md +++ b/packages/netx-mcp/README.md @@ -38,7 +38,7 @@ python -m netx_mcp [`mcp.json`](./mcp.json) — `command: python`，`args: ["-m", "netx_mcp"]`，`env` 见文件。 -## 工具（14）— skill 组 **ops**（`netx-ops`） +## 工具（15）— skill 组 **ops**（`netx-ops`） **NMS**（模型面通用名；当前适配器 zte-ume，REST 仍 `/v1/ume/*`）： `queryNmsAlarms`, `aggregateNmsAlarms`, `runNmsDiagnostics`, `queryNmsNeInventory`, `getNmsNe`, `queryNmsAlarmsRaw`, `aggregateNmsAlarmsRaw`, `listNmsAlarmFields`, `sqlQueryNms` @@ -46,6 +46,8 @@ python -m netx_mcp **Managed CLI + paths**（问「能否登录」必须走这里，不要只查 inventory）： `listManagedNe`, `getManagedNe`, `execManagedNe`, `getNeExecJob`, `listCliTargets`, `findTopologyPaths` +0.3.1：`getNmsNe` 可直接传 host_name/IP；输出统一 `nms_ne_id` / `source=nms`；参数类型错误返回工具错误而非协议崩溃。 + 参数优先 `nms_ne_id` / `nms_ne_ids`（保留 `ume_*` 别名）。 拓扑画布 / Fabric → skill **topology** / [`netx-topology-mcp`](../netx-topology-mcp)（见 [docs/MCP_TOPOLOGY.md](../../docs/MCP_TOPOLOGY.md)）。 diff --git a/packages/netx-mcp/pyproject.toml b/packages/netx-mcp/pyproject.toml index 56d3e79..090358e 100644 --- a/packages/netx-mcp/pyproject.toml +++ b/packages/netx-mcp/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "netx-mcp" -version = "0.3.0" +version = "0.3.1" description = "stdio MCP server for netx REST API (NMS alarms, managed NE CLI)" readme = "README.md" requires-python = ">=3.11" diff --git a/packages/netx-mcp/src/netx_mcp/http_client.py b/packages/netx-mcp/src/netx_mcp/http_client.py index 60a153c..3a6914e 100644 --- a/packages/netx-mcp/src/netx_mcp/http_client.py +++ b/packages/netx-mcp/src/netx_mcp/http_client.py @@ -117,7 +117,9 @@ def mcp_text_result(payload: Any, *, is_error: bool = False) -> dict[str, Any]: def mcp_from_handler_result(result: dict[str, Any]) -> dict[str, Any]: - if not result.get("ok"): + # API envelopes may nest a failure: {"ok": true, "data": {"ok": false, ...}} (e.g. job_not_found). + data = result.get("data") + if not result.get("ok") or (isinstance(data, dict) and data.get("ok") is False): return mcp_text_result(result, is_error=True) return mcp_text_result(result) diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py index 65e3bcd..e604cbf 100644 --- a/packages/netx-mcp/src/netx_mcp/http_tools.py +++ b/packages/netx-mcp/src/netx_mcp/http_tools.py @@ -75,6 +75,80 @@ def _should_run_exec_async(args: dict[str, Any]) -> bool: return _count_exec_ne_targets(args) >= min_n +class ToolArgError(ValueError): + """Bad tool argument; surfaced to the model as a tool error (not a JSON-RPC crash).""" + + +def _int_arg(args: dict[str, Any], key: str, default: int, lo: int, hi: int) -> int: + raw = args.get(key) + if raw is None or (isinstance(raw, str) and not raw.strip()): + return default + try: + val = int(raw) + except (TypeError, ValueError): + raise ToolArgError(f"{key} must be an integer, got {raw!r}") from None + return max(lo, min(hi, val)) + + +def _severity(args: dict[str, Any]) -> str: + # Stored values are lowercase (critical/major/minor/warning); exact match server-side. + return str(args.get("severity") or "").strip().lower() + + +def _is_me_placeholder(ne_name: Any, ne_id: Any) -> bool: + return bool(ne_id) and str(ne_name or "") == f"ME{{{ne_id}}}" + + +def _compact_rows(out: dict[str, Any], *, name_key: str, id_key: str) -> dict[str, Any]: + """Drop ZTE ``ME{}`` ne_name placeholders: they repeat ne_id and carry no meaning.""" + data = out.get("data") if isinstance(out, dict) else None + items = data.get("items") if isinstance(data, dict) else None + if isinstance(items, list): + for row in items: + if isinstance(row, dict) and _is_me_placeholder(row.get(name_key), row.get(id_key)): + row.pop(name_key, None) + return out + + +def _strip_meta_field_lists(out: dict[str, Any]) -> dict[str, Any]: + """Raw endpoints echo ~40 field names per call; listNmsAlarmFields already serves them.""" + data = out.get("data") if isinstance(out, dict) else None + meta = data.get("meta") if isinstance(data, dict) else None + if isinstance(meta, dict): + for k in ("available_fields", "group_by_allowed", "order_by_allowed"): + meta.pop(k, None) + return out + + +def _nms_naming(row: dict[str, Any]) -> None: + """REST still speaks ume_*; model-facing params are nms_* -- keep output consistent.""" + if "ume_ne_id" in row: + row["nms_ne_id"] = row.pop("ume_ne_id") + if row.get("source") == "ume": + row["source"] = "nms" + + +_HOP_KEYS = ( + "hop_vendor", + "hop_host", + "hop_port", + "hop_protocol", + "hop_username", + "hop_command_template", + "hop_vrf", + "hop_target_auth_mode", + "hop_enter_system_view", +) + + +def _compact_managed_ne(row: dict[str, Any], *, keep_detail: bool) -> None: + if row.get("hop_enabled") is False: + for k in _HOP_KEYS: + row.pop(k, None) + if not keep_detail: + row.pop("connect_detail", None) + + UME_RAW_GROUP_FIELDS = [ "alarm_alarm_key", "alarm_host_name", @@ -121,7 +195,6 @@ _UME_RAW_FIELD_PRESETS: dict[str, list[str]] = { "alarm_last_seen_at", "ne_host_name", "ne_user_label", - "ne_ne_name", "ne_ip_address", "ne_exists", ], @@ -137,7 +210,6 @@ _UME_RAW_FIELD_PRESETS: dict[str, list[str]] = { "alarm_last_seen_at", "ne_host_name", "ne_user_label", - "ne_ne_name", "ne_ip_address", "ne_connection_status", "ne_exists", @@ -148,7 +220,6 @@ _UME_RAW_FIELD_PRESETS: dict[str, list[str]] = { "alarm_perceived_severity", "alarm_last_seen_at", "ne_user_label", - "ne_ne_name", "ne_ip_address", "ne_ipv6_address", "ne_device_level", @@ -163,13 +234,14 @@ _UME_RAW_FIELD_PRESETS: dict[str, list[str]] = { def _query_ume_alarms(args: dict[str, Any]) -> dict[str, Any]: - page = max(1, int(args.get("page") or 1)) - if page > 2: - page = 2 - 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("severity") or "").strip(): - params["severity"] = str(args.get("severity")).strip() + params: dict[str, Any] = { + "page": _int_arg(args, "page", 1, 1, 100000), + "page_size": _int_arg(args, "page_size", 50, 1, 500), + } + if _severity(args): + params["severity"] = _severity(args) + if str(args.get("is_cleared") or "").strip(): + params["is_cleared"] = str(args.get("is_cleared")).strip() keyword = str(args.get("keyword") or "").strip() ne_name = str(args.get("ne_name") or "").strip() if keyword: @@ -183,7 +255,7 @@ def _query_ume_alarms(args: dict[str, Any]) -> dict[str, Any]: for k in ("time_from", "time_to"): if str(args.get(k) or "").strip(): params[k] = str(args.get(k)).strip() - return http_json("GET", "/v1/ume/alarms", params=params) + return _compact_rows(http_json("GET", "/v1/ume/alarms", params=params), name_key="ne_name", id_key="ne_id") def _aggregate_ume_alarms(args: dict[str, Any]) -> dict[str, Any]: @@ -192,11 +264,12 @@ def _aggregate_ume_alarms(args: dict[str, Any]) -> dict[str, Any]: if group_by: return _aggregate_ume_alarms_raw(args) # Default top 50 named NEs — missing host buckets reported separately. - top_ne = max(0, min(500, int(args.get("top_ne") if args.get("top_ne") is not None else 50))) - params: dict[str, Any] = {"top_ne": top_ne} + params: dict[str, Any] = {"top_ne": _int_arg(args, "top_ne", 50, 0, 500)} if "exclude_missing_host" in args: params["exclude_missing_host"] = bool(args.get("exclude_missing_host")) - for k in ("severity", "time_from", "time_to"): + if _severity(args): + params["severity"] = _severity(args) + for k in ("time_from", "time_to"): if str(args.get(k) or "").strip(): params[k] = str(args.get(k)).strip() return http_json("GET", "/v1/ume/alarms/aggregate", params=params) @@ -208,26 +281,58 @@ def _run_ume_diagnostics(args: dict[str, Any]) -> dict[str, Any]: def _query_ume_ne_inventory(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} + params: dict[str, Any] = { + "page": _int_arg(args, "page", 1, 1, 100000), + "page_size": _int_arg(args, "page_size", 50, 1, 500), + } 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 _compact_rows(http_json("GET", "/v1/ume/inventory/ne", params=params), name_key="ne_name", id_key="ne_id") + + +def _looks_like_uuid(text: str) -> bool: + parts = text.split("-") + return len(parts) == 5 and all(p and all(c in "0123456789abcdefABCDEF" for c in p) for p in parts) def _get_ume_ne(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/ume/inventory/ne/{quote_ne_id(ne_id)}", params=None) + out = http_json("GET", f"/v1/ume/inventory/ne/{quote_ne_id(ne_id)}", params=None) + if out.get("ok") or out.get("error") != "netx_http_404" or _looks_like_uuid(ne_id): + return out + # Agents usually know the host_name (alarms show it), not the UUID: resolve it. + found = http_json("GET", "/v1/ume/inventory/ne", params={"keyword": ne_id, "page": 1, "page_size": 20}) + items = ((found.get("data") or {}).get("items") or []) if found.get("ok") else [] + needle = ne_id.lower() + exact = [ + r + for r in items + if isinstance(r, dict) + and needle in {str(r.get(k) or "").lower() for k in ("host_name", "user_label", "ip_address", "ne_name")} + ] + if len(exact) == 1 and exact[0].get("ne_id"): + return http_json("GET", f"/v1/ume/inventory/ne/{quote_ne_id(str(exact[0]['ne_id']))}", params=None) + out = dict(out) + out["hint"] = "ne_id is the NMS UUID. Resolve host_name/IP via queryNmsNeInventory(keyword=...)." + if items: + out["candidates"] = [ + {"ne_id": r.get("ne_id"), "host_name": r.get("host_name"), "ip_address": r.get("ip_address")} + for r in items[:10] + if isinstance(r, dict) + ] + return out def _query_ume_alarms_raw(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} - for k in ("severity", "is_cleared", "ne_id", "event_type", "keyword", "time_from", "time_to", "order_by", "order"): + params: dict[str, Any] = { + "page": _int_arg(args, "page", 1, 1, 100000), + "page_size": _int_arg(args, "page_size", 50, 1, 500), + } + if _severity(args): + params["severity"] = _severity(args) + for k in ("is_cleared", "ne_id", "event_type", "keyword", "time_from", "time_to", "order_by", "order"): v = str(args.get(k) or "").strip() if v: params[k] = v @@ -240,7 +345,15 @@ def _query_ume_alarms_raw(args: dict[str, Any]) -> dict[str, Any]: fields = list(_UME_RAW_FIELD_PRESETS.get(preset) or []) if fields: params["select_fields"] = ",".join(fields) - return http_json("GET", "/v1/ume/alarms/raw", params=params) + out = http_json("GET", "/v1/ume/alarms/raw", params=params) + data = out.get("data") if isinstance(out.get("data"), dict) else None + if data and isinstance(data.get("items"), list): + for row in data["items"]: + if isinstance(row, dict) and _is_me_placeholder( + row.get("ne_ne_name"), row.get("ne_ne_id") or row.get("alarm_ne_id") + ): + row.pop("ne_ne_name", None) + return _strip_meta_field_lists(out) def _aggregate_ume_alarms_raw(args: dict[str, Any]) -> dict[str, Any]: @@ -248,7 +361,6 @@ def _aggregate_ume_alarms_raw(args: dict[str, Any]) -> dict[str, Any]: for k in ( "group_by", "group_by2", - "severity", "is_cleared", "ne_id", "event_type", @@ -263,9 +375,13 @@ def _aggregate_ume_alarms_raw(args: dict[str, Any]) -> dict[str, Any]: sv = str(v).strip() if sv: params[k] = sv + if _severity(args): + params["severity"] = _severity(args) + if "limit" in params: + params["limit"] = _int_arg(args, "limit", 200, 1, 2000) if "exclude_missing_host" in args: params["exclude_missing_host"] = bool(args.get("exclude_missing_host")) - return http_json("GET", "/v1/ume/alarms/aggregate/raw", params=params) + return _strip_meta_field_lists(http_json("GET", "/v1/ume/alarms/aggregate/raw", params=params)) def _list_ume_alarm_fields(args: dict[str, Any]) -> dict[str, Any]: @@ -275,8 +391,8 @@ def _list_ume_alarm_fields(args: dict[str, Any]) -> dict[str, Any]: def _sql_query_ume(args: dict[str, Any]) -> dict[str, Any]: sql = str(args.get("sql") or "").strip() - limit = max(1, min(2000, int(args.get("limit") or 200))) - statement_timeout_ms = max(0, min(30000, int(args.get("statement_timeout_ms") or 0))) + limit = _int_arg(args, "limit", 200, 1, 2000) + statement_timeout_ms = _int_arg(args, "statement_timeout_ms", 0, 0, 30000) if not sql: return {"ok": False, "error": "sql_required"} return http_post_json( @@ -287,13 +403,18 @@ def _sql_query_ume(args: dict[str, Any]) -> dict[str, Any]: def _list_managed_ne(args: dict[str, Any]) -> dict[str, Any]: - page = max(1, int(args.get("page") or 1)) - page_size = min(100, max(1, int(args.get("page_size") or 20))) + page = _int_arg(args, "page", 1, 1, 100000) + page_size = _int_arg(args, "page_size", 20, 1, 100) keyword = str(args.get("keyword") or "").strip() vendor = str(args.get("vendor") or "").strip() connect_status = str(args.get("connect_status") or "").strip() if not (keyword or vendor or connect_status): - return {"ok": False, "error": "managed_ne_filter_required", "error_code": "managed_ne_filter_required"} + return { + "ok": False, + "error": "managed_ne_filter_required", + "error_code": "managed_ne_filter_required", + "hint": "Pass keyword (name/IP, >=2 chars), vendor, or connect_status (e.g. pass).", + } if keyword and len(keyword) < 2: return {"ok": False, "error": "managed_ne_keyword_too_short", "error_code": "managed_ne_keyword_too_short"} params: dict[str, Any] = {"page": page, "page_size": page_size} @@ -303,7 +424,13 @@ def _list_managed_ne(args: dict[str, Any]) -> dict[str, Any]: params["vendor"] = vendor if connect_status: params["connect_status"] = connect_status - return http_json("GET", "/v1/managed-ne", params=params) + out = http_json("GET", "/v1/managed-ne", params=params) + items = (out.get("data") or {}).get("items") if out.get("ok") else None + if isinstance(items, list): + for row in items: + if isinstance(row, dict): + _compact_managed_ne(row, keep_detail=False) + return out def _first_str(*vals: Any) -> str: @@ -330,7 +457,9 @@ def _get_managed_ne(args: dict[str, Any]) -> dict[str, Any]: "example": {"ne_id": ""}, } # NMS ne_id is typically a UUID; managed NE may differ. Soft-guide when callers mix them. - out = http_json("GET", f"/v1/managed-ne/{ne_id}", params=None) + out = http_json("GET", f"/v1/managed-ne/{quote_ne_id(ne_id)}", params=None) + if out.get("ok") and isinstance(out.get("data"), dict): + _compact_managed_ne(out["data"], keep_detail=True) if isinstance(out, dict) and out.get("ok") is False: detail = str(out.get("detail") or out.get("error") or "") low = detail.lower() @@ -391,11 +520,9 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: if len(shared_commands) > exec_max_commands(): return {"ok": False, "error": "too_many_commands", "error_code": "too_many_commands"} body["commands"] = shared_commands - rts = args.get("read_timeout_sec") - body["read_timeout_sec"] = int(rts) if rts is not None else 60 - conc = args.get("concurrency") - if conc is not None: - body["concurrency"] = int(conc) + body["read_timeout_sec"] = _int_arg(args, "read_timeout_sec", 60, 10, 120) + if args.get("concurrency") is not None: + body["concurrency"] = _int_arg(args, "concurrency", 4, 1, 8) if want_async: out = http_post_json("/v1/managed-ne/exec-jobs", body, timeout=60.0) if not out.get("ok"): @@ -434,8 +561,7 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: if nms_ne_id: body["ume_ne_id"] = nms_ne_id # Default 60s matches netx API default; slow show commands often exceed 30s. - rts = args.get("read_timeout_sec") - body["read_timeout_sec"] = int(rts) if rts is not None else 60 + body["read_timeout_sec"] = _int_arg(args, "read_timeout_sec", 60, 10, 120) if want_async: out = http_post_json("/v1/managed-ne/exec-jobs", body, timeout=60.0) if not out.get("ok"): @@ -464,16 +590,23 @@ def _get_ne_exec_job(args: dict[str, Any]) -> dict[str, Any]: def _list_cli_targets(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} + params: dict[str, Any] = { + "page": _int_arg(args, "page", 1, 1, 100000), + "page_size": _int_arg(args, "page_size", 50, 1, 500), + } source = str(args.get("source") or "").strip() if source: # API still uses source=ume; accept model-facing "nms". params["source"] = "ume" if source.lower() == "nms" else source if str(args.get("keyword") or "").strip(): params["keyword"] = str(args.get("keyword")).strip() - return http_json("GET", "/v1/cli/targets", params=params) + out = http_json("GET", "/v1/cli/targets", params=params) + items = (out.get("data") or {}).get("items") if out.get("ok") else None + if isinstance(items, list): + for row in items: + if isinstance(row, dict): + _nms_naming(row) + return out def _find_topology_paths(args: dict[str, Any]) -> dict[str, Any]: @@ -489,8 +622,8 @@ def _find_topology_paths(args: dict[str, Any]) -> dict[str, Any]: if detail not in {"summary", "full"}: detail = "summary" body: dict[str, Any] = { - "max_paths": max(1, min(10, int(args.get("max_paths") or 3))), - "max_hops": max(1, min(12, int(args.get("max_hops") or 6))), + "max_paths": _int_arg(args, "max_paths", 3, 1, 10), + "max_hops": _int_arg(args, "max_hops", 6, 1, 12), "layer": str(args.get("layer") or "physical").strip() or "physical", "detail": detail, } @@ -502,7 +635,14 @@ def _find_topology_paths(args: dict[str, Any]) -> dict[str, Any]: body["to_ume_ne_id"] = to_uid else: body["to_managed_ne_id"] = to_mid - return http_post_json("/v1/topology/fabric/paths", body, timeout=30.0) + out = http_post_json("/v1/topology/fabric/paths", body, timeout=30.0) + paths = (out.get("data") or {}).get("paths") if out.get("ok") else None + if isinstance(paths, list): + for p in paths: + for node in (p.get("nodes") or []) if isinstance(p, dict) else []: + if isinstance(node, dict): + _nms_naming(node) + return out HTTP_MCP_TOOLS: list[dict[str, Any]] = [ @@ -520,7 +660,8 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "inputSchema": { "type": "object", "properties": { - "severity": {"type": "string"}, + "severity": {"type": "string", "description": "critical/major/minor/warning"}, + "is_cleared": {"type": "string", "description": "\"False\" = active only; \"True\" = cleared."}, "ne_id": {"type": "string", "description": "Filter only; do not show UUID to users"}, "host_name": {"type": "string", "description": "Filter by NE host_name"}, "ne_name": {"type": "string", "description": "Legacy alias mapped to keyword"}, @@ -626,10 +767,10 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ }, { "name": "getNmsNe", - "description": "Get single NMS NE detail by ne_id (UUID).", + "description": "Get single NMS NE detail by ne_id (UUID); exact host_name / user_label / IP also resolve.", "inputSchema": { "type": "object", - "properties": {"ne_id": {"type": "string"}}, + "properties": {"ne_id": {"type": "string", "description": "NMS UUID, or exact host_name / IP."}}, "required": ["ne_id"], "additionalProperties": False, }, @@ -907,7 +1048,8 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ { "name": "findTopologyPaths", "description": ( - "Find up to max_paths simple paths between two fabric nodes for troubleshooting. " + "Find up to max_paths simple paths between two fabric nodes for troubleshooting " + "(parallel links between the same NEs count as one hop; see edges[].parallel_links). " "For each endpoint provide exactly one of nms_ne_id (from NMS alarm ne_id; alias " "from_ume_ne_id) or managed_ne_id — resolved to fabric node internally. Returns " "shortest paths first with compact label + node/edge summary (detail=summary default). " @@ -1014,4 +1156,8 @@ def call_http_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: fn = _HANDLERS.get(str(name or "").strip()) if not fn: raise ValueError(f"unknown tool: {name}") - return mcp_from_handler_result(fn(dict(args or {}))) + try: + result = fn(dict(args or {})) + except ToolArgError as exc: + result = {"ok": False, "error": "invalid_argument", "error_code": "invalid_argument", "detail": str(exc)} + return mcp_from_handler_result(result) diff --git a/packages/netx-mcp/src/netx_mcp/server.py b/packages/netx-mcp/src/netx_mcp/server.py index 9f4b9ae..16b13ac 100644 --- a/packages/netx-mcp/src/netx_mcp/server.py +++ b/packages/netx-mcp/src/netx_mcp/server.py @@ -43,6 +43,15 @@ def _err(rid: Any, code: int, message: str) -> None: _UNSET = object() +def _version() -> str: + try: + from importlib.metadata import version + + return version("netx-mcp") + except Exception: + return "0.0.0" + + def _fetch_scopes() -> list[str] | None: """Return granted scopes from /v1/auth/me, or None if the call fails (show all tools).""" try: @@ -81,6 +90,7 @@ def run_stdio_loop() -> None: except Exception: continue rid = req.get("id") + is_notification = "id" not in req method = str(req.get("method") or "") params = req.get("params") if isinstance(req.get("params"), dict) else {} @@ -91,11 +101,15 @@ def run_stdio_loop() -> None: { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, - "serverInfo": {"name": "netx-mcp", "version": "0.3.0", "mode": "http"}, + "serverInfo": {"name": "netx-mcp", "version": _version(), "mode": "http"}, }, ) continue - if method == "notifications/initialized": + if is_notification: + # JSON-RPC: notifications (initialized, cancelled, ...) never get a response. + continue + if method == "ping": + _ok(rid, {}) continue if method == "tools/list": _ok(rid, {"tools": tools_for_scopes(scopes())}) diff --git a/packages/netx-mcp/tests/test_mcp_http.py b/packages/netx-mcp/tests/test_mcp_http.py index 50e5b44..5b80df7 100644 --- a/packages/netx-mcp/tests/test_mcp_http.py +++ b/packages/netx-mcp/tests/test_mcp_http.py @@ -74,7 +74,7 @@ def test_call_aggregate_nms_alarms_group_by_routes_to_raw() -> None: params = mock_http.call_args[1]["params"] assert params["group_by"] == "alarm_host_name" assert params["severity"] == "critical" - assert params["limit"] == "20" + assert params["limit"] == 20 payload = json.loads(out["content"][0]["text"]) assert payload["ok"] is True @@ -285,6 +285,14 @@ def test_stdio_initialize_and_tools_list() -> None: assert msg.get("id") == 1 assert "result" in msg + # Notifications must not be answered; the next line belongs to ping. + for note in ("notifications/initialized", "notifications/cancelled"): + proc.stdin.write(json.dumps({"jsonrpc": "2.0", "method": note, "params": {}}) + "\n") + proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 9, "method": "ping"}) + "\n") + proc.stdin.flush() + pong = json.loads(proc.stdout.readline()) + assert pong == {"jsonrpc": "2.0", "id": 9, "result": {}} + proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + "\n") proc.stdin.flush() line2 = proc.stdout.readline() @@ -297,3 +305,92 @@ def test_stdio_initialize_and_tools_list() -> None: finally: proc.kill() proc.wait(timeout=5) + + +def test_bad_integer_arg_is_tool_error_not_crash() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + out = call_http_tool("queryNmsAlarms", {"page": "two"}) + mock_http.assert_not_called() + assert out["isError"] is True + payload = json.loads(out["content"][0]["text"]) + assert payload["error"] == "invalid_argument" + assert "page" in payload["detail"] + + +def test_query_nms_alarms_pages_past_two_and_normalizes_severity() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + mock_http.return_value = { + "ok": True, + "data": {"items": [{"ne_id": "u-1", "ne_name": "ME{u-1}", "host_name": "H1"}]}, + } + out = call_http_tool("queryNmsAlarms", {"page": 5, "severity": "Critical", "is_cleared": "False"}) + params = mock_http.call_args[1]["params"] + assert params["page"] == 5 + assert params["severity"] == "critical" + assert params["is_cleared"] == "False" + row = json.loads(out["content"][0]["text"])["data"]["items"][0] + assert "ne_name" not in row + assert row["host_name"] == "H1" + + +def test_nested_api_failure_is_marked_error() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + mock_http.return_value = {"ok": True, "data": {"ok": False, "error": "job_not_found"}} + out = call_http_tool("getNeExecJob", {"job_id": "nope"}) + assert out.get("isError") is True + + +def test_raw_query_strips_echoed_field_lists() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + mock_http.return_value = { + "ok": True, + "data": { + "items": [], + "meta": {"available_fields": ["a"], "order_by_allowed": ["b"], "time_filter_field": "last_seen_at"}, + }, + } + out = call_http_tool("queryNmsAlarmsRaw", {"field_preset": "brief"}) + meta = json.loads(out["content"][0]["text"])["data"]["meta"] + assert meta == {"time_filter_field": "last_seen_at"} + + +def test_get_nms_ne_resolves_host_name() -> None: + calls: list[str] = [] + + def fake(method: str, path: str, params: dict | None = None) -> dict: + calls.append(path) + if path == "/v1/ume/inventory/ne": + return {"ok": True, "data": {"items": [{"ne_id": "uuid-1", "host_name": "MDN-X-AN1"}]}} + if path.endswith("/uuid-1"): + return {"ok": True, "data": {"ne_id": "uuid-1"}} + return {"ok": False, "error": "netx_http_404", "detail": "ume_ne_not_found"} + + with patch("netx_mcp.http_tools.http_json", side_effect=fake): + out = call_http_tool("getNmsNe", {"ne_id": "mdn-x-an1"}) + assert json.loads(out["content"][0]["text"])["data"]["ne_id"] == "uuid-1" + assert calls[-1] == "/v1/ume/inventory/ne/uuid-1" + + +def test_cli_targets_and_paths_use_nms_naming() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + mock_http.return_value = {"ok": True, "data": {"items": [{"source": "ume", "ume_ne_id": "u-1"}]}} + out = call_http_tool("listCliTargets", {"source": "nms"}) + row = json.loads(out["content"][0]["text"])["data"]["items"][0] + assert row == {"source": "nms", "nms_ne_id": "u-1"} + + with patch("netx_mcp.http_tools.http_post_json") as mock_post: + mock_post.return_value = {"ok": True, "data": {"paths": [{"nodes": [{"name": "A", "ume_ne_id": "u-1"}]}]}} + out = call_http_tool("findTopologyPaths", {"from_nms_ne_id": "u-1", "to_nms_ne_id": "u-2"}) + node = json.loads(out["content"][0]["text"])["data"]["paths"][0]["nodes"][0] + assert node == {"name": "A", "nms_ne_id": "u-1"} + + +def test_managed_ne_list_drops_disabled_hop_noise() -> None: + with patch("netx_mcp.http_tools.http_json") as mock_http: + mock_http.return_value = { + "ok": True, + "data": {"items": [{"id": "m1", "hop_enabled": False, "hop_vrf": "", "connect_detail": "long"}]}, + } + out = call_http_tool("listManagedNe", {"connect_status": "pass"}) + row = json.loads(out["content"][0]["text"])["data"]["items"][0] + assert row == {"id": "m1", "hop_enabled": False} diff --git a/tests/test_mcp_http.py b/tests/test_mcp_http.py index 88f5ff2..e0ac008 100644 --- a/tests/test_mcp_http.py +++ b/tests/test_mcp_http.py @@ -12,9 +12,9 @@ import pytest from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool -def test_http_mcp_tool_list_has_fourteen_tools() -> None: +def test_http_mcp_tool_list_has_fifteen_tools() -> None: names = [str(t.get("name") or "") for t in HTTP_MCP_TOOLS] - assert len(names) == 14 + assert len(names) == 15 assert "queryNmsAlarms" in names assert "queryNmsAlarmsRaw" in names assert "execManagedNe" in names @@ -92,7 +92,7 @@ def test_stdio_initialize_and_tools_list() -> None: assert "error" not in list_resp, list_resp tools = list_resp["result"]["tools"] names = {t["name"] for t in tools} - assert len(tools) == 14 + assert len(tools) == 15 assert "queryNmsAlarms" in names assert "findTopologyPaths" in names diff --git a/tests/test_topology.py b/tests/test_topology.py index eeb85bc..00de82b 100644 --- a/tests/test_topology.py +++ b/tests/test_topology.py @@ -3404,6 +3404,27 @@ Management Addresses: self.assertEqual(full["detail"], "full") self.assertIn("attrs", full["paths"][0]["nodes"][0]) + # A parallel A—B link must not spawn a duplicate 1-hop path. + svc.upsert_fabric_edge( + self.db, + a_node_id=nodes[1].id, + b_node_id=nodes[0].id, + a_port="Gi0/9", + b_port="Gi0/9", + source="manual", + ) + self.db.commit() + dedup = svc.find_fabric_paths( + self.db, + from_managed_ne_id=nes[0].id, + to_managed_ne_id=nes[1].id, + max_paths=3, + max_hops=6, + ) + self.assertEqual([p["hops"] for p in dedup["paths"]], [1, 2]) + self.assertEqual(dedup["paths"][0]["edges"][0]["parallel_links"], 2) + self.assertNotIn("parallel_links", dedup["paths"][1]["edges"][0]) + if __name__ == "__main__": unittest.main()