Tools¶

Define the functions an agent can call — the @tool decorator, execution strategies, and JSON Schema generation.

For the concepts, start with Tools.

Decorator¶

The primary entry point — wrap a Python function and you get a Tool the agent can call. Parameters are introspected from the type hints and the JSON Schema is generated automatically.

tool ¶

tool(fn: Callable[P, R]) -> Tool

tool(fn: None = None, *, name: str | None = None, description: str | None = None, idempotent: bool = False, labels: Iterable[str] | None = None, sandbox: SandboxSpec | ToolSandbox | str | bool | None = None, deferred: bool = False, emits_progress: bool = False) -> Callable[[Callable[P, R]], Tool]

tool(fn: Callable[P, R] | None = None, *, name: str | None = None, description: str | None = None, idempotent: bool = False, labels: Iterable[str] | None = None, sandbox: SandboxSpec | ToolSandbox | str | bool | None = None, deferred: bool = False, emits_progress: bool = False) -> Tool | Callable[[Callable[P, R]], Tool]

Decorator to create a tool from a function.

Usage

@tool def search(query: str) -> str: '''Search the knowledge base.''' return "results..."

@tool(name="custom_name", description="Custom description") def my_tool(x: int) -> int: return x * 2

@tool(idempotent=True) def open_incident(alert_id: str, severity: str) -> dict: '''Open an incident — safe to mark idempotent because repeated calls for the same alert would open duplicate incidents, which we never want.''' ...

@tool(sandbox=True, labels={"code-exec"}) def run_snippet(source: str) -> dict: '''Evaluate a snippet — runs in an isolated box, never in the host process. Self-contained: imports live inside the body.''' import ast return {"parsed": ast.dump(ast.parse(source))}

Parameters:

Name	Type	Description	Default
`fn`	`Callable[P, R] \| None`	The function to wrap	`None`
`name`	`str \| None`	Override tool name (defaults to function name)	`None`
`description`	`str \| None`	Override description (defaults to docstring)	`None`
`idempotent`	`bool`	If True, the ReAct loop deduplicates calls with matching (name, arguments) within a single agent run. Prevents duplicate side-effects when a model re-issues a tool call it has already made this turn.	`False`
`labels`	`Iterable[str] \| None`	Policy-matching labels the tool declares; matched against :class:`~locus.control.policy.ControlPolicy` label sets (e.g. `require_sandbox_for`).	`None`
`sandbox`	`SandboxSpec \| ToolSandbox \| str \| bool \| None`	Run the tool in an isolated box instead of the host process. `True` uses the default provider (`$LOCUS_SANDBOX` or the built-in subprocess box); a name, provider object, or :class:`~locus.tools.sandbox.SandboxSpec` selects/configures one. The function must be synchronous and self-contained; this is validated at decoration time. See :mod:`locus.tools.sandbox`.	`None`
`emits_progress`	`bool`	Stream the progress this tool reports with :func:`locus.tools.context.report_progress` as live :class:`~locus.core.events.ToolProgressEvent` s.	`False`

Returns:

Type	Description
`Tool \| Callable[[Callable[P, R]], Tool]`	Tool instance

Source code in src/locus/tools/decorator.py

def tool(
    fn: Callable[P, R] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    idempotent: bool = False,
    labels: Iterable[str] | None = None,
    sandbox: SandboxSpec | ToolSandbox | str | bool | None = None,
    deferred: bool = False,
    emits_progress: bool = False,
) -> Tool | Callable[[Callable[P, R]], Tool]:
    """
    Decorator to create a tool from a function.

    Usage:
        @tool
        def search(query: str) -> str:
            '''Search the knowledge base.'''
            return "results..."

        @tool(name="custom_name", description="Custom description")
        def my_tool(x: int) -> int:
            return x * 2

        @tool(idempotent=True)
        def open_incident(alert_id: str, severity: str) -> dict:
            '''Open an incident — safe to mark idempotent because repeated
            calls for the same alert would open duplicate incidents,
            which we never want.'''
            ...

        @tool(sandbox=True, labels={"code-exec"})
        def run_snippet(source: str) -> dict:
            '''Evaluate a snippet — runs in an isolated box, never in the
            host process. Self-contained: imports live inside the body.'''
            import ast
            return {"parsed": ast.dump(ast.parse(source))}

    Args:
        fn: The function to wrap
        name: Override tool name (defaults to function name)
        description: Override description (defaults to docstring)
        idempotent: If True, the ReAct loop deduplicates calls with
            matching (name, arguments) within a single agent run. Prevents
            duplicate side-effects when a model re-issues a tool call it
            has already made this turn.
        labels: Policy-matching labels the tool declares; matched against
            :class:`~locus.control.policy.ControlPolicy` label sets (e.g.
            ``require_sandbox_for``).
        sandbox: Run the tool in an isolated box instead of the host
            process. ``True`` uses the default provider (``$LOCUS_SANDBOX``
            or the built-in subprocess box); a name, provider object, or
            :class:`~locus.tools.sandbox.SandboxSpec` selects/configures
            one. The function must be synchronous and self-contained; this
            is validated at decoration time. See :mod:`locus.tools.sandbox`.
        emits_progress: Stream the progress this tool reports with
            :func:`locus.tools.context.report_progress` as live
            :class:`~locus.core.events.ToolProgressEvent` s.

    Returns:
        Tool instance
    """

    def decorator(func: Callable[P, R]) -> Tool:
        spec = normalize_sandbox(sandbox)
        if spec is not None:
            validate_sandboxable(func)

        # Generate schema
        schema = generate_schema(func, description)
        func_schema = schema["function"]

        return Tool(
            name=name or func_schema["name"],
            description=func_schema["description"],
            parameters=func_schema["parameters"],
            fn=func,
            idempotent=idempotent,
            labels=frozenset(labels or ()),
            sandbox=spec,
            deferred=deferred,
            emits_progress=emits_progress,
        )

    if fn is not None:
        # Called without arguments: @tool
        return decorator(fn)

    # Called with arguments: @tool(name="...")
    return decorator

Tool ¶

Bases: BaseModel

A tool that can be called by agents.

Created via the @tool decorator.

idempotent `class-attribute` `instance-attribute` ¶

idempotent: bool = False

When True, the ReAct loop deduplicates calls: if the model emits the same (tool_name, arguments) combination that has already been executed earlier in the current agent run, the prior result is reused and the tool function is not invoked again. Use for tools that either have side-effects you don't want duplicated (bookings, transfers, writes) or whose output is stable across the run (config/date lookups).

labels `class-attribute` `instance-attribute` ¶

labels: frozenset[str] = frozenset()

Policy-matching labels this tool declares (e.g. {"code-exec"}). Matched against :class:~locus.control.policy.ControlPolicy label sets such as require_sandbox_for by governance hooks like :class:~locus.tools.sandbox.SandboxEnforcerHook.

sandbox `class-attribute` `instance-attribute` ¶

sandbox: SandboxSpec | None = None

When set, :meth:execute ships the function's source into the configured sandbox and runs it there — the host process never executes the body. The function must be self-contained: synchronous, imports inside its body, JSON-serializable arguments and return value, and no ctx/context parameter. See :mod:locus.tools.sandbox.

deferred `class-attribute` `instance-attribute` ¶

deferred: bool = False

When True, the tool's schema is NOT sent to the model until it is activated — the model reaches it through the tool_search builtin (auto-registered when any deferred tool exists). Deferral is a context economy: only visibility changes. The tool is registered, gated, and policy-matched exactly as an eager tool; activating it adds its schema to the next model call, nothing more (#177).

native `class-attribute` `instance-attribute` ¶

native: dict[str, dict[str, Any]] = {}

Provider-native definitions of this tool, keyed by adapter ("anthropic", "openai_responses"). An adapter that finds its key sends that definition instead of the function schema; the others keep using :attr:parameters. Keys starting with _ are directives to the adapter (a beta header, a request option) and are never sent as part of the tool. See :mod:locus.tools.computer.

emits_progress `class-attribute` `instance-attribute` ¶

emits_progress: bool = False

When True, the agent runtime streams progress this tool reports (:func:locus.tools.context.report_progress, or an MCP server's notifications/progress) as live :class:~locus.core.events.ToolProgressEvent s while the call runs. Tools created from an MCP server set it.

output_schema `class-attribute` `instance-attribute` ¶

output_schema: dict[str, Any] | None = None

JSON Schema of the structured content the tool returns, when it declares one (an MCP tool's outputSchema). Informational: it is not sent to the model and does not change how the result is handled.

func `property` ¶

func: Callable[..., Any]

Alias for :attr:fn. Some samples and downstream code reach for .func (the LangChain/LangGraph idiom); keep both names pointed at the same underlying callable so users don't have to write getattr(t, 'fn', None) or getattr(t, 'func', t).

execute `async` ¶

execute(ctx: ToolContext | None = None, **kwargs: Any) -> Any

Execute the tool with given arguments.

Parameters:

Name	Type	Description	Default
`ctx`	`ToolContext \| None`	Optional tool context (injected if function accepts it)	`None`
`**kwargs`	`Any`	Tool arguments	`{}`

Returns:

Type	Description
`Any`	Tool result

Source code in src/locus/tools/decorator.py

async def execute(self, ctx: ToolContext | None = None, **kwargs: Any) -> Any:
    """
    Execute the tool with given arguments.

    Args:
        ctx: Optional tool context (injected if function accepts it)
        **kwargs: Tool arguments

    Returns:
        Tool result
    """
    if self.sandbox is not None:
        return await self._execute_sandboxed(self.sandbox, kwargs)

    # Check if function accepts context
    sig = inspect.signature(self.fn)
    accepts_ctx = any(name in ("ctx", "context") for name in sig.parameters)

    if accepts_ctx and ctx is not None:
        # Find the context parameter name
        ctx_param = next(name for name in sig.parameters if name in ("ctx", "context"))
        kwargs[ctx_param] = ctx

    # Bind before calling, so a caller mistake is distinguishable from a
    # bug inside the tool. Both raise TypeError, but only one is worth
    # telling a model about — and "search() missing 1 required positional
    # argument: 'title'" is a Python signature error, not an instruction it
    # can act on. Naming the tool and the missing parameters is.
    try:
        sig.bind(**kwargs)
    except TypeError as exc:
        missing = [
            name
            for name, param in sig.parameters.items()
            if param.default is inspect.Parameter.empty
            and param.kind
            not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
            and name not in kwargs
            and name not in ("ctx", "context")
        ]
        detail = f"missing required argument(s): {', '.join(missing)}" if missing else str(exc)
        raise TypeError(
            f"{self.name} was called with the wrong arguments — {detail}. "
            f"Call {self.name} again with every required argument filled in; "
            f"if a value is unknown, ask for it rather than guessing."
        ) from exc

    # Execute function
    if asyncio.iscoroutinefunction(self.fn):
        result = await self.fn(**kwargs)
    else:
        # Run sync function in thread pool. Propagate the current
        # contextvars context so observability emits (run_id) and
        # any other contextvar-driven instrumentation see the same
        # state inside the worker thread.
        import contextvars  # noqa: PLC0415

        loop = asyncio.get_event_loop()
        ctxvars_snapshot = contextvars.copy_context()
        result = await loop.run_in_executor(
            None,
            lambda: ctxvars_snapshot.run(self.fn, **kwargs),
        )

    return self._format_result(result)

to_openai_schema ¶

to_openai_schema() -> dict[str, Any]

Get OpenAI-compatible tool schema.

A tool with :attr:native definitions carries them under a top-level native key for the adapters that understand it.

Source code in src/locus/tools/decorator.py

def to_openai_schema(self) -> dict[str, Any]:
    """Get OpenAI-compatible tool schema.

    A tool with :attr:`native` definitions carries them under a top-level
    ``native`` key for the adapters that understand it.
    """
    schema: dict[str, Any] = {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }
    if self.native:
        schema["native"] = self.native
    return schema

call ¶

__call__(*args: Any, **kwargs: Any) -> Any

Direct invocation of the tool.

A sandboxed tool is sandboxed here too — direct calls are not a bypass. The recovered (JSON round-tripped) value is returned raw, mirroring a plain function call.

Source code in src/locus/tools/decorator.py

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Direct invocation of the tool.

    A sandboxed tool is sandboxed here too — direct calls are not a
    bypass. The recovered (JSON round-tripped) value is returned raw,
    mirroring a plain function call.
    """
    if self.sandbox is None:
        return self.fn(*args, **kwargs)
    bound = inspect.signature(self.fn).bind(*args, **kwargs)
    spec = self.sandbox
    emit_sync(
        EV_TOOL_SANDBOX_STARTED,
        tool=self.name,
        provider=provider_label(spec),
        timeout=spec.timeout,
    )
    result = run_tool_sandboxed(spec, self.fn, dict(bound.arguments))
    emit_sync(
        EV_TOOL_SANDBOX_COMPLETED,
        tool=self.name,
        ok=result.ok,
        exit_code=result.exit_code,
        timed_out=result.timed_out,
        duration_ms=result.duration_ms,
    )
    if not result.ok:
        raise SandboxExecutionError(failure_message(self.name, result), result=result)
    return result.value

Tool context¶

Inject per-call context (the agent's state, custom metadata, the hook orchestrator) into a tool by declaring a ToolContext parameter.

ToolContext ¶

Bases: BaseModel

Context passed to tools during execution.

Provides access to agent state, metadata, and utilities.

messages `property` ¶

messages: list[Any]

Get conversation messages (if state available).

confidence `property` ¶

confidence: float

Get current confidence score (if state available).

get_metadata ¶

get_metadata(key: str, default: Any = None) -> Any

Get a metadata value (ephemeral metadata included).

Source code in src/locus/tools/context.py

def get_metadata(self, key: str, default: Any = None) -> Any:
    """Get a metadata value (ephemeral metadata included)."""
    if key in self.invocation_metadata:
        return self.invocation_metadata[key]
    return self.ephemeral_metadata.get(key, default)

get_config ¶

get_config(key: str, default: Any = None) -> Any

Get a tool config value.

Source code in src/locus/tools/context.py

def get_config(self, key: str, default: Any = None) -> Any:
    """Get a tool config value."""
    return self.tool_config.get(key, default)

Registry¶

The agent's compiled tool collection. Built once during Agent.__init__; mutating config.tools directly afterwards has no effect (use Agent.add_tool / add_tools instead).

ToolRegistry ¶

Bases: BaseModel

Registry for managing available tools.

Handles tool registration, lookup, and schema generation.

activated `class-attribute` `instance-attribute` ¶

activated: set[str] = Field(default_factory=set)

Names of deferred tools whose schemas have been surfaced to the model (via tool_search or :meth:activate). Deferral is visibility only: an unactivated deferred tool is still registered, still executable, and still gated — it just costs no context until the model asks for it.