Tools

Decorator

tool

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

tool(fn: None = None, *, name: str | None = None, description: str | None = None, idempotent: 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) -> 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 book_flight(flight_id: str, customer_id: str) -> dict: '''Book a flight — safe to mark idempotent because repeated calls with the same flight/customer would create duplicate bookings, which we never want.''' ...

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

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,
) -> 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 book_flight(flight_id: str, customer_id: str) -> dict:
            '''Book a flight — safe to mark idempotent because repeated
            calls with the same flight/customer would create duplicate
            bookings, which we never want.'''
            ...

    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.

    Returns:
        Tool instance
    """

    def decorator(func: Callable[P, R]) -> Tool:
        # 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,
        )

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

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

Tool class

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

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
    """
    # 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

    # 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.

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

def to_openai_schema(self) -> dict[str, Any]:
    """Get OpenAI-compatible tool schema."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }

__call__

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

Direct invocation of the tool.

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

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

Tool context

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.

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

def get_metadata(self, key: str, default: Any = None) -> Any:
    """Get a metadata value."""
    return self.invocation_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)

Built-in tools

get_today_date

get_today_date() -> dict

Return today's date plus common reference points for date arithmetic.

Call this whenever the user mentions a relative or partial date ("tomorrow", "next Monday", "in ten days", "April 20") so you can convert to an explicit YYYY-MM-DD before calling a date-sensitive tool.

Returns:

Type	Description
dict	A dict with:
dict	today — today's date (YYYY-MM-DD)
dict	weekday — e.g. "Saturday"
dict	year — current year
dict	tomorrow / day_after_tomorrow
dict	next_7_days_by_weekday — map of lower-cased weekday → ISO date for the next seven days, so "Monday" / "Friday" resolve without further arithmetic
dict	one_week_from_now / two_weeks_from_now

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

@tool(idempotent=True)
def get_today_date() -> dict:
    """Return today's date plus common reference points for date arithmetic.

    Call this whenever the user mentions a relative or partial date
    ("tomorrow", "next Monday", "in ten days", "April 20") so you can
    convert to an explicit YYYY-MM-DD before calling a date-sensitive tool.

    Returns:
        A dict with:

        - ``today`` — today's date (YYYY-MM-DD)
        - ``weekday`` — e.g. ``"Saturday"``
        - ``year`` — current year
        - ``tomorrow`` / ``day_after_tomorrow``
        - ``next_7_days_by_weekday`` — map of lower-cased weekday → ISO date
            for the next seven days, so "Monday" / "Friday" resolve without
            further arithmetic
        - ``one_week_from_now`` / ``two_weeks_from_now``
    """
    now = datetime.now().astimezone()
    today = now.date()
    return {
        "today": today.isoformat(),
        "weekday": now.strftime("%A"),
        "year": today.year,
        "tomorrow": (today + timedelta(days=1)).isoformat(),
        "day_after_tomorrow": (today + timedelta(days=2)).isoformat(),
        "next_7_days_by_weekday": {
            (today + timedelta(days=n)).strftime("%A").lower(): (
                today + timedelta(days=n)
            ).isoformat()
            for n in range(1, 8)
        },
        "one_week_from_now": (today + timedelta(days=7)).isoformat(),
        "two_weeks_from_now": (today + timedelta(days=14)).isoformat(),
    }