Index

tools

Tools primitive — tool system with ABC interface and built-in tools.

Classes

BaseTool

Bases: ABC

Base class for all tool implementations.

Subclasses must be registered via @ToolRegistry.register("name") to become discoverable.

Attributes

spec abstractmethod property

spec: ToolSpec

Return the tool specification.

Functions

execute abstractmethod

execute(**params: Any) -> ToolResult

Execute the tool with the given parameters.

Source code in src/openjarvis/tools/_stubs.py

@abstractmethod
def execute(self, **params: Any) -> ToolResult:
    """Execute the tool with the given parameters."""

to_openai_function

to_openai_function() -> Dict[str, Any]

Convert to OpenAI function-calling format.

Source code in src/openjarvis/tools/_stubs.py

def to_openai_function(self) -> Dict[str, Any]:
    """Convert to OpenAI function-calling format."""
    s = self.spec
    return {
        "type": "function",
        "function": {
            "name": s.name,
            "description": s.description,
            "parameters": s.parameters,
        },
    }

ToolExecutor

ToolExecutor(tools: List[BaseTool], bus: Optional[EventBus] = None, *, interactive: bool = False, confirm_callback: Optional[Callable[[str], bool]] = None, default_timeout: float = 30.0, capability_policy: Optional[Any] = None, agent_id: str = '')

Dispatch tool calls to registered tools with event bus integration.

PARAMETER	DESCRIPTION
tools	List of tool instances to make available. TYPE: List[BaseTool]
bus	Optional event bus for publishing TOOL_CALL_START/TOOL_CALL_END. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/openjarvis/tools/_stubs.py

def __init__(
    self,
    tools: List[BaseTool],
    bus: Optional[EventBus] = None,
    *,
    interactive: bool = False,
    confirm_callback: Optional[Callable[[str], bool]] = None,
    default_timeout: float = 30.0,
    capability_policy: Optional[Any] = None,
    agent_id: str = "",
) -> None:
    self._tools: Dict[str, BaseTool] = {t.spec.name: t for t in tools}
    self._bus = bus
    self._interactive = interactive
    self._confirm_callback = confirm_callback
    self._default_timeout = default_timeout
    self._capability_policy = capability_policy
    self._agent_id = agent_id

Functions

execute

execute(tool_call: ToolCall) -> ToolResult

Parse arguments, dispatch to tool, measure latency, emit events.

Source code in src/openjarvis/tools/_stubs.py

def execute(self, tool_call: ToolCall) -> ToolResult:
    """Parse arguments, dispatch to tool, measure latency, emit events."""
    tool = self._tools.get(tool_call.name)
    if tool is None:
        return ToolResult(
            tool_name=tool_call.name,
            content=f"Unknown tool: {tool_call.name}",
            success=False,
        )

    # Parse arguments
    try:
        params = json.loads(tool_call.arguments) if tool_call.arguments else {}
    except json.JSONDecodeError as exc:
        return ToolResult(
            tool_name=tool_call.name,
            content=f"Invalid arguments JSON: {exc}",
            success=False,
        )

    # RBAC capability check
    if self._capability_policy and tool.spec.required_capabilities:
        for cap in tool.spec.required_capabilities:
            if not self._capability_policy.check(
                self._agent_id, cap, tool_call.name,
            ):
                if self._bus:
                    self._bus.publish(
                        EventType.CAPABILITY_DENIED,
                        {
                            "agent_id": self._agent_id,
                            "capability": cap,
                            "tool": tool_call.name,
                        },
                    )
                return ToolResult(
                    tool_name=tool_call.name,
                    content=(
                        f"Capability '{cap}' denied for"
                        f" agent '{self._agent_id}'"
                        f" on tool '{tool_call.name}'."
                    ),
                    success=False,
                )

    # Taint checking (sink policy)
    taint_set = params.get("_taint") if isinstance(params, dict) else None
    if taint_set is not None:
        try:
            from openjarvis.security.taint import TaintSet, check_taint

            if isinstance(taint_set, TaintSet):
                violation = check_taint(tool_call.name, taint_set)
                if violation:
                    if self._bus:
                        self._bus.publish(
                            EventType.TAINT_VIOLATION,
                            {
                                "tool": tool_call.name,
                                "violation": violation,
                            },
                        )
                    return ToolResult(
                        tool_name=tool_call.name,
                        content=f"Taint violation: {violation}",
                        success=False,
                    )
        except ImportError:
            pass
        # Remove internal taint key before passing to tool
        if isinstance(params, dict):
            params.pop("_taint", None)

    # Confirmation check for sensitive tools
    if tool.spec.requires_confirmation:
        if not self._interactive or self._confirm_callback is None:
            return ToolResult(
                tool_name=tool_call.name,
                content=(
                    f"Tool '{tool_call.name}' requires"
                    " confirmation but no confirmation"
                    " callback is available."
                ),
                success=False,
            )
        prompt = (
            f"Allow execution of tool"
            f" '{tool_call.name}' with args {params}?"
        )
        if not self._confirm_callback(prompt):
            return ToolResult(
                tool_name=tool_call.name,
                content=f"Tool '{tool_call.name}' execution denied by user.",
                success=False,
            )

    # Emit start event
    if self._bus:
        self._bus.publish(
            EventType.TOOL_CALL_START,
            {"tool": tool_call.name, "arguments": params},
        )

    # Execute with timeout
    timeout = tool.spec.timeout_seconds or self._default_timeout
    t0 = time.time()
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(tool.execute, **params)
            result = future.result(timeout=timeout)
    except concurrent.futures.TimeoutError:
        if self._bus:
            self._bus.publish(
                EventType.TOOL_TIMEOUT,
                {"tool": tool_call.name, "timeout": timeout},
            )
        result = ToolResult(
            tool_name=tool_call.name,
            content=(
                f"Tool '{tool_call.name}' timed out"
                f" after {timeout:.0f}s."
            ),
            success=False,
        )
    except Exception as exc:
        result = ToolResult(
            tool_name=tool_call.name,
            content=f"Tool execution error: {exc}",
            success=False,
        )
    latency = time.time() - t0
    result.latency_seconds = latency

    # Auto-detect taints in results
    if result.success:
        try:
            from openjarvis.security.taint import auto_detect_taint

            detected = auto_detect_taint(result.content)
            if detected and detected.labels:
                result.metadata["_taint"] = detected
        except ImportError:
            pass

    # Emit end event
    if self._bus:
        self._bus.publish(
            EventType.TOOL_CALL_END,
            {
                "tool": tool_call.name,
                "success": result.success,
                "latency": latency,
            },
        )

    return result

available_tools

available_tools() -> List[ToolSpec]

Return specs for all available tools.

Source code in src/openjarvis/tools/_stubs.py

def available_tools(self) -> List[ToolSpec]:
    """Return specs for all available tools."""
    return [t.spec for t in self._tools.values()]

get_openai_tools

get_openai_tools() -> List[Dict[str, Any]]

Return tools in OpenAI function-calling format.

Source code in src/openjarvis/tools/_stubs.py

def get_openai_tools(self) -> List[Dict[str, Any]]:
    """Return tools in OpenAI function-calling format."""
    return [t.to_openai_function() for t in self._tools.values()]

ToolSpec dataclass

ToolSpec(name: str, description: str, parameters: Dict[str, Any] = dict(), category: str = '', cost_estimate: float = 0.0, latency_estimate: float = 0.0, requires_confirmation: bool = False, timeout_seconds: float = 30.0, required_capabilities: List[str] = list(), metadata: Dict[str, Any] = dict())

Declarative description of a tool's interface and characteristics.