diff --git a/runtime/chat/agent.py b/runtime/chat/agent.py index 3119970f..165f9705 100644 --- a/runtime/chat/agent.py +++ b/runtime/chat/agent.py @@ -95,8 +95,14 @@ class Agent: t0 = time.perf_counter() tool = self.tools.get(tc.name) if not tool: - msg = f"Unregistered tool: {tc.name}" if self.lang.startswith("en") else f"未注册的工具: {tc.name}" - return {"ok": False, "error": msg}, int((time.perf_counter() - t0) * 1000) + from runtime.tools.tool_error_hints import format_unregistered_tool_error + + available = [t.name for t in self.tools.list()] if hasattr(self.tools, "list") else [] + return format_unregistered_tool_error( + str(tc.name or ""), + available, + lang=str(self.lang or "zh"), + ), int((time.perf_counter() - t0) * 1000) ok, v_err = validate_tool_arguments(tool.parameters, tc.arguments) if not ok: diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index c7eae5d7..7d09b318 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -534,8 +534,14 @@ class ToolExecutor: tool = ctx.tools.get(tc.name) if not tool: - msg = f"Unregistered tool: {tc.name}" if ctx.lang.startswith("en") else f"未注册的工具: {tc.name}" - return {"ok": False, "error_code": "tool_not_registered", "error": msg}, int((time.perf_counter() - t0) * 1000) + from runtime.tools.tool_error_hints import format_unregistered_tool_error + + available = [t.name for t in ctx.tools.list()] if hasattr(ctx.tools, "list") else [] + return format_unregistered_tool_error( + str(tc.name or ""), + available, + lang=str(ctx.lang or "zh"), + ), int((time.perf_counter() - t0) * 1000) from runtime.tools.context_inject import enrich_tool_arguments from runtime.tools.tool_validation import filter_arguments_to_schema diff --git a/runtime/tools/mcp/adapter.py b/runtime/tools/mcp/adapter.py index 696185f8..4ca96e6e 100644 --- a/runtime/tools/mcp/adapter.py +++ b/runtime/tools/mcp/adapter.py @@ -127,6 +127,9 @@ class _McpBoundTool: return {"ok": False, "error_code": "mcp_runtime_invalid_payload", "error": "invalid_response"} if "ok" not in res: res["ok"] = False + from runtime.tools.tool_error_hints import enrich_mcp_scope_error + + res = enrich_mcp_scope_error(res) if tool_name == "listCliTargets" and res.get("ok") is not False and cache_key: _set_list_cli_cache(cache_key, res) res = dict(res) diff --git a/runtime/tools/tool_error_hints.py b/runtime/tools/tool_error_hints.py new file mode 100644 index 00000000..920dc9e3 --- /dev/null +++ b/runtime/tools/tool_error_hints.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import difflib +from typing import Any, Iterable + + +def suggest_tool_names(requested: str, available: Iterable[str], *, n: int = 5) -> list[str]: + """Suggest close tool names for unregistered-tool errors.""" + name = str(requested or "").strip() + pool = [str(x).strip() for x in available if str(x).strip()] + if not name or not pool: + return [] + # Prefer substring matches (mcp__netx__foo vs netx__foo / queryUmeAlarmsRaw). + low = name.lower() + substr = [p for p in pool if low in p.lower() or p.lower() in low] + close = difflib.get_close_matches(name, pool, n=max(n, 8), cutoff=0.45) + out: list[str] = [] + for x in substr + close: + if x not in out: + out.append(x) + if len(out) >= n: + break + return out + + +def format_unregistered_tool_error( + tool_name: str, + available: Iterable[str], + *, + lang: str = "zh", +) -> dict[str, Any]: + suggestions = suggest_tool_names(tool_name, available) + if str(lang or "").startswith("en"): + err = f"Unregistered tool: {tool_name}" + hint = ( + "Use one of suggestions, or refresh MCP tools for this specialist." + if suggestions + else "Tool is not in the current registry; check MCP enablement / specialist binding." + ) + else: + err = f"未注册的工具: {tool_name}" + hint = ( + "请改用 suggestions 中的工具名，或检查该专家的 MCP 绑定/启用状态。" + if suggestions + else "当前工具目录中无此工具；请检查 MCP 是否启用及专家绑定。" + ) + out: dict[str, Any] = { + "ok": False, + "error_code": "tool_not_registered", + "error": err, + "hint": hint, + } + if suggestions: + out["suggestions"] = suggestions + return out + + +def enrich_mcp_scope_error(result: dict[str, Any]) -> dict[str, Any]: + """Rewrite insufficient_scope MCP errors into actionable ops guidance.""" + if not isinstance(result, dict) or result.get("ok") is not False: + return result + err = str(result.get("error") or "") + code = str(result.get("error_code") or "") + blob = f"{err} {code}".lower() + if "insufficient_scope" not in blob and "insufficient_scope" not in err: + return result + scope = "" + marker = "insufficient_scope:" + if marker in err: + scope = err.split(marker, 1)[1].strip().split()[0].strip("\"'") + elif marker in code: + scope = code.split(marker, 1)[1].strip() + out = dict(result) + out["error_code"] = "insufficient_scope" + if scope: + out["required_scope"] = scope + out["error"] = f"insufficient_scope:{scope}" if scope else "insufficient_scope" + if scope == "sql:query": + out["hint"] = ( + "Current netx token lacks sql:query. Prefer aggregateUmeAlarms / queryUmeAlarmsRaw / " + "ume_alarm_xlsx_report; ask an admin to grant sql:query only if SQL is required." + ) + out["fallback_tools"] = [ + "mcp__netx__aggregateUmeAlarms", + "mcp__netx__queryUmeAlarmsRaw", + "ume_alarm_xlsx_report", + ] + else: + out["hint"] = ( + f"Current netx token lacks scope {scope or '(unknown)'}. " + "Ask an admin to grant it on the netx API token, or use tools that do not need this scope." + ) + return out + + +__all__ = [ + "enrich_mcp_scope_error", + "format_unregistered_tool_error", + "suggest_tool_names", +] diff --git a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md index d17f1e8c..ac493e15 100644 --- a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md @@ -78,7 +78,9 @@ description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查 - 体积：`page_size` 默认 50；聚合 `top_ne` 默认 50；raw 用 preset；动态聚合 `limit≤200`。 - **Top 网元**：默认忽略 `(host_name missing)`；结论中说明 missing 数量，勿把 UUID/`unknown`/空串当网元名。 - SQL：建议 `statement_timeout_ms=8000`；非 `count(*)` 应带过滤；时间窗相对 **数据新鲜度**，不是盲目 `now()`。 + - 若返回 `insufficient_scope:sql:query`：改用 `aggregateUmeAlarms` / `queryUmeAlarmsRaw` / `ume_alarm_xlsx_report`，勿盲重试 SQL。 - `WITH` CTE 可用；**禁止** `WITH RECURSIVE`。 +- `getManagedNe` 只要 **纳管 ne_id**（来自 listManagedNe）；UME UUID 用 `getUmeNe` / `execManagedNe(ume_ne_id=...)`。 ## 输出约定 diff --git a/tests/test_tool_error_hints.py b/tests/test_tool_error_hints.py new file mode 100644 index 00000000..c1060422 --- /dev/null +++ b/tests/test_tool_error_hints.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from runtime.tools.tool_error_hints import ( + enrich_mcp_scope_error, + format_unregistered_tool_error, + suggest_tool_names, +) + + +def test_suggest_tool_names_prefers_substring() -> None: + pool = [ + "mcp__netx__queryUmeAlarms", + "mcp__netx__queryUmeAlarmsRaw", + "mcp__netx__aggregateUmeAlarms", + "write_xlsx", + ] + hits = suggest_tool_names("mcp__netx__queryUmeAlarmsRaw", pool) + assert "mcp__netx__queryUmeAlarmsRaw" in hits or "mcp__netx__queryUmeAlarms" in hits + + +def test_format_unregistered_includes_suggestions() -> None: + out = format_unregistered_tool_error( + "mcp__netx__queryUmeAlarmsRaw", + ["mcp__netx__queryUmeAlarms", "mcp__netx__aggregateUmeAlarms", "ume_alarm_xlsx_report"], + lang="en", + ) + assert out["error_code"] == "tool_not_registered" + assert out.get("suggestions") + assert any("queryUmeAlarms" in s for s in out["suggestions"]) + + +def test_enrich_mcp_scope_sql() -> None: + raw = {"ok": False, "error_code": "mcp_rpc_error_-32001", "error": "insufficient_scope:sql:query"} + out = enrich_mcp_scope_error(raw) + assert out["error_code"] == "insufficient_scope" + assert out["required_scope"] == "sql:query" + assert "fallback_tools" in out + assert "ume_alarm_xlsx_report" in out["fallback_tools"]