Reduce WhatsApp ops tool friction with looser xlsx/file schemas and short-intent playbooks.

Production usage showed schema mismatches and multi-tool loops; accept common arg aliases and route agents to <=3-call alarm recipes.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-10 22:32:44 +08:00
parent 685e3ffa8d
commit aff11c1eae
11 changed files with 276 additions and 24 deletions

View file

@ -6,12 +6,30 @@ from runtime.tools.base import ToolSpec
from runtime.tools.path_guard import resolve_workspace_path
def _resolve_write_path(args: dict[str, Any]) -> str:
for key in ("path", "file", "filename", "file_path", "filepath", "name"):
raw = str(args.get(key) or "").strip().strip('"').strip("'")
if raw:
return raw
return ""
def write_file_tool() -> ToolSpec:
def _handler(args: dict[str, Any]) -> dict[str, Any]:
raw = str(args.get("path") or "").strip().strip('"').strip("'")
raw = _resolve_write_path(args)
if not raw:
return {"ok": False, "error": "path_required"}
content = str(args.get("content") or "")
return {
"ok": False,
"error": "path_required",
"hint": "Pass path (or file/filename) relative to workspace root.",
"example": {"path": "tmp/notes.txt", "content": "hello", "mode": "overwrite"},
}
content = args.get("content")
if content is None:
content = args.get("text")
if content is None:
content = args.get("body")
content_s = "" if content is None else str(content)
mode = str(args.get("mode") or "overwrite").strip().lower()
try:
p = resolve_workspace_path(raw)
@ -21,22 +39,32 @@ def write_file_tool() -> ToolSpec:
if mode not in ("overwrite", "append"):
return {"ok": False, "error": "invalid_mode", "allowed": ["overwrite", "append"]}
if mode == "append":
p.write_text(p.read_text(encoding="utf-8", errors="replace") + content, encoding="utf-8")
p.write_text(p.read_text(encoding="utf-8", errors="replace") + content_s, encoding="utf-8")
else:
p.write_text(content, encoding="utf-8")
p.write_text(content_s, encoding="utf-8")
return {"ok": True, "path": str(p), "bytes": p.stat().st_size}
return ToolSpec(
name="write_file",
description="Write text content to a workspace file (overwrite or append).",
description=(
"Write text content to a workspace file (overwrite or append). "
"Path aliases: file, filename, file_path."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "File path, relative to workspace root."},
"file": {"type": "string", "description": "Alias for path."},
"filename": {"type": "string", "description": "Alias for path."},
"file_path": {"type": "string", "description": "Alias for path."},
"filepath": {"type": "string", "description": "Alias for path."},
"name": {"type": "string", "description": "Alias for path (when it looks like a relative file path)."},
"content": {"type": "string", "description": "Full text content to write."},
"text": {"type": "string", "description": "Alias for content."},
"body": {"type": "string", "description": "Alias for content."},
"mode": {"type": "string", "enum": ["overwrite", "append"], "default": "overwrite"},
},
"required": ["path", "content"],
"required": [],
"additionalProperties": False,
},
handler=_handler,

View file

