From 8f9a64a266a707096be99188165c80486cd41d17 Mon Sep 17 00:00:00 2001 From: oliver Date: Fri, 16 Oct 2026 23:37:16 +0800 Subject: [PATCH] feat(mcp): tell skills which tools the token cannot call; cache skill descriptions Also make the HTTP-error test independent of the developer .env token. Co-Authored-By: Claude Opus 5.5 --- docs/MCP.md | 4 +- netx_api/mcp_remote/server.py | 2 +- netx_api/mcp_remote/skills.py | 78 ++++++++++++++++++++++++----- tests/mcp_remote/test_mcp_http.py | 2 + tests/mcp_remote/test_mcp_skills.py | 30 +++++++++++ 5 files changed, 101 insertions(+), 15 deletions(-) diff --git a/docs/MCP.md b/docs/MCP.md index 14110f2..2cd4b66 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -69,7 +69,9 @@ http://127.0.0.1:8890/mcp?groups=ops | `prompts/list` · `prompts/get` | 每个可见 skill 一个 prompt（名=skill 名）；Claude Code 显示为 `/mcp__netx__` | | `initialize.instructions` | 在原有说明后追加当前可见 skill 目录，并提示先调 `getNetxSkill` | -可见规则：该组下当前 token 至少有一个可见工具，且遵守 `?groups=`。改 skill 文件后无需重启 netx（每次请求读盘）。 +可见规则：该组下当前 token 至少有一个可见工具，且遵守 `?groups=`。改 skill 文件后无需重启 netx（每次请求读盘，description 按文件修改时间缓存）。 + +skill 原文不随 token 改写。当 token 只有该组部分工具的权限时：`getNetxSkill` 和 prompts 会在末尾附上"当前 token 无权使用以下工具"清单；`resources/list` 里 SKILL.md 条目的 `_meta.netx.unavailableTools` 也给出同一份清单（DSH 插件据此在 skill 末尾追加"不可用工具"提示）。 ### Cursor / Claude Code 怎么读 diff --git a/netx_api/mcp_remote/server.py b/netx_api/mcp_remote/server.py index cdff91a..4693a68 100644 --- a/netx_api/mcp_remote/server.py +++ b/netx_api/mcp_remote/server.py @@ -131,7 +131,7 @@ async def _on_list_resources( scopes, groups = _scopes_and_groups() resources: list[Resource] = [] for sf in list_visible_skill_files(scopes, groups): - d = resource_dict(sf) + d = resource_dict(sf, scopes=scopes) resources.append( Resource( uri=d["uri"], diff --git a/netx_api/mcp_remote/skills.py b/netx_api/mcp_remote/skills.py index ac9666c..47b2f18 100644 --- a/netx_api/mcp_remote/skills.py +++ b/netx_api/mcp_remote/skills.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import Any from .registry import ( + ALL_TOOLS, GROUP_BIZ, GROUP_OPS, GROUP_TOPOLOGY, @@ -74,6 +75,26 @@ def _is_text_file(path: Path) -> bool: return path.is_file() and path.suffix.lower() in _TEXT_SUFFIXES +# abs_path -> (mtime_ns, size, description); SKILL.md is re-parsed only when it changes. +_DESC_CACHE: dict[Path, tuple[int, int, str]] = {} + + +def _cached_description(skill_file: "SkillFile") -> str: + try: + st = skill_file.abs_path.stat() + except OSError: + return "" + hit = _DESC_CACHE.get(skill_file.abs_path) + if hit and hit[0] == st.st_mtime_ns and hit[1] == st.st_size: + return hit[2] + try: + desc = _parse_description(read_skill_text(skill_file)) + except OSError: + desc = "" + _DESC_CACHE[skill_file.abs_path] = (st.st_mtime_ns, st.st_size, desc) + return desc + + def _parse_description(text: str) -> str: m = _FRONTMATTER_RE.match(text) if not m: @@ -170,12 +191,7 @@ def list_visible_skills( key = (f.group, f.skill) if key in by_key and not f.entry: continue - desc = "" - if f.entry: - try: - desc = _parse_description(read_skill_text(f)) - except OSError: - desc = "" + desc = _cached_description(f) if f.entry else "" by_key[key] = (f.skill, f.group, desc) order = {GROUP_OPS: 0, GROUP_TOPOLOGY: 1, GROUP_BIZ: 2} return sorted(by_key.values(), key=lambda x: (order.get(x[1], 9), x[0].lower())) @@ -219,13 +235,39 @@ def resolve_skill_file( return None -def resource_dict(skill_file: SkillFile, *, description: str | None = None) -> dict[str, Any]: +def unavailable_tools( + group: str, + scopes: frozenset[str] | list[str] | None, +) -> list[str]: + """Tools of ``group`` the skill may mention but this token cannot call.""" + granted = {str(t["name"]) for t in capability_tools_for_request(scopes, frozenset({group}))} + return [ + str(t["name"]) + for t in ALL_TOOLS + if t.get("_group") == group and str(t["name"]) not in granted + ] + + +def unavailable_note(group: str, scopes: frozenset[str] | list[str] | None) -> str: + names = unavailable_tools(group, scopes) + if not names: + return "" + return ( + "\n---\n当前 token 无权使用以下工具，本 skill 中涉及它们的步骤请跳过，不要调用：" + + ", ".join(names) + + "\n" + ) + + +def resource_dict( + skill_file: SkillFile, + *, + description: str | None = None, + scopes: frozenset[str] | list[str] | None = None, +) -> dict[str, Any]: desc = description if desc is None and skill_file.entry: - try: - desc = _parse_description(read_skill_text(skill_file)) - except OSError: - desc = "" + desc = _cached_description(skill_file) out: dict[str, Any] = { "uri": skill_file.uri, "name": skill_file.rel_path, @@ -235,6 +277,11 @@ def resource_dict(skill_file: SkillFile, *, description: str | None = None) -> d "group": skill_file.group, "skill": skill_file.skill, "entry": skill_file.entry, + **( + {"unavailableTools": unavailable_tools(skill_file.group, scopes)} + if skill_file.entry + else {} + ), } }, } @@ -315,7 +362,7 @@ def should_advertise_get_netx_skill( *, root: Path | None = None, ) -> bool: - return bool(list_visible_skills(scopes, groups, root=root)) + return any(f.entry for f in list_visible_skill_files(scopes, groups, root=root)) def available_skill_names( @@ -371,6 +418,7 @@ def call_get_netx_skill( text = text.rstrip() + "\n\n---\n其他文件：\n" + "\n".join(f"- {s}" for s in siblings) + "\n" else: text = text.rstrip() + "\n\n---\n其他文件：（无）\n" + text += unavailable_note(skill_file.group, scopes) return {"content": [{"type": "text", "text": text}], "isError": False} @@ -388,4 +436,8 @@ def prompt_skill_md_text( body = read_skill_text(skill_file) except OSError: return None - return body.rstrip() + "\n\n其他文件用 getNetxSkill 读取。\n" + return ( + body.rstrip() + + "\n\n其他文件用 getNetxSkill 读取。\n" + + unavailable_note(skill_file.group, scopes) + ) diff --git a/tests/mcp_remote/test_mcp_http.py b/tests/mcp_remote/test_mcp_http.py index 1315ea2..7d43e0a 100644 --- a/tests/mcp_remote/test_mcp_http.py +++ b/tests/mcp_remote/test_mcp_http.py @@ -463,6 +463,8 @@ def test_http_errors_preserve_status_and_bound_detail(method: str, status: int) client = httpx.Client(transport=transport, base_url="http://example.test") with ( patch("netx_api.mcp_remote.asgi_client.get_app", return_value=None), + # Independent of the developer's .env: the fallback path needs an outbound token. + patch("netx_api.mcp_remote.asgi_client.api_token", return_value="test-token"), patch("netx_api.mcp_remote.asgi_client.httpx.Client", return_value=client), ): result = http_json("GET", "/v1/ume/alarms") if method == "GET" else http_post_json("/v1/managed-ne/exec", {}) diff --git a/tests/mcp_remote/test_mcp_skills.py b/tests/mcp_remote/test_mcp_skills.py index 18ad60a..4757436 100644 --- a/tests/mcp_remote/test_mcp_skills.py +++ b/tests/mcp_remote/test_mcp_skills.py @@ -116,6 +116,7 @@ def test_resources_list_and_read_via_http(): "group": "ops", "skill": "netx-ops", "entry": True, + "unavailableTools": [], } ref = next(r for r in resources if r["uri"].endswith("/reference.md")) assert "description" not in ref or not ref.get("description") @@ -471,3 +472,32 @@ def test_mcp_tcp_resources_list(live_skills_port: int): uris = {x["uri"] for x in data["result"]["resources"]} assert "netx://skills/ops/netx-ops/SKILL.md" in uris assert all("biz-state" not in u for u in uris) + + +def test_skill_lists_tools_the_token_cannot_call(): + from netx_api.mcp_remote.skills import ( + call_get_netx_skill, + prompt_skill_md_text, + resource_dict, + unavailable_tools, + ) + + limited = frozenset({"ne:read"}) + missing = unavailable_tools("ops", limited) + assert "execManagedNe" in missing + assert "listManagedNe" not in missing + assert unavailable_tools("ops", None) == [] + + out = call_get_netx_skill({"name": "netx-ops"}, limited) + text = out["content"][0]["text"] + assert out["isError"] is False + assert "当前 token 无权使用以下工具" in text and "execManagedNe" in text + + assert "execManagedNe" in (prompt_skill_md_text("netx-ops", limited) or "") + + entry = next(f for f in list_visible_skill_files(limited) if f.skill == "netx-ops" and f.entry) + meta = resource_dict(entry, scopes=limited)["_meta"]["netx"] + assert "execManagedNe" in meta["unavailableTools"] + + full = call_get_netx_skill({"name": "netx-ops"}, None)["content"][0]["text"] + assert "当前 token 无权使用以下工具" not in full