kiln_ai.sandbox.tools_surface

  1"""Shared definition of the synthetic ``kiln.tools`` module surface.
  2
  3Stdlib only — no Pydantic / Kiln-model / DB / UI imports.
  4
  5This is the single source of truth for the ``kiln`` / ``kiln.tools`` /
  6``kiln.async_tools`` surface that user code (``from kiln import tools``) sees.
  7Both the sandbox runtime (``sandbox/tools_api.py``) and the test shim
  8(``tool_testing/``) build the exact same module objects from here, so runtime
  9and tests can never drift — a test that catches ``kiln.tools.ToolCallError``
 10catches the same class the runtime raises.
 11
 12The surface is parameterized over a *bridge* (see :class:`ToolBridge`): the
 13sandbox injects its real IPC bridge, the shim injects an in-process fake. This
 14module contains no execution, IPC, or subprocess behavior of its own.
 15"""
 16
 17from __future__ import annotations
 18
 19import asyncio
 20import sys
 21import types
 22from typing import Any, Protocol
 23
 24# ---------------------------------------------------------------------------
 25# Typed exceptions
 26# ---------------------------------------------------------------------------
 27
 28
 29class ToolNotAllowed(Exception):
 30    """The requested tool is not in this code tool's allowlist."""
 31
 32    def __init__(self, tool: str, message: str) -> None:
 33        self.tool = tool
 34        self.message = message
 35        self.raw: str | None = None
 36        super().__init__(message)
 37
 38
 39class ToolTimeout(Exception):
 40    """A nested tool call timed out."""
 41
 42    def __init__(self, tool: str, message: str) -> None:
 43        self.tool = tool
 44        self.message = message
 45        self.raw: str | None = None
 46        super().__init__(message)
 47
 48
 49class ToolCallError(Exception):
 50    """Catch-all for nested tool-call failures."""
 51
 52    def __init__(self, tool: str, message: str, raw: str | None = None) -> None:
 53        self.tool = tool
 54        self.message = message
 55        self.raw = raw
 56        super().__init__(message)
 57
 58
 59# ---------------------------------------------------------------------------
 60# Bridge protocol
 61# ---------------------------------------------------------------------------
 62
 63
 64class ToolBridge(Protocol):
 65    """Behavior the synthetic modules require of a bridge.
 66
 67    ``call`` receives the tool name plus the positional/keyword arguments the
 68    user passed to the proxy and returns the tool's raw output string (or raises
 69    one of the typed exceptions). ``list_tools`` returns the tool declarations.
 70    """
 71
 72    def call(self, name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str: ...
 73
 74    def list_tools(self) -> list[dict[str, Any]]: ...
 75
 76
 77# ---------------------------------------------------------------------------
 78# Synthetic module objects
 79# ---------------------------------------------------------------------------
 80
 81
 82class _SyncToolsModule(types.ModuleType):
 83    """``kiln.tools`` — sync callable proxies for tool calls."""
 84
 85    def __getattr__(self, name: str) -> Any:
 86        if name.startswith("_"):
 87            raise AttributeError(name)
 88        bridge: ToolBridge = self._bridge  # type: ignore[attr-defined]
 89        return lambda *args, **kw: bridge.call(name, args, kw)
 90
 91
 92class _AsyncToolsModule(types.ModuleType):
 93    """``kiln.async_tools`` — awaitable proxies for tool calls."""
 94
 95    def __getattr__(self, name: str) -> Any:
 96        if name.startswith("_"):
 97            raise AttributeError(name)
 98        bridge: ToolBridge = self._bridge  # type: ignore[attr-defined]
 99
100        async def _async_proxy(*args: Any, **kw: Any) -> str:
101            return await asyncio.to_thread(bridge.call, name, args, kw)
102
103        return _async_proxy
104
105
106def build_tools_modules(
107    bridge: ToolBridge,
108) -> tuple[types.ModuleType, _SyncToolsModule, _AsyncToolsModule]:
109    """Build the ``kiln`` / ``kiln.tools`` / ``kiln.async_tools`` modules.
110
111    Wires the modules to *bridge* and returns them WITHOUT installing them into
112    ``sys.modules`` (see :func:`install_tools_modules_for_bridge`).
113    """
114    # -- kiln --
115    kiln_mod = types.ModuleType("kiln")
116    kiln_mod.__path__ = []  # type: ignore[attr-defined]  # make it a package so `from kiln import ...` works
117
118    # -- kiln.tools (sync) --
119    tools_mod = _SyncToolsModule("kiln.tools")
120    tools_mod._bridge = bridge  # type: ignore[attr-defined]
121    tools_mod.ToolNotAllowed = ToolNotAllowed  # type: ignore[attr-defined]
122    tools_mod.ToolTimeout = ToolTimeout  # type: ignore[attr-defined]
123    tools_mod.ToolCallError = ToolCallError  # type: ignore[attr-defined]
124    tools_mod.list_tools = bridge.list_tools  # type: ignore[attr-defined]
125
126    # -- kiln.async_tools (async) --
127    async_tools_mod = _AsyncToolsModule("kiln.async_tools")
128    async_tools_mod._bridge = bridge  # type: ignore[attr-defined]
129    async_tools_mod.ToolNotAllowed = ToolNotAllowed  # type: ignore[attr-defined]
130    async_tools_mod.ToolTimeout = ToolTimeout  # type: ignore[attr-defined]
131    async_tools_mod.ToolCallError = ToolCallError  # type: ignore[attr-defined]
132
133    async def _async_list_tools() -> list[dict[str, Any]]:
134        return await asyncio.to_thread(bridge.list_tools)
135
136    async_tools_mod.list_tools = _async_list_tools  # type: ignore[attr-defined]
137
138    kiln_mod.tools = tools_mod  # type: ignore[attr-defined]
139    kiln_mod.async_tools = async_tools_mod  # type: ignore[attr-defined]
140
141    return kiln_mod, tools_mod, async_tools_mod
142
143
144def install_tools_modules_for_bridge(
145    bridge: ToolBridge,
146) -> tuple[types.ModuleType, _SyncToolsModule, _AsyncToolsModule]:
147    """Build and install the synthetic ``kiln`` package into ``sys.modules``.
148
149    Returns the built modules. After this call ``from kiln import tools`` (and
150    ``kiln.async_tools``) resolves against these objects.
151    """
152    kiln_mod, tools_mod, async_tools_mod = build_tools_modules(bridge)
153
154    sys.modules["kiln"] = kiln_mod
155    sys.modules["kiln.tools"] = tools_mod
156    sys.modules["kiln.async_tools"] = async_tools_mod
157
158    return kiln_mod, tools_mod, async_tools_mod