From 229e8b1368f323a8cccd483d3a356a4588caff0a Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 10 Aug 2026 22:40:17 +0800 Subject: [PATCH] Raise MCP exec timeouts and cache listCliTargets for WhatsApp ops. Production logs showed execManagedNe glued to 30s wall-clock kills; override long tools, TTL-cache CLI target lists, and return schema examples on invalid args. Co-authored-by: Cursor --- runtime/chat/agent.py | 9 +- runtime/chat/tool_runtime.py | 36 ++++-- runtime/tools/mcp/adapter.py | 109 ++++++++++++++++- runtime/tools/tool_validation.py | 67 +++++++++- runtime/workspaces/ops/ROLE_SYSTEM.en.md | 4 +- runtime/workspaces/ops/ROLE_SYSTEM.md | 4 +- .../ops/ops-netx-managed-ne-playbook/SKILL.md | 2 +- tests/test_mcp_timeout_and_cache.py | 114 ++++++++++++++++++ 8 files changed, 325 insertions(+), 20 deletions(-) create mode 100644 tests/test_mcp_timeout_and_cache.py diff --git a/runtime/chat/agent.py b/runtime/chat/agent.py index 9cff7209..3119970f 100644 --- a/runtime/chat/agent.py +++ b/runtime/chat/agent.py @@ -23,7 +23,7 @@ from svc.llm.chat_models import ( gemini_openai_compat_client, ) from runtime.prompt_templates import render_prompt_for_lang -from runtime.tools.tool_validation import validate_tool_arguments +from runtime.tools.tool_validation import format_invalid_arguments_error, validate_tool_arguments logger = logging.getLogger(__name__) @@ -100,8 +100,11 @@ class Agent: ok, v_err = validate_tool_arguments(tool.parameters, tc.arguments) if not ok: - msg = f"Invalid arguments: {v_err}" if self.lang.startswith("en") else f"参数不合法: {v_err}" - return {"ok": False, "error": msg}, int((time.perf_counter() - t0) * 1000) + return format_invalid_arguments_error( + tool.parameters or {}, + str(v_err or "invalid"), + lang=str(self.lang or "zh"), + ), int((time.perf_counter() - t0) * 1000) try: result = tool.handler(tc.arguments) diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index 8f7f5206..c7eae5d7 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -20,7 +20,7 @@ from typing import Any, Callable, Optional from svc.persistence.sqlite_store import SqliteStore from runtime.tools.base import ToolRegistry from svc.llm.chat_models import LLMToolCall -from runtime.tools.tool_validation import validate_tool_arguments +from runtime.tools.tool_validation import format_invalid_arguments_error, validate_tool_arguments from runtime.chat.media_redact import ingest_embedded_image_blobs_as_refs from runtime.tools.path_guard import ( workspace_path_access_scope, @@ -641,8 +641,11 @@ class ToolExecutor: ok, v_err = validate_tool_arguments(tool.parameters, tool_args) if not ok: - msg = f"Invalid arguments: {v_err}" if ctx.lang.startswith("en") else f"参数不合法: {v_err}" - return {"ok": False, "error_code": "tool_invalid_arguments", "error": msg}, int((time.perf_counter() - t0) * 1000) + return format_invalid_arguments_error( + tool.parameters or {}, + str(v_err or "invalid"), + lang=str(ctx.lang or "zh"), + ), int((time.perf_counter() - t0) * 1000) try: timeout_s = getattr(tool, "timeout_s", None) @@ -697,17 +700,32 @@ class ToolExecutor: ex.shutdown(wait=False, cancel_futures=True) except Exception: ex.shutdown(wait=False) - return {"ok": False, "error_code": "tool_timeout_or_failed", "error": "tool_timeout_or_failed", "detail": f"{type(e).__name__}: {e}"}, int( - (time.perf_counter() - t0) * 1000 - ) + return { + "ok": False, + "error_code": "tool_timeout_or_failed", + "error": "tool_timeout_or_failed", + "detail": f"{type(e).__name__}: {e}", + "timeout_s": float(timeout_s), + "tool": str(tc.name or ""), + "hint": ( + "Tool wall-clock timeout. For mcp__netx__execManagedNe raise read_timeout_sec " + "and/or reduce commands; do not blind-retry the same call." + if str(ctx.lang or "").startswith("en") + else "工具墙钟超时。对 mcp__netx__execManagedNe 请提高 read_timeout_sec 或减少命令条数，禁止相同参数盲重试。" + ), + }, int((time.perf_counter() - t0) * 1000) except Exception as e: try: ex.shutdown(wait=False, cancel_futures=True) except Exception: ex.shutdown(wait=False) - return {"ok": False, "error_code": "tool_timeout_or_failed", "error": "tool_timeout_or_failed", "detail": f"{type(e).__name__}: {e}"}, int( - (time.perf_counter() - t0) * 1000 - ) + return { + "ok": False, + "error_code": "tool_timeout_or_failed", + "error": "tool_timeout_or_failed", + "detail": f"{type(e).__name__}: {e}", + "tool": str(tc.name or ""), + }, int((time.perf_counter() - t0) * 1000) else: try: ex.shutdown(wait=False, cancel_futures=True) diff --git a/runtime/tools/mcp/adapter.py b/runtime/tools/mcp/adapter.py index 9e336bad..696185f8 100644 --- a/runtime/tools/mcp/adapter.py +++ b/runtime/tools/mcp/adapter.py @@ -3,6 +3,8 @@ from __future__ import annotations from dataclasses import dataclass import json import os +import threading +import time from typing import Any from runtime.tools.mcp.env_config import mcp_row_env_config @@ -17,6 +19,71 @@ def _mcp_row_env_config(row: dict[str, Any]) -> tuple[list[str], dict[str, str]] return mcp_row_env_config(row) +# Long-running netx tools exceed the generic MCP row timeout (often 30s). +# Production WA ops showed execManagedNe p90/p95 glued to ~30000ms timeouts. +_MCP_TOOL_TIMEOUT_OVERRIDES_S: dict[str, float] = { + "execManagedNe": 320.0, + "sqlQueryUme": 90.0, + "findTopologyPaths": 60.0, + "aggregateUmeAlarmsRaw": 60.0, + "queryUmeAlarmsRaw": 60.0, + "aggregateUmeAlarms": 60.0, +} + +_LIST_CLI_CACHE_TTL_S = 120.0 +_LIST_CLI_CACHE_LOCK = threading.Lock() +_LIST_CLI_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} + + +def mcp_timeout_for_tool(tool_name: str, row_timeout_s: float | None = None) -> float: + """Resolve effective MCP tool wall-clock timeout (oclaw-side).""" + name = str(tool_name or "").strip() + override = _MCP_TOOL_TIMEOUT_OVERRIDES_S.get(name) + base = float(row_timeout_s) if row_timeout_s is not None else 30.0 + if override is not None: + return max(base, float(override)) + return max(5.0, base) + + +def _list_cli_cache_key(server_id: str, args: dict[str, Any]) -> str: + payload = { + "server_id": server_id, + "source": str(args.get("source") or "all"), + "keyword": str(args.get("keyword") or ""), + "page": int(args.get("page") or 1), + "page_size": int(args.get("page_size") or 50), + } + return json.dumps(payload, sort_keys=True, ensure_ascii=False) + + +def _get_list_cli_cache(key: str) -> dict[str, Any] | None: + now = time.monotonic() + with _LIST_CLI_CACHE_LOCK: + hit = _LIST_CLI_CACHE.get(key) + if not hit: + return None + ts, payload = hit + if now - ts > _LIST_CLI_CACHE_TTL_S: + _LIST_CLI_CACHE.pop(key, None) + return None + return dict(payload) + + +def _set_list_cli_cache(key: str, payload: dict[str, Any]) -> None: + with _LIST_CLI_CACHE_LOCK: + # Bound memory: drop oldest when large. + if len(_LIST_CLI_CACHE) >= 64: + oldest = sorted(_LIST_CLI_CACHE.items(), key=lambda kv: kv[1][0])[:16] + for k, _ in oldest: + _LIST_CLI_CACHE.pop(k, None) + _LIST_CLI_CACHE[key] = (time.monotonic(), dict(payload)) + + +def clear_list_cli_targets_cache() -> None: + with _LIST_CLI_CACHE_LOCK: + _LIST_CLI_CACHE.clear() + + @dataclass class _McpBoundTool: server_id: str @@ -36,13 +103,47 @@ class _McpBoundTool: env_allowlist=self.env_allowlist, env_defaults=self.env_defaults, ) + tool_name = self.tool_name + server_id = self.server_id def _handler(args: dict[str, Any]) -> dict[str, Any]: - res = rt.call_tool(tool_name=self.tool_name, arguments=args or {}) + call_args = dict(args or {}) + cache_key = "" + if tool_name == "listCliTargets": + cache_key = _list_cli_cache_key(server_id, call_args) + cached = _get_list_cli_cache(cache_key) + if cached is not None: + out = dict(cached) + out["cache_hit"] = True + out["cache_ttl_s"] = _LIST_CLI_CACHE_TTL_S + out["hint"] = ( + out.get("hint") + or "Reused listCliTargets result from short TTL cache; do not re-list before every execManagedNe." + ) + return out + + res = rt.call_tool(tool_name=tool_name, arguments=call_args) if not isinstance(res, dict): return {"ok": False, "error_code": "mcp_runtime_invalid_payload", "error": "invalid_response"} if "ok" not in res: res["ok"] = False + if tool_name == "listCliTargets" and res.get("ok") is not False and cache_key: + _set_list_cli_cache(cache_key, res) + res = dict(res) + res["cache_hit"] = False + res["hint"] = ( + "Cache listCliTargets ids for this session; call execManagedNe with ne_id/ume_ne_id " + "instead of listing again." + ) + if tool_name == "execManagedNe" and res.get("ok") is False: + err = str(res.get("error") or res.get("error_code") or "") + low = err.lower() + if "timeout" in low or res.get("error_code") == "tool_timeout_or_failed": + res = dict(res) + res["hint"] = ( + "CLI timed out. Raise read_timeout_sec (60–120), reduce commands, " + "or reuse prior listCliTargets ids — do not blind-retry identical calls." + ) return res return ToolSpec( @@ -156,7 +257,7 @@ def materialize_mcp_tools_for_specialist( tags=frozenset({"mcp", "plugin", "compat"}), version="v1", risk_level="high", - timeout_s=float(row.get("timeout_s") or 30.0), + timeout_s=mcp_timeout_for_tool(tname, float(row.get("timeout_s") or 30.0)), required_permissions=frozenset(str(x) for x in (row.get("required_permissions") or [])), execution_mode="subprocess", ) @@ -168,7 +269,7 @@ def materialize_mcp_tools_for_specialist( description=str(t.get("description") or f"MCP tool {t.get('tool_name') or ''}"), parameters=t.get("parameters") if isinstance(t.get("parameters"), dict) else {}, command=command, - timeout_s=float(row.get("timeout_s") or 30.0), + timeout_s=mcp_timeout_for_tool(tname, float(row.get("timeout_s") or 30.0)), required_permissions=frozenset(str(x) for x in (row.get("required_permissions") or [])), env_allowlist=env_allowlist, env_defaults=env_defaults, @@ -196,8 +297,10 @@ def materialize_mcp_skills_for_specialist( __all__ = [ + "clear_list_cli_targets_cache", "materialize_mcp_tools", "materialize_mcp_tools_for_specialist", "materialize_mcp_skills_for_specialist", + "mcp_timeout_for_tool", ] diff --git a/runtime/tools/tool_validation.py b/runtime/tools/tool_validation.py index be258ae9..34b8d210 100644 --- a/runtime/tools/tool_validation.py +++ b/runtime/tools/tool_validation.py @@ -1,5 +1,6 @@ from __future__ import annotations +import time from typing import Any try: @@ -21,6 +22,65 @@ def filter_arguments_to_schema(parameters: dict[str, Any], arguments: dict[str, return dict(arguments) +def _example_from_schema(parameters: dict[str, Any]) -> dict[str, Any]: + """Build a minimal example object from JSON Schema properties/required.""" + props = parameters.get("properties") if isinstance(parameters.get("properties"), dict) else {} + required = parameters.get("required") if isinstance(parameters.get("required"), list) else [] + keys = [str(k) for k in required if str(k) in props] + if not keys: + # Prefer a few representative optional keys so agents see the shape. + keys = list(props.keys())[:4] + example: dict[str, Any] = {} + for key in keys: + spec = props.get(key) if isinstance(props.get(key), dict) else {} + if "default" in spec: + example[key] = spec["default"] + continue + t = spec.get("type") + if t == "string" or (isinstance(t, list) and "string" in t): + enum = spec.get("enum") + example[key] = enum[0] if isinstance(enum, list) and enum else f"<{key}>" + elif t == "integer" or t == "number": + example[key] = int(spec.get("minimum") or 1) + elif t == "boolean": + example[key] = bool(spec.get("default") if "default" in spec else True) + elif t == "array": + example[key] = [] + elif t == "object": + example[key] = {} + else: + example[key] = f"<{key}>" + return example + + +def format_invalid_arguments_error( + parameters: dict[str, Any], + message: str, + *, + lang: str = "zh", +) -> dict[str, Any]: + """Rich invalid-arg payload so the model can self-correct without blind retries.""" + props = parameters.get("properties") if isinstance(parameters.get("properties"), dict) else {} + required = [str(x) for x in (parameters.get("required") or []) if str(x)] + example = _example_from_schema(parameters or {}) + if str(lang or "").startswith("en"): + err = f"Invalid arguments: {message}" + hint = "Fix arguments to match the schema example; do not retry with the same payload." + else: + err = f"参数不合法: {message}" + hint = "请按 example 修正参数后重试，不要用相同参数盲目重试。" + return { + "ok": False, + "error_code": "tool_invalid_arguments", + "error": err, + "validation_message": message, + "required": required, + "properties": sorted(str(k) for k in props.keys()), + "example": example, + "hint": hint, + } + + def validate_tool_arguments(parameters: dict[str, Any], arguments: dict[str, Any]) -> tuple[bool, str | None]: """校验模型给出的 arguments 是否符合工具的 JSON Schema（OpenAI function parameters）。""" if not isinstance(arguments, dict): @@ -44,5 +104,8 @@ def validate_tool_arguments(parameters: dict[str, Any], arguments: dict[str, Any return True, None -__all__ = ["filter_arguments_to_schema", "validate_tool_arguments"] - +__all__ = [ + "filter_arguments_to_schema", + "format_invalid_arguments_error", + "validate_tool_arguments", +] diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index 37971054..53643dcd 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -36,7 +36,9 @@ You are the ops specialist (network operations expert). ## WhatsApp interaction (mandatory) - Short ops intents follow `ops-netx-ume-playbook` WhatsApp recipes; target **≤3 tool calls** per user message. For Excel exports prefer `ume_alarm_xlsx_report`. - Spreadsheet delivery: `ume_alarm_xlsx_report` or `write_xlsx(deliverable=true)` — never claim a file was sent without deliverable marking. -- Call `listCliTargets` at most once per session and reuse ids; batch `execManagedNe` commands; on timeout raise `read_timeout_sec` — no blind retries. +- Call `listCliTargets` at most once per session and reuse ids; batch `execManagedNe` commands; default `read_timeout_sec=60` — on timeout raise it, no blind retries. +- Replies like `YES` / `confirm` / `继续` / `please continue`: continue the previous unfinished task — do **not** re-ask for confirmation or restart the query. +- On `tool_invalid_arguments`, fix args using the returned `example`; on timeout hints, raise `read_timeout_sec` or shrink commands. ## Required skills - For every netx/UME **alarm or NE** request, load and follow skill: `ops-netx-ume-playbook` (skill text may be Chinese; **user-facing output must still match the user's language**). diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index baa8aeab..028d0653 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -27,7 +27,9 @@ ## WhatsApp 交互（强制） - 短句优先走 `ops-netx-ume-playbook` 的「WhatsApp 短指令配方」，控制在 ≤3 次工具调用；要 Excel 时优先 `ume_alarm_xlsx_report`。 - 用户要表格/Excel：`ume_alarm_xlsx_report` 或 `write_xlsx(deliverable=true)`；禁止只写文件不投递。 -- `listCliTargets` 每会话最多查一次并复用 id；`execManagedNe` 合并 commands，超时调 `read_timeout_sec`，禁止盲重试。 +- `listCliTargets` 每会话最多查一次并复用 id；`execManagedNe` 合并 commands，超时调 `read_timeout_sec`（默认 60），禁止盲重试。 +- 用户回复 `YES` / `confirm` / `确认` / `可以` / `继续` / `please continue`：直接承接上一未完成任务继续执行，**不要**再问一遍确认或重开查询。 +- 工具返回 `tool_invalid_arguments` 时按返回的 `example` 修正参数；返回超时 hint 时提高 `read_timeout_sec` 或减命令，禁止相同参数重试。 ## 必须加载技能 - 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。 diff --git a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md index ae7ced64..555f48ba 100644 --- a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md @@ -22,7 +22,7 @@ description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手 2. **登录查信息** - `mcp__netx__execManagedNe`：`ne_id` **或** `ume_ne_id` + `commands`（默认最多 5 条，可由 `NETX_NE_EXEC_MAX_COMMANDS` 调高，硬上限 50） - **一次会话内**：`listCliTargets` 最多调用一次，缓存返回的 id；多条 show 合并进同一次 `commands`，禁止「list→exec→list→exec」循环 - - 超时：提高 `read_timeout_sec`（60–120）或减少命令条数，禁止对同一命令盲重试 + - 超时：提高 `read_timeout_sec`（默认 60，慢命令 90–120）或减少命令条数，禁止对同一命令盲重试 ## CLI 约束（服务端强制） diff --git a/tests/test_mcp_timeout_and_cache.py b/tests/test_mcp_timeout_and_cache.py new file mode 100644 index 00000000..93aef352 --- /dev/null +++ b/tests/test_mcp_timeout_and_cache.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from svc.persistence.sqlite_store import SqliteStore +from runtime.tools.mcp.adapter import ( + clear_list_cli_targets_cache, + materialize_mcp_tools, + mcp_timeout_for_tool, +) +from runtime.tools.tool_validation import format_invalid_arguments_error, validate_tool_arguments + + +class McpTimeoutAndCacheTests(unittest.TestCase): + def test_exec_managed_ne_timeout_override(self) -> None: + self.assertEqual(mcp_timeout_for_tool("execManagedNe", 30.0), 320.0) + self.assertEqual(mcp_timeout_for_tool("ping", 30.0), 30.0) + self.assertGreaterEqual(mcp_timeout_for_tool("sqlQueryUme", 30.0), 90.0) + + def test_materialize_applies_exec_timeout(self) -> None: + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: + store = SqliteStore(str(Path(td) / "ops.sqlite")) + store.upsert_mcp_server( + server_id="netx", + source_type="github", + source_ref="local", + entry_command="python", + entry_args=["-m", "netx_mcp"], + enabled=True, + timeout_s=30.0, + ) + store.replace_mcp_server_tools( + server_id="netx", + tools=[ + { + "tool_name": "execManagedNe", + "description": "exec", + "parameters": {"type": "object", "properties": {}}, + }, + { + "tool_name": "listCliTargets", + "description": "list", + "parameters": {"type": "object", "properties": {}}, + }, + ], + ) + specs = {s.name: s for s in materialize_mcp_tools(store)} + self.assertEqual(specs["mcp__netx__execManagedNe"].timeout_s, 320.0) + self.assertEqual(specs["mcp__netx__listCliTargets"].timeout_s, 30.0) + + def test_list_cli_targets_ttl_cache(self) -> None: + clear_list_cli_targets_cache() + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: + store = SqliteStore(str(Path(td) / "ops.sqlite")) + store.upsert_mcp_server( + server_id="netx", + source_type="github", + source_ref="local", + entry_command="python", + entry_args=["-m", "netx_mcp"], + enabled=True, + ) + store.replace_mcp_server_tools( + server_id="netx", + tools=[ + { + "tool_name": "listCliTargets", + "description": "list", + "parameters": {"type": "object", "properties": {}}, + } + ], + ) + spec = next(s for s in materialize_mcp_tools(store) if s.name.endswith("listCliTargets")) + calls = {"n": 0} + + def fake_call_tool(self, tool_name, arguments=None): # type: ignore[no-untyped-def] + calls["n"] += 1 + return {"ok": True, "data": {"items": [{"ne_id": "1"}]}} + + with patch("runtime.tools.mcp.adapter.McpProcessRuntime.call_tool", fake_call_tool): + first = spec.handler({"keyword": "PE", "source": "ume"}) + second = spec.handler({"keyword": "PE", "source": "ume"}) + self.assertEqual(calls["n"], 1) + self.assertFalse(first.get("cache_hit")) + self.assertTrue(second.get("cache_hit")) + self.assertEqual(second.get("data", {}).get("items", [])[0]["ne_id"], "1") + clear_list_cli_targets_cache() + + +class InvalidArgFormatTests(unittest.TestCase): + def test_format_includes_example(self) -> None: + schema = { + "type": "object", + "properties": { + "command": {"type": "string"}, + "timeout": {"type": "integer", "default": 30}, + }, + "required": ["command"], + "additionalProperties": False, + } + ok, err = validate_tool_arguments(schema, {}) + self.assertFalse(ok) + payload = format_invalid_arguments_error(schema, str(err), lang="en") + self.assertEqual(payload["error_code"], "tool_invalid_arguments") + self.assertIn("example", payload) + self.assertIn("command", payload["example"]) + self.assertIn("required", payload) + + +if __name__ == "__main__": + unittest.main()