5ce2056521
Part of #3629 (the `admin_tools.py` bullet). Moves the config/integration admin tools off the legacy elif dispatch chain in tool_implementations.py onto the agent_tools registry: manage_endpoints, manage_mcp, manage_webhooks, manage_tokens, manage_settings The do_* implementations (and manage_mcp's command-allowlist / RCE guard: _validate_mcp_command, _mcp_allowed_commands, and the _MCP_* constants) move verbatim into the new src/agent_tools/admin_tools.py. They register through a single ADMIN_TOOL_HANDLERS map that TOOL_HANDLERS.update()s, and the five elif branches plus their imports are dropped from tool_execution.py, so these tools now flow through _direct_fallback like the other migrated clusters. The names are re-exported from src.agent_tools for back-compat. Dedup: - _parse_tool_args was duplicated in tool_implementations.py and document_tools.py. It now lives once in src.tool_utils (which imports nothing from the project beyond src.constants, so this introduces no cycle) and both call sites import it from there. The orphaned `import json` in document_tools is removed with it. - The five tools share one _owner_adapter(fn) factory that threads ctx["owner"] into the owner-taking do_* signature, instead of five near-identical wrappers. Tests: new tests/test_admin_tools_registry.py pins the registration, the re-export back-compat, the owner-threading adapter, and the single-source _parse_tool_args (across admin_tools and document_tools). Existing MCP / settings / webhook suites are repointed at the new module.
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
"""
|
|
This module intentionally imports NOTHING from the project (except
|
|
src.constants which imports nothing from src). Adding a project import here
|
|
will reintroduce the circular dependency that this module exists to break.
|
|
"""
|
|
|
|
import json
|
|
|
|
from src.constants import MAX_OUTPUT_CHARS
|
|
|
|
_mcp_manager = None
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP Manager singleton
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def set_mcp_manager(manager):
|
|
"""Set the global MCP manager instance."""
|
|
global _mcp_manager
|
|
_mcp_manager = manager
|
|
|
|
def get_mcp_manager():
|
|
"""Get the global MCP manager instance."""
|
|
return _mcp_manager
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
|
|
"""
|
|
Truncate text to *limit* characters with a suffix note.
|
|
|
|
Callers treat the result as text, so always return a string: coerce a
|
|
non-string (None -> "", otherwise str(...)) instead of returning it raw,
|
|
which would just move the crash downstream.
|
|
"""
|
|
if not isinstance(text, str):
|
|
text = "" if text is None else str(text)
|
|
if len(text) > limit:
|
|
return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
|
|
return text
|
|
|
|
|
|
def _parse_tool_args(content):
|
|
"""Parse a tool-call argument blob.
|
|
|
|
Accepts either a JSON string or an already-decoded dict. Unwraps the
|
|
common `{"body": {...}}` envelope that smaller models emit when they
|
|
read tool descriptions like "Body is JSON: {...}" literally and
|
|
pass `body` as a field name rather than treating it as a noun.
|
|
|
|
Returns a dict on success, raises ValueError on bad JSON.
|
|
"""
|
|
if isinstance(content, str):
|
|
try:
|
|
args = json.loads(content) if content.strip() else {}
|
|
except (json.JSONDecodeError, TypeError) as e:
|
|
raise ValueError(str(e))
|
|
elif isinstance(content, dict):
|
|
args = content
|
|
else:
|
|
args = {}
|
|
# Unwrap {"body": {...}} envelope, but only if `body` is the sole key
|
|
# and points at a dict. We don't want to clobber a legitimate `body`
|
|
# field on tools where it's a real arg (e.g. send_email body text).
|
|
if (
|
|
isinstance(args, dict)
|
|
and len(args) == 1
|
|
and "body" in args
|
|
and isinstance(args["body"], dict)
|
|
and "action" in args["body"] # extra safety: only unwrap if the inner dict looks like a tool call
|
|
):
|
|
args = args["body"]
|
|
return args
|