@ -97,7 +97,7 @@ def _build_workbook(sheets: list[dict[str, Any]], *, freeze_header: bool, auto_w
for i, sheet in enumerate(sheets, start=1):
if not isinstance(sheet, dict):
continue
rows, inferred_cols = _normalize_rows(sheet.get("rows"))
rows, inferred_cols = _normalize_rows(sheet.get("rows") if sheet.get("rows") is not None else sheet.get("data"))
headers_raw = sheet.get("headers")
if isinstance(headers_raw, list) and headers_raw:
col_count = max(len(headers_raw), inferred_cols, 1)
@ -129,15 +129,62 @@ def _build_workbook(sheets: list[dict[str, Any]], *, freeze_header: bool, auto_w
return buf.getvalue(), summary
_XLSX_EXAMPLE = {
"name": "alarm_summary.xlsx",
"sheets": [
{
"name": "by_host",
"headers": ["host_name", "severity", "count"],
"rows": [["NE-A", "critical", 12], ["NE-B", "major", 5]],
}
],
"freeze_header": True,
"auto_width": True,
}
def _coerce_sheets(args: dict[str, Any]) -> list[Any] | None:
"""Accept common agent shapes: sheets[], or top-level headers/rows[/name]."""
sheets_raw = args.get("sheets")
if isinstance(sheets_raw, list) and sheets_raw:
return sheets_raw
# Single sheet mistaken as top-level object
if isinstance(sheets_raw, dict) and (
"rows" in sheets_raw or "headers" in sheets_raw or "data" in sheets_raw
):
return [sheets_raw]
headers = args.get("headers")
rows = args.get("rows")
if rows is None:
rows = args.get("data")
if isinstance(rows, list):
sheet: dict[str, Any] = {"rows": rows}
if isinstance(headers, list):
sheet["headers"] = headers
title = str(args.get("sheet_name") or args.get("sheet") or "").strip()
if title:
sheet["name"] = title
return [sheet]
return None
def write_xlsx_tool() -> ToolSpec:
def _handler(args: dict[str, Any]) -> dict[str, Any]:
sheets_raw = args.get("sheets")
if not isinstance(sheets_raw, list) or not sheets_raw:
return {"ok": False, "error": "sheets_required"}
sheets_raw = _coerce_sheets(args)
if not sheets_raw:
return {
"ok": False,
"error": "sheets_required",
"hint": "Pass sheets=[{name, headers, rows}] (or top-level headers+rows).",
"example": _XLSX_EXAMPLE,
}
if len(sheets_raw) > _MAX_SHEETS:
return {"ok": False, "error": "too_many_sheets", "max_sheets": _MAX_SHEETS}
filename = str(args.get("name") or "").strip() or "report.xlsx"
filename = (
str(args.get("name") or args.get("filename") or args.get("file_name") or "").strip()
or "report.xlsx"
)
if not filename.lower().endswith(".xlsx"):
filename = f"{filename}.xlsx"
freeze_header = args.get("freeze_header") is not False
@ -200,7 +247,10 @@ def write_xlsx_tool() -> ToolSpec:
"properties": {
"sheets": {
"type": "array",
"description": "One or more sheets. Each item: {name, headers[], rows[][]}.",
"description": (
"Preferred: one or more sheets [{name, headers[], rows[][]}]. "
"If omitted, top-level headers+rows are accepted as a single sheet."
),
"items": {
"type": "object",
"properties": {
@ -213,17 +263,72 @@ def write_xlsx_tool() -> ToolSpec:
"rows": {
"type": "array",
"description": "Data rows: each row is an array of cell values (string/number/bool/null).",
"items": {"type": "array"},
"items": {
"anyOf": [
{"type": "array"},
{"type": "object"},
]
},
},
"data": {
"type": "array",
"description": "Alias for rows.",
"items": {
"anyOf": [
{"type": "array"},
{"type": "object"},
]
},
},
},
"required": ["rows"],
"additionalProperties": False,
"additionalProperties": True,
},
},
"headers": {
"type": "array",
"items": {"type": "string"},
"description": "Shortcut when sheets omitted: column headers for a single sheet.",
},
"rows": {
"type": "array",
"description": "Shortcut when sheets omitted: data rows for a single sheet.",
"items": {
"anyOf": [
{"type": "array"},
{"type": "object"},
]
},
},
"data": {
"type": "array",
"description": "Alias for top-level rows.",
"items": {
"anyOf": [
{"type": "array"},
{"type": "object"},
]
},
},
"sheet_name": {
"type": "string",
"description": "Shortcut sheet tab name when using top-level headers/rows.",
},
"sheet": {
"type": "string",
"description": "Alias for sheet_name.",
},
"name": {
"type": "string",
"description": "Download filename, e.g. alarm_summary.xlsx",
},
"filename": {
"type": "string",
"description": "Alias for name.",
},
"file_name": {
"type": "string",
"description": "Alias for name.",
},
"path": {
"type": "string",
"description": "Optional workspace path to also write the .xlsx file (mirror).",
@ -239,7 +344,7 @@ def write_xlsx_tool() -> ToolSpec:
"description": "Best-effort column width from sample cells (default true).",
},
},
"required": ["sheets"],
"required": [],
"additionalProperties": False,
},
handler=_handler,

View file

@ -28,9 +28,15 @@ You are the ops specialist (network operations expert).
- **Use `host_name` as the primary key for every NE dimension** (first table column, Top-N keys, group-by, and how you refer to an NE in prose). After sync, netx stores it on the alarm row — prefer:
- List/paged alarms: **`host_name`** from `mcp__netx__queryUmeAlarms`
- Raw/SQL: **`alarm_host_name`** (over `ne_host_name` when both exist)
- Aggregate: default `by_ne` from `mcp__netx__aggregateUmeAlarms`; custom dims via `group_by=alarm_host_name` (routes to raw aggregate)
- **Never** use `ne_id` / `alarm_ne_id` (UUID) as the user-facing primary key; `ne_id` is for filters and joins only.
- If `host_name` is empty, fall back to `user_label` / `ne_name` with a "host_name missing" note — never bare `ne_id`.
- NE stats/aggregates: prefer `group_by=alarm_host_name` or `group_by=ne_host_name`; do not group by `alarm_ne_id` / `ne_ne_id` for user output.
- NE stats/aggregates: prefer `aggregateUmeAlarms(group_by=alarm_host_name)` or `aggregateUmeAlarmsRaw`; do not group by `alarm_ne_id` / `ne_ne_id` for user output.
## WhatsApp interaction (mandatory)
- Short ops intents follow `ops-netx-ume-playbook` WhatsApp recipes; target **≤3 tool calls** per user message.
- Spreadsheet delivery: `write_xlsx` then `save_deliverable_attachment` — never claim a file was sent without the deliverable step.
- Call `listCliTargets` at most once per session and reuse ids; batch `execManagedNe` commands; on timeout raise `read_timeout_sec` — no blind retries.
## 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**).

View file

@ -19,10 +19,15 @@
## 告警与网元展示(强制)
- **网元维度一律以 `host_name` 为主键展示**(表格首列、Top 排名键、分组维度、结论中的网元指称)。告警同步后 netx 已把 `host_name` 写入告警表,优先读:
- 列表/分页:`mcp__netx__queryUmeAlarms`(或 legacy `netx_query_ume_alarms`)返回的 **`host_name`**
- 聚合:`mcp__netx__aggregateUmeAlarms` 的网元维度字段
- 聚合:`mcp__netx__aggregateUmeAlarms` 的 `by_ne`(默认按 host);自定义维度用 `group_by=alarm_host_name`(会路由到 Raw 聚合)
- **禁止**用 `ne_id` / `alarm_ne_id`(UUID)作为对用户的主展示键;`ne_id` 仅用于工具过滤或内部关联。
- 若 `host_name` 为空,再用 `user_label` / `ne_name` 并标注「host_name 缺失」;仍不得用裸 `ne_id`。
- 按网元统计/聚合:优先 `group_by=alarm_host_name` 或 `group_by=ne_host_name`,勿按 `alarm_ne_id` / `ne_ne_id` 对外展示。
- 按网元统计/聚合:优先 `aggregateUmeAlarms(group_by=alarm_host_name)` 或 `aggregateUmeAlarmsRaw`;勿按 `alarm_ne_id` / `ne_ne_id` 对外展示。
## WhatsApp 交互(强制)
- 短句优先走 `ops-netx-ume-playbook` 的「WhatsApp 短指令配方」,控制在 ≤3 次工具调用。
- 用户要表格/Excel:`write_xlsx` → `save_deliverable_attachment`;禁止只写文件不投递。
- `listCliTargets` 每会话最多查一次并复用 id;`execManagedNe` 合并 commands,超时调 `read_timeout_sec`,禁止盲重试。
## 必须加载技能
- 每次处理 netx/UME **告警或网元** 问题时,必须加载并遵循技能:`ops-netx-ume-playbook`。