"""Shared helpers used across tool implementation domains. Extracted from tool_implementations.py as part of slice 1 (#4082/#4071). Domain modules under src/tools/ import from here. """ import json from typing import Dict, Optional from core.constants import internal_api_base 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 — they 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 # In-process loopback base for agent tools that call Odysseus's own API # (cookbook state, model serve, gallery, email, calendar). We ride the # per-process internal token so require_admin lets us through. See # core/middleware.py. Resolution (override / APP_PORT / 7000) lives in # core.constants.internal_api_base(). _INTERNAL_BASE = internal_api_base() def _internal_headers(owner: Optional[str] = None) -> Dict[str, str]: from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN headers = {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN} if owner: headers["X-Odysseus-Owner"] = owner return headers