Tools¶

def to_definition(self) -> ToolDefinition:
    """Convert this tool to a ``ToolDefinition`` for LLM providers.

    Returns:
        A ``ToolDefinition`` with name, description, and schema.
    """
    return ToolDefinition(
        name=self._name,
        description=self._description,
        parameters_schema=self.parameters_schema or {},
    )

@abstractmethod
async def execute(
    self,
    *,
    arguments: dict[str, Any],
) -> ToolExecutionResult:
    """Execute the tool with the given arguments.

    Arguments are pre-validated against the tool's JSON Schema (if
    one is defined) by the ``ToolInvoker`` before reaching this
    method.  Implementations with a schema can assume compliance
    when invoked through the invoker; tools without a schema
    receive unvalidated arguments.

    Args:
        arguments: Parsed arguments matching the parameters schema.

    Returns:
        A ``ToolExecutionResult`` with the tool output.
    """

def get(self, name: str) -> BaseTool:
    """Look up a tool by name.

    Args:
        name: Tool name.

    Returns:
        The registered tool instance.

    Raises:
        ToolNotFoundError: If no tool is registered with that name.
    """
    tool = self._tools.get(name)
    if tool is None:
        available = sorted(self._tools) or ["(none)"]
        logger.warning(
            TOOL_NOT_FOUND,
            tool_name=name,
            available=available,
        )
        msg = (
            f"Tool {name!r} is not registered. "
            f"Available tools: {', '.join(available)}"
        )
        raise ToolNotFoundError(msg, context={"tool": name})
    return tool

def list_tools(self) -> tuple[str, ...]:
    """Return sorted tuple of registered tool names."""
    return tuple(sorted(self._tools))

def all_tools(self) -> tuple[BaseTool, ...]:
    """Return all registered tool instances, sorted by name."""
    return tuple(self._tools[name] for name in sorted(self._tools))

def to_definitions(self) -> tuple[ToolDefinition, ...]:
    """Return all tool definitions as a sorted tuple, ordered by name.

    Returns:
        Sorted tuple of tool definitions for LLM providers.
    """
    return tuple(self._tools[name].to_definition() for name in sorted(self._tools))

def __contains__(self, name: object) -> bool:
    """Check whether a tool name is registered."""
    if not isinstance(name, str):
        logger.debug(
            TOOL_REGISTRY_CONTAINS_TYPE_ERROR,
            name_type=type(name).__name__,
        )
        return False
    return name in self._tools

def __len__(self) -> int:
    """Return the number of registered tools."""
    return len(self._tools)

def get_permitted_definitions(self) -> tuple[ToolDefinition, ...]:
    """Return tool definitions filtered by the permission checker.

    When no permission checker is set, returns all definitions.

    Returns:
        Tuple of permitted tool definitions, sorted by name.
    """
    if self._permission_checker is None:
        return self._registry.to_definitions()
    return self._permission_checker.filter_definitions(self._registry)

async def invoke(self, tool_call: ToolCall) -> ToolResult:
    """Execute a single tool call.

    Steps:
        1. Look up the tool in the registry.
        2. Check permissions against the permission checker (if any).
        3. Validate arguments against the tool's JSON Schema (if any).
        4. Run security interceptor pre-tool check (if any).
        5. Call ``tool.execute(arguments=...)``.
        6. Scan tool output for sensitive data (if interceptor is set).
        7. Return a ``ToolResult`` with the output.

    Recoverable errors produce ``ToolResult(is_error=True)``.
    Non-recoverable errors are re-raised.

    Args:
        tool_call: The tool call from the LLM.

    Returns:
        A ``ToolResult`` with the tool's output or error message.
    """
    self._pending_escalations.clear()
    return await self._invoke_single(tool_call)

async def invoke_all(
    self,
    tool_calls: Iterable[ToolCall],
    *,
    max_concurrency: int | None = None,
) -> tuple[ToolResult, ...]:
    """Execute multiple tool calls concurrently.

    Args:
        tool_calls: Tool calls to execute.
        max_concurrency: Max concurrent invocations (``>= 1``).

    Returns:
        Tuple of results in the same order as the input.

    Raises:
        ValueError: If *max_concurrency* < 1.
        MemoryError: Re-raised if a single fatal error occurred.
        RecursionError: Re-raised if a single fatal error occurred.
        ExceptionGroup: If multiple fatal errors occurred.
    """
    self._pending_escalations.clear()

    if max_concurrency is not None and max_concurrency < 1:
        msg = f"max_concurrency must be >= 1, got {max_concurrency}"
        raise ValueError(msg)

    calls = list(tool_calls)
    if not calls:
        return ()

    logger.info(
        TOOL_INVOKE_ALL_START,
        count=len(calls),
        max_concurrency=max_concurrency,
    )

    results: dict[int, ToolResult] = {}
    fatal_errors: list[Exception] = []
    semaphore = (
        asyncio.Semaphore(max_concurrency) if max_concurrency is not None else None
    )

    async with asyncio.TaskGroup() as tg:
        for idx, call in enumerate(calls):
            tg.create_task(
                self._run_guarded(
                    idx,
                    call,
                    results,
                    fatal_errors,
                    semaphore,
                ),
            )

    logger.info(
        TOOL_INVOKE_ALL_COMPLETE,
        count=len(calls),
        fatal_count=len(fatal_errors),
    )

    self._raise_fatal_errors(fatal_errors)

    # Sort escalations by tool-call index for deterministic ordering.
    if len(self._pending_escalations) > 1:
        call_id_order = {tc.id: idx for idx, tc in enumerate(calls)}
        self._pending_escalations.sort(
            key=lambda e: call_id_order.get(e.tool_call_id, len(calls)),
        )

    return tuple(results[i] for i in range(len(calls)))

@classmethod
def from_permissions(cls, permissions: ToolPermissions) -> Self:
    """Create a checker from an agent's ``ToolPermissions`` model.

    Args:
        permissions: Agent tool permissions.

    Returns:
        Configured permission checker.
    """
    return cls(
        access_level=permissions.access_level,
        allowed=frozenset(permissions.allowed),
        denied=frozenset(permissions.denied),
    )

def is_permitted(self, tool_name: str, category: ToolCategory) -> bool:
    """Check whether a tool is permitted.

    Args:
        tool_name: Name of the tool.
        category: Category of the tool.

    Returns:
        ``True`` if the tool is permitted, ``False`` otherwise.
    """
    name_lower = tool_name.strip().casefold()
    if name_lower in self._denied:
        return False
    if name_lower in self._allowed:
        return True
    if self._access_level == ToolAccessLevel.CUSTOM:
        return False
    allowed_cats = self._LEVEL_CATEGORIES[self._access_level]
    return category in allowed_cats

def check(self, tool_name: str, category: ToolCategory) -> None:
    """Assert that a tool is permitted, raising on denial.

    Args:
        tool_name: Name of the tool.
        category: Category of the tool.

    Raises:
        ToolPermissionDeniedError: If the tool is not permitted.
    """
    if not self.is_permitted(tool_name, category):
        reason = self.denial_reason(tool_name, category)
        logger.warning(
            TOOL_PERMISSION_DENIED,
            tool_name=tool_name,
            category=category.value,
            reason=reason,
        )
        raise ToolPermissionDeniedError(
            reason,
            context={"tool": tool_name, "category": category.value},
        )

def denial_reason(self, tool_name: str, category: ToolCategory) -> str:
    """Return a human-readable reason why a tool would be denied.

    Intended for use after confirming the tool is denied via
    ``is_permitted`` or via ``check``.  If the tool is actually
    permitted, the returned string does not apply and should not
    be shown to users.

    Args:
        tool_name: Name of the tool.
        category: Category of the tool.

    Returns:
        Explanation string suitable for error messages.
    """
    name_lower = tool_name.strip().casefold()
    if name_lower in self._denied:
        return f"Tool {tool_name!r} is explicitly denied"
    if self._access_level == ToolAccessLevel.CUSTOM:
        return (
            f"Tool {tool_name!r} is not in the allowed list (access level: custom)"
        )
    return (
        f"Category {category.value!r} is not permitted "
        f"at access level {self._access_level.value!r}"
    )

def filter_definitions(self, registry: ToolRegistry) -> tuple[ToolDefinition, ...]:
    """Return only permitted tool definitions from a registry.

    Args:
        registry: Tool registry to filter.

    Returns:
        Tuple of permitted tool definitions, sorted by tool name.
    """
    tool_names = registry.list_tools()
    result: list[ToolDefinition] = []
    for name in tool_names:
        tool = registry.get(name)
        if self.is_permitted(name, tool.category):
            result.append(tool.to_definition())
    result.sort(key=lambda d: d.name)
    excluded = len(tool_names) - len(result)
    if excluded:
        logger.debug(
            TOOL_PERMISSION_FILTERED,
            access_level=self._access_level.value,
            total=len(tool_names),
            permitted=len(result),
            excluded=excluded,
        )
    return tuple(result)

def __str__(self) -> str:
    """Format error with optional context metadata."""
    if self.context:
        ctx = ", ".join(f"{k}={v!r}" for k, v in self.context.items())
        return f"{self.message} ({ctx})"
    return self.message

async def execute(
    self,
    *,
    arguments: dict[str, Any],
) -> ToolExecutionResult:
    """Execute a code snippet in the sandbox.

    Args:
        arguments: Must contain ``code`` (str), ``language`` (str),
            and optionally ``timeout`` (float).

    Returns:
        A ``ToolExecutionResult`` with execution output.
    """
    code: str = arguments["code"]
    language: str = arguments["language"]
    timeout: float | None = arguments.get("timeout")

    if language not in _LANGUAGE_COMMANDS:
        logger.warning(
            CODE_RUNNER_INVALID_LANGUAGE,
            language=language,
        )
        return ToolExecutionResult(
            content=f"Unsupported language: {language!r}. "
            f"Supported: {sorted(_LANGUAGE_COMMANDS)}",
            is_error=True,
        )

    command, flag = _LANGUAGE_COMMANDS[language]

    logger.debug(
        CODE_RUNNER_EXECUTE_START,
        language=language,
        timeout=timeout,
        code_length=len(code),
    )

    try:
        result = await self._sandbox.execute(
            command=command,
            args=(flag, code),
            timeout=timeout,
        )
    except SandboxError as exc:
        logger.warning(
            CODE_RUNNER_EXECUTE_FAILED,
            language=language,
            error=str(exc),
        )
        return ToolExecutionResult(
            content=f"Sandbox error: {exc}",
            is_error=True,
            metadata={"language": language},
        )

    if result.success:
        logger.debug(
            CODE_RUNNER_EXECUTE_SUCCESS,
            language=language,
        )
        return ToolExecutionResult(
            content=result.stdout or "(no output)",
            metadata={
                "returncode": result.returncode,
                "language": language,
            },
        )

    logger.warning(
        CODE_RUNNER_EXECUTE_FAILED,
        language=language,
        returncode=result.returncode,
        timed_out=result.timed_out,
    )
    error_msg = result.stderr or result.stdout or "Execution failed"
    if result.timed_out:
        error_msg = f"Execution timed out. {error_msg}"
    return ToolExecutionResult(
        content=error_msg,
        is_error=True,
        metadata={
            "returncode": result.returncode,
            "timed_out": result.timed_out,
            "language": language,
        },
    )

async def execute(
    self,
    *,
    arguments: dict[str, Any],
) -> ToolExecutionResult:
    """Run ``git status``.

    Args:
        arguments: Optional ``short`` and ``porcelain`` flags.

    Returns:
        A ``ToolExecutionResult`` with the status output.
    """
    args = ["status"]
    if arguments.get("porcelain"):
        args.append("--porcelain")
    elif arguments.get("short"):
        args.append("--short")
    return await self._run_git(args)