ryan
d2f21a6d23
Pure functions build tool_start/tool_end payloads with run_id, friendly label, truncated input (config dropped), status, duration and artifact ref. Labels cover every agent tool, both old and #55 YNAB names, and the homebox_/bookstack_ MCP prefixes, falling back to "Using <tool>…". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
"""Pure shaping of LangChain tool events into SSE `tool_start` / `tool_end` payloads."""
|
|
|
|
import json
|
|
import re
|
|
|
|
from .tool_labels import tool_label
|
|
|
|
INPUT_VALUE_MAX_CHARS = 200
|
|
# Injected by LangChain rather than chosen by the model; never shown to the client.
|
|
_HIDDEN_INPUT_KEYS = frozenset({"config", "callbacks", "run_manager"})
|
|
|
|
_TOOL_HEADER_RE = re.compile(r"^\[(\S+) result\]")
|
|
|
|
|
|
def _truncate(text: str) -> str:
|
|
if len(text) > INPUT_VALUE_MAX_CHARS:
|
|
return text[: INPUT_VALUE_MAX_CHARS - 1] + "…"
|
|
return text
|
|
|
|
|
|
def summarize_tool_input(raw) -> dict[str, str | int | float | bool]:
|
|
"""Reduce a tool's input to flat, display-safe scalars."""
|
|
if raw is None:
|
|
return {}
|
|
if not isinstance(raw, dict):
|
|
return {"input": _truncate(str(raw))}
|
|
summary: dict[str, str | int | float | bool] = {}
|
|
for key, value in raw.items():
|
|
if key in _HIDDEN_INPUT_KEYS or value is None:
|
|
continue
|
|
if isinstance(value, (bool, int, float)):
|
|
summary[key] = value
|
|
elif isinstance(value, str):
|
|
summary[key] = _truncate(value)
|
|
else:
|
|
try:
|
|
summary[key] = _truncate(json.dumps(value, default=str))
|
|
except (TypeError, ValueError):
|
|
summary[key] = _truncate(str(value))
|
|
return summary
|
|
|
|
|
|
def tool_start_payload(event: dict) -> dict:
|
|
name = event["name"]
|
|
return {
|
|
"type": "tool_start",
|
|
"run_id": event["run_id"],
|
|
"tool": name,
|
|
"label": tool_label(name),
|
|
"input": summarize_tool_input(event.get("data", {}).get("input")),
|
|
}
|
|
|
|
|
|
def tool_end_payload(
|
|
event: dict,
|
|
started_at: float | None,
|
|
artifact_ref: dict | None,
|
|
*,
|
|
now: float,
|
|
status: str = "ok",
|
|
) -> dict:
|
|
name = event["name"]
|
|
return {
|
|
"type": "tool_end",
|
|
"run_id": event["run_id"],
|
|
"tool": name,
|
|
"label": tool_label(name),
|
|
"status": status,
|
|
"ms": round((now - started_at) * 1000) if started_at is not None else 0,
|
|
"artifact": artifact_ref,
|
|
}
|
|
|
|
|
|
def tool_name_from_message(text: str) -> str:
|
|
"""Recover the tool name from a persisted `[tool_name result]` header."""
|
|
match = _TOOL_HEADER_RE.match(text)
|
|
return match.group(1) if match else "tool"
|