nanobot/nanobot/agent/tools/cli_apps.py

"""Controlled runner for installed CLI Apps."""

# pyright: reportIncompatibleMethodOverride=false

from __future__ import annotations

from pathlib import Path

from pydantic import Field

from nanobot.agent.tools.base import Tool, ToolResult, tool_parameters
from nanobot.agent.tools.context import RequestContext, ToolContext
from nanobot.agent.tools.schema import (
    ArraySchema,
    BooleanSchema,
    IntegerSchema,
    StringSchema,
    tool_parameters_schema,
)
from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig
from nanobot.apps.cli.utils import runtime_lines_for_request
from nanobot.config_base import Base
from nanobot.runtime_context import RuntimeContextBlock, wrap_runtime_context_lines
from nanobot.security.workspace_access import current_tool_workspace


class CliAppsToolConfig(Base):
    """CLI Apps tool configuration."""

    enable: bool = True
    install_timeout: int = Field(default=300, ge=1, le=3600)
    run_timeout: int = Field(default=60, ge=1, le=600)
    catalog_ttl_seconds: int = Field(default=3600, ge=60, le=86_400)


@tool_parameters(
    tool_parameters_schema(
        required=["name"],
        name=StringSchema("Installed CLI app registry name, for example gimp, safari, or obsidian."),
        args=ArraySchema(
            StringSchema("One command-line argument."),
            description="Arguments to pass to the CLI entry point. Do not include the entry point itself.",
            nullable=True,
        ),
        json=BooleanSchema(
            description="Whether to prepend --json when supported by the CLI.",
            default=False,
            nullable=True,
        ),
        working_dir=StringSchema("Optional working directory for the CLI call.", nullable=True),
        timeout=IntegerSchema(
            description="Timeout in seconds for this CLI call.",
            minimum=1,
            maximum=600,
            nullable=True,
        ),
    )
)
class CliAppsTool(Tool):
    """Run an installed CLI-Anything or public CLI app through a controlled argv subprocess."""

    config_key = "cli_apps"
    _scopes = {"core", "subagent"}

    @classmethod
    def config_cls(cls):
        return CliAppsToolConfig

    @classmethod
    def enabled(cls, ctx: ToolContext) -> bool:
        return ctx.config.cli_apps.enable

    @classmethod
    def create(cls, ctx: ToolContext) -> Tool:
        cfg = ctx.config.cli_apps
        return cls(
            workspace=Path(ctx.workspace),
            restrict_to_workspace=ctx.config.restrict_to_workspace,
            runtime=CliAppsRuntimeConfig(
                install_timeout=cfg.install_timeout,
                run_timeout=cfg.run_timeout,
                catalog_ttl_seconds=cfg.catalog_ttl_seconds,
            ),
        )

    def __init__(
        self,
        *,
        workspace: Path,
        restrict_to_workspace: bool = False,
        runtime: CliAppsRuntimeConfig | None = None,
    ) -> None:
        self.workspace = workspace
        self.restrict_to_workspace = restrict_to_workspace
        self.runtime = runtime or CliAppsRuntimeConfig()

    @property
    def name(self) -> str:
        return "run_cli_app"

    @property
    def description(self) -> str:
        try:
            installed = CliAppManager(workspace=self.workspace, runtime=self.runtime).installed_names()
        except Exception:
            installed = []
        installed_note = (
            f" Installed Settings CLI Apps: {', '.join(installed)}."
            if installed
            else " No Settings CLI Apps are currently installed."
        )
        return (
            "Run a CLI App that the user explicitly installed in Settings or attached as @app. "
            "Do not use this for ordinary system CLIs such as git, gh, python, npm, or brew; "
            "unknown names are rejected. Execution uses argv, not shell."
            + installed_note
        )

    def runtime_context_provider(self):
        return self._provide_runtime_context

    async def _provide_runtime_context(
        self,
        request: RequestContext,
    ) -> RuntimeContextBlock | None:
        lines = runtime_lines_for_request(
            request.original_user_text or "",
            request.metadata,
            request.workspace or self.workspace,
        )
        content = wrap_runtime_context_lines(lines)
        if not content:
            return None
        return RuntimeContextBlock(source="cli_apps", content=content)

    async def execute(
        self,
        name: str,
        args: list[str] | None = None,
        json: bool | None = False,
        working_dir: str | None = None,
        timeout: int | None = None,
    ) -> str:
        access = current_tool_workspace(
            self.workspace,
            restrict_to_workspace=self.restrict_to_workspace,
        )
        workspace = access.project_path or self.workspace
        manager = CliAppManager(workspace=workspace, runtime=self.runtime)
        try:
            return manager.run(
                name,
                args=args or [],
                json_output=bool(json),
                working_dir=working_dir,
                timeout=timeout,
                restrict_to_workspace=access.restrict_to_workspace,
            )
        except CliAppError as exc:
            return ToolResult.error(f"Error: {exc.message}")