Clarify getManagedNe vs UME ids and humanize insufficient_scope errors.

Agents often passed UME UUIDs into getManagedNe and hit SQL without scope; aliases, richer hints, and scope error text reduce these dead ends.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-10 22:42:38 +08:00
parent 84f16fcb3c
commit dd008c1919
3 changed files with 67 additions and 12 deletions

View file

@ -252,10 +252,33 @@ def _list_managed_ne(args: dict[str, Any]) -> dict[str, Any]:
def _get_managed_ne(args: dict[str, Any]) -> dict[str, Any]: def _get_managed_ne(args: dict[str, Any]) -> dict[str, Any]:
ne_id = str(args.get("ne_id") or "").strip() ne_id = str(
args.get("ne_id") or args.get("managed_ne_id") or args.get("id") or ""
).strip()
if not ne_id: if not ne_id:
return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} return {
return http_json("GET", f"/v1/managed-ne/{ne_id}", params=None) "ok": False,
"error": "ne_id_required",
"error_code": "ne_id_required",
"hint": (
"Pass managed NE id from listManagedNe/listCliTargets (source=managed). "
"For UME inventory UUIDs use execManagedNe(ume_ne_id=...) or getUmeNe, not getManagedNe."
),
"example": {"ne_id": "<managed-ne-uuid-from-listManagedNe>"},
}
# UME 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)
if isinstance(out, dict) and out.get("ok") is False:
detail = str(out.get("detail") or out.get("error") or "")
low = detail.lower()
if "404" in low or "not_found" in low or "not found" in low or out.get("error") == "netx_http_404":
out = dict(out)
out["hint"] = (
"Managed NE not found for this ne_id. Call listManagedNe(keyword=...) or "
"listCliTargets(source=managed) first. If this is a UME ne_id, use "
"execManagedNe(ume_ne_id=...) / getUmeNe instead of getManagedNe."
)
return out
def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]:
@ -513,7 +536,11 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [
}, },
{ {
"name": "sqlQueryUme", "name": "sqlQueryUme",
"description": "Read-only SELECT on UME tables (ume_alarms_current/ume_inventory_ne); server enforces limits.", "description": (
"Read-only SELECT on UME tables (ume_alarms_current/ume_inventory_ne); server enforces limits. "
"Requires netx scope sql:query. If insufficient_scope, use aggregateUmeAlarms / "
"queryUmeAlarmsRaw / ume_alarm_xlsx_report instead."
),
"inputSchema": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
@ -543,11 +570,21 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [
}, },
{ {
"name": "getManagedNe", "name": "getManagedNe",
"description": "Get single managed NE metadata (connect_status, hop config summary).", "description": (
"Get one **managed** NE by managed ne_id (from listManagedNe / listCliTargets source=managed). "
"Do NOT pass UME inventory UUID here — use getUmeNe or execManagedNe(ume_ne_id=...) instead."
),
"inputSchema": { "inputSchema": {
"type": "object", "type": "object",
"properties": {"ne_id": {"type": "string"}}, "properties": {
"required": ["ne_id"], "ne_id": {
"type": "string",
"description": "Managed NE id (not UME host_name / not UME ne_id unless they coincide).",
},
"managed_ne_id": {"type": "string", "description": "Alias for ne_id."},
"id": {"type": "string", "description": "Alias for ne_id."},
},
"required": [],
"additionalProperties": False, "additionalProperties": False,
}, },
}, },

View file

@ -105,7 +105,15 @@ def run_stdio_loop() -> None:
need = TOOL_REQUIRED_SCOPE.get(name) need = TOOL_REQUIRED_SCOPE.get(name)
granted = scopes() granted = scopes()
if need and granted is not None and need not in {str(s).lower() for s in granted}: if need and granted is not None and need not in {str(s).lower() for s in granted}:
_err(rid, -32001, f"insufficient_scope:{need}") _err(
rid,
-32001,
(
f"insufficient_scope:{need}. "
"Ask a netx admin to grant this scope on the API token; "
"for sql:query prefer aggregateUmeAlarms/queryUmeAlarmsRaw/ume_alarm_xlsx_report instead."
),
)
continue continue
args = params.get("arguments") if isinstance(params.get("arguments"), dict) else {} args = params.get("arguments") if isinstance(params.get("arguments"), dict) else {}
_ok(rid, call_http_tool(name, args)) _ok(rid, call_http_tool(name, args))

View file

@ -114,12 +114,22 @@ def test_call_exec_managed_ne_defaults_read_timeout() -> None:
assert payload["ok"] is True assert payload["ok"] is True
def test_call_get_ume_ne_requires_id() -> None: def test_get_managed_ne_requires_id_with_hint() -> None:
out = call_http_tool("getUmeNe", {}) out = call_http_tool("getManagedNe", {})
assert out.get("isError") is True assert out.get("isError") is True
text = out["content"][0]["text"] payload = json.loads(out["content"][0]["text"])
payload = json.loads(text)
assert payload["error"] == "ne_id_required" assert payload["error"] == "ne_id_required"
assert "hint" in payload
def test_get_managed_ne_accepts_managed_ne_id_alias() -> None:
with patch("netx_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"ne_id": "m1"}}
out = call_http_tool("getManagedNe", {"managed_ne_id": "m1"})
mock_http.assert_called_once()
assert mock_http.call_args[0][1].endswith("/m1")
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
def test_call_exec_managed_ne_posts_body() -> None: def test_call_exec_managed_ne_posts_body() -> None: