Make tool failures actionable with suggestions and scope hints.

Unregistered tools now suggest close matches; MCP insufficient_scope is rewritten with fallbacks so ops agents stop blind-retrying SQL without sql:query.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-10 22:42:37 +08:00
parent 229e8b1368
commit d245e463c0
6 changed files with 159 additions and 4 deletions

View file

@ -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:

View file

@ -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

View file

@ -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)

View file

@ -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",
]

View file

@ -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=...)`。
## 输出约定

View file

@ -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"]