"""Registry-based tool manager for agent execution. Registers builtin, client-specified, and local-python tools. MCP lifecycle lives on the runner — see designs/RUNNER_MCP.md. """ from __future__ import annotations import logging import shutil from pathlib import Path from typing import Any from omnigent.errors import ErrorCode, OmnigentError from omnigent.inner.os_env import OSEnvironment from omnigent.spec import AgentSpec from omnigent.spec.types import SharePolicy, ToolRuntime from omnigent.tools._srt import is_srt_available from omnigent.tools.base import Tool, ToolContext, is_valid_tool_name from omnigent.tools.builtins import ( ListCommentsTool, LoadSkillTool, ReadSkillFileTool, SysAdviseModelsTool, SysAgentDownloadTool, SysAgentGetTool, SysAgentListTool, SysCallAsyncTool, SysCancelAsyncTool, SysListModelsTool, SysReadInboxTool, SysSessionCloseTool, SysSessionCreateTool, SysSessionGetHistoryTool, SysSessionGetInfoTool, SysSessionListTool, SysSessionSendTool, SysSessionShareTool, SysTimerCancelTool, SysTimerSetTool, UpdateCommentTool, any_skill_has_resources, get_builtin_tool, ) from omnigent.tools.client_specified import ClientSideTool, ClientSideToolSpec from omnigent.tools.local import load_local_python_tools # MCP lifecycle moved to runner; see designs/RUNNER_MCP.md. _logger = logging.getLogger(__name__) class _UCFunctionSchemaTool(Tool): """Schema-only tool entry for UC function tools. UC function tools are dispatched by the runner via the SQL Statement Execution API — the tool manager only exposes the schema to the LLM. This avoids the ``ClientSideTool`` path (which routes to ``action_required`` / client tunneling) and the ``LocalCallableTool`` path (which requires a server-side callable). :param tool_name: Tool name shown to the LLM, e.g. ``"classify_text"``. :param schema: OpenAI-format tool schema dict. """ def __init__(self, tool_name: str, schema: dict[str, object]) -> None: self._name = tool_name self._schema = schema def name(self) -> str: # type: ignore[override] """Return the tool name. :returns: The tool name, e.g. ``"classify_text"``. """ return self._name def description(self) -> str: # type: ignore[override] """Return the tool description from the schema. :returns: The description string, or empty string if none was provided in the schema. """ func = self._schema.get("function", {}) return func.get("description", "") if isinstance(func, dict) else "" def get_schema(self) -> dict[str, object]: """Return the OpenAI-format tool schema. :returns: A dict with ``"type": "function"`` and the ``"function"`` sub-dict. """ return self._schema class ToolManager: """Registry-based tool manager for a single workflow execution. Registers at init: skill tools (if any skills), builtins, local-python tools, and client-specified tools. MCP tools are runner-owned (designs/RUNNER_MCP.md) and never registered here. """ def __init__( self, spec: AgentSpec, client_tool_specs: list[ClientSideToolSpec] | None = None, workdir: Path | None = None, sandbox_enabled: bool = True, os_env: OSEnvironment | None = None, ) -> None: """ Initialize the tool manager and register built-in, client-specified, and local tools. MCP tools are not registered until ``start()`` is called. :param spec: The parsed AgentSpec defining which tools (skills, MCP servers) are available. :param client_tool_specs: Optional list of :class:`ClientSideToolSpec` objects supplied by the API caller at request time, e.g. ``[ClientSideToolSpec(name="get_weather", ...)]``. ``None`` and ``[]`` are equivalent (no client tools). :param workdir: The extracted agent image directory on disk. Required for local tool loading. ``None`` skips local tool registration, e.g. ``Path("/tmp/cache/ag_abc123")``. :param sandbox_enabled: Runtime policy for ``srt`` sandboxing. ``True`` enables sandboxing when ``srt`` is on PATH. This is a deployment decision from ``RuntimeCaps``, not an agent config setting. :param os_env: Pre-resolved primary OSEnvironment from the ``SessionResourceRegistry``. When provided, ``sys_os_*`` tools use this shared instance instead of creating their own. ``None`` falls back to per-call creation via ``create_os_environment()``. """ self._spec = spec self._workdir = workdir self._sandbox_enabled = sandbox_enabled self._pre_resolved_os_env = os_env self._started = False self._tools: dict[str, Tool] = {} self._srt_available = is_srt_available() self._uv_available = shutil.which("uv") is not None # Track the OS environment instance the spec opts into so # ``shutdown`` can close it cleanly. ``None`` when the spec # didn't declare ``os_env``. self._os_env: OSEnvironment | None = None self._register_skill_tools() self._register_builtin_tools() self._register_sub_agent_tools() self._register_agent_mgmt_tools() self._register_os_env_tools() self._register_terminal_tools() self._register_local_tools(workdir) self._register_client_tools(client_tool_specs or []) # Step 11a: register the async-dispatch builtins # (sys_call_async, sys_read_inbox, sys_cancel_async). # ``AgentSpec.async_enabled`` defaults to True (matches # the legacy inner stack) — agents that want to suppress # the surface set ``async: false`` explicitly. Lookup of # the dispatch target uses the runtime ContextVar, so # registration only cares about the gate — no # construction-time wiring. self._register_async_inbox_tools() # Step 10: register sys_timer_set / sys_timer_cancel when # the spec opts in via ``timers: true``. Defaults off # (matches the inner stack's ``AgentDef.timers`` default). # Firings ride the same ``async_work_complete`` drain path # as ``sys_call_async``, but with ``kind="timer"`` so they # don't block end-of-turn auto-collect. self._register_timer_tools() # Task lifecycle builtin (sys_cancel_task) is auto-enabled # at the end so any AP-created background handle can point # at a stable cancel tool. self._register_task_lifecycle_tools() # Comment tools are always auto-registered so agents can # list and update review comments without the spec opting in. self._register_comment_tools() # Policy tool is always auto-registered so agents can add # inline CEL policies at runtime without spec changes. self._register_policy_tools() # Embedded-browser tools are always auto-registered so any agent # can drive the desktop app's browser without the spec opting in # (framework-owned). self._register_browser_tools() def _register_policy_tools(self) -> None: """ Auto-register ``sys_add_policy`` and ``sys_policy_registry``. Always available so the agent can browse available policy templates and add CEL or builtin policies to the current session at runtime. The runner dispatches both tools via the Omnigent server's REST endpoints. """ from omnigent.tools.builtins.policy import SysAddPolicyTool, SysPolicyRegistryTool self._tools[SysAddPolicyTool.name()] = SysAddPolicyTool() self._tools[SysPolicyRegistryTool.name()] = SysPolicyRegistryTool() def _register_async_inbox_tools(self) -> None: """ Register the async-dispatch builtins when the agent spec has ``async`` enabled (parsed onto :attr:`AgentSpec.async_enabled`, which defaults to ``True``). Registers (in order): * :class:`SysCallAsyncTool` — fire-and-forget dispatcher for any local Python tool (11a.i). * :class:`SysReadInboxTool` — pull-mode drain of completed async-work payloads (11a.ii). * :class:`SysCancelAsyncTool` — cancel a dispatched task by handle id; thin alias over the always-registered ``sys_cancel_task`` (11a.iii). When the spec sets ``async: false`` explicitly, none of the async-namespace builtins register — the flag is the kill-switch for agents that want a minimal-tools surface. See ``designs/SERVER_HARNESS_CONTRACT.md`` §Async work + inbox for the design rationale (including why the default flipped to ``True`` post-step-11a.iii). """ if not self._spec.async_enabled: return self._tools[SysCallAsyncTool.name()] = SysCallAsyncTool() self._tools[SysReadInboxTool.name()] = SysReadInboxTool() self._tools[SysCancelAsyncTool.name()] = SysCancelAsyncTool() def _register_timer_tools(self) -> None: """ Register the timer builtins when the agent spec opts in. Gated on :attr:`AgentSpec.timers` (defaults to ``False`` to match ``omnigent/inner/datamodel.py::AgentDef.timers``). Agents that want timer scheduling declare ``timers: true`` at the top level of their YAML. Registers (in order): * :class:`SysTimerSetTool` — schedules a timer; returns the ``timer_id`` synchronously, fires later via the ``async_work_complete`` drain. * :class:`SysTimerCancelTool` — cancels a scheduled timer by id. Step 10 of the harness contract migration. See ``designs/SERVER_HARNESS_CONTRACT.md`` §Timers and the runner-side timer task implementation for the firing workflow. :raises ValueError: If a ``sys_timer_*`` name collides with an already-registered tool. Defensive — should not happen given the standard registration order. """ if not self._spec.timers: return for tool in (SysTimerSetTool(), SysTimerCancelTool()): if tool.name() in self._tools: raise ValueError( f"sys_timer_* tool {tool.name()!r} collides with an already-registered tool" ) self._tools[tool.name()] = tool def _register_skill_tools(self) -> None: """ Register built-in skill tools. Always registers ``load_skill`` — it discovers host-scope skills (``~/.claude/skills/``, ``.agents/skills/``, etc.) at init time even when the agent has no bundled skills. Registers ``read_skill_file`` only when at least one skill has bundled resource files. """ load_tool = LoadSkillTool( self._spec.skills, agent_root=self._workdir, skills_filter=self._spec.skills_filter, ) self._tools[load_tool.name()] = load_tool # Combine bundled + discovered skills for resource check. all_skills = load_tool.skills if any_skill_has_resources(all_skills): read_tool = ReadSkillFileTool(all_skills) self._tools[read_tool.name()] = read_tool def _register_builtin_tools(self) -> None: """ Register built-in tools declared in ``tools.builtins``. Most tools are looked up in the built-in registry and instantiated with spec-level config. Some (``web_search``, ``web_fetch``, ``upload_file``) need runtime context the registry doesn't have and are dispatched through :meth:`_create_builtin` instead. """ for entry in self._spec.tools.builtins: tool = self._create_builtin(entry.name, entry.config) if tool is None: _logger.warning( "Unknown built-in tool %r — skipping. " "Available: web_search, web_fetch, upload_file, " "list_files, download_file, " "search_conversations, export_agent. " "Note: list_comments and update_comment are " "framework-owned and auto-registered; they cannot " "be declared in the spec. " "(Interactive shells now go through the " "spec-level `terminals:` block + sys_terminal_*; " "one-shot shells use sys_os_shell.)", entry.name, ) continue self._tools[tool.name()] = tool def _create_builtin( self, name: str, config: dict[str, str] | None, ) -> Tool | None: """ Instantiate a built-in tool by name. :param name: The builtin name from the spec. :param config: Optional spec-level config dict. :returns: A :class:`Tool` instance, or ``None``. """ if name == "web_search": return self._create_web_search(config) if name == "web_fetch": return self._create_web_fetch() if name == "upload_file": from omnigent.tools.builtins.upload_file import UploadFileTool return UploadFileTool() return get_builtin_tool(name, config=config) def _create_web_search(self, config: dict[str, str] | None) -> Tool: """ Build a :class:`WebSearchTool` for the parent's LLM. Uses ``parse_model_string`` to infer the provider, except for ``databricks-*`` models which don't support the native ``web_search_preview`` schema and fall back to function-tool mode. :param config: Spec-level tool config dict, e.g. ``{"api_key": "...", "engine_id": "..."}``. :returns: A configured :class:`WebSearchTool`. """ from omnigent.tools.builtins.web_search import WebSearchTool llm_provider = None if self._spec.executor.model: model = self._spec.executor.model # Databricks doesn't support web_search_preview; skip # OpenAI provider inference for all databricks-* models. if not model.startswith("databricks-"): from omnigent.llms.routing import parse_model_string llm_provider = parse_model_string(model).provider return WebSearchTool(config=config, llm_provider=llm_provider) def _create_web_fetch(self) -> Tool: """ Build a WebFetchTool with the parent's spec. :returns: A WebFetchTool that inherits the parent's LLM config. """ from omnigent.tools.builtins.web_fetch import WebFetchTool return WebFetchTool(parent_spec=self._spec) def _register_sub_agent_tools(self) -> None: """ Register the sub-agent tool surface. The read-only discovery tools — ``sys_session_list``, ``sys_session_get_history``, and ``sys_session_get_info`` — are registered for **every** agent. Any agent can be part of a multi-agent session (most importantly a user-added agent that declares no sub-agents of its own but needs to read ``main``/siblings for context), so listing, peeking, and metadata reads are always available; access is enforced server-side (each proxies an auth-gated ``GET`` endpoint). ``peek`` / ``get_info`` take a ``conversation_id`` (from ``sys_session_list`` or a prior ``sys_session_send`` handle), so they need no ``sub_specs`` map. ``sys_session_share`` is gated by its OWN dedicated ``agent_session_sharing:`` flag (:class:`SharePolicy`), independent of the spawn grants (and unrelated to sharing via the server API or CLI). Sharing MUTATES access control — it can expose a session to a third party or, via ``__public__``, to anonymous read of the full transcript — and the server can confirm the caller holds manage-level access but cannot distinguish "the owner intended this" from "the agent was prompt-injected into sharing". So it is off unless the spec opts in: ``none`` (default) leaves it unregistered; ``non-public`` registers it for granting named users; ``public`` additionally lets it grant ``__public__`` (the tool advertises and enforces that extra tier via ``allow_public``). The spawn-lifecycle tools are a SEPARATE opt-in, gated behind ``tools.agents`` (declared sub-agents) or the top-level ``spawn: true`` flag: - ``tools.agents`` registers ``sys_session_send`` (named ``(agent, title)`` mode limited to the declared-type enum, plus ``session_id`` mode for driving existing children), ``sys_session_close``, and ``sys_list_models`` (per-worker model availability for picking a valid ``args.model``) — the agent may spawn THE SPECIFIED LIST of sub-agents, nothing else. - ``spawn: true`` additionally registers ``sys_session_create`` — launching arbitrary children from an existing agent_id or a custom locally-authored bundle (``config_path``). It also registers send/close (an agent must be able to drive and tombstone the children it creates); without declared sub-agents, send's schema omits the named-mode parameters. The spawn writes are child-only and ``sys_session_share`` is owner-authority-bounded, both enforced at dispatch/server level; the opt-ins control advertisement, not authority. (Cancellation uses the unified ``sys_cancel_task``; inspection is via inbox auto-delivery.) """ # Read-only discovery tools: always available. self._tools[SysSessionListTool.name()] = SysSessionListTool() self._tools[SysSessionGetHistoryTool.name()] = SysSessionGetHistoryTool() self._tools[SysSessionGetInfoTool.name()] = SysSessionGetInfoTool() # Session sharing: opt-in via the dedicated # ``agent_session_sharing`` flag, independent of spawn / declared # sub-agents. ``none`` leaves it unregistered; ``public`` # additionally permits __public__ grants (the tool reflects that # in its schema and the runner enforces it). It is its own flag — # not folded into the spawn grant — because letting the agent # expose a session is a distinct authority from spawning # children, and the public tier warrants an explicit extra opt-in # given the prompt-injection exposure. if self._spec.agent_session_sharing is not SharePolicy.NONE: self._tools[SysSessionShareTool.name()] = SysSessionShareTool( allow_public=self._spec.agent_session_sharing is SharePolicy.PUBLIC, ) # send + close: opt-in via declared sub-agents or spawn: true. if not (self._spec.tools.agents or self._spec.spawn): return sub_specs = {sa.name: sa for sa in self._spec.sub_agents if sa.name is not None} self._tools[SysSessionSendTool.name()] = SysSessionSendTool( sub_specs=sub_specs, ) self._tools[SysSessionCloseTool.name()] = SysSessionCloseTool() # Model awareness pairs with the dispatch grant: the per-worker # listing exists to pick a valid ``args.model`` for send. self._tools[SysListModelsTool.name()] = SysListModelsTool(spec=self._spec) # Advise-models is registered unconditionally alongside the # dispatch grant — same pattern as sys_list_models. The server's # MCP intercept returns router_on:false when routing is off, # giving the model a clear signal without hiding the tool. self._tools[SysAdviseModelsTool.name()] = SysAdviseModelsTool() # create: spawning OUTSIDE the declared list (existing agents # by id, or custom bundles via config_path) requires the # explicit ``spawn: true`` grant — declaring tools.agents # alone only permits the specified sub-agent types. if self._spec.spawn: self._tools[SysSessionCreateTool.name()] = SysSessionCreateTool() def _register_agent_mgmt_tools(self) -> None: """ Register the read-only ``sys_agent_*`` discovery tools. ``sys_agent_get``, ``sys_agent_download``, and ``sys_agent_list`` are registered for **every** agent, mirroring the always-on session reads in :meth:`_register_sub_agent_tools`. All three are global reads bounded by the server's per-user permission model — they proxy auth-gated ``GET /v1/sessions/{id}/agent``, ``.../agent/contents``, and ``GET /v1/agents`` + ``/v1/sessions`` endpoints (``sys_agent_list`` also scans the agent's own local config dir), so registration grants no access the caller didn't already have. They are runner-dispatched and need no construction-time wiring. """ self._tools[SysAgentGetTool.name()] = SysAgentGetTool() self._tools[SysAgentDownloadTool.name()] = SysAgentDownloadTool() self._tools[SysAgentListTool.name()] = SysAgentListTool() def _register_task_lifecycle_tools(self) -> None: """ Auto-register ``sys_cancel_task``. ``sys_cancel_task`` is registered unconditionally on every agent: any dispatched handle's system message references it — that promise only holds if it's always in the schema. ``list_tasks`` has been removed (the tasks table was removed; it always returned an empty list in production). ``check_task`` was dropped per design step 11 in favour of the inbox-based pattern (``sys_call_async`` + ``sys_read_inbox`` + auto-delivery). Idempotent: explicit registrations elsewhere win. """ from omnigent.tools.builtins.async_inbox import SysCancelTaskTool # Idempotent: explicit registration wins. if SysCancelTaskTool.name() not in self._tools: self._tools[SysCancelTaskTool.name()] = SysCancelTaskTool() def _register_comment_tools(self) -> None: """ Auto-register ``list_comments`` and ``update_comment``. Both tools are framework-owned and always available to the agent so it can read and update review comments left by the user without the spec explicitly opting in. They are session- scoped at invoke time via ``ToolContext.conversation_id`` (W1/W2 multi-user guard — the agent cannot query another session's comments by supplying a different id). """ self._tools[ListCommentsTool.name()] = ListCommentsTool() self._tools[UpdateCommentTool.name()] = UpdateCommentTool() def _register_browser_tools(self) -> None: """ Auto-register the embedded-browser tools (``browser_navigate`` / ``browser_snapshot`` / ``browser_click`` / ``browser_type`` / ``browser_screenshot``). Framework-owned and always available so any agent can drive the desktop app's embedded browser without the spec opting in. The classes here are schema-only (``name`` / ``description`` / ``get_schema``); execution lives in the runner ``_BROWSER_TOOLS`` dispatch branch (``omnigent/runner/tool_dispatch.py``), which needs the runner's ``server_client`` that ``ToolContext`` does not carry. """ from omnigent.tools.builtins.browser import ( BrowserClickTool, BrowserNavigateTool, BrowserScreenshotTool, BrowserSnapshotTool, BrowserTypeTool, ) for _cls in ( BrowserNavigateTool, BrowserSnapshotTool, BrowserClickTool, BrowserTypeTool, BrowserScreenshotTool, ): self._tools[_cls.name()] = _cls() def _register_os_env_tools(self) -> None: """ Register ``sys_os_*`` tools when the spec declares ``os_env``. When a pre-resolved :class:`OSEnvironment` was provided to the constructor (from ``SessionResourceRegistry``), that shared instance is used directly. Otherwise, falls back to creating a new instance via ``create_os_environment()``. When the spec declares no os_env and no pre-resolved env was provided, this is a no-op. :raises ValueError: If a ``sys_os_*`` name collides with an already-registered tool. """ from omnigent.tools.builtins.os_env import build_os_env_tools os_env = self._pre_resolved_os_env if os_env is None: from omnigent.inner.os_env import create_os_environment os_env_spec_obj = self._spec.os_env if os_env_spec_obj is None: return os_env = create_os_environment(os_env_spec_obj) if os_env is None: return self._os_env = os_env for tool in build_os_env_tools(os_env): if tool.name() in self._tools: raise ValueError( f"sys_os_* tool {tool.name()!r} collides with an " f"already-registered tool — investigate the offending " f"earlier registration before re-enabling os_env." ) self._tools[tool.name()] = tool def _register_terminal_tools(self) -> None: """ Register ``sys_terminal_*`` tools when the spec declares ``terminals``. Per ``designs/OMNIGENT_TERMINAL_BRIDGE.md`` §4.3, the five ``sys_terminal_launch`` / ``send`` / ``read`` / ``list`` / ``close`` tools register together when the spec carries a non-empty ``terminals`` block. They share the AP-process :class:`TerminalRegistry` (looked up via :func:`omnigent.runtime.get_terminal_registry`) so state survives across turns within a conversation. When the spec declares no terminals, this is a no-op. :raises ValueError: If a ``sys_terminal_*`` name collides with an already-registered tool — defensive; none of AP's other builtins use that prefix today. """ if not self._spec.terminals: return from omnigent.runtime import get_terminal_registry from omnigent.tools.builtins.sys_terminal import ( SysTerminalCloseTool, SysTerminalLaunchTool, SysTerminalListTool, SysTerminalReadTool, SysTerminalSendTool, ) registry = get_terminal_registry() for tool in ( SysTerminalLaunchTool(spec=self._spec, registry=registry), SysTerminalSendTool(registry=registry), SysTerminalReadTool(registry=registry), SysTerminalListTool(registry=registry), SysTerminalCloseTool(registry=registry), ): if tool.name() in self._tools: raise ValueError( f"sys_terminal_* tool {tool.name()!r} collides with an " f"already-registered tool — investigate the offending " f"earlier registration." ) self._tools[tool.name()] = tool def _register_local_tools(self, workdir: Path | None) -> None: """ Load and register local Python tools from the agent image. Each ``@tool``-decorated function in ``tools/python/*.py`` becomes one tool. Name collisions with already-registered tools (built-ins or earlier local tools) fail loud at load time per G27. If ``workdir`` is ``None`` or the spec has no local tools, this is a no-op. :param workdir: The agent image directory, or ``None``. :raises LocalToolLoadError: If any tool file fails to load or any name collides with a built-in. """ if not self._spec.local_tools: return # Split local tools by runtime. ``runtime: client`` entries # have no server-side callable (no ``path``, no ``callable:``), # so the file-based / dotted-callable loaders below would # crash on them. They're registered separately as # :class:`ClientSideTool` instances using the explicit # ``parameters`` block from the spec — schema visible to the # LLM, dispatch short-circuited to ``action_required`` via # :meth:`is_client_side_tool`. server_local_tools = [ t for t in self._spec.local_tools if t.runtime not in (ToolRuntime.CLIENT, ToolRuntime.UC_FUNCTION) ] client_local_tools = [t for t in self._spec.local_tools if t.runtime == ToolRuntime.CLIENT] uc_function_tools = [ t for t in self._spec.local_tools if t.runtime == ToolRuntime.UC_FUNCTION ] for client_info in client_local_tools: if not is_valid_tool_name(client_info.name): _logger.warning( "Spec-declared client local tool %r has invalid name — skipping", client_info.name, ) continue if client_info.name in self._tools: # Collision with a builtin or earlier registration — # fail loud rather than silently override (matches # the server-side path's G27 collision discipline). raise ValueError( f"spec-declared client local tool {client_info.name!r} " f"collides with an already-registered tool" ) # The validator (see ``omnigent/spec/validator.py``) # enforces that ``runtime: client`` tools carry an # explicit ``parameters`` block — no callable to # introspect, so the schema must be authoritative. # Fail loud here if the invariant is broken rather # than substituting a synthetic empty-parameters # default that would mask a validator regression. if client_info.parameters is None: raise ValueError( f"spec-declared client local tool {client_info.name!r} " f"has no ``parameters`` block — the spec validator " f"is supposed to require one for runtime: client tools" ) schema: dict[str, Any] = { "type": "function", "function": { "name": client_info.name, "parameters": client_info.parameters, }, } self._tools[client_info.name] = ClientSideTool( ClientSideToolSpec(name=client_info.name, schema=schema), ) # UC function tools — dispatched by the runner via the SQL # Statement Execution API. The tool manager only needs to # expose the schema to the LLM; actual execution goes # through ``_execute_uc_function_tool`` in tool_dispatch. for uc_info in uc_function_tools: if not is_valid_tool_name(uc_info.name): _logger.warning( "UC function tool %r has invalid name — skipping", uc_info.name, ) continue if uc_info.name in self._tools: raise ValueError( f"UC function tool {uc_info.name!r} collides with an already-registered tool" ) uc_schema: dict[str, Any] = { "type": "function", "function": { "name": uc_info.name, "parameters": uc_info.parameters or { "type": "object", "properties": {}, }, }, } if uc_info.description: uc_schema["function"]["description"] = uc_info.description self._tools[uc_info.name] = _UCFunctionSchemaTool( tool_name=uc_info.name, schema=uc_schema, ) # Native Omnigent local tools (``language == "python"``) need a # workdir on disk so the subprocess loader can locate the # ``tools/python/*.py`` files. Omnigent-style tools # (``language == "omnigent-python-callable"``) come # from a dotted import path with no on-disk presence and # don't need ``workdir``. if workdir is not None: for tool in load_local_python_tools( server_local_tools, workdir, sandbox_config=self._spec.tools.sandbox, srt_available=self._srt_available, uv_available=self._uv_available, sandbox_enabled=self._sandbox_enabled, agent_name=self._spec.name, # Pass the names of already-registered tools (builtins # at this point) so the loader can detect collisions # at G27 strictness — fail loud, not silent shadowing. builtin_tool_names=frozenset(self._tools.keys()), ): if not is_valid_tool_name(tool.name()): _logger.warning( "Local tool %r has invalid name — skipping", tool.name(), ) continue self._tools[tool.name()] = tool # Omnigent-style callable tools — sibling loader for the # ``omnigent-python-callable`` language entries the YAML # translator emits. See # :mod:`omnigent.tools.local_callable` for the reasoning; # in short, these wrap dotted Python callables imported # in-process and don't need the subprocess sandbox the # file-based path uses. from omnigent.tools.local_callable import load_local_callable_tools for callable_tool in load_local_callable_tools(server_local_tools): if not is_valid_tool_name(callable_tool.name()): _logger.warning( "Omnigent callable tool %r has invalid name — skipping", callable_tool.name(), ) continue if callable_tool.name() in self._tools: # Collision with a builtin or earlier registration — # fail loud rather than silently override (matches # the file-based path's G27 collision discipline). raise ValueError( f"omnigent callable tool {callable_tool.name()!r} " f"collides with an already-registered tool" ) self._tools[callable_tool.name()] = callable_tool def _register_client_tools( self, specs: list[ClientSideToolSpec], ) -> None: """ Register client-specified tools. Raises :class:`OmnigentError` if a tool name violates the OpenAI function-calling constraint (``^[a-zA-Z0-9_-]{1,256}$``). If a client tool name collides with an already-registered tool (e.g. a built-in skill tool), the client tool wins and a warning is logged. :param specs: List of :class:`ClientSideToolSpec` objects to register, e.g. ``[ClientSideToolSpec(name="get_weather", ...)]``. :raises OmnigentError: If any tool name is invalid. """ for spec in specs: if not is_valid_tool_name(spec.name): raise OmnigentError( f"Invalid client tool name {spec.name!r}: must match [a-zA-Z0-9_-]{{1,256}}", code=ErrorCode.INVALID_INPUT, ) if spec.name in self._tools: _logger.warning( "Client-specified tool %r shadows existing tool — overwriting", spec.name, ) self._tools[spec.name] = ClientSideTool(spec) def start(self) -> None: """No-op marker; all tools registered at init. MCP lives on runner.""" self._started = True def shutdown(self) -> None: """Idempotent teardown: close OS environment and shut down tools. Closes the OS environment that was created during registration (skipped when the environment was pre-resolved from the ``SessionResourceRegistry`` — the registry owns that lifecycle). Then calls ``shutdown()`` on every registered tool so subprocess-backed tools can kill lingering child processes. Safe to call multiple times; the second call is a no-op. """ self._started = False os_env = self._os_env if os_env is not None and self._pre_resolved_os_env is None: self._os_env = None try: os_env.close() except Exception: _logger.warning("os_env.close() failed during shutdown", exc_info=True) for tool in self._tools.values(): try: tool.shutdown() except Exception: _logger.warning("tool %s shutdown failed", tool.name(), exc_info=True) def get_tool_names(self) -> list[str]: """ Return the names of all registered tools. :returns: Tool names, e.g. ``["sys_session_send", "load_skill", "web_search"]``. """ return list(self._tools.keys()) def get_tool_schemas(self) -> list[dict[str, Any]]: """ Return OpenAI-format tool schemas for all registered tools. Each tool's schema is built independently. If a single tool's ``get_schema()`` raises — e.g. a ``type: function`` tool whose dotted ``callable`` path is unimportable — that one tool is skipped with a warning rather than aborting the whole list. This keeps one bad tool from silently dropping the agent's entire tool surface (#378); the remaining, valid tools are still advertised. :returns: A list of OpenAI tool schema dicts, each with ``"type": "function"`` and a ``"function"`` sub-dict describing the tool. """ schemas: list[dict[str, Any]] = [] for name, tool in self._tools.items(): try: schemas.append(tool.get_schema()) except Exception: _logger.warning( "Skipping tool %r: schema build failed; other tools are unaffected.", name, exc_info=True, ) return schemas def get_tool(self, name: str) -> Tool | None: """ Look up a registered tool by name. :param name: The tool function name, e.g. ``"load_skill"``. :returns: The :class:`Tool` instance, or ``None`` if not registered. """ return self._tools.get(name) def call_tool( self, name: str, arguments: str, ctx: ToolContext, ) -> str: """ Dispatch a tool call to the registered handler. :param name: The tool function name, e.g. ``"load_skill"`` or ``"github_list_issues"``. :param arguments: JSON-encoded arguments string from the LLM, e.g. ``'{"name": "summarize"}'``. :param ctx: Server-side execution context with task and agent identity. :returns: The tool's string result, or an error message if the tool is not registered. """ tool = self._tools.get(name) if tool is None: return f"Error: tool {name!r} not found. Registered tools: {list(self._tools.keys())}" return tool.invoke(arguments, ctx) def get_client_tool_schemas(self) -> list[dict[str, Any]]: """ Return the raw OpenAI-format schemas for all registered client-side tools. A tool is client-side if it was registered as a :class:`ClientSideTool` instance. Two sources produce :class:`ClientSideTool` registrations: - Request-time: each entry in the ``client_tool_specs`` constructor argument (from the API request body). - Spec-declared: each :attr:`AgentSpec.local_tools` entry whose ``runtime`` is :attr:`ToolRuntime.CLIENT`. Both paths produce indistinguishable :class:`ClientSideTool` registrations from the workflow's perspective. Used by :class:`SpawnTool` to propagate client tools to sub-agent workflows — the sub-agent's LLM needs the schemas so it knows which tools are available. Each client tool's schema is built independently. If a single tool's ``get_schema()`` raises, that one tool is skipped with a warning rather than aborting the whole list, mirroring :meth:`get_tool_schemas` (#378). This keeps one bad client tool from silently dropping every client tool propagated to a sub-agent; the remaining, valid tools are still advertised. :returns: List of tool schema dicts, e.g. ``[{"type": "function", "function": {"name": "Read", ...}}]``. Empty list if no client tools are registered. """ schemas: list[dict[str, Any]] = [] for name, tool in self._tools.items(): if not isinstance(tool, ClientSideTool): continue try: schemas.append(tool.get_schema()) except Exception: _logger.warning( "Skipping client tool %r: schema build failed; other tools are unaffected.", name, exc_info=True, ) return schemas def is_client_side_tool(self, name: str) -> bool: """ Return ``True`` if the named tool should be dispatched as ``action_required`` instead of executed server-side. A tool is client-side iff it's registered as a :class:`ClientSideTool` instance — the same predicate :meth:`get_client_tool_schemas` uses. Both the request-time ``client_tool_specs`` path and the spec-declared ``runtime: client`` path register :class:`ClientSideTool` instances, so this single check covers both. Used by the agent loop to detect when the LLM has invoked a client-side tool. On detection, the workflow persists the ``function_call`` items, streams them to the caller, and completes the response without executing any tools server-side. :param name: The tool function name, e.g. ``"get_weather"``. :returns: ``True`` if the tool is registered as a :class:`ClientSideTool`, ``False`` if the tool is not registered or is server-side. """ return isinstance(self._tools.get(name), ClientSideTool)