Tools Module

The tools module implements the tool system used by agents for executing actions such as calculations, memory search, file reading, and sub-model queries. Each tool implements the BaseTool ABC and is registered via @ToolRegistry.register("name"). The ToolExecutor dispatches tool calls with JSON argument parsing, latency tracking, and event bus integration.

Abstract Base Class

BaseTool

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,
        },
    }

ToolSpec

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, metadata: Dict[str, Any] = dict())

Declarative description of a tool's interface and characteristics.

ToolExecutor

ToolExecutor

ToolExecutor(tools: List[BaseTool], bus: Optional[EventBus] = None, *, interactive: bool = False, confirm_callback: Optional[Callable[[str], bool]] = None)

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,
) -> 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

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,
        )

    # 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 timing
    t0 = time.time()
    try:
        result = tool.execute(**params)
    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

    # 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()]

build_tool_descriptions

build_tool_descriptions

build_tool_descriptions(tools: List[BaseTool], *, include_category: bool = True, include_cost: bool = False) -> str

Build rich text descriptions from a list of tools.

This is the single source of truth for all text-based agents that need to describe available tools in their system prompts.

PARAMETER	DESCRIPTION
tools	List of tool instances. TYPE: List[BaseTool]
include_category	Whether to include the Category: line. TYPE: bool DEFAULT: True
include_cost	Whether to include Cost estimate: and Latency estimate: lines. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str	Formatted multi-tool description, or "No tools available." if tools is empty.

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

def build_tool_descriptions(
    tools: List[BaseTool],
    *,
    include_category: bool = True,
    include_cost: bool = False,
) -> str:
    """Build rich text descriptions from a list of tools.

    This is the single source of truth for all text-based agents that need
    to describe available tools in their system prompts.

    Parameters
    ----------
    tools:
        List of tool instances.
    include_category:
        Whether to include the ``Category:`` line.
    include_cost:
        Whether to include ``Cost estimate:`` and ``Latency estimate:`` lines.

    Returns
    -------
    str
        Formatted multi-tool description, or ``"No tools available."`` if
        *tools* is empty.
    """
    if not tools:
        return "No tools available."

    sections: list[str] = []
    for t in tools:
        s = t.spec
        lines = [f"### {s.name}", s.description]

        if include_category and s.category:
            lines.append(f"Category: {s.category}")

        if include_cost:
            if s.cost_estimate:
                lines.append(f"Cost estimate: ${s.cost_estimate:.4f}")
            if s.latency_estimate:
                lines.append(f"Latency estimate: {s.latency_estimate:.1f}s")

        # Parameter descriptions
        props = s.parameters.get("properties", {})
        required = set(s.parameters.get("required", []))
        if props:
            lines.append("Parameters:")
            for pname, pinfo in props.items():
                ptype = pinfo.get("type", "any")
                req_mark = ", required" if pname in required else ""
                desc = pinfo.get("description", "")
                if desc:
                    lines.append(f"  - {pname} ({ptype}{req_mark}): {desc}")
                else:
                    lines.append(f"  - {pname} ({ptype}{req_mark})")

        sections.append("\n".join(lines))

    return "\n\n".join(sections)

---

Built-in Tools

CalculatorTool

CalculatorTool

Bases: BaseTool

Safe math calculator using AST-based evaluation.

safe_eval

safe_eval

safe_eval(expression: str) -> float

Evaluate a math expression safely using AST parsing.

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

def safe_eval(expression: str) -> float:
    """Evaluate a math expression safely using AST parsing."""
    tree = ast.parse(expression, mode="eval")
    return _safe_eval_node(tree)

ThinkTool

ThinkTool

Bases: BaseTool

Reasoning scratchpad that echoes input for chain-of-thought.

RetrievalTool

RetrievalTool

RetrievalTool(backend: Optional[MemoryBackend] = None, *, top_k: int = 5)

Bases: BaseTool

Search the memory backend and return formatted context.

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

def __init__(
    self,
    backend: Optional[MemoryBackend] = None,
    *,
    top_k: int = 5,
) -> None:
    self._backend = backend
    self._top_k = top_k

LLMTool

LLMTool

LLMTool(engine: Optional[InferenceEngine] = None, *, model: str = '')

Bases: BaseTool

Delegate a sub-query to an inference engine for generation.

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

def __init__(
    self,
    engine: Optional[InferenceEngine] = None,
    *,
    model: str = "",
) -> None:
    self._engine = engine
    self._model = model

FileReadTool

FileReadTool

FileReadTool(allowed_dirs: Optional[List[str]] = None)

Bases: BaseTool

Read file contents with optional directory restrictions.

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

def __init__(
    self,
    allowed_dirs: Optional[List[str]] = None,
) -> None:
    self._allowed_dirs = [Path(d).resolve() for d in (allowed_dirs or [])]

---

Storage Tools

Tools that expose memory storage operations for use by agents via the tool system. These wrap the memory backend methods as BaseTool implementations.

MemoryStoreTool

MemoryStoreTool

MemoryStoreTool(backend: MemoryBackend | None = None)

Bases: BaseTool

MCP-exposed tool: store content into memory backend.

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

def __init__(self, backend: MemoryBackend | None = None) -> None:
    self._backend = backend

MemoryRetrieveTool

MemoryRetrieveTool

MemoryRetrieveTool(backend: MemoryBackend | None = None)

Bases: BaseTool

MCP-exposed tool: retrieve from memory backend.

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

def __init__(self, backend: MemoryBackend | None = None) -> None:
    self._backend = backend

MemorySearchTool

MemorySearchTool

MemorySearchTool(backend: MemoryBackend | None = None)

Bases: BaseTool

MCP-exposed tool: search memory with agent-friendly formatting.

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

def __init__(self, backend: MemoryBackend | None = None) -> None:
    self._backend = backend

MemoryIndexTool

MemoryIndexTool

MemoryIndexTool(backend: MemoryBackend | None = None)

Bases: BaseTool

MCP-exposed tool: index a file or directory into memory.

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

def __init__(self, backend: MemoryBackend | None = None) -> None:
    self._backend = backend

---

Scheduler Tools

MCP tools for scheduling agent tasks for future or recurring execution. All five tools are in the scheduler category and require a TaskScheduler instance to be injected at _scheduler. For usage and schedule type examples see the Scheduler Tools user guide.

ScheduleTaskTool

ScheduleTaskTool

Bases: BaseTool

Schedule a new task for future or recurring execution.

ListScheduledTasksTool

ListScheduledTasksTool

Bases: BaseTool

List all scheduled tasks.

PauseScheduledTaskTool

PauseScheduledTaskTool

Bases: BaseTool

Pause a scheduled task.

ResumeScheduledTaskTool

ResumeScheduledTaskTool

Bases: BaseTool

Resume a paused scheduled task.

CancelScheduledTaskTool

CancelScheduledTaskTool

Bases: BaseTool

Cancel a scheduled task.

---

MCP Adapter

Integration with the Model Context Protocol (MCP). The MCP adapter exposes OpenJarvis tools as MCP-compatible resources and allows consuming external MCP tool providers. Supports MCP protocol version 2025-11-25.

MCPToolAdapter

MCPToolAdapter

MCPToolAdapter(client: MCPClient, tool_spec: ToolSpec)

Bases: BaseTool

Wraps a single MCP-hosted tool as a native BaseTool.

This adapter enables tools discovered from external MCP servers to be used seamlessly within OpenJarvis agents via the ToolExecutor.

PARAMETER	DESCRIPTION
client	The MCPClient connected to the external MCP server. TYPE: MCPClient
tool_spec	The ToolSpec describing this tool (from MCPClient.list_tools()). TYPE: ToolSpec

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

def __init__(self, client: MCPClient, tool_spec: ToolSpec) -> None:
    self._client = client
    self._spec = tool_spec

Functions

execute

execute(**params: Any) -> ToolResult

Execute the remote MCP tool and return a ToolResult.

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

def execute(self, **params: Any) -> ToolResult:
    """Execute the remote MCP tool and return a ToolResult."""
    try:
        result = self._client.call_tool(self._spec.name, params)
        content_parts = result.get("content", [])
        text = "\n".join(
            p.get("text", "")
            for p in content_parts
            if isinstance(p, dict)
        )
        return ToolResult(
            tool_name=self._spec.name,
            content=text,
            success=not result.get("isError", False),
        )
    except Exception as exc:
        return ToolResult(
            tool_name=self._spec.name,
            content=f"MCP tool error: {exc}",
            success=False,
        )

MCPToolProvider

MCPToolProvider

MCPToolProvider(client: MCPClient)

Discovers tools from an MCP server and returns BaseTool adapters.

PARAMETER	DESCRIPTION
client	The MCPClient connected to the MCP server. TYPE: MCPClient

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

def __init__(self, client: MCPClient) -> None:
    self._client = client

Functions

discover

discover() -> List[BaseTool]

Discover available tools and return them as BaseTool adapters.

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

def discover(self) -> List[BaseTool]:
    """Discover available tools and return them as BaseTool adapters."""
    specs = self._client.list_tools()
    return [MCPToolAdapter(self._client, s) for s in specs]