ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tools/infra): split cronjob god-methods into per-action helpers; extract tool_search catalog/validation; compact registry, lazy_deps, tool_backend_helpers, desktop_ui

Browse Source
This commit is contained in:
Teknium
2026-09-02 14:41:59 -07:00
parent 6723628de9
commit ae67178be1
12 changed files with 793 additions and 855 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/apply_layout_tool.py
+2
View File
@@ -12,6 +12,8 @@ model can self-correct without a registry-listing tool.
from tools import desktop_ui
from tools.registry import registry, tool_error
# Renderer answers via the blocking-prompt bridge; layout apply is synchronous
# there, so the bridge timeout is generous.
_TIMEOUT_NOTE = "Layout apply is only available in the Hermes desktop app."
tools/cronjob_job_args.py
+17 -11
View File
@@ -42,7 +42,7 @@ def _origin_from_env() -> Optional[Dict[str, str]]:
"thread_id": thread_id,
# Lets an opt-in delivery mirror resolve the exact participant's
# session in per-user-isolated group chats (parity with send_message).
"user_id": get_session_env("HERMES_SESSION_USER_ID") or None,
"user_id": get_session_env("HERMES_SESSION_USER_ID") or None, # harmless for DMs
# Workspace/server scope (Slack team, Discord guild...). Slack session
# keys embed it, so a continuable cron seed built without it would
# create a row no scoped reply ever resolves to.
@@ -138,7 +138,9 @@ def _split_monitor_arg(
"""Resolve the single model-facing ``monitor`` field into the stored
``(monitor_script, monitor_url)`` pair.
Shape decides transport: http(s):// is a URL, anything else a script path.
Shape decides transport: http(s):// is a URL, anything else a script path
(a legal script path can never start with a URL scheme). Storage keeps the
two fields separate — interface merge, not a storage migration.
Update semantics: None = unchanged, '' = clear; setting one source clears
the other so switching transports never trips mutual exclusion. An
explicit ``monitor`` wins over the legacy alias fields.
@@ -154,8 +156,8 @@ def _split_monitor_arg(
def _repeat_display(job: Dict[str, Any]) -> str:
times = (job.get("repeat") or {}).get("times")
completed = (job.get("repeat") or {}).get("completed", 0)
rep = job.get("repeat") or {}
times, completed = rep.get("times"), rep.get("completed", 0)
if times is None:
return "forever"
if times == 1:
@@ -174,12 +176,11 @@ def _clean_str_list(items: Any) -> List[str]:
def _canonical_skills(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]:
if skills is None:
raw_items = [skill] if skill else []
skills = [skill] if skill else []
elif isinstance(skills, str):
raw_items = [skills]
else:
raw_items = list(skills)
return list(dict.fromkeys(_clean_str_list(item or "" for item in raw_items)))
skills = [skills]
# `item or ""`: a None entry must drop out, not stringify to "None".
return list(dict.fromkeys(_clean_str_list(item or "" for item in skills)))
def _normalize_optional_job_value(value: Optional[Any], *, strip_trailing_slash: bool = False) -> Optional[str]:
@@ -209,7 +210,8 @@ def _validate_bot_chat_deliver(deliver: Optional[str]) -> Optional[str]:
"""Validate ``bot-chat[:<profile>]`` deliver elements at create time.
Bot Chat delivery is machine-local: the profile must exist where the
scheduler fires. Fail loudly here rather than as a per-run delivery error.
scheduler fires (Desktop multi-gateway rosters may show same-named profiles
from other machines). Fail loudly here rather than as a per-run delivery error.
Returns an error string or None.
"""
if not deliver:
@@ -245,6 +247,7 @@ def _resolve_cron_context_deliver(deliver: Optional[str]) -> Optional[str]:
(or an omitted value) becomes the creating run's ``platform:chat_id[:thread]``
from the HERMES_CRON_AUTO_DELIVER_* contextvars, or ``local`` when the
creating run has no concrete target; other elements pass through verbatim.
Without this the scheduler would fall back to guessing a home channel.
"""
from gateway.session_context import get_session_env
from utils import is_truthy_value
@@ -301,6 +304,8 @@ def _validate_cron_base_url(
return f"Unable to validate base_url override for provider {prov!r}; refused."
if prov.lower() == "custom":
# Pure BYOK: key comes from a pool keyed by THIS base_url or host-gated
# env vars, never an arbitrary stored secret.
return None
if has_named_custom_provider(prov):
# A NAMED custom provider carries a STORED key that the runtime still
@@ -377,7 +382,8 @@ def _apply_continuity(
def _validate_context_from_refs(refs: List[Any]) -> Optional[str]:
"""Error string if any non-"self" ref names a job that doesn't exist.
("self" resolves to the job's own id at run time.)"""
("self" resolves to the job's own id at run time, so it can't be checked
against the store — the job doesn't exist yet at create time.)"""
from cron.jobs import get_job as _get_job
for ref_id in refs:
if isinstance(ref_id, str) and ref_id.strip().lower() == "self":
tools/cronjob_prompt_scan.py
+4 -12
View File
@@ -82,15 +82,6 @@ def _zwj_has_emoji_neighbour(text: str, idx: int) -> bool:
)
def _strip_legitimate_emoji_zwj(prompt: str) -> str:
if '\u200d' not in prompt:
return prompt
return ''.join(
ch for idx, ch in enumerate(prompt)
if not (ch == '\u200d' and _zwj_has_emoji_neighbour(prompt, idx))
)
def _strip_cron_safe_constructs(prompt: str) -> str:
"""Scrub the bundled GitHub skill's `Authorization: token $GITHUB_TOKEN` +
api.github.com curl so it doesn't trip the auth-header exfil rule.
@@ -112,10 +103,11 @@ def _strip_cron_safe_constructs(prompt: str) -> str:
def _check_invisible_unicode(prompt: str) -> str:
"""Error string if the prompt holds invisible-unicode markers (emoji ZWJ allowed)."""
prompt_for_invisible_scan = _strip_legitimate_emoji_zwj(prompt)
"""Error string if the prompt holds invisible-unicode markers (emoji ZWJ allowed).
Reports the first offender in ``_CRON_INVISIBLE_CHARS`` order."""
removed = set(_strip_invisible_unicode(prompt)[1])
for char in _CRON_INVISIBLE_CHARS:
if char in prompt_for_invisible_scan:
if f"U+{ord(char):04X}" in removed:
return f"Blocked: prompt contains invisible unicode U+{ord(char):04X} (possible injection)."
return ""
tools/cronjob_tools.py
+345 -301
View File
@@ -5,6 +5,7 @@ Expose a single compressed action-oriented tool to avoid schema/context bloat.
Compatibility wrappers remain for direct Python callers and legacy tests.
"""
import contextlib
import json
import logging
import sys
@@ -19,7 +20,8 @@ logger = logging.getLogger(__name__)
# Heartbeat cadence that keeps the calling agent's inactivity watchdog at bay
# while a manual `cronjob(action="run")` executes synchronously in-process
# (mirrors tools/environments/base.py::touch_activity_if_due).
# (mirrors tools/environments/base.py::touch_activity_if_due; comfortably
# below the default HERMES_AGENT_TIMEOUT).
_CRON_RUN_HEARTBEAT_INTERVAL = 10.0
# Hard ceiling on the heartbeat: with HERMES_CRON_TIMEOUT=0 (unlimited) a truly
@@ -54,6 +56,7 @@ from tools.cronjob_prompt_scan import ( # noqa: F401 (re-exported)
from tools.cronjob_job_args import ( # noqa: F401 (re-exported)
_apply_continuity,
_canonical_skills,
_clean_str_list,
_format_job,
_gateway_liveness_notice,
_local_delivery_notice,
@@ -117,7 +120,47 @@ def _forward_relay_fronted_run(
"""
if not _relay_fronted_delivery_platforms(job):
return None
job_id = job["id"]
url = f"{_api_server_base_url()}/api/jobs/{job['id']}/run"
from agent.secret_scope import get_secret
key = get_secret("API_SERVER_KEY", "") or ""
try:
import httpx
resp = httpx.post(
url,
headers={"Authorization": f"Bearer {key}"},
json=({"prompt": extra_prompt} if extra_prompt else {}),
timeout=10.0,
)
except Exception:
resp = None
if resp is not None and resp.status_code < 300:
return _dumps({
"success": True,
"forwarded_to_gateway": True,
"note": (
"This job targets a relay-fronted platform; it was dispatched "
"to the running gateway, whose live relay adapter owns that "
"delivery."
),
})
return _dumps({
"success": False,
"error": (
"This job targets a relay-fronted platform, which has no "
"standalone sender. Start the gateway — its ticker will "
"deliver the job on schedule via the live relay adapter."
),
})
def _api_server_base_url() -> str:
"""``http://host:port`` of the local api_server, mirroring its bind
resolution (extra.host -> API_SERVER_HOST -> 127.0.0.1); a wildcard bind
listens on loopback too."""
import os
port_raw = os.getenv("API_SERVER_PORT", "").strip()
@@ -125,9 +168,6 @@ def _forward_relay_fronted_run(
port = int(port_raw) if port_raw else 8642
except ValueError:
port = 8642
# Mirror the api_server's bind resolution (extra.host -> API_SERVER_HOST
# -> 127.0.0.1); a wildcard bind listens on loopback too.
host = ""
try:
from hermes_cli.config import cfg_get, load_config_readonly
@@ -146,49 +186,7 @@ def _forward_relay_fronted_run(
host = "127.0.0.1"
if ":" in host and not host.startswith("["):
host = f"[{host}]" # bare IPv6 literal
url = f"http://{host}:{port}/api/jobs/{job_id}/run"
from agent.secret_scope import get_secret
key = get_secret("API_SERVER_KEY", "") or ""
resp = None
try:
import httpx
resp = httpx.post(
url,
headers={"Authorization": f"Bearer {key}"},
json=({"prompt": extra_prompt} if extra_prompt else {}),
timeout=10.0,
)
except Exception:
resp = None
if resp is not None and resp.status_code < 300:
return json.dumps(
{
"success": True,
"forwarded_to_gateway": True,
"note": (
"This job targets a relay-fronted platform; it was dispatched "
"to the running gateway, whose live relay adapter owns that "
"delivery."
),
},
indent=2,
)
return json.dumps(
{
"success": False,
"error": (
"This job targets a relay-fronted platform, which has no "
"standalone sender. Start the gateway — its ticker will "
"deliver the job on schedule via the live relay adapter."
),
},
indent=2,
)
return f"http://{host}:{port}"
def _manual_run_delivery_note(deliver: str, refreshed: Dict[str, Any]) -> str:
@@ -258,6 +256,68 @@ def _execute_job_now(
return _run_claimed_job(claimed_job, extra_prompt=extra_prompt)
def _gateway_adapters_and_loop():
"""Live gateway adapter map + event loop when running inside the gateway
process, else ``(None, None)``. Manual runs from a gateway agent must
deliver on the loop that owns clients such as Matrix/aiohttp (a standalone
asyncio.run() loop breaks them)."""
runner_ref = getattr(sys.modules.get("gateway.run"), "_gateway_runner_ref", None)
runner = runner_ref() if callable(runner_ref) else None
if runner is None:
return None, None
return getattr(runner, "adapters", None), getattr(runner, "_gateway_loop", None)
@contextlib.contextmanager
def _run_heartbeat(job_name: str):
"""Heartbeat into the caller's activity tracker while a manual run
executes; stops and joins the thread on exit.
A manual run executes synchronously on the caller's thread and can take
minutes; without tool activity the gateway inactivity watchdog would kill
the parent turn. Best-effort: no callback -> no thread, unchanged behavior.
"""
stop = threading.Event()
thread = None
try:
from tools.environments.base import get_activity_callback
# Capture on THIS thread: the callback is thread-local (installed by
# the tool executor), so a freshly spawned thread cannot read it.
activity_cb = get_activity_callback()
except Exception:
activity_cb = None
def _heartbeat_loop() -> None:
started = time.monotonic()
while not stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL):
elapsed = time.monotonic() - started
if elapsed > _CRON_RUN_HEARTBEAT_CEILING:
# A run this long with an unlimited child watchdog is
# likely wedged — stop masking the gateway watchdog.
logger.warning(
"cronjob run heartbeat ceiling reached for job "
"'%s' (%.0fs) — stopping heartbeat; gateway "
"watchdog regains authority",
job_name, elapsed,
)
return
try:
activity_cb(f"cronjob: running job '{job_name}' ({int(elapsed)}s elapsed)")
except Exception:
continue # one transient callback error must not drop protection
if activity_cb is not None:
thread = threading.Thread(target=_heartbeat_loop, daemon=True, name="cronjob-run-heartbeat")
thread.start()
try:
yield
finally:
stop.set()
if thread is not None:
thread.join(timeout=_CRON_RUN_HEARTBEAT_INTERVAL + 1)
def _run_claimed_job(
job: Dict[str, Any], extra_prompt: Optional[str] = None
) -> Dict[str, Any]:
@@ -287,73 +347,16 @@ def _run_claimed_job(
claim = job.get("fire_claim")
fire_owner = str(claim.get("by") or "") if isinstance(claim, dict) else None
# A manual run executes synchronously on the caller's thread and can
# take minutes; without tool activity the gateway inactivity watchdog
# would kill the parent turn. Heartbeat into the caller's activity
# tracker while the job runs (best-effort: no callback -> unchanged).
adapters, gateway_loop = _gateway_adapters_and_loop()
try:
from tools.environments.base import get_activity_callback
# Capture on THIS thread: the callback is thread-local.
activity_cb = get_activity_callback()
except Exception:
activity_cb = None
_heartbeat_stop = threading.Event()
_heartbeat_thread = None
if activity_cb is not None:
job_name = str(job.get("name") or job_id)
def _heartbeat_loop() -> None:
started = time.monotonic()
while not _heartbeat_stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL):
elapsed = time.monotonic() - started
if elapsed > _CRON_RUN_HEARTBEAT_CEILING:
logger.warning(
"cronjob run heartbeat ceiling reached for job "
"'%s' (%.0fs) — stopping heartbeat; gateway "
"watchdog regains authority",
job_name, elapsed,
)
return
try:
activity_cb(
f"cronjob: running job '{job_name}' ({int(elapsed)}s elapsed)"
)
except Exception:
continue # one transient callback error must not drop protection
_heartbeat_thread = threading.Thread(
target=_heartbeat_loop,
daemon=True,
name="cronjob-run-heartbeat",
)
_heartbeat_thread.start()
# Manual runs from a gateway agent share the process with live platform
# adapters: pass the gateway adapter map + event loop so delivery runs
# on the loop that owns clients such as Matrix/aiohttp (a standalone
# asyncio.run() loop breaks them).
gateway_module = sys.modules.get("gateway.run")
runner_ref = getattr(gateway_module, "_gateway_runner_ref", None)
runner = runner_ref() if callable(runner_ref) else None
adapters = getattr(runner, "adapters", None) if runner is not None else None
gateway_loop = getattr(runner, "_gateway_loop", None) if runner is not None else None
try:
try:
# run_one_job records last_run_at/last_status via mark_job_run;
# `job` is the owner-bearing claimed snapshot, so terminal writes
# stay fenced by that owner.
# run_one_job records last_run_at/last_status via mark_job_run;
# `job` is the owner-bearing claimed snapshot, so terminal writes
# stay fenced by that owner.
with _run_heartbeat(str(job.get("name") or job_id)):
processed = run_one_job(
job, adapters=adapters, loop=gateway_loop,
extra_prompt=extra_prompt,
)
finally:
_heartbeat_stop.set()
if _heartbeat_thread is not None:
_heartbeat_thread.join(timeout=_CRON_RUN_HEARTBEAT_INTERVAL + 1)
finally:
_registered = False
release_running_job(job_id)
@@ -373,28 +376,23 @@ def _run_claimed_job(
except Exception as e:
logger.error("Failed to execute cron job %s immediately: %s", job_id, e)
if _registered:
# Only release registrations WE took — a bare discard could erase
# a ticker-owned entry.
# We raised before the run's own release (e.g. heartbeat setup);
# don't leave the job marked in-flight. Only release registrations
# WE took — a bare discard could erase a ticker-owned entry.
try:
from cron.scheduler import release_running_job as _release
_release(job_id)
release_running_job(job_id)
except Exception:
pass
try:
mark_job_run(
job_id,
False,
str(e),
expected_fire_owner=fire_owner,
)
mark_job_run(job_id, False, str(e), expected_fire_owner=fire_owner)
except Exception:
pass
return {"claimed": True, "success": False, "error": str(e)}
def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[str]:
"""Best-effort excerpt of the job's most recent saved output file. Never raises."""
"""Best-effort excerpt of the job's most recent saved output file, for the
background-run completion block (parent sees what the job produced). Never raises."""
try:
from cron.jobs import get_cron_output_dir
@@ -412,6 +410,72 @@ def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[s
return None
def _reap_stale_executions(job_name: str) -> None:
"""Reap execution rows left 'claimed'/'running' by a dead owner process
(e.g. a prior one-shot `hermes cron run` that exited mid-run). The ticker
does this at its own startup; one-shot invocations have no such moment, so
a stale claim would block every later manual run. Only provably-dead
owners are reaped. Best-effort self-heal: must not block dispatch."""
try:
from cron.executions import recover_interrupted_executions
_reclaimed = recover_interrupted_executions()
if _reclaimed:
logger.warning(
"Reclaimed %d stale cron execution(s) from dead owner(s) "
"before dispatching job '%s'",
_reclaimed,
job_name,
)
except Exception as _reap_exc:
logger.debug("Stale execution reclaim failed: %s", _reap_exc)
def _background_session_key(session_id: Optional[str]) -> str:
"""Routing key for a detached completion, captured on THIS thread
(contextvars don't cross the pool). Empty string = no durable consumer."""
try:
from tools.approval import get_current_session_key
session_key = get_current_session_key(default="")
except Exception:
session_key = ""
if not session_key and session_id:
# CLI path: the approval contextvar is only bound during gateway/TUI
# turns; the CLI drain filters completions by the durable session id,
# and an empty key would fail closed (completion never claimable).
session_key = str(session_id)
return session_key
def _manual_run_completion(
res: Dict[str, Any], job_id: str, job_name: str, deliver: str, started_at: float
) -> Dict[str, Any]:
"""Async-delegation completion block for a finished background manual run."""
duration = round(time.time() - started_at, 2)
refreshed = get_job(job_id) or {}
lines = [
f"Cron job '{job_name}' ({job_id}) finished its manual run.",
f"Result: {'ok' if res.get('success') else 'FAILED'}"
+ (f" — {res.get('error')}" if res.get("error") else ""),
f"Delivery target: {deliver}"
+ _manual_run_delivery_note(deliver, refreshed),
]
if refreshed.get("next_run_at"):
lines.append(f"Next scheduled run: {refreshed['next_run_at']}")
excerpt = _latest_job_output_excerpt(job_id)
if excerpt:
lines.append("--- JOB OUTPUT ---")
lines.append(excerpt)
return {
"status": "completed" if res.get("success") else "error",
"summary": "\n".join(lines),
"error": res.get("error"),
"api_calls": 0,
"duration_seconds": duration,
}
def _try_dispatch_background_run(
job: Dict[str, Any], session_id: Optional[str] = None,
extra_prompt: Optional[str] = None,
@@ -431,7 +495,8 @@ def _try_dispatch_background_run(
running in the background; ``{"claimed": True, "dispatched": False, ...}``
when the pool was full and the run executed inline (claim already taken).
"""
# Finite sessions cannot route a detached result back after the turn ends.
# Finite sessions cannot route a detached result back after the turn ends
# — mirror delegate_task's gate and fall back to sync execution.
try:
from gateway.session_context import async_delivery_supported
@@ -442,42 +507,14 @@ def _try_dispatch_background_run(
job_id = job["id"]
job_name = str(job.get("name") or job_id)
_reap_stale_executions(job_name)
# Reap execution rows left 'claimed'/'running' by a dead owner process
# (e.g. a prior one-shot `hermes cron run` that exited mid-run). The
# ticker does this at its own startup; one-shot invocations have no such
# moment, so a stale claim would block every later manual run. Only
# provably-dead owners are reaped.
try:
from cron.executions import recover_interrupted_executions
_reclaimed = recover_interrupted_executions()
if _reclaimed:
logger.warning(
"Reclaimed %d stale cron execution(s) from dead owner(s) "
"before dispatching job '%s'",
_reclaimed,
job_name,
)
except Exception as _reap_exc:
logger.debug("Stale execution reclaim failed: %s", _reap_exc)
# Routing capture on THIS thread (contextvars don't cross the pool), and
# BEFORE the claim: with no routable session there is no durable consumer
# for a detached completion, so we must not claim-and-dispatch.
try:
from tools.approval import get_current_session_key
session_key = get_current_session_key(default="")
except Exception:
session_key = ""
if not session_key and session_id:
# CLI path: the approval contextvar is only bound during gateway/TUI
# turns; the CLI drain filters completions by the durable session id.
session_key = str(session_id)
# Routing capture BEFORE the claim: with no routable session there is no
# durable consumer for a detached completion, so we must not
# claim-and-dispatch. Direct Python callers (`hermes cron run`, tests):
# process exits right after the tool returns, so run synchronously.
session_key = _background_session_key(session_id)
if not session_key:
# Direct Python callers (`hermes cron run`, tests): process exits right
# after the tool returns, so run synchronously.
return None
# Best-effort early dedupe so a mid-run job reports in THIS tool response
@@ -537,28 +574,7 @@ def _try_dispatch_background_run(
def _runner() -> Dict[str, Any]:
res = _run_claimed_job(claimed_job, extra_prompt=extra_prompt)
duration = round(time.time() - started_at, 2)
refreshed = get_job(job_id) or {}
lines = [
f"Cron job '{job_name}' ({job_id}) finished its manual run.",
f"Result: {'ok' if res.get('success') else 'FAILED'}"
+ (f" — {res.get('error')}" if res.get("error") else ""),
f"Delivery target: {deliver}"
+ _manual_run_delivery_note(deliver, refreshed),
]
if refreshed.get("next_run_at"):
lines.append(f"Next scheduled run: {refreshed['next_run_at']}")
excerpt = _latest_job_output_excerpt(job_id)
if excerpt:
lines.append("--- JOB OUTPUT ---")
lines.append(excerpt)
return {
"status": "completed" if res.get("success") else "error",
"summary": "\n".join(lines),
"error": res.get("error"),
"api_calls": 0,
"duration_seconds": duration,
}
return _manual_run_completion(res, job_id, job_name, deliver, started_at)
dispatch = dispatch_async_delegation(
goal=f"Manual run of cron job '{job_name}' ({job_id})",
@@ -605,7 +621,8 @@ def _dumps(payload: Dict[str, Any]) -> str:
def _action_create(a: Dict[str, Any]) -> str:
prompt, script, deliver = a["prompt"], a["script"], a["deliver"]
prompt, script = a["prompt"], a["script"]
deliver = _normalize_deliver_param(a["deliver"])
if not a["schedule"]:
return tool_error("schedule is required for create", success=False)
canonical_skills = _canonical_skills(a["skill"], a["skills"])
@@ -633,7 +650,7 @@ def _action_create(a: Dict[str, Any]) -> str:
# credential to an attacker endpoint.
or _validate_cron_base_url(a["provider"], a["base_url"])
# bot-chat targets are machine-local: fail the CREATE, not the run.
or _validate_bot_chat_deliver(_normalize_deliver_param(deliver))
or _validate_bot_chat_deliver(deliver)
# failure_deliver shares deliver's grammar and validators (NS-788).
or _validate_bot_chat_deliver(_normalize_deliver_param(a["failure_deliver"]))
or (a["context_from"] and _validate_context_from_refs(
@@ -658,7 +675,7 @@ def _action_create(a: Dict[str, Any]) -> str:
schedule=a["schedule"],
name=a["name"],
repeat=a["repeat"],
deliver=_resolve_cron_context_deliver(_normalize_deliver_param(deliver)),
deliver=_resolve_cron_context_deliver(deliver),
origin=_origin_from_env(),
skills=canonical_skills,
model=_normalize_optional_job_value(a["model"]),
@@ -683,11 +700,12 @@ def _action_create(a: Dict[str, Any]) -> str:
_partial = exc.to_dict()
return tool_error(_partial.pop("error"), success=False, **_partial)
_create_message = f"Cron job '{job['name']}' created."
_local_notice = _local_delivery_notice(job, _normalize_deliver_param(deliver))
_local_notice = _local_delivery_notice(job, deliver)
if _local_notice:
_create_message = f"{_create_message} {_local_notice}"
# A job created with no gateway running is stored but never fires — tell
# the model, which otherwise reports a clean success.
# The builtin ticker lives in the gateway process: a job created with no
# gateway running is stored but never fires — tell the model (the CLI
# already warns; the agent path otherwise reports a clean success).
_result = {
"success": True,
"job_id": job["id"],
@@ -702,10 +720,15 @@ def _action_create(a: Dict[str, Any]) -> str:
"message": _create_message,
**_gateway_liveness_notice(),
}
_notes = _mode_guidance_notes(job, _normalize_deliver_param(deliver))
return _dumps(_with_guidance(_result, job, deliver))
def _with_guidance(result: Dict[str, Any], job: Dict[str, Any], deliver: Optional[str]) -> Dict[str, Any]:
"""Attach mode/delivery guidance (create and update echo the same notes)."""
_notes = _mode_guidance_notes(job, deliver)
if _notes:
_result["guidance"] = _notes
return _dumps(_result)
result["guidance"] = _notes
return result
def _action_list(a: Dict[str, Any]) -> str:
@@ -734,17 +757,23 @@ def _action_remove(job: Dict[str, Any], a: Dict[str, Any]) -> str:
def _action_pause(job: Dict[str, Any], a: Dict[str, Any]) -> str:
updated = pause_job(job["id"], reason=a["reason"])
_notify_provider_jobs_changed_safe()
return _dumps({"success": True, "job": _format_job(updated)})
return _job_state_result(pause_job(job["id"], reason=a["reason"]))
def _action_resume(job: Dict[str, Any], a: Dict[str, Any]) -> str:
updated = resume_job(job["id"])
return _job_state_result(resume_job(job["id"]))
def _job_state_result(updated: Dict[str, Any]) -> str:
_notify_provider_jobs_changed_safe()
return _dumps({"success": True, "job": _format_job(updated)})
def _refreshed_job_view(job_id: str) -> Dict[str, Any]:
"""Re-read so the response reflects the post-run last_run_at/last_status."""
return _format_job(get_job(job_id) or {"id": job_id})
def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str:
job_id = job["id"]
# `prompt` on run is transient per-fire context appended to the stored
@@ -762,7 +791,7 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str:
)
if bg is not None and bg.get("dispatched"):
_notify_provider_jobs_changed_safe()
result = _format_job(get_job(job_id) or {"id": job_id})
result = _refreshed_job_view(job_id)
result["executed"] = True
result["execution_mode"] = "background"
result["delegation_id"] = bg.get("delegation_id")
@@ -786,12 +815,12 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str:
return forwarded
exec_result = _execute_job_now(job, extra_prompt=extra_prompt)
# A claimed direct run advances next_run_at and may race an external
# provider's one-shot for the same occurrence — reconcile after the run.
# provider's one-shot for the same occurrence; a lost consumed fire cannot
# re-arm itself, so reconcile after the run has persisted its final state.
claimed = exec_result.get("claimed", False)
if claimed:
_notify_provider_jobs_changed_safe()
# Re-read so the response reflects the post-run last_run_at/last_status.
result = _format_job(get_job(job_id) or {"id": job_id})
result = _refreshed_job_view(job_id)
result["executed"] = claimed
result["execution_success"] = exec_result.get("success", False)
if not claimed:
@@ -803,16 +832,18 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str:
return _dumps({"success": True, "job": result})
def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
job_id = job["id"]
updates: Dict[str, Any] = {}
def _pick(updates: Dict[str, Any], job: Dict[str, Any], key: str) -> Any:
"""Effective value of ``key`` after this update: pending update wins over the stored job."""
return updates[key] if key in updates else job.get(key)
def _update_core_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]:
"""prompt / name / deliver / skills / model pins; returns an error string or None."""
prompt, deliver, skill, skills = a["prompt"], a["deliver"], a["skill"], a["skills"]
script, monitor_script, monitor_url = a["script"], a["monitor_script"], a["monitor_url"]
context_from, continuity, no_agent = a["context_from"], a["continuity"], a["no_agent"]
if prompt is not None:
scan_error = _scan_cron_prompt(prompt)
if scan_error:
return tool_error(scan_error, success=False)
return scan_error
updates["prompt"] = prompt
if a["name"] is not None and a["name"].strip():
# Blank name is a no-op, not a clear: a model re-sending the whole
@@ -821,7 +852,7 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
if deliver is not None:
bot_chat_error = _validate_bot_chat_deliver(_normalize_deliver_param(deliver))
if bot_chat_error:
return tool_error(bot_chat_error, success=False)
return bot_chat_error
updates["deliver"] = _resolve_cron_context_deliver(_normalize_deliver_param(deliver))
if a["failure_deliver"] is not None:
# '' clears the override (job falls back to deliver on failures);
@@ -832,7 +863,7 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
if _norm_fd:
bot_chat_error = _validate_bot_chat_deliver(_norm_fd)
if bot_chat_error:
return tool_error(bot_chat_error, success=False)
return bot_chat_error
_norm_fd = _resolve_cron_context_deliver(_norm_fd)
updates["failure_deliver"] = _norm_fd
if skills is not None or skill is not None:
@@ -850,50 +881,55 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
updates["reasoning_effort"] = a["reasoning_effort"]
# Re-validate the EFFECTIVE provider/base_url on EVERY update: a job
# persisted before this guard may already hold an unsafe pair, and editing
# an unrelated field must not leave it schedulable.
base_url_error = _validate_cron_base_url(
updates["provider"] if "provider" in updates else job.get("provider"),
updates["base_url"] if "base_url" in updates else job.get("base_url"),
)
if base_url_error:
return tool_error(base_url_error, success=False)
# Empty string clears script / monitor fields.
for field, value in (("script", script), ("monitor_script", monitor_script)):
# an unrelated field must not leave it schedulable. Merging this update's
# values over the stored job lets an operator remediate in the same call.
return _validate_cron_base_url(_pick(updates, job, "provider"), _pick(updates, job, "base_url"))
def _update_script_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]:
"""script / monitor_script / monitor_url (empty string clears); returns an error string or None."""
monitor_script, monitor_url = a["monitor_script"], a["monitor_url"]
for field, value in (("script", a["script"]), ("monitor_script", monitor_script)):
if value is not None:
if value:
path_error = _validate_cron_script_path(value)
if path_error:
return tool_error(path_error, success=False)
return path_error
updates[field] = _normalize_optional_job_value(value) if value else None
if monitor_url is not None:
updates["monitor_url"] = _normalize_optional_job_value(monitor_url) if monitor_url else None
if monitor_script is not None or monitor_url is not None:
eff_mon_script = updates["monitor_script"] if "monitor_script" in updates else job.get("monitor_script")
eff_mon_url = updates["monitor_url"] if "monitor_url" in updates else job.get("monitor_url")
if eff_mon_script and eff_mon_url:
return tool_error(
"monitor_script and monitor_url are mutually exclusive — "
"clear one before setting the other.",
success=False,
)
if context_from is not None or continuity is not None:
# Empty string / list clears; otherwise every ref must exist. Stored
# as a list (or None) to match create_job().
if context_from is None:
# continuity-only update: start from the job's stored refs.
existing = job.get("context_from") or []
refs = [str(j).strip() for j in existing if str(j).strip()]
elif isinstance(context_from, str):
refs = [context_from.strip()] if context_from.strip() else []
else:
refs = [str(j).strip() for j in context_from if str(j).strip()]
if continuity is not None:
refs = _apply_continuity(refs, continuity) or []
if refs:
ref_error = _validate_context_from_refs(refs)
if ref_error:
return tool_error(ref_error, success=False)
updates["context_from"] = refs or None
if (monitor_script is not None or monitor_url is not None) and (
_pick(updates, job, "monitor_script") and _pick(updates, job, "monitor_url")
):
return (
"monitor_script and monitor_url are mutually exclusive — "
"clear one before setting the other."
)
return None
def _update_context_from(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]:
"""context_from / continuity: empty string / list clears; otherwise every
ref must exist. Stored as a list (or None) to match create_job()."""
context_from, continuity = a["context_from"], a["continuity"]
if context_from is None and continuity is None:
return None
if context_from is None:
# continuity-only update: start from the job's stored refs.
context_from = list(job.get("context_from") or [])
refs = _clean_str_list(context_from)
if continuity is not None:
refs = _apply_continuity(refs, continuity) or []
if refs:
ref_error = _validate_context_from_refs(refs)
if ref_error:
return ref_error
updates["context_from"] = refs or None
return None
def _update_run_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]:
"""enabled_toolsets / attach_to_session / workdir / no_agent / repeat / schedule."""
if a["enabled_toolsets"] is not None:
updates["enabled_toolsets"] = a["enabled_toolsets"] or None
if a["attach_to_session"] is not None:
@@ -901,22 +937,19 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
if a["workdir"] is not None:
# Empty string clears; otherwise update_job() validates/normalizes.
updates["workdir"] = _normalize_optional_job_value(a["workdir"]) or None
if no_agent is not None:
if a["no_agent"] is not None:
# Flipping to True needs a script on the job or in this same update,
# otherwise the next tick would error out.
target_no_agent = bool(no_agent)
if target_no_agent:
effective_script = updates.get("script") if "script" in updates else job.get("script")
if not effective_script:
return tool_error(
"Cannot set no_agent=True on a job without a script. "
"Set `script` in the same update, or on the job first.",
success=False,
)
target_no_agent = bool(a["no_agent"])
if target_no_agent and not _pick(updates, job, "script"):
return (
"Cannot set no_agent=True on a job without a script. "
"Set `script` in the same update, or on the job first."
)
updates["no_agent"] = target_no_agent
if a["repeat"] is not None:
# Shared chokepoint coerces string forms ('forever'/'once'/'3') and
# 0/negative values.
# 0/negative values (a bare `repeat <= 0` raised TypeError on strings).
from cron.jobs import normalize_repeat_value
repeat_state = dict(job.get("repeat") or {})
repeat_state["times"] = normalize_repeat_value(a["repeat"])
@@ -928,16 +961,27 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
if job.get("state") != "paused":
updates["state"] = "scheduled"
updates["enabled"] = True
return None
# Validation order is behavior (first failing field wins): keep this sequence.
_UPDATE_STEPS = (_update_core_fields, _update_script_fields, _update_context_from, _update_run_fields)
def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str:
updates: Dict[str, Any] = {}
for step in _UPDATE_STEPS:
error = step(job, a, updates)
if error:
return tool_error(error, success=False)
if not updates:
return tool_error("No updates provided.", success=False)
updated = update_job(job_id, updates)
updated = update_job(job["id"], updates)
_notify_provider_jobs_changed_safe()
_upd_result: Dict[str, Any] = {"success": True, "job": _format_job(updated)}
# An update can switch modes or delivery — echo the same guidance as create.
_upd_notes = _mode_guidance_notes(updated, _normalize_deliver_param(deliver))
if _upd_notes:
_upd_result["guidance"] = _upd_notes
return _dumps(_upd_result)
return _dumps(_with_guidance(
{"success": True, "job": _format_job(updated)}, updated, _normalize_deliver_param(a["deliver"])
))
# Actions that need no job_id, and job-bound actions (job resolved first).
@@ -953,6 +997,31 @@ _JOB_ACTIONS = {
}
def _resolve_job_or_error(job_id: str):
"""``(job, None)`` or ``(None, json_error)`` for a job_id/name reference."""
try:
job = resolve_job_ref(job_id)
except AmbiguousJobReference as exc:
return None, _dumps({
"success": False,
"error": str(exc),
"matches": [
{
"id": m["id"],
"name": m.get("name"),
"schedule": m.get("schedule_display"),
"next_run_at": m.get("next_run_at"),
}
for m in exc.matches
],
})
if not job:
return None, _dumps(
{"success": False, "error": f"Job with ID or name '{job_id}' not found. Use cronjob(action='list') to inspect jobs."},
)
return job, None
def cronjob(
action: str,
job_id: Optional[str] = None,
@@ -995,27 +1064,11 @@ def cronjob(
if not job_id:
return tool_error(f"job_id is required for action '{normalized}'", success=False)
try:
job = resolve_job_ref(job_id)
except AmbiguousJobReference as exc:
return _dumps({
"success": False,
"error": str(exc),
"matches": [
{
"id": m["id"],
"name": m.get("name"),
"schedule": m.get("schedule_display"),
"next_run_at": m.get("next_run_at"),
}
for m in exc.matches
],
})
if not job:
return _dumps(
{"success": False, "error": f"Job with ID or name '{job_id}' not found. Use cronjob(action='list') to inspect jobs."},
)
# Job resolution precedes the action check (an unknown action on a
# missing job reports the missing job) — preserved ordering.
job, error = _resolve_job_or_error(job_id)
if error is not None:
return error
handler = _JOB_ACTIONS.get(normalized)
if handler is None:
@@ -1130,6 +1183,18 @@ def check_cronjob_requirements() -> bool:
from tools.registry import registry, tool_error
# Agent-facing arguments forwarded verbatim to cronjob(). model / provider /
# base_url are intentionally NOT in this list: per-job inference pins are
# user-owned (dashboard, `hermes cron create/edit --model`, or hand-edited
# jobs). The agent must not be able to point unattended spend at a different
# model. Programmatic callers of cronjob() itself retain the parameters.
_HANDLER_FORWARDED_ARGS = (
"job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills",
"reason", "script", "context_from", "continuity", "enabled_toolsets", "workdir",
"no_agent", "attach_to_session",
)
def _cronjob_handler(args, **kw):
"""Model-tool dispatch for ``cronjob``: resolves the one model-facing
``monitor`` field into the stored ``monitor_script``/``monitor_url`` pair
@@ -1139,33 +1204,12 @@ def _cronjob_handler(args, **kw):
)
return cronjob(
action=args.get("action", ""),
job_id=args.get("job_id"),
prompt=args.get("prompt"),
schedule=args.get("schedule"),
name=args.get("name"),
repeat=args.get("repeat"),
deliver=args.get("deliver"),
failure_deliver=args.get("failure_deliver"),
include_disabled=args.get("include_disabled", True),
skill=args.get("skill"),
skills=args.get("skills"),
# model / provider / base_url are intentionally NOT read from the
# agent's arguments: per-job inference pins are user-owned (dashboard,
# `hermes cron create/edit --model`, or hand-edited jobs). The agent
# must not be able to point unattended spend at a different model.
# Programmatic callers of cronjob() itself retain the parameters.
reason=args.get("reason"),
script=args.get("script"),
context_from=args.get("context_from"),
continuity=args.get("continuity"),
enabled_toolsets=args.get("enabled_toolsets"),
workdir=args.get("workdir"),
no_agent=args.get("no_agent"),
attach_to_session=args.get("attach_to_session"),
monitor_script=_mon_script,
monitor_url=_mon_url,
task_id=kw.get("task_id"),
session_id=kw.get("session_id"),
**{key: args.get(key) for key in _HANDLER_FORWARDED_ARGS},
)
tools/desktop_ui.py
+3 -1
View File
@@ -33,7 +33,9 @@ def user_enabled(setting: str, default: bool) -> bool:
The renderer mirrors these toggles onto the CONNECTED gateway's config, so this
reads the user's real answer for local/SSH/URL/cloud gateways alike (an env var
would only describe the process). ``check_fn``s use it to withdraw a tool from the
schema when the user switched the feature off. Unreadable config -> ``default``.
schema when the user switched the feature off — Hermes should not be told about
a surface it may not use. Unreadable config -> ``default`` so a shipped-on
feature does not vanish on a transient read error.
"""
try:
from hermes_cli.config import load_config_readonly
tools/lazy_deps.py
+7 -2
View File
@@ -198,7 +198,7 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
# path for lean/partial installs. Call sites use prompt=False so read_file /
# vision can never block on an input() prompt mid-session.
"tool.vision": ("Pillow==12.3.0",),
"tool.doc_extract": ("firecrawl-anydoc==0.2.4",), # lockstep with pyproject
"tool.doc_extract": ("firecrawl-anydoc==0.2.4",), # imports as `anydoc`; lockstep with pyproject
# MCP client SDK for the cua-driver; covers lean/broken-extra installs so
# computer_use never dead-ends on `No module named 'mcp'`.
"tool.computer_use": (
@@ -512,7 +512,10 @@ def _warm_installed_bytecode(specs: tuple[str, ...], target: Optional[Path]) ->
on an installer. Best-effort; never invalidates a successful install."""
if sys.dont_write_bytecode:
return
import compileall
try:
import compileall
except Exception: # pragma: no cover — stdlib, but never break an install
return
for spec in specs:
try:
@@ -579,6 +582,8 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install
try:
# --compile-bytecode: uv writes no __pycache__ by default, so the
# first import would recompile the backend AND its transitives.
# Covers the whole install; _warm_installed_bytecode is the
# belt-and-braces pass for the spec's own roots on any tier.
r = _run([uv_bin, "pip", "install", "--compile-bytecode", *extra_args, *specs],
timeout=timeout, env=uv_env)
if r.returncode != 0:
tools/registry.py
+168 -284
View File
@@ -22,6 +22,7 @@ import logging
import sys
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Dict, List, Optional, Set
@@ -208,36 +209,26 @@ def _save_discovery_cache(cache: Dict[str, list]) -> None:
logger.debug("Could not write tool discovery cache %s: %s", path, e)
@dataclass(eq=False, slots=True)
class ToolEntry:
"""Metadata for a single registered tool."""
"""Metadata for a single registered tool (identity semantics: registry
restore/CAS paths compare entries with ``is``)."""
__slots__ = (
"name", "toolset", "schema", "handler", "check_fn",
"requires_env", "is_async", "description", "emoji",
"max_result_size_chars", "dynamic_schema_overrides",
)
def __init__(self, name, toolset, schema, handler, check_fn,
requires_env, is_async, description, emoji,
max_result_size_chars=None, dynamic_schema_overrides=None):
self.name = name
self.toolset = toolset
self.schema = schema
self.handler = handler
self.check_fn = check_fn
self.requires_env = requires_env
self.is_async = is_async
self.description = description
self.emoji = emoji
self.max_result_size_chars = max_result_size_chars
# Optional zero-arg callable returning a dict of schema overrides
# applied at get_definitions() time. Use for fields that depend on
# runtime config (e.g. delegate_task's description must reflect the
# user's current delegation.max_concurrent_children / max_spawn_depth
# so the model isn't told the wrong limits). The callable is invoked
# on every get_definitions() call; results are merged shallow on top
# of the base schema before the {"type": "function", ...} wrap.
self.dynamic_schema_overrides = dynamic_schema_overrides
name: str
toolset: str
schema: dict
handler: Callable
check_fn: Optional[Callable]
requires_env: list
is_async: bool
description: str
emoji: str
max_result_size_chars: int | float | None = None
# Zero-arg callable returning schema overrides merged (shallow) on top of
# the base schema at every get_definitions() call — for fields that track
# runtime config (e.g. delegate_task's description must reflect the current
# delegation.max_concurrent_children / max_spawn_depth).
dynamic_schema_overrides: Optional[Callable] = None
class _PluginOverridePolicy:
@@ -249,33 +240,32 @@ class _PluginOverridePolicy:
self.allowed = bool(allowed)
_OVERRIDE_DENIED_MSG = (
"Plugin module {owner!r} cannot override built-in tool {name!r} "
"without operator opt-in (allow_tool_override)."
)
# ---------------------------------------------------------------------------
# check_fn TTL cache
#
# external state (Docker daemon, Modal SDK install, playwright binary
# availability). For a long-lived CLI or gateway process, calling them on
# every get_definitions() is pure waste — external state changes on human
# timescales. Cache results for ~30 s so env-var flips via ``hermes tools``
# or live credential file changes propagate within a turn or two without
# requiring any explicit invalidation.
# check_fns probe external state (Docker daemon, Modal SDK, playwright binary)
# that changes on human timescales, so results are cached ~30 s: env-var flips
# via ``hermes tools`` still propagate within a turn or two with no explicit
# invalidation.
#
#
# Transient-failure suppression (issue #21658 / #5304): these probes can flap.
# A single ``subprocess.run([docker, "version"], timeout=5)`` that times out
# under load returns False for one call, which would silently strip the entire
# terminal+file toolset from whatever agent is being built at that instant —
# most visibly a delegate_task subagent, which then reports "Tool read_file
# does not exist". To absorb such flakes WITHOUT pinning a permanently-stale
# "available" verdict, we remember the last time each check returned True and,
# when a fresh probe fails within a short grace window of that last success,
# we serve the last-good True instead of caching the failure. A failure that
# persists past the grace window is honored normally, so a backend that really
# went down stops advertising its tools.
# Transient-failure suppression: probes can flap (a ``docker version`` that
# times out under load), which would silently strip a whole toolset from the
# agent being built at that instant — most visibly a delegate_task subagent
# reporting "Tool read_file does not exist". So we remember each check's last
# success and, when a fresh probe fails within a short grace window of it,
# serve the last-good True WITHOUT caching the failure. A failure persisting
# past the window is honored, so a backend that really went down stops
# advertising its tools.
# ---------------------------------------------------------------------------
_CHECK_FN_TTL_SECONDS = 30.0
# How long after a successful check a subsequent transient failure is treated
# as a flake (last-good True is served) rather than a real outage. Kept short
# Grace window after a success in which a failure counts as a flake; kept short
# so a genuinely-down backend is reflected within a couple of turns.
_CHECK_FN_FAILURE_GRACE_SECONDS = 60.0
_CHECK_FN_CACHE_MAX = 512
@@ -292,6 +282,10 @@ def no_cache_check_fn(fn: Callable) -> Callable:
return fn
def _fn_label(fn: Callable) -> object:
return getattr(fn, "__qualname__", fn)
def _prune_check_fn_caches(now: float) -> None:
"""Expire stale entries and cap profile-dimensional cache growth.
@@ -313,15 +307,13 @@ def check_fn_cache_scope() -> Optional[str]:
"""Return the active profile key when availability is profile-scoped.
Browser-controller availability is request-bound and can change on every
attach/detach. A fully bound browser-control request therefore bypasses both
this check cache and model_tools' outer definition cache; the same sentinel
is consumed by both layers. This prevents one Browser session's live tools
from leaking into any unrelated session.
attach/detach, so a fully bound browser-control request bypasses both this
cache and model_tools' outer definition cache (same sentinel for both
layers) — one Browser session's live tools must not leak into another.
Single-profile processes intentionally keep the historical process-wide
cache. A multiplex gateway installs a Hermes-home override for every
profile turn, so the canonical profile key is the stable isolation
boundary across repeated turns for that profile.
Single-profile processes keep the historical process-wide cache. A
multiplex gateway installs a Hermes-home override per profile turn, so the
canonical profile key is the stable isolation boundary.
"""
try:
from gateway.session_context import get_session_env
@@ -365,12 +357,12 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b
# check_fns run before any profile secret scope exists, so
# get_secret raises by design. The tool re-probes on the first
# scoped turn — log without a traceback so this cannot be
# mistaken for a crashed check_fn (#100697).
# mistaken for a crashed check_fn.
logger.debug(
"check_fn %s hit the multiplex fail-closed path with no "
"profile secret scope active; dependent tools re-probe on "
"the first scoped turn",
getattr(fn, "__qualname__", fn),
_fn_label(fn),
)
return False
# The scope resolved but the read still failed closed: a genuinely
@@ -378,7 +370,7 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b
logger.warning(
"check_fn %s raised UnscopedSecretError while the profile cache "
"scope was resolved; dependent tools will be unavailable this turn",
getattr(fn, "__qualname__", fn),
_fn_label(fn),
exc_info=True,
)
return False
@@ -386,7 +378,7 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b
detail = " while profile cache scope was unresolved" if unresolved_scope else ""
logger.warning(
"check_fn %s raised%s; dependent tools will be unavailable this turn",
getattr(fn, "__qualname__", fn),
_fn_label(fn),
detail,
exc_info=True,
)
@@ -403,19 +395,17 @@ def _check_fn_cached(fn: Callable) -> bool:
return _run_check_fn_uncached(fn, unresolved_scope=True)
cache_key = (fn, scope)
with _check_fn_cache_lock:
_prune_check_fn_caches(now)
_prune_check_fn_caches(now) # leaves only entries within TTL
cached = _check_fn_cache.get(cache_key)
if cached is not None:
ts, value = cached
if now - ts < _CHECK_FN_TTL_SECONDS:
return value
return cached[1]
raised = False
try:
value = bool(fn())
outcome = "returned False"
except Exception:
value = False
raised = True
outcome = "raised"
with _check_fn_cache_lock:
_prune_check_fn_caches(now)
@@ -432,9 +422,7 @@ def _check_fn_cached(fn: Callable) -> bool:
logger.warning(
"check_fn %s failed (%s) within %.0fs of last success; "
"treating as transient and keeping tool(s) available",
getattr(fn, "__qualname__", fn),
"raised" if raised else "returned False",
_CHECK_FN_FAILURE_GRACE_SECONDS,
_fn_label(fn), outcome, _CHECK_FN_FAILURE_GRACE_SECONDS,
)
return True
@@ -442,8 +430,7 @@ def _check_fn_cached(fn: Callable) -> bool:
# silent tool loss in quiet mode (subagents) is diagnosable.
logger.warning(
"check_fn %s %s; dependent tools will be unavailable this turn",
getattr(fn, "__qualname__", fn),
"raised" if raised else "returned False",
_fn_label(fn), outcome,
)
_check_fn_cache[cache_key] = (now, False)
return False
@@ -458,27 +445,21 @@ def invalidate_check_fn_cache() -> None:
def get_cached_check_fn_result(fn: Callable) -> Optional[bool]:
"""Return the current cached verdict for *fn* if its TTL is still valid.
"""Return the cached verdict for *fn* if its TTL is still valid, else None.
Unlike :func:`_check_fn_cached`, this NEVER executes the probe. It is for
read-only surfaces (e.g. dashboard status panels) that need the last-known
availability without triggering network / auth / SDK work inside a request
path. Returns ``None`` when there is no fresh cached verdict.
NEVER executes the probe: for read-only surfaces (dashboard status panels)
that must not trigger network / auth / SDK work inside a request path.
"""
now = time.monotonic()
scope = check_fn_cache_scope()
if scope == CHECK_FN_CACHE_BYPASS:
# Unresolved profile identity bypasses the cache entirely; there is no
# trustworthy cached verdict to report.
# Unresolved profile identity bypasses the cache; nothing trustworthy to report.
return None
with _check_fn_cache_lock:
cached = _check_fn_cache.get((fn, scope))
if cached is None:
return None
ts, value = cached
if now - ts < _CHECK_FN_TTL_SECONDS:
return value
return None
if cached is not None and now - cached[0] < _CHECK_FN_TTL_SECONDS:
return cached[1]
return None
class ToolRegistry:
@@ -517,6 +498,21 @@ class ToolRegistry:
"""Return the active profile's canonical registry scope."""
return hermes_home_key()
def _slot(self, scope: Optional[str], *, create: bool = False) -> Dict[str, ToolEntry]:
"""The registration map for *scope*: global when None, else that profile's overlay."""
if scope is None:
return self._tools
if create:
return self._scoped_tools.setdefault(scope, {})
return self._scoped_tools.get(scope, {})
def _drop_toolset_aliases(self, toolset: str) -> None:
self._toolset_aliases = {
alias: target
for alias, target in self._toolset_aliases.items()
if target != toolset
}
def _merged_tools(self, scope: Optional[str] = None) -> Dict[str, ToolEntry]:
"""Return global tools overlaid with one profile's plugin tools."""
active_scope = scope or self.current_scope_key()
@@ -583,8 +579,7 @@ class ToolRegistry:
) -> Optional[ToolEntry]:
"""Return the local slot state without following global fallback."""
with self._lock:
target = self._tools if scope is None else self._scoped_tools.get(scope, {})
return target.get(name)
return self._slot(scope).get(name)
def get_registered_toolset_names(self) -> List[str]:
"""Return sorted unique toolset names present in the registry."""
@@ -685,21 +680,14 @@ class ToolRegistry:
return bool(policy and policy.allowed)
def _plugin_owner_of(self, handler: Callable) -> Optional[str]:
"""Return the plugin module namespace that defined *handler*, or None
if it was not defined in a loaded plugin module.
"""Plugin namespace that DEFINED *handler* (None for built-in/MCP handlers).
Authorization is bound to where the handler was DEFINED
(``handler.__globals__["__name__"]``), which is fixed at definition
time and cannot drift with the call site, thread, or timing. Lambdas
and nested functions inherit the defining module's globals, so a
plugin cannot launder an override through a callback. Built-in/MCP
handlers live outside the plugin namespace and return None (unchanged
behavior).
Bound to ``handler.__globals__["__name__"]``, fixed at definition time so
it cannot drift with call site, thread, or timing; lambdas and nested
functions inherit it, so a plugin cannot launder an override via a callback.
"""
mod = self._callable_module(handler)
if not mod:
return None
return self._plugin_namespace_of_module(mod)
return self._plugin_namespace_of_module(mod) if mod else None
@staticmethod
def _callable_module(handler: Callable) -> str:
@@ -778,13 +766,11 @@ class ToolRegistry:
@staticmethod
def _caller_module() -> str:
"""Best-effort module name of whoever called the registry method that
invoked this helper (two frames up: this helper, then the registry
method itself, then the actual caller).
"""Best-effort module name of the registry method's caller (two frames up).
``deregister()`` takes only a tool name — unlike ``register()`` it has
no handler argument to bind authorization to via ``_plugin_owner_of``.
Frame inspection is the only way to know who is asking.
``deregister()`` takes only a tool name — no handler to bind authorization
to via ``_plugin_owner_of`` — so frame inspection is the only way to know
who is asking.
"""
try:
frame = sys._getframe(2)
@@ -822,16 +808,15 @@ class ToolRegistry:
if scope is None and owner is not None:
scope = self._plugin_scope_of(owner)
with self._lock:
target = (
self._tools
if scope is None
else self._scoped_tools.setdefault(scope, {})
)
target = self._slot(scope, create=True)
existing = (
self._tools.get(name)
if scope is None
else self._merged_tools(scope).get(name)
)
plugin_override_denied = (
owner is not None and not self._plugin_override_allowed(scope, owner)
)
shadows_global = (
owner is not None
and scope is not None
@@ -847,17 +832,11 @@ class ToolRegistry:
name,
)
return
if not self._plugin_override_allowed(scope, owner):
raise PermissionError(
f"Plugin module {owner!r} cannot override built-in "
f"tool {name!r} without operator opt-in "
f"(allow_tool_override)."
)
if plugin_override_denied:
raise PermissionError(_OVERRIDE_DENIED_MSG.format(owner=owner, name=name))
if existing and existing.toolset != toolset:
if override:
if owner is not None and not self._plugin_override_allowed(
scope, owner
):
if plugin_override_denied:
logger.error(
"Tool registration REJECTED: plugin %r attempted to "
"override built-in tool %r (existing toolset %r) without "
@@ -866,11 +845,7 @@ class ToolRegistry:
"in config.yaml to allow it.",
owner, name, existing.toolset,
)
raise PermissionError(
f"Plugin module {owner!r} cannot override built-in "
f"tool {name!r} without operator opt-in "
f"(allow_tool_override)."
)
raise PermissionError(_OVERRIDE_DENIED_MSG.format(owner=owner, name=name))
# Explicit opt-in (or non-plugin caller): replace the tool.
# Logged at INFO so the override is auditable in agent.log.
logger.info(
@@ -903,36 +878,27 @@ class ToolRegistry:
max_result_size_chars=max_result_size_chars,
dynamic_schema_overrides=dynamic_schema_overrides,
)
# Availability is now derived per-tool (_toolset_has_exposable_tools),
# so this map no longer gates a toolset. It is still consumed by
# Availability is derived per-tool (_toolset_has_exposable_tools), so
# this map no longer gates a toolset. It still feeds
# get_toolset_requirements -> TOOLSET_REQUIREMENTS["check_fn"], which
# banner.py reads (presence only, never called) to classify an
# already-unavailable toolset as lazy-init vs disabled. Keep the
# write path for that classification.
# already-unavailable toolset as lazy-init vs disabled.
if scope is None and check_fn and toolset not in self._toolset_checks:
self._toolset_checks[toolset] = check_fn
self._generation += 1
def deregister(self, name: str, *, scope: Optional[str] = None) -> None:
"""Remove a tool from the registry.
"""Remove a tool; also drops the toolset check/aliases if it was the last
tool in its toolset (MCP nuke-and-repave on ``tools/list_changed``).
Also cleans up the toolset check if no other tools remain in the
same toolset. Used by MCP dynamic tool discovery to nuke-and-repave
when a server sends ``notifications/tools/list_changed``.
``scope`` selects a profile overlay explicitly (multiplexed MCP tools live
in the owning profile's overlay). Plugin callers may not name another
scope; non-plugin callers without ``scope`` target the process-global map.
``scope`` selects a profile overlay explicitly (multiplexed MCP tools
live in the owning profile's overlay). Plugin callers keep their own
scope and may not name another one; non-plugin callers without
``scope`` keep the historical process-global target.
Gated by the same operator opt-in policy ``register(override=True)``
enforces. Without this, a plugin could bypass that gate entirely by
deregistering a tool it doesn't own and then calling plain
``register()`` over the now-empty slot — ``register()`` only runs its
override check when an ``existing`` entry is present, so removing it
first skips the check altogether. MCP toolsets (``mcp-*``) are exempt:
dynamic tool discovery legitimately nukes-and-repaves its own tools on
every refresh and has no plugin-override concept.
Gated by the same opt-in as ``register(override=True)``: otherwise a plugin
could deregister a tool it doesn't own and re-register over the empty slot,
skipping the override check (which only runs when an entry exists).
``mcp-*`` toolsets are exempt — discovery repaves its own tools per refresh.
"""
with self._lock:
caller_mod = self._caller_module()
@@ -949,31 +915,21 @@ class ToolRegistry:
)
if scope is None:
scope = caller_scope
target = (
self._scoped_tools.get(scope, {})
if scope is not None
else self._tools
)
target = self._slot(scope)
entry = target.get(name)
if entry is None and scope is not None:
if caller_owner is not None and name in self._tools:
if entry is None:
if scope is not None and caller_owner is not None and name in self._tools:
raise PermissionError(
f"Scoped plugin module {caller_mod!r} cannot deregister "
f"process-global tool {name!r}; register a scoped "
"override instead."
)
return
if entry is None:
return
if not entry.toolset.startswith("mcp-"):
owner = self._plugin_owner_of(entry.handler)
# Ownership check: bind to the plugin package root
# (``hermes_plugins.{name}``), not the exact module string.
# A handler defined in ``hermes_plugins.pkg.handlers`` is
# still owned by the ``hermes_plugins.pkg`` package — exact
# string equality would wrongly block root-module cleanup code
# from removing tools registered by a submodule of the same
# plugin (egilewski review on #55840).
# Ownership binds to the plugin package root (``hermes_plugins.{name}``),
# not the exact module string: a handler defined in a submodule is
# still owned by the package, so root-module cleanup may remove it.
same_plugin = bool(owner and caller_owner == owner)
if (
caller_owner is not None
@@ -1006,11 +962,7 @@ class ToolRegistry:
)
if not toolset_still_exists:
self._toolset_checks.pop(entry.toolset, None)
self._toolset_aliases = {
alias: target
for alias, target in self._toolset_aliases.items()
if target != entry.toolset
}
self._drop_toolset_aliases(entry.toolset)
self._generation += 1
logger.debug("Deregistered tool: %s", name)
@@ -1031,11 +983,7 @@ class ToolRegistry:
must leave the newer entry untouched.
"""
with self._lock:
target = (
self._tools
if scope is None
else self._scoped_tools.setdefault(scope, {})
)
target = self._slot(scope, create=True)
if target.get(name) is not current:
return False
@@ -1072,11 +1020,7 @@ class ToolRegistry:
for entries in self._scoped_tools.values()
for entry in entries.values()
):
self._toolset_aliases = {
alias: target
for alias, target in self._toolset_aliases.items()
if target != toolset
}
self._drop_toolset_aliases(toolset)
self._generation += 1
logger.debug("Restored tool registration: %s", name)
return True
@@ -1086,20 +1030,12 @@ class ToolRegistry:
# ------------------------------------------------------------------
def get_definitions(self, tool_names: Set[str], quiet: bool = False) -> List[dict]:
"""Return OpenAI-format tool schemas for the requested tool names.
Only tools whose ``check_fn()`` returns True (or have no check_fn)
are included. ``check_fn()`` results are cached for ~30 s via
:func:`_check_fn_cached` to amortize repeat probes (check_terminal_
requirements probes modal/docker, browser checks probe playwright,
etc.); TTL chosen so env-var changes (``hermes tools enable foo``)
still take effect in near-real-time without forcing a full cache
flush on every call.
"""Return OpenAI-format schemas for the requested tools whose ``check_fn``
passes (or is absent). Probes go through the ~30 s TTL cache
(:func:`_check_fn_cached`) so ``hermes tools enable`` still lands quickly.
"""
result = []
# Per-call cache on top of the 30 s TTL — handles repeat probes of the
# same check_fn within one definitions pass without re-reading the
# TTL clock.
# Per-call memo on top of the TTL: one probe per distinct check_fn per pass.
check_results: Dict[Callable, bool] = {}
entries_by_name = {entry.name: entry for entry in self._snapshot_entries()}
for name in sorted(tool_names):
@@ -1113,13 +1049,10 @@ class ToolRegistry:
if not quiet:
logger.debug("Tool %s unavailable (check failed)", name)
continue
# Ensure schema always has a "name" field — use entry.name as fallback
schema_with_name = {**entry.schema, "name": entry.name}
# Apply runtime-dynamic overrides (e.g. delegate_task description
# depends on current delegation.max_concurrent_children /
# max_spawn_depth). Caller side (model_tools.get_tool_definitions)
# already keys its memo on config.yaml mtime + size, so changes
# to delegation.* in config invalidate the cache automatically.
# Runtime-dynamic overrides (e.g. delegate_task limits). The caller's
# memo (model_tools.get_tool_definitions) is keyed on config.yaml
# mtime+size, so config changes invalidate it automatically.
if entry.dynamic_schema_overrides is not None:
try:
overrides = entry.dynamic_schema_overrides()
@@ -1140,13 +1073,9 @@ class ToolRegistry:
@staticmethod
def _normalize_handler_result(name: str, result):
"""Enforce the result shapes supported by the agent tool pipeline.
Normal tool results are strings. The sole structured exception is the
multimodal envelope consumed by the agent executor. Returning every
other value as a string error keeps logging, hooks, budgeting, and
persistence from receiving values they cannot safely slice or size.
"""
"""Results must be a string or the multimodal envelope; anything else
becomes a string error so logging/hooks/budgeting/persistence never
receive values they cannot slice or size."""
if isinstance(result, str):
return _bound_json_error_result(result)
if (
@@ -1177,14 +1106,9 @@ class ToolRegistry:
scope: Optional[str] = None,
**kwargs,
) -> str | dict:
"""Execute a tool handler by name.
* Async handlers are bridged automatically via ``_run_async()``.
* Handler results are normalized to a string or supported multimodal
envelope before leaving the registry.
* All exceptions are caught and returned as ``{"error": "..."}``
for consistent error format.
"""
"""Execute a tool handler by name: async handlers bridged via
``_run_async()``, results normalized, every exception returned as
``{"error": ...}``."""
entry = self.get_entry(name, scope=scope)
if not entry:
return tool_error(f"Unknown tool: {name}")
@@ -1200,9 +1124,8 @@ class ToolRegistry:
logger.exception(
"Tool %s dispatch error: %s", name, _bound_error_text(str(e))
)
# Route through the sanitizer so framing tokens / CDATA / fences
# in exception strings don't reach the model as structural noise.
# See model_tools._sanitize_tool_error for rationale.
# Sanitize so framing tokens / CDATA / fences in exception strings
# don't reach the model as structural noise.
raw = f"Tool execution failed: {type(e).__name__}: {e}"
try:
from model_tools import _sanitize_tool_error
@@ -1253,41 +1176,32 @@ class ToolRegistry:
return {entry.name: entry.toolset for entry in self._snapshot_entries()}
def is_toolset_available(self, toolset: str) -> bool:
"""Check if a toolset has at least one exposable tool.
Returns False (rather than crashing) when a per-tool check raises
an unexpected exception (e.g. network error, missing import, bad config).
"""
entries, _ = self._snapshot_state()
return self._toolset_has_exposable_tools(toolset, entries)
"""True when a toolset has at least one exposable tool (never raises)."""
return self._toolset_has_exposable_tools(toolset, self._snapshot_entries())
def check_toolset_requirements(self) -> Dict[str, bool]:
"""Return ``{toolset: available_bool}`` for every toolset."""
entries, _ = self._snapshot_state()
toolsets = sorted({entry.toolset for entry in entries})
entries = self._snapshot_entries()
return {
toolset: self._toolset_has_exposable_tools(toolset, entries)
for toolset in toolsets
for toolset in sorted({entry.toolset for entry in entries})
}
def get_available_toolsets(self) -> Dict[str, dict]:
"""Return toolset metadata for UI display."""
toolsets: Dict[str, dict] = {}
entries, _ = self._snapshot_state()
entries = self._snapshot_entries()
for entry in entries:
ts = entry.toolset
if ts not in toolsets:
toolsets[ts] = {
"available": self._toolset_has_exposable_tools(ts, entries),
info = toolsets.get(entry.toolset)
if info is None:
info = toolsets[entry.toolset] = {
"available": self._toolset_has_exposable_tools(entry.toolset, entries),
"tools": [],
"description": "",
"requirements": [],
}
toolsets[ts]["tools"].append(entry.name)
if entry.requires_env:
for env in entry.requires_env:
if env not in toolsets[ts]["requirements"]:
toolsets[ts]["requirements"].append(env)
info["tools"].append(entry.name)
_extend_unique(info["requirements"], entry.requires_env or [])
return toolsets
def get_toolset_requirements(self) -> Dict[str, dict]:
@@ -1295,27 +1209,22 @@ class ToolRegistry:
result: Dict[str, dict] = {}
entries, toolset_checks = self._snapshot_state()
for entry in entries:
ts = entry.toolset
if ts not in result:
result[ts] = {
"name": ts,
"env_vars": [],
"check_fn": toolset_checks.get(ts),
"setup_url": None,
"tools": [],
}
if entry.name not in result[ts]["tools"]:
result[ts]["tools"].append(entry.name)
for env in entry.requires_env:
if env not in result[ts]["env_vars"]:
result[ts]["env_vars"].append(env)
info = result.setdefault(entry.toolset, {
"name": entry.toolset,
"env_vars": [],
"check_fn": toolset_checks.get(entry.toolset),
"setup_url": None,
"tools": [],
})
_extend_unique(info["tools"], [entry.name])
_extend_unique(info["env_vars"], entry.requires_env)
return result
def check_tool_availability(self, quiet: bool = False):
"""Return (available_toolsets, unavailable_info) like the old function."""
available = []
unavailable = []
entries, _ = self._snapshot_state()
entries = self._snapshot_entries()
for ts in sorted({entry.toolset for entry in entries}):
ts_entries = [entry for entry in entries if entry.toolset == ts]
if self._toolset_has_exposable_tools(ts, entries):
@@ -1329,51 +1238,26 @@ class ToolRegistry:
return available, unavailable
def _extend_unique(target: list, items) -> None:
for item in items:
if item not in target:
target.append(item)
# Module-level singleton
registry = ToolRegistry()
# ---------------------------------------------------------------------------
# Helpers for tool response serialization
# ---------------------------------------------------------------------------
# Every tool handler must return a JSON string. These helpers eliminate the
# boilerplate ``json.dumps({"error": msg}, ensure_ascii=False)`` that appears
# hundreds of times across tool files.
#
# Usage:
# from tools.registry import registry, tool_error, tool_result
#
# return tool_error("something went wrong")
# return tool_error("not found", code=404)
# return tool_result(success=True, data=payload)
# return tool_result(items) # pass a dict directly
# Tool handlers must return JSON strings; these replace the ubiquitous
# ``json.dumps({"error": msg}, ensure_ascii=False)`` boilerplate.
def tool_error(message, **extra) -> str:
"""Return a JSON error string for tool handlers.
>>> tool_error("file not found")
'{"error": "file not found"}'
>>> tool_error("bad input", success=False)
'{"error": "bad input", "success": false}'
"""
# Bound the context-bound copy so a raw exception can't bloat history across retries.
result = {"error": _bound_error_text(str(message))}
if extra:
result.update(extra)
return json.dumps(result, ensure_ascii=False)
"""``'{"error": "<message>", **extra}'`` — the error body is bounded so a raw
exception can't bloat history across retries."""
return json.dumps({"error": _bound_error_text(str(message)), **extra}, ensure_ascii=False)
def tool_result(data=None, **kwargs) -> str:
"""Return a JSON result string for tool handlers.
Accepts a dict positional arg *or* keyword arguments (not both):
>>> tool_result(success=True, count=42)
'{"success": true, "count": 42}'
>>> tool_result({"key": "value"})
'{"key": "value"}'
"""
if data is not None:
return json.dumps(data, ensure_ascii=False)
return json.dumps(kwargs, ensure_ascii=False)
"""JSON-encode a dict positional arg *or* keyword arguments (not both)."""
return json.dumps(data if data is not None else kwargs, ensure_ascii=False)
tools/tool_backend_helpers.py
+13 -5
View File
@@ -18,7 +18,8 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"}
def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
"""True when the user is entitled to the Nous Tool Gateway (coarse gate).
"""True when the user is entitled to the Nous Tool Gateway (coarse gate:
paid Portal service access OR a live free tool pool — "is ANY managed tool usable").
Fails closed on unknown/error entitlement — never blocks startup.
Per-category coverage is narrowed by callers via ``tool_gateway_entitled_for``.
@@ -125,7 +126,9 @@ def resolve_modal_backend_state(
def _scoped_credential(name: str) -> str:
"""Read a credential env var under the active profile secret scope."""
"""Read a credential env var under the active profile secret scope. Raw env
fallback only if ``agent.secret_scope`` cannot import — a packaging edge must
never leave the caller without a key."""
try:
from agent.secret_scope import get_secret
@@ -280,6 +283,8 @@ def read_selection(section: str) -> str | None:
text = str(value).strip().lower()
if text:
return text
# use_gateway: false with no name key is not a usable selection shape;
# per-capability web keys still count as configured via selection_exists().
return None
@@ -295,8 +300,10 @@ def selection_exists(section: str) -> bool:
return any(str(raw.get(key) or "").strip() for key in extra)
# Backends that once shipped in-tree but were removed; consulted by the startup
# config check and selection_error() so a stale selection gets a real message.
# Backends that once shipped in-tree but were removed. A config still pointing at
# one would otherwise fail silently at the FIRST tool call with a generic "no
# registered provider has that name" — no migration, no startup notice. Consulted
# by the startup config check and selection_error() so the user gets a real message.
# Add removals here, never as one-off string checks at call sites, e.g.
# "web": {"<name>": "the <Name> backend was removed in vX.Y.Z (...)"},
REMOVED_BACKENDS: Dict[str, Dict[str, str]] = {}
@@ -324,7 +331,8 @@ def selection_error(section: str, selection_name: str, failure: str) -> str:
def fal_key_is_configured() -> bool:
"""True when FAL_KEY is set (scope/env, else ``.env`` for CLI paths that
run before dotenv loads) to a non-whitespace value."""
run before dotenv loads) to a non-whitespace value — so tool-side and CLI
setup-time checks agree; whitespace-only counts as unset everywhere."""
value = _scoped_credential("FAL_KEY") or None
if value is None:
try:
tools/tool_search.py
+173 -196
View File
@@ -38,11 +38,14 @@ from tools.tool_search_catalog import ( # noqa: F401 — re-exported public/tes
_classify_source,
_corpus_stats,
_entry_search_text,
_fn,
_listing_group_label,
_registry_entry,
_short_desc,
_stem,
_tokenize,
build_catalog,
build_catalog_listing,
build_catalog_listing_with_form,
search_catalog,
)
@@ -77,7 +80,8 @@ class ToolSearchConfig:
search_default_limit: int
max_search_limit: int
# Embedded name + short-description manifest so deferred tools stay
# discoverable. "auto"/"on" = include when it fits; "off" = bare bridge.
# DISCOVERABLE (like the skills listing) while full schemas stay deferred.
# "auto"/"on" = include when it fits (names-only, then bare); "off" = bare bridge.
listing: str = "auto" # "auto" | "on" | "off"
# Effective budget = min(listing_max_tokens, threshold_pct% of context).
listing_max_tokens: int = 4000
@@ -98,32 +102,21 @@ class ToolSearchConfig:
return cls(enabled="off" if raw is False else "auto", threshold_pct=5.0,
search_default_limit=5, max_search_limit=25)
enabled = _tri_state(raw.get("enabled", "auto"))
threshold_pct = max(0.0, min(100.0, _safe_float(raw.get("threshold_pct"), 5.0)))
max_search_limit = max(1, min(50, _safe_int(raw.get("max_search_limit"), 25)))
search_default_limit = max(1, min(max_search_limit,
_safe_int(raw.get("search_default_limit"), 5)))
listing = _tri_state(raw.get("listing", "auto"))
listing_max_tokens = max(200, min(60000, _safe_int(raw.get("listing_max_tokens"), 4000)))
max_search_limit = _clamped_int(raw.get("max_search_limit"), 25, 1, 50)
defer_raw = raw.get("defer")
if isinstance(defer_raw, (list, tuple, set)):
defer_tools = frozenset(
str(n).strip() for n in defer_raw if str(n).strip()
)
else:
defer_tools = None # curated default
return cls(
enabled=enabled,
threshold_pct=threshold_pct,
search_default_limit=search_default_limit,
enabled=_tri_state(raw.get("enabled", "auto")),
threshold_pct=max(0.0, min(100.0, _safe_float(raw.get("threshold_pct"), 5.0))),
search_default_limit=_clamped_int(
raw.get("search_default_limit"), 5, 1, max_search_limit),
max_search_limit=max_search_limit,
listing=listing,
listing_max_tokens=listing_max_tokens,
defer_tools=defer_tools,
listing=_tri_state(raw.get("listing", "auto")),
listing_max_tokens=_clamped_int(raw.get("listing_max_tokens"), 4000, 200, 60000),
# A list replaces the curated default wholesale; anything else = curated.
defer_tools=(
frozenset(str(n).strip() for n in defer_raw if str(n).strip())
if isinstance(defer_raw, (list, tuple, set)) else None
),
)
@@ -144,6 +137,11 @@ def _safe_int(value: Any, fallback: int) -> int:
return fallback
def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int:
"""``_safe_int`` clamped to ``[lo, hi]`` (the fallback is clamped too)."""
return max(lo, min(hi, _safe_int(value, fallback)))
def _safe_float(value: Any, fallback: float) -> float:
try:
return float(value)
@@ -218,13 +216,12 @@ def is_deferrable_tool_name(name: str, defer_tools: Optional[frozenset] = None)
return True
if name in _core_tool_names():
return False
entry = _registry_entry(name)
if entry is None:
return False
try:
from tools.registry import registry
entry = registry.get_entry(name)
if entry is None:
return False
return entry.toolset.startswith("mcp-") or entry.toolset not in _DIRECT_SURFACE_TOOLSETS
except Exception:
except Exception: # malformed entry (no str toolset) is never deferrable
return False
@@ -233,11 +230,7 @@ def _describe_classification(
defer_tools: Optional[frozenset] = None,
) -> Literal["available", "not_found", "not_deferrable"]:
"""Classify a describe name without treating unknown names as errors."""
try:
from tools.registry import registry
entry = registry.get_entry(name)
except Exception:
return "not_found"
entry = _registry_entry(name)
if entry is None:
return "not_found"
if defer_tools is not None and name in defer_tools:
@@ -251,6 +244,11 @@ def _describe_classification(
return "available"
def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
"""Function names of a tool-defs list (``""`` for a nameless def)."""
return (_fn(td).get("name", "") for td in tool_defs)
def classify_tools(
tool_defs: List[Dict[str, Any]],
defer_tools: Optional[frozenset] = None,
@@ -259,18 +257,19 @@ def classify_tools(
dropped (they are re-added after classification)."""
visible: List[Dict[str, Any]] = []
deferrable: List[Dict[str, Any]] = []
for td in tool_defs:
fn = td.get("function") or {}
name = fn.get("name", "")
for td, name in zip(tool_defs, _tool_def_names(tool_defs)):
if name in BRIDGE_TOOL_NAMES:
continue
if is_deferrable_tool_name(name, defer_tools):
deferrable.append(td)
else:
visible.append(td)
(deferrable if is_deferrable_tool_name(name, defer_tools) else visible).append(td)
return visible, deferrable
def _deferrable_in(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Deferrable subset of a pre-assembly ``tool_defs`` list under the current
(read-only) user config — the universe the bridge tools operate on."""
return classify_tools(tool_defs, load_config_readonly().effective_defer_tools)[1]
# ---------------------------------------------------------------------------
# Token estimation and threshold gate
# ---------------------------------------------------------------------------
@@ -319,6 +318,57 @@ def listing_token_budget(
# ---------------------------------------------------------------------------
def _bridge_schema(name: str, description: str, properties: Dict[str, Any],
required: List[str]) -> Dict[str, Any]:
"""One OpenAI-style function schema (key order is part of the frozen bytes)."""
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {"type": "object", "properties": properties, "required": required},
},
}
def _search_description(deferred_count: int, listing: Optional[str], listing_form: str) -> str:
"""tool_search bridge description with the listing embedded; ``listing_form``
picks the framing (see :func:`bridge_tool_schemas`)."""
desc = (
f"Search {deferred_count} additional tools that are loaded on demand. "
"Takes a list of queries searched in parallel against the same "
"catalog; send one query per distinct capability you need. Returns "
"matching tool names grouped per query plus a shared map with each "
"tool's description. Follow with "
f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, "
f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this "
"system prompt are already available and do not need to be searched."
)
if not listing:
return desc
if listing_form == "groups":
return desc + (
"\n\nThe servers below are connected and their tools ARE available "
"through this bridge. For any request in these domains, search "
"here FIRST — do not claim the capability is unavailable and do "
"not substitute a generic tool (terminal/browser) without "
"searching.\n\n" + listing
)
desc += (
"\n\nEvery deferred capability is listed below. If a tool name "
"appears here, do NOT claim it is unavailable — load it with "
f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you "
"already see the exact name)."
)
if listing_form == "mixed":
desc += (
" For servers marked 'names not listed', the tools exist "
f"too — find them with `{TOOL_SEARCH_NAME}` before "
"concluding anything is missing."
)
return desc + "\n\n" + listing
def bridge_tool_schemas(
deferred_count: int,
listing: Optional[str] = None,
@@ -329,111 +379,51 @@ def bridge_tool_schemas(
tool_search description; ``listing_form`` picks the framing (per-tool forms
say "skip search when you see the exact name", the "groups" summary says
which domains exist and that search is mandatory)."""
desc_search = (
f"Search {deferred_count} additional tools that are loaded on demand. "
"Takes a list of queries searched in parallel against the same "
"catalog; send one query per distinct capability you need. Returns "
"matching tool names grouped per query plus a shared map with each "
"tool's description. Follow with "
f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, "
f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this "
"system prompt are already available and do not need to be searched."
)
if listing and listing_form == "groups":
desc_search += (
"\n\nThe servers below are connected and their tools ARE available "
"through this bridge. For any request in these domains, search "
"here FIRST — do not claim the capability is unavailable and do "
"not substitute a generic tool (terminal/browser) without "
"searching.\n\n" + listing
)
elif listing:
desc_search += (
"\n\nEvery deferred capability is listed below. If a tool name "
"appears here, do NOT claim it is unavailable — load it with "
f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you "
"already see the exact name)."
)
if listing_form == "mixed":
desc_search += (
" For servers marked 'names not listed', the tools exist "
f"too — find them with `{TOOL_SEARCH_NAME}` before "
"concluding anything is missing."
)
desc_search += "\n\n" + listing
desc_describe = (
f"Load the full JSON schemas for tools returned by `{TOOL_SEARCH_NAME}`. "
f"Required before `{TOOL_CALL_NAME}` if a tool's parameters are unknown. "
"Batch every schema you need into one call."
)
desc_call = (
"Invoke a deferred tool by name with the given arguments. Argument shape "
f"matches the tool's schema (see `{TOOL_DESCRIBE_NAME}`). Policy, hooks, "
"and approvals run exactly as for any directly-listed tool."
)
return [
{
"type": "function",
"function": {
"name": TOOL_SEARCH_NAME,
"description": desc_search,
"parameters": {
"type": "object",
"properties": {
"queries": {
"type": "array",
"items": {"type": "string"},
"description": "Search queries, each a few keywords describing one capability (e.g. ['create github issue', 'send slack message']). Searched in parallel; results come back grouped per query. A single string is accepted and treated as one query.",
},
"limit": {
"type": "integer",
"description": "Maximum number of matches per query. Defaults to 5 and is clamped to the configured maximum (25 by default).",
},
},
"required": ["queries"],
_bridge_schema(
TOOL_SEARCH_NAME,
_search_description(deferred_count, listing, listing_form),
{
"queries": {
"type": "array",
"items": {"type": "string"},
"description": "Search queries, each a few keywords describing one capability (e.g. ['create github issue', 'send slack message']). Searched in parallel; results come back grouped per query. A single string is accepted and treated as one query.",
},
"limit": {
"type": "integer",
"description": "Maximum number of matches per query. Defaults to 5 and is clamped to the configured maximum (25 by default).",
},
},
},
{
"type": "function",
"function": {
"name": TOOL_DESCRIBE_NAME,
"description": desc_describe,
"parameters": {
"type": "object",
"properties": {
"names": {
"type": "array",
"items": {"type": "string"},
"description": "Exact tool names (as returned by tool_search). A single string is accepted and treated as one name.",
},
},
"required": ["names"],
["queries"],
),
_bridge_schema(
TOOL_DESCRIBE_NAME,
f"Load the full JSON schemas for tools returned by `{TOOL_SEARCH_NAME}`. "
f"Required before `{TOOL_CALL_NAME}` if a tool's parameters are unknown. "
"Batch every schema you need into one call.",
{
"names": {
"type": "array",
"items": {"type": "string"},
"description": "Exact tool names (as returned by tool_search). A single string is accepted and treated as one name.",
},
},
},
{
"type": "function",
"function": {
"name": TOOL_CALL_NAME,
"description": desc_call,
"parameters": {
["names"],
),
_bridge_schema(
TOOL_CALL_NAME,
"Invoke a deferred tool by name with the given arguments. Argument shape "
f"matches the tool's schema (see `{TOOL_DESCRIBE_NAME}`). Policy, hooks, "
"and approvals run exactly as for any directly-listed tool.",
{
"name": {"type": "string", "description": "Exact tool name to invoke."},
"arguments": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Exact tool name to invoke.",
},
"arguments": {
"type": "object",
"description": "Arguments for the tool, matching its schema.",
},
},
"required": ["name", "arguments"],
"description": "Arguments for the tool, matching its schema.",
},
},
},
["name", "arguments"],
),
]
@@ -452,7 +442,8 @@ class AssemblyResult:
deferred_tokens: int = 0
threshold_tokens: int = 0
# 0 = passthrough; 1 = bridge + per-tool listing (full/names/mixed);
# 2 = bare bridge / server-summary only.
# 2 = bare bridge / server-summary only — domains stay visible but
# individual tools are reachable only via tool_search.
tier: int = 0
listing_form: str = "none" # "full" | "names" | "mixed" | "groups" | "none"
@@ -469,8 +460,8 @@ def assemble_tool_defs(
if config is None:
config = load_config()
incoming = [td for td in tool_defs
if (td.get("function") or {}).get("name") not in BRIDGE_TOOL_NAMES]
incoming = [td for td, name in zip(tool_defs, _tool_def_names(tool_defs))
if name not in BRIDGE_TOOL_NAMES]
visible, deferrable = classify_tools(incoming, config.effective_defer_tools)
if not deferrable:
@@ -487,15 +478,13 @@ def assemble_tool_defs(
tier=0,
)
listing = None
listing_form = "none"
listing, listing_form = None, "none"
listing_budget = listing_token_budget(config, context_length)
if config.listing != "off":
listing, listing_form = build_catalog_listing_with_form(
deferrable, max_tokens=listing_budget)
bridge = bridge_tool_schemas(len(deferrable), listing=listing,
listing_form=listing_form)
result = visible + bridge
tier = 1 if listing_form in ("full", "names", "mixed") else 2
logger.info(
@@ -506,7 +495,7 @@ def assemble_tool_defs(
)
return AssemblyResult(
tool_defs=result,
tool_defs=visible + bridge,
activated=True,
deferred_count=len(deferrable),
deferred_tokens=deferrable_tokens,
@@ -531,12 +520,8 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
trivial call without a ``tool_describe`` round-trip."""
schema = entry.schema if isinstance(entry.schema, dict) else {}
fn = schema.get("function")
if not isinstance(fn, dict):
fn = {}
params = fn.get("parameters")
if not isinstance(params, dict):
params = {}
required = params.get("required")
params = fn.get("parameters") if isinstance(fn, dict) else None
required = params.get("required") if isinstance(params, dict) else None
if not isinstance(required, list):
required = []
return {
@@ -549,24 +534,36 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any]]:
"""Deterministic ``[{name, tool_count}]`` of connected sources, attached to
empty query groups so a lexical miss is not read as a missing capability."""
empty query groups so a lexical miss is not read as a missing capability
(adds nothing to the fixed per-turn prompt)."""
counts = Counter(_listing_group_label(entry.source_name) for entry in catalog)
return [{"name": name, "tool_count": counts[name]} for name in sorted(counts)]
def _string_list(raw: Any, *, dedupe: bool) -> Optional[List[str]]:
"""Normalize a list-of-strings argument. A bare string (a common model slip)
counts as a one-item list; non-list input returns None."""
def _string_list_arg(
args: Dict[str, Any], key: str, *, dedupe: bool, max_items: int, retry_hint: str,
) -> Tuple[Optional[List[str]], Optional[str]]:
"""Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare
string (a common model slip) counts as a one-item list. Rejects non-list
input, an empty list (after stripping blanks), and more than ``max_items``
(bounds the work one bridge call can request)."""
raw = args.get(key)
if isinstance(raw, str):
raw = [raw]
if not isinstance(raw, list):
return None
return None, tool_error(f"{key} is required and must be an array of strings")
out: List[str] = []
for item in raw:
text = str(item or "").strip()
if text and (not dedupe or text not in out):
out.append(text)
return out
if not out:
return None, tool_error(
f"{key} is required and must contain at least one non-empty string")
if len(out) > max_items:
return None, tool_error(
f"too many {key}: {len(out)} > max {max_items}. {retry_hint}")
return out, None
def dispatch_tool_search(args: Dict[str, Any],
@@ -585,27 +582,19 @@ def dispatch_tool_search(args: Dict[str, Any],
if config is None:
config = load_config()
queries = _string_list(args.get("queries"), dedupe=False)
if queries is None:
return tool_error("queries is required and must be an array of strings")
if not queries:
return tool_error("queries is required and must contain at least one non-empty string")
if len(queries) > _MAX_QUERIES_PER_CALL:
return tool_error(
f"too many queries: {len(queries)} > max {_MAX_QUERIES_PER_CALL}. "
"Retry with fewer, more targeted queries."
)
queries, err = _string_list_arg(
args, "queries", dedupe=False, max_items=_MAX_QUERIES_PER_CALL,
retry_hint="Retry with fewer, more targeted queries.")
if err:
return err
raw_limit = args.get("limit")
if raw_limit is None:
limit = config.search_default_limit
else:
limit = max(1, min(config.max_search_limit, _safe_int(raw_limit, config.search_default_limit)))
limit = _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit)
_, deferrable = classify_tools(
current_tool_defs, load_config_readonly().effective_defer_tools
)
catalog = build_catalog(deferrable)
catalog = build_catalog(_deferrable_in(current_tool_defs))
results: List[Dict[str, Any]] = []
tools_map: Dict[str, Dict[str, Any]] = {}
@@ -614,8 +603,7 @@ def dispatch_tool_search(args: Dict[str, Any],
for query in queries:
hits = search_catalog(catalog, query, limit=limit, corpus_stats=corpus_stats)
for h in hits:
if h.name not in tools_map:
tools_map[h.name] = _shared_tool_record(h)
tools_map.setdefault(h.name, _shared_tool_record(h))
group: Dict[str, Any] = {"query": query, "matches": [h.name for h in hits]}
if not hits and catalog:
group["available_sources"] = available_sources
@@ -627,13 +615,12 @@ def dispatch_tool_search(args: Dict[str, Any],
)
results.append(group)
result: Dict[str, Any] = {
return json.dumps({
"queries": queries,
"total_available": len(catalog),
"results": results,
"tools": tools_map,
}
return json.dumps(result, ensure_ascii=False)
}, ensure_ascii=False)
def dispatch_tool_describe(args: Dict[str, Any],
@@ -651,25 +638,14 @@ def dispatch_tool_describe(args: Dict[str, Any],
if config is None:
config = load_config_readonly()
names = _string_list(args.get("names"), dedupe=True)
if names is None:
return tool_error("names is required and must be an array of strings")
if not names:
return tool_error("names is required and must contain at least one non-empty string")
if len(names) > _MAX_DESCRIBE_NAMES_PER_CALL:
return tool_error(
f"too many names: {len(names)} > max {_MAX_DESCRIBE_NAMES_PER_CALL}. "
"Retry with fewer names per call."
)
names, err = _string_list_arg(
args, "names", dedupe=True, max_items=_MAX_DESCRIBE_NAMES_PER_CALL,
retry_hint="Retry with fewer names per call.")
if err:
return err
_, deferrable = classify_tools(
current_tool_defs, load_config_readonly().effective_defer_tools
)
by_name: Dict[str, Dict[str, Any]] = {}
for td in deferrable:
fn = td.get("function") or {}
if fn.get("name"):
by_name[fn["name"]] = fn
deferrable = _deferrable_in(current_tool_defs)
by_name = {name: _fn(td) for td, name in zip(deferrable, _tool_def_names(deferrable)) if name}
tools: Dict[str, Dict[str, Any]] = {}
not_found: List[str] = []
@@ -707,7 +683,7 @@ def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
out-of-scope tool via the bridge."""
defer_tools = load_config_readonly().effective_defer_tools
return frozenset(
name for name in ((td.get("function") or {}).get("name", "") for td in tool_defs)
name for name in _tool_def_names(tool_defs)
if name and is_deferrable_tool_name(name, defer_tools)
)
@@ -753,6 +729,7 @@ __all__ = [
"estimate_tokens_from_schemas",
"should_activate",
"build_catalog",
"build_catalog_listing",
"build_catalog_listing_with_form",
"listing_token_budget",
"search_catalog",
tools/tool_search_catalog.py
+56 -38
View File
@@ -63,6 +63,22 @@ def _tokenize(text: str) -> List[str]:
return [_stem(token.lower()) for token in _TOKEN_RE.findall(text)]
def _fn(td: Dict[str, Any]) -> Dict[str, Any]:
"""The ``function`` block of a tool-def (``{}`` when absent/None)."""
return td.get("function") or {}
def _registry_entry(name: str) -> Any:
"""Registry entry for ``name``; None when unregistered OR when the registry
is unavailable/raises (lookup failures must never fail a bridge call).
The import stays lazy: tests patch ``tools.registry.registry``."""
try:
from tools.registry import registry
return registry.get_entry(name)
except Exception:
return None
def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
"""Search-text blob: split name words + source label + description +
top-level parameter names. Schema bodies are excluded (noise, no recall
@@ -70,7 +86,7 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
its IDF is ~0. The source label lets a service-name query ("linear") reach
a tool whose own name omits the vendor.
"""
fn = td.get("function") or {}
fn = _fn(td)
name = fn.get("name", "")
if name.startswith("mcp__"):
name = name[len("mcp__"):]
@@ -84,15 +100,12 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
def _classify_source(name: str) -> Tuple[str, str]:
"""Return (source_kind, source_name) for a registered tool name."""
entry = _registry_entry(name)
if entry is None:
return ("other", "")
try:
from tools.registry import registry
entry = registry.get_entry(name)
if entry is None:
return ("other", "")
if entry.toolset.startswith("mcp-"):
return ("mcp", entry.toolset)
return ("plugin", entry.toolset)
except Exception:
return ("mcp" if entry.toolset.startswith("mcp-") else "plugin", entry.toolset)
except Exception: # malformed entry (no str toolset)
return ("other", "")
@@ -100,7 +113,7 @@ def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]:
"""Build the deferred-tool catalog from the deferrable subset of tool-defs."""
catalog: List[CatalogEntry] = []
for td in tool_defs:
fn = td.get("function") or {}
fn = _fn(td)
name = fn.get("name", "")
if not name:
continue
@@ -123,7 +136,7 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str],
doc_freq: Dict[str, int], n_docs: int,
k1: float = 1.5, b: float = 0.75) -> float:
"""Standard BM25 score for one query against one document (inlined; the
catalog is small enough that a dependency is not worth it)."""
catalog is bounded — typically < 500 tools — so a dependency is not worth it)."""
if not doc_tokens:
return 0.0
score = 0.0
@@ -204,7 +217,8 @@ _SENTENCE_END_RE = re.compile(r"(?<!\be\.g)(?<!\bi\.e)(?<!\betc)[.!?](?=\s|$)")
def _short_desc(description: str, max_chars: int = 60) -> str:
"""First sentence of a tool description, clipped to ``max_chars`` on a
word boundary. Linear-time on hostile input."""
word boundary. ``e.g.``/``i.e.``/``etc.`` do not end a sentence; whitespace
normalization and the regex search stay linear-time on hostile input."""
text = " ".join((description or "").split())
if not text:
return ""
@@ -222,9 +236,17 @@ def _short_desc(description: str, max_chars: int = 60) -> str:
def _listing_group_label(source_name: str) -> str:
"""Human-facing group heading for a toolset, e.g. ``mcp-github`` -> ``github``."""
label = source_name or "other"
if label.startswith("mcp-"):
label = label[4:]
return label
return label[4:] if label.startswith("mcp-") else label
def build_catalog_listing(
deferrable: List[Dict[str, Any]],
*,
max_tokens: int = 4000,
) -> Optional[str]:
"""Render the deferred-catalog manifest; text only (see
:func:`build_catalog_listing_with_form` for the degradation ladder)."""
return build_catalog_listing_with_form(deferrable, max_tokens=max_tokens)[0]
def build_catalog_listing_with_form(
@@ -250,12 +272,13 @@ def build_catalog_listing_with_form(
groups: Dict[str, List[Tuple[str, str]]] = {}
for td in deferrable:
fn = td.get("function") or {}
fn = _fn(td)
name = fn.get("name", "")
if not name:
continue
source, source_name = _classify_source(name)
label = _listing_group_label(source_name if source != "other" else "other")
# ``_classify_source`` returns ("other", "") for unregistered names and
# ``_listing_group_label("")`` is "other", so one call covers both.
label = _listing_group_label(_classify_source(name)[1])
groups.setdefault(label, []).append((name, _short_desc(fn.get("description", ""))))
if not groups:
@@ -269,8 +292,7 @@ def build_catalog_listing_with_form(
f"discover via `{TOOL_SEARCH_NAME}`)")
lines = [f"{label} tools ({len(tools)}):"]
if mode == "full":
for name, desc in tools:
lines.append(f"- {name}: {desc}" if desc else f"- {name}")
lines.extend(f"- {name}: {desc}" if desc else f"- {name}" for name, desc in tools)
else:
lines.append(", ".join(name for name, _ in tools))
return "\n".join(lines)
@@ -278,30 +300,26 @@ def build_catalog_listing_with_form(
header = ("Deferred tool catalog (call schemas via "
f"`{TOOL_DESCRIBE_NAME}`, invoke via `{TOOL_CALL_NAME}`):")
def assemble(modes: Dict[str, str]) -> str:
return "\n".join([header] + [render_group(lbl, modes[lbl])
for lbl in sorted(groups)])
def assemble_if_fits(modes: Dict[str, str]) -> Optional[str]:
text = "\n".join([header] + [render_group(lbl, modes[lbl]) for lbl in sorted(groups)])
return text if math.ceil(len(text) / CHARS_PER_TOKEN) <= max_tokens else None
def fits(text: str) -> bool:
return math.ceil(len(text) / CHARS_PER_TOKEN) <= max_tokens
# 1. Everything full.
modes = {lbl: "full" for lbl in groups}
if fits(assemble(modes)):
return assemble(modes), "full"
# 2. Everything names-only.
modes = {lbl: "names" for lbl in groups}
if fits(assemble(modes)):
return assemble(modes), "names"
# 1. Everything full. 2. Everything names-only.
for mode in ("full", "names"):
modes = {lbl: mode for lbl in groups}
text = assemble_if_fits(modes)
if text is not None:
return text, mode
# 3. Per-server degradation: collapse the LARGEST rendered groups first
# (deterministic: size then label).
# (deterministic: size then label) so one oversized server does not
# cost a small co-attached server its per-tool names.
by_size = sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl))
for lbl in by_size:
modes[lbl] = "summary"
if fits(assemble(modes)):
text = assemble_if_fits(modes)
if text is not None:
form = "groups" if all(m == "summary" for m in modes.values()) else "mixed"
return assemble(modes), form
return text, form
return None, "none"
tools/tool_search_names.py
+2
View File
@@ -1,5 +1,7 @@
"""Reserved bridge tool names shared by tool_search and its catalog module."""
# Reserved: a user/plugin/MCP tool may not take these names — the registry's
# override protection rejects such registrations.
TOOL_SEARCH_NAME = "tool_search"
TOOL_DESCRIBE_NAME = "tool_describe"
TOOL_CALL_NAME = "tool_call"
tools/tool_search_validation.py
+3 -5
View File
@@ -107,13 +107,11 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
if not isinstance(schema, dict):
return None
fn = schema.get("function") if schema.get("type") == "function" else schema
if not isinstance(fn, dict):
return None
params = fn.get("parameters")
params = fn.get("parameters") if isinstance(fn, dict) else None
if not isinstance(params, dict):
return None
required = params.get("required")
if isinstance(required, list) and required:
if isinstance(required, list):
missing = [r for r in required if isinstance(r, str) and r not in args]
if missing:
return _validation_error(
@@ -131,7 +129,7 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
return None
# Validate the repaired shape dispatch will see; copy because
# coerce_tool_args may normalize in place.
# coerce_tool_args may normalize in place (dispatch re-coerces canonically).
candidate_args = dict(args)
try:
from model_tools import coerce_tool_args