tools

Class info

Classes

Name	Children	Inherits
Tool llmling_agent.tools.base Information about a registered tool.		object
ToolCallInfo llmling_agent.tools.tool_call_info Information about an executed tool call.		BaseModel
ToolContext llmling_agent.tools.base Context for tool execution confirmation.		object
ToolManager llmling_agent.tools.manager Manages tool registration, enabling/disabling and access.		BaseRegistry

🛈 DocStrings

Tool implementations and related classes / functions.

Tool dataclass

Information about a registered tool.

Source code in src/llmling_agent/tools/base.py

@dataclass
class Tool:
    """Information about a registered tool."""

    callable: LLMCallableTool
    """The actual tool implementation"""

    enabled: bool = True
    """Whether the tool is currently enabled"""

    source: ToolSource = "runtime"
    """Where the tool came from."""

    priority: int = 100
    """Priority for tool execution (lower = higher priority)"""

    requires_confirmation: bool = False
    """Whether tool execution needs explicit confirmation"""

    requires_capability: str | None = None
    """Optional capability required to use this tool"""

    agent_name: str | None = None
    """The agent name as an identifier for agent-as-a-tool."""

    metadata: dict[str, str] = field(default_factory=dict)
    """Additional tool metadata"""

    cache_enabled: bool = False
    """Whether to enable caching for this tool."""

    @property
    def schema(self) -> py2openai.OpenAIFunctionTool:
        """Get the OpenAI function schema for the tool."""
        return self.callable.get_schema()

    @property
    def name(self) -> str:
        """Get tool name."""
        return self.callable.name

    @property
    def description(self) -> str | None:
        """Get tool description."""
        return self.callable.description

    def matches_filter(self, state: Literal["all", "enabled", "disabled"]) -> bool:
        """Check if tool matches state filter."""
        match state:
            case "all":
                return True
            case "enabled":
                return self.enabled
            case "disabled":
                return not self.enabled

    @property
    def parameters(self) -> list[ToolParameter]:
        """Get information about tool parameters."""
        schema = self.schema["function"]
        properties: dict[str, Property] = schema.get("properties", {})  # type: ignore
        required: list[str] = schema.get("required", [])  # type: ignore

        return [
            ToolParameter(
                name=name,
                required=name in required,
                type_info=details.get("type"),
                description=details.get("description"),
            )
            for name, details in properties.items()
        ]

    def format_info(self, indent: str = "  ") -> str:
        """Format complete tool information."""
        lines = [f"{indent}→ {self.name}"]
        if self.description:
            lines.append(f"{indent}  {self.description}")
        if self.parameters:
            lines.append(f"{indent}  Parameters:")
            lines.extend(f"{indent}    {param}" for param in self.parameters)
        if self.metadata:
            lines.append(f"{indent}  Metadata:")
            lines.extend(f"{indent}    {k}: {v}" for k, v in self.metadata.items())
        return "\n".join(lines)

    async def execute(self, *args: Any, **kwargs: Any) -> Any:
        """Execute tool, handling both sync and async cases."""
        fn = track_tool(self.name)(self.callable.callable)
        return await execute(fn, *args, **kwargs, use_thread=True)

    @classmethod
    def from_code(
        cls,
        code: str,
        name: str | None = None,
        description: str | None = None,
    ) -> Self:
        """Create a tool from a code string."""
        namespace: dict[str, Any] = {}
        exec(code, namespace)
        func = next((v for v in namespace.values() if callable(v)), None)
        if not func:
            msg = "No callable found in provided code"
            raise ValueError(msg)
        return cls.from_callable(
            func, name_override=name, description_override=description
        )

    @classmethod
    def from_callable(
        cls,
        fn: Callable[..., Any] | str,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: py2openai.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Self:
        tool = LLMCallableTool.from_callable(
            fn,
            name_override=name_override,
            description_override=description_override,
            schema_override=schema_override,
        )
        return cls(tool, **kwargs)

    @classmethod
    def from_crewai_tool(
        cls,
        tool: Any,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: py2openai.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Self:
        """Allows importing crewai tools."""
        # vaidate_import("crewai_tools", "crewai")
        try:
            from crewai.tools import BaseTool as CrewAiBaseTool
        except ImportError as e:
            msg = "crewai package not found. Please install it with 'pip install crewai'"
            raise ImportError(msg) from e

        if not isinstance(tool, CrewAiBaseTool):
            msg = f"Expected CrewAI BaseTool, got {type(tool)}"
            raise TypeError(msg)

        return cls.from_callable(
            tool._run,
            name_override=name_override or tool.__class__.__name__.removesuffix("Tool"),
            description_override=description_override or tool.description,
            schema_override=schema_override,
            **kwargs,
        )

    @classmethod
    def from_langchain_tool(
        cls,
        tool: Any,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: py2openai.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Self:
        """Create a tool from a LangChain tool."""
        # vaidate_import("langchain_core", "langchain")
        try:
            from langchain_core.tools import BaseTool as LangChainBaseTool
        except ImportError as e:
            msg = "langchain-core package not found."
            raise ImportError(msg) from e

        if not isinstance(tool, LangChainBaseTool):
            msg = f"Expected LangChain BaseTool, got {type(tool)}"
            raise TypeError(msg)

        return cls.from_callable(
            tool.invoke,
            name_override=name_override or tool.name,
            description_override=description_override or tool.description,
            schema_override=schema_override,
            **kwargs,
        )

    @classmethod
    def from_autogen_tool(
        cls,
        tool: Any,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: py2openai.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Self:
        """Create a tool from a AutoGen tool."""
        # vaidate_import("autogen_core", "autogen")
        try:
            from autogen_core import CancellationToken
            from autogen_core.tools import BaseTool
        except ImportError as e:
            msg = "autogent_core package not found."
            raise ImportError(msg) from e

        if not isinstance(tool, BaseTool):
            msg = f"Expected AutoGent BaseTool, got {type(tool)}"
            raise TypeError(msg)
        token = CancellationToken()

        input_model = tool.__class__.__orig_bases__[0].__args__[0]  # type: ignore

        name = name_override or tool.name or tool.__class__.__name__.removesuffix("Tool")
        description = (
            description_override
            or tool.description
            or inspect.getdoc(tool.__class__)
            or ""
        )

        async def wrapper(**kwargs: Any) -> Any:
            # Convert kwargs to the expected input model
            model = input_model(**kwargs)
            return await tool.run(model, cancellation_token=token)

        return cls.from_callable(
            wrapper,  # type: ignore
            name_override=name,
            description_override=description,
            schema_override=schema_override,
            **kwargs,
        )

agent_name class-attribute instance-attribute

agent_name: str | None = None

The agent name as an identifier for agent-as-a-tool.

cache_enabled class-attribute instance-attribute

cache_enabled: bool = False

Whether to enable caching for this tool.

callable instance-attribute

callable: LLMCallableTool

The actual tool implementation

description property

description: str | None

Get tool description.

enabled class-attribute instance-attribute

enabled: bool = True

Whether the tool is currently enabled

metadata class-attribute instance-attribute

metadata: dict[str, str] = field(default_factory=dict)

Additional tool metadata

name property

name: str

Get tool name.

parameters property

parameters: list[ToolParameter]

Get information about tool parameters.

priority class-attribute instance-attribute

priority: int = 100

Priority for tool execution (lower = higher priority)

requires_capability class-attribute instance-attribute

requires_capability: str | None = None

Optional capability required to use this tool

requires_confirmation class-attribute instance-attribute

requires_confirmation: bool = False

Whether tool execution needs explicit confirmation

schema property

schema: OpenAIFunctionTool

Get the OpenAI function schema for the tool.

source class-attribute instance-attribute

source: ToolSource = 'runtime'

Where the tool came from.

execute async

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

Execute tool, handling both sync and async cases.

Source code in src/llmling_agent/tools/base.py

async def execute(self, *args: Any, **kwargs: Any) -> Any:
    """Execute tool, handling both sync and async cases."""
    fn = track_tool(self.name)(self.callable.callable)
    return await execute(fn, *args, **kwargs, use_thread=True)

format_info

format_info(indent: str = '  ') -> str

Format complete tool information.

Source code in src/llmling_agent/tools/base.py

def format_info(self, indent: str = "  ") -> str:
    """Format complete tool information."""
    lines = [f"{indent}→ {self.name}"]
    if self.description:
        lines.append(f"{indent}  {self.description}")
    if self.parameters:
        lines.append(f"{indent}  Parameters:")
        lines.extend(f"{indent}    {param}" for param in self.parameters)
    if self.metadata:
        lines.append(f"{indent}  Metadata:")
        lines.extend(f"{indent}    {k}: {v}" for k, v in self.metadata.items())
    return "\n".join(lines)

from_autogen_tool classmethod

from_autogen_tool(
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self

Create a tool from a AutoGen tool.

Source code in src/llmling_agent/tools/base.py

@classmethod
def from_autogen_tool(
    cls,
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: py2openai.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self:
    """Create a tool from a AutoGen tool."""
    # vaidate_import("autogen_core", "autogen")
    try:
        from autogen_core import CancellationToken
        from autogen_core.tools import BaseTool
    except ImportError as e:
        msg = "autogent_core package not found."
        raise ImportError(msg) from e

    if not isinstance(tool, BaseTool):
        msg = f"Expected AutoGent BaseTool, got {type(tool)}"
        raise TypeError(msg)
    token = CancellationToken()

    input_model = tool.__class__.__orig_bases__[0].__args__[0]  # type: ignore

    name = name_override or tool.name or tool.__class__.__name__.removesuffix("Tool")
    description = (
        description_override
        or tool.description
        or inspect.getdoc(tool.__class__)
        or ""
    )

    async def wrapper(**kwargs: Any) -> Any:
        # Convert kwargs to the expected input model
        model = input_model(**kwargs)
        return await tool.run(model, cancellation_token=token)

    return cls.from_callable(
        wrapper,  # type: ignore
        name_override=name,
        description_override=description,
        schema_override=schema_override,
        **kwargs,
    )

from_code classmethod

from_code(code: str, name: str | None = None, description: str | None = None) -> Self

Create a tool from a code string.

Source code in src/llmling_agent/tools/base.py

@classmethod
def from_code(
    cls,
    code: str,
    name: str | None = None,
    description: str | None = None,
) -> Self:
    """Create a tool from a code string."""
    namespace: dict[str, Any] = {}
    exec(code, namespace)
    func = next((v for v in namespace.values() if callable(v)), None)
    if not func:
        msg = "No callable found in provided code"
        raise ValueError(msg)
    return cls.from_callable(
        func, name_override=name, description_override=description
    )

from_crewai_tool classmethod

from_crewai_tool(
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self

Allows importing crewai tools.

Source code in src/llmling_agent/tools/base.py

@classmethod
def from_crewai_tool(
    cls,
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: py2openai.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self:
    """Allows importing crewai tools."""
    # vaidate_import("crewai_tools", "crewai")
    try:
        from crewai.tools import BaseTool as CrewAiBaseTool
    except ImportError as e:
        msg = "crewai package not found. Please install it with 'pip install crewai'"
        raise ImportError(msg) from e

    if not isinstance(tool, CrewAiBaseTool):
        msg = f"Expected CrewAI BaseTool, got {type(tool)}"
        raise TypeError(msg)

    return cls.from_callable(
        tool._run,
        name_override=name_override or tool.__class__.__name__.removesuffix("Tool"),
        description_override=description_override or tool.description,
        schema_override=schema_override,
        **kwargs,
    )

from_langchain_tool classmethod

from_langchain_tool(
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self

Create a tool from a LangChain tool.

Source code in src/llmling_agent/tools/base.py

@classmethod
def from_langchain_tool(
    cls,
    tool: Any,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: py2openai.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Self:
    """Create a tool from a LangChain tool."""
    # vaidate_import("langchain_core", "langchain")
    try:
        from langchain_core.tools import BaseTool as LangChainBaseTool
    except ImportError as e:
        msg = "langchain-core package not found."
        raise ImportError(msg) from e

    if not isinstance(tool, LangChainBaseTool):
        msg = f"Expected LangChain BaseTool, got {type(tool)}"
        raise TypeError(msg)

    return cls.from_callable(
        tool.invoke,
        name_override=name_override or tool.name,
        description_override=description_override or tool.description,
        schema_override=schema_override,
        **kwargs,
    )

matches_filter

matches_filter(state: Literal['all', 'enabled', 'disabled']) -> bool

Check if tool matches state filter.

Source code in src/llmling_agent/tools/base.py

def matches_filter(self, state: Literal["all", "enabled", "disabled"]) -> bool:
    """Check if tool matches state filter."""
    match state:
        case "all":
            return True
        case "enabled":
            return self.enabled
        case "disabled":
            return not self.enabled

ToolCallInfo

Bases: BaseModel

Information about an executed tool call.

Source code in src/llmling_agent/tools/tool_call_info.py

class ToolCallInfo(BaseModel):
    """Information about an executed tool call."""

    tool_name: str
    """Name of the tool that was called."""

    args: dict[str, Any]
    """Arguments passed to the tool."""

    result: Any
    """Result returned by the tool."""

    agent_name: str
    """Name of the calling agent."""

    tool_call_id: str = Field(default_factory=lambda: str(uuid4()))
    """ID provided by the model (e.g. OpenAI function call ID)."""

    timestamp: datetime = Field(default_factory=get_now)
    """When the tool was called."""

    message_id: str | None = None
    """ID of the message that triggered this tool call."""

    context_data: Any | None = None
    """Optional context data that was passed to the agent's run() method."""

    error: str | None = None
    """Error message if the tool call failed."""

    timing: float | None = None
    """Time taken for this specific tool call in seconds."""

    agent_tool_name: str | None = None
    """If this tool is agent-based, the name of that agent."""

    model_config = ConfigDict(use_attribute_docstrings=True, extra="forbid")

    def format(
        self,
        style: FormatStyle = "simple",
        *,
        template: str | None = None,
        variables: dict[str, Any] | None = None,
        show_timing: bool = True,
        show_ids: bool = False,
    ) -> str:
        """Format tool call information with configurable style.

        Args:
            style: Predefined style to use:
                - simple: Compact single-line format
                - detailed: Multi-line with all details
                - markdown: Formatted markdown with syntax highlighting
            template: Optional custom template (required if style="custom")
            variables: Additional variables for template rendering
            show_timing: Whether to include execution timing
            show_ids: Whether to include tool_call_id and message_id

        Returns:
            Formatted tool call information

        Raises:
            ValueError: If style is invalid or custom template is missing
        """
        from jinjarope import Environment

        # Select template
        if template:
            template_str = template
        elif style in TEMPLATES:
            template_str = TEMPLATES[style]
        else:
            msg = f"Invalid style: {style}"
            raise ValueError(msg)

        # Prepare template variables
        vars_ = {
            "tool_name": self.tool_name,
            "args": self.args,  # No pre-formatting needed
            "result": self.result,
            "error": self.error,
            "agent_name": self.agent_name,
            "timestamp": self.timestamp,
            "timing": self.timing if show_timing else None,
            "agent_tool_name": self.agent_tool_name,
        }

        if show_ids:
            vars_.update({
                "tool_call_id": self.tool_call_id,
                "message_id": self.message_id,
            })

        if variables:
            vars_.update(variables)

        # Render template
        env = Environment(trim_blocks=True, lstrip_blocks=True)
        env.filters["repr"] = repr  # Add repr filter
        template_obj = env.from_string(template_str)
        return template_obj.render(**vars_)

agent_name instance-attribute

agent_name: str

Name of the calling agent.

agent_tool_name class-attribute instance-attribute

agent_tool_name: str | None = None

If this tool is agent-based, the name of that agent.

args instance-attribute

args: dict[str, Any]

Arguments passed to the tool.

context_data class-attribute instance-attribute

context_data: Any | None = None

Optional context data that was passed to the agent's run() method.

error class-attribute instance-attribute

error: str | None = None

Error message if the tool call failed.

message_id class-attribute instance-attribute

message_id: str | None = None

ID of the message that triggered this tool call.

result instance-attribute

result: Any

Result returned by the tool.

timestamp class-attribute instance-attribute

timestamp: datetime = Field(default_factory=get_now)

When the tool was called.

timing class-attribute instance-attribute

timing: float | None = None

Time taken for this specific tool call in seconds.

tool_call_id class-attribute instance-attribute

tool_call_id: str = Field(default_factory=lambda: str(uuid4()))

ID provided by the model (e.g. OpenAI function call ID).

tool_name instance-attribute

tool_name: str

Name of the tool that was called.

format

format(
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_timing: bool = True,
    show_ids: bool = False,
) -> str

Format tool call information with configurable style.

Parameters:

Name	Type	Description	Default
style	FormatStyle	Predefined style to use: - simple: Compact single-line format - detailed: Multi-line with all details - markdown: Formatted markdown with syntax highlighting	'simple'
template	str | None	Optional custom template (required if style="custom")	None
variables	dict[str, Any] | None	Additional variables for template rendering	None
show_timing	bool	Whether to include execution timing	True
show_ids	bool	Whether to include tool_call_id and message_id	False

Returns:

Type	Description
str	Formatted tool call information

Raises:

Type	Description
ValueError	If style is invalid or custom template is missing

Source code in src/llmling_agent/tools/tool_call_info.py

def format(
    self,
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_timing: bool = True,
    show_ids: bool = False,
) -> str:
    """Format tool call information with configurable style.

    Args:
        style: Predefined style to use:
            - simple: Compact single-line format
            - detailed: Multi-line with all details
            - markdown: Formatted markdown with syntax highlighting
        template: Optional custom template (required if style="custom")
        variables: Additional variables for template rendering
        show_timing: Whether to include execution timing
        show_ids: Whether to include tool_call_id and message_id

    Returns:
        Formatted tool call information

    Raises:
        ValueError: If style is invalid or custom template is missing
    """
    from jinjarope import Environment

    # Select template
    if template:
        template_str = template
    elif style in TEMPLATES:
        template_str = TEMPLATES[style]
    else:
        msg = f"Invalid style: {style}"
        raise ValueError(msg)

    # Prepare template variables
    vars_ = {
        "tool_name": self.tool_name,
        "args": self.args,  # No pre-formatting needed
        "result": self.result,
        "error": self.error,
        "agent_name": self.agent_name,
        "timestamp": self.timestamp,
        "timing": self.timing if show_timing else None,
        "agent_tool_name": self.agent_tool_name,
    }

    if show_ids:
        vars_.update({
            "tool_call_id": self.tool_call_id,
            "message_id": self.message_id,
        })

    if variables:
        vars_.update(variables)

    # Render template
    env = Environment(trim_blocks=True, lstrip_blocks=True)
    env.filters["repr"] = repr  # Add repr filter
    template_obj = env.from_string(template_str)
    return template_obj.render(**vars_)

ToolContext dataclass

Context for tool execution confirmation.

Source code in src/llmling_agent/tools/base.py

@dataclass(frozen=True)
class ToolContext:
    """Context for tool execution confirmation."""

    name: str
    """Name of the tool being executed"""

    args: dict[str, Any]
    """Arguments being passed to the tool"""

    schema: py2openai.OpenAIFunctionTool
    """Complete OpenAI function schema"""

    runtime_ctx: AgentContext
    """Runtime context from agent"""

    @property
    def description(self) -> str | None:
        """Get tool description from schema."""
        return self.schema["function"].get("description")

    @property
    def parameters(self) -> ToolParameters:
        """Get parameter definitions from schema."""
        return self.schema["function"].get("parameters", {})  # type: ignore

    def __str__(self) -> str:
        """Format tool context for logging/display."""
        return (
            f"Tool: {self.name}\n"
            f"Arguments: {self.args}\n"
            f"Description: {self.description or 'N/A'}"
        )

args instance-attribute

args: dict[str, Any]

Arguments being passed to the tool

description property

description: str | None

Get tool description from schema.

name instance-attribute

name: str

Name of the tool being executed

parameters property

parameters: ToolParameters

Get parameter definitions from schema.

runtime_ctx instance-attribute

runtime_ctx: AgentContext

Runtime context from agent

schema instance-attribute

schema: OpenAIFunctionTool

Complete OpenAI function schema

__str__

__str__() -> str

Format tool context for logging/display.

Source code in src/llmling_agent/tools/base.py

def __str__(self) -> str:
    """Format tool context for logging/display."""
    return (
        f"Tool: {self.name}\n"
        f"Arguments: {self.args}\n"
        f"Description: {self.description or 'N/A'}"
    )

ToolManager

Bases: BaseRegistry[str, Tool]

Manages tool registration, enabling/disabling and access.

Inherits from BaseRegistry providing: - Dict-like access: manager["tool_name"] -> Tool - Async startup/shutdown: await manager.startup() - Event observation: manager.add_observer(observer) - Registration: manager.register("tool_name", tool) - Listing: manager.list_items() - State check: manager.is_empty, manager.has_item() - Async iteration: async for name, tool in manager: ...

Source code in src/llmling_agent/tools/manager.py

class ToolManager(BaseRegistry[str, Tool]):
    """Manages tool registration, enabling/disabling and access.

    Inherits from BaseRegistry providing:
    - Dict-like access: manager["tool_name"] -> Tool
    - Async startup/shutdown: await manager.startup()
    - Event observation: manager.add_observer(observer)
    - Registration: manager.register("tool_name", tool)
    - Listing: manager.list_items()
    - State check: manager.is_empty, manager.has_item()
    - Async iteration: async for name, tool in manager: ...
    """

    @dataclass(frozen=True)
    class ToolStateReset:
        """Emitted when tool states are reset."""

        previous_tools: dict[str, bool]
        new_tools: dict[str, bool]
        timestamp: datetime = field(default_factory=get_now)

    tool_states_reset = Signal(ToolStateReset)

    def __init__(
        self,
        tools: Sequence[Tool | ToolType | dict[str, Any]] | None = None,
    ):
        """Initialize tool manager.

        Args:
            tools: Initial tools to register
        """
        super().__init__()
        self.providers: list[ResourceProvider] = []

        # Register initial tools
        for tool in tools or []:
            t = self._validate_item(tool)
            self.register(t.name, t)

    def __prompt__(self) -> str:
        enabled_tools = [t.name for t in self.values() if t.enabled]
        if not enabled_tools:
            return "No tools available"
        return f"Available tools: {', '.join(enabled_tools)}"

    def add_provider(
        self,
        provider: ResourceProvider | ResourceCallable,
        owner: str | None = None,
    ):
        """Add a resource provider or tool callable.

        Args:
            provider: Either a ResourceProvider instance or a callable
                     returning tools. Callables are automatically wrapped.
            owner: Optional owner for the provider
        """
        from llmling_agent.resource_providers.base import ResourceProvider

        if not isinstance(provider, ResourceProvider):
            # Wrap old-style callable in ResourceProvider
            prov: ResourceProvider = CallableResourceProvider(
                name=provider.__name__,
                tool_callable=provider,
            )
        else:
            prov = provider
        if owner:
            prov.owner = owner
        self.providers.append(prov)

    def remove_provider(
        self, provider: ResourceProvider | ResourceCallable | ProviderName
    ):
        """Remove a resource provider."""
        from llmling_agent.resource_providers.base import ResourceProvider

        match provider:
            case ResourceProvider():
                self.providers.remove(provider)
            case Callable():
                # Find and remove wrapped callable
                for p in self.providers:
                    if (
                        isinstance(p, CallableResourceProvider)
                        and p.tool_callable == provider
                    ):
                        self.providers.remove(p)
            case str():
                for p in self.providers:
                    if p.name == provider:
                        self.providers.remove(p)
            case _:
                msg = f"Invalid provider type: {type(provider)}"
                raise ValueError(msg)

    def reset_states(self):
        """Reset all tools to their default enabled states."""
        for info in self.values():
            info.enabled = True

    @property
    def _error_class(self) -> type[ToolError]:
        """Error class for tool operations."""
        return ToolError

    def _validate_item(self, item: Tool | ToolType | dict[str, Any]) -> Tool:
        """Validate and convert items before registration."""
        match item:
            case Tool():
                return item
            case str():
                if item.startswith("crewai_tools"):
                    obj = import_class(item)()
                    return Tool.from_crewai_tool(obj)
                if item.startswith("langchain"):
                    obj = import_class(item)()
                    return Tool.from_langchain_tool(obj)
                return Tool.from_callable(item)
            case Callable():
                return Tool.from_callable(item)
            case {"callable": callable_item, **config} if callable(callable_item):
                valid_keys = {f.name for f in fields(Tool)} - {"callable"}
                tool_config = {k: v for k, v in config.items() if k in valid_keys}
                return Tool.from_callable(callable_item, **tool_config)  # type: ignore
            case _:
                typ = type(item)
                msg = f"Item must be Tool or callable. Got {typ}"
                raise ToolError(msg)

    def enable_tool(self, tool_name: str):
        """Enable a previously disabled tool."""
        if tool_name not in self:
            msg = f"Tool not found: {tool_name}"
            raise ToolError(msg)
        tool_info = self[tool_name]
        tool_info.enabled = True
        self.events.changed(tool_name, tool_info)
        logger.debug("Enabled tool: %s", tool_name)

    def disable_tool(self, tool_name: str):
        """Disable a tool."""
        if tool_name not in self:
            msg = f"Tool not found: {tool_name}"
            raise ToolError(msg)
        tool_info = self[tool_name]
        tool_info.enabled = False
        self.events.changed(tool_name, tool_info)
        logger.debug("Disabled tool: %s", tool_name)

    def is_tool_enabled(self, tool_name: str) -> bool:
        """Check if a tool is currently enabled."""
        return self[tool_name].enabled if tool_name in self else False

    async def list_tools(self) -> dict[str, bool]:
        """Get a mapping of all tools and their enabled status."""
        return {tool.name: tool.enabled for tool in await self.get_tools()}

    async def get_tools(
        self,
        state: ToolState = "all",
        names: str | list[str] | None = None,
    ) -> list[Tool]:
        """Get tool objects based on filters."""
        tools = [t for t in self.values() if t.matches_filter(state)]
        match names:
            case str():
                tools = [t for t in tools if t.name == names]
            case list():
                tools = [t for t in tools if t.name in names]
        # Get tools from providers
        for provider in self.providers:
            try:
                provider_tools = await provider.get_tools()
                tools.extend(t for t in provider_tools if t.matches_filter(state))
            except Exception:
                logger.exception("Failed to get tools from provider: %r", provider)
                continue
        # Sort by priority if any have non-default priority
        if any(t.priority != 100 for t in tools):  # noqa: PLR2004
            tools.sort(key=lambda t: t.priority)

        return tools

    async def get_tool_names(self, state: ToolState = "all") -> set[str]:
        """Get tool names based on state."""
        return {t.name for t in await self.get_tools() if t.matches_filter(state)}

    def register_tool(
        self,
        tool: ToolType | Tool,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        enabled: bool = True,
        source: ToolSource = "runtime",
        priority: int = 100,
        requires_confirmation: bool = False,
        requires_capability: str | None = None,
        metadata: dict[str, str] | None = None,
    ) -> Tool:
        """Register a new tool with custom settings.

        Args:
            tool: Tool to register (callable, LLMCallableTool, or import path)
            enabled: Whether tool is initially enabled
            name_override: Optional name override for the tool
            description_override: Optional description override for the tool
            source: Tool source (runtime/agent/builtin/dynamic)
            priority: Execution priority (lower = higher priority)
            requires_confirmation: Whether tool needs confirmation
            requires_capability: Optional capability needed to use tool
            metadata: Additional tool metadata

        Returns:
            Created Tool instance
        """
        # First convert to basic Tool
        match tool:
            case LLMCallableTool():
                llm_tool = tool
                llm_tool.name = name_override or llm_tool.name
                llm_tool.description = description_override or llm_tool.description
            case Tool():
                llm_tool = tool.callable
                llm_tool.description = description_override or llm_tool.description
                llm_tool.name = name_override or llm_tool.name
            case _:
                llm_tool = LLMCallableTool.from_callable(
                    tool,
                    name_override=name_override,
                    description_override=description_override,
                )

        if llm_tool.description and len(llm_tool.description) > MAX_LEN_DESCRIPTION:
            msg = f"Too long description for {tool}"
            raise ToolError(msg)
        tool_info = Tool(
            llm_tool,
            enabled=enabled,
            source=source,
            priority=priority,
            requires_confirmation=requires_confirmation,
            requires_capability=requires_capability,
            metadata=metadata or {},
        )
        # Register the tool
        self.register(tool_info.name, tool_info)
        return tool_info

    def register_worker(
        self,
        worker: MessageNode[Any, Any],
        *,
        name: str | None = None,
        reset_history_on_run: bool = True,
        pass_message_history: bool = False,
        share_context: bool = False,
        parent: AnyAgent[Any, Any] | None = None,
    ) -> Tool:
        """Register an agent as a worker tool.

        Args:
            worker: Agent to register as worker
            name: Optional name override for the worker tool
            reset_history_on_run: Whether to clear history before each run
            pass_message_history: Whether to pass parent's message history
            share_context: Whether to pass parent's context/deps
            parent: Optional parent agent for history/context sharing
        """
        from llmling_agent import Agent, BaseTeam, StructuredAgent

        match worker:
            case BaseTeam():
                tool = worker.to_tool(name=name)
            case Agent() | StructuredAgent():
                tool = worker.to_tool(
                    parent=parent,
                    name=name,
                    reset_history_on_run=reset_history_on_run,
                    pass_message_history=pass_message_history,
                    share_context=share_context,
                )
            case _:
                msg = f"Unsupported worker type: {type(worker)}"
                raise ValueError(msg)
        msg = "Registering worker %s as tool %s"
        logger.debug(msg, worker.name, tool.name)
        return self.register_tool(tool, source="agent", metadata={"agent": worker.name})

    def reset(self):
        """Reset tool states."""
        old_tools = {i.name: i.enabled for i in self._items.values()}
        self.reset_states()
        new_tools = {i.name: i.enabled for i in self._items.values()}

        event = self.ToolStateReset(old_tools, new_tools)
        self.tool_states_reset.emit(event)

    @contextmanager
    def temporary_tools(
        self,
        tools: ToolType | Tool | Sequence[ToolType | Tool],
        *,
        exclusive: bool = False,
    ) -> Iterator[list[Tool]]:
        """Temporarily register tools.

        Args:
            tools: Tool(s) to register
            exclusive: Whether to temporarily disable all other tools

        Yields:
            List of registered tool infos

        Example:
            ```python
            with tool_manager.temporary_tools([tool1, tool2], exclusive=True) as tools:
                # Only tool1 and tool2 are available
                await agent.run(prompt)
            # Original tool states are restored
            ```
        """
        # Normalize inputs to lists
        tools_list: list[ToolType | Tool] = (
            [tools] if not isinstance(tools, Sequence) else list(tools)
        )

        # Store original tool states if exclusive
        original_states: dict[str, bool] = {}
        if exclusive:
            original_states = {name: tool.enabled for name, tool in self.items()}
            # Disable all existing tools
            for t in self.values():
                t.enabled = False

        # Register all tools
        registered_tools: list[Tool] = []
        try:
            for tool in tools_list:
                tool_info = self.register_tool(tool)
                registered_tools.append(tool_info)
            yield registered_tools

        finally:
            # Remove temporary tools
            for tool_info in registered_tools:
                del self[tool_info.name]

            # Restore original tool states if exclusive
            if exclusive:
                for name_, was_enabled in original_states.items():
                    if t := self.get(name_):
                        t.enabled = was_enabled

    def tool(
        self,
        name: str | None = None,
        *,
        description: str | None = None,
        enabled: bool = True,
        source: ToolSource = "runtime",
        priority: int = 100,
        requires_confirmation: bool = False,
        requires_capability: str | None = None,
        metadata: dict[str, str] | None = None,
    ) -> Callable[[AnyCallable], AnyCallable]:
        """Decorator to register a function as a tool.

        Args:
            name: Optional override for tool name (defaults to function name)
            description: Optional description override
            enabled: Whether tool is initially enabled
            source: Tool source type
            priority: Execution priority (lower = higher)
            requires_confirmation: Whether tool needs confirmation
            requires_capability: Optional required capability
            metadata: Additional tool metadata

        Returns:
            Decorator function that registers the tool

        Example:
            @tool_manager.register(
                name="search_docs",
                description="Search documentation",
                requires_confirmation=True
            )
            async def search(query: str) -> str:
                '''Search the docs.'''
                return "Results..."
        """

        def decorator(func: AnyCallable) -> AnyCallable:
            self.register_tool(
                func,
                name_override=name,
                description_override=description,
                enabled=enabled,
                source=source,
                priority=priority,
                requires_confirmation=requires_confirmation,
                requires_capability=requires_capability,
                metadata=metadata,
            )
            return func

        return decorator

_error_class property

_error_class: type[ToolError]

Error class for tool operations.

ToolStateReset dataclass

Emitted when tool states are reset.

Source code in src/llmling_agent/tools/manager.py

@dataclass(frozen=True)
class ToolStateReset:
    """Emitted when tool states are reset."""

    previous_tools: dict[str, bool]
    new_tools: dict[str, bool]
    timestamp: datetime = field(default_factory=get_now)

__init__

__init__(tools: Sequence[Tool | ToolType | dict[str, Any]] | None = None)

Initialize tool manager.

Parameters:

Name	Type	Description	Default
tools	Sequence[Tool | ToolType | dict[str, Any]] | None	Initial tools to register	None

Source code in src/llmling_agent/tools/manager.py

def __init__(
    self,
    tools: Sequence[Tool | ToolType | dict[str, Any]] | None = None,
):
    """Initialize tool manager.

    Args:
        tools: Initial tools to register
    """
    super().__init__()
    self.providers: list[ResourceProvider] = []

    # Register initial tools
    for tool in tools or []:
        t = self._validate_item(tool)
        self.register(t.name, t)

_validate_item

_validate_item(item: Tool | ToolType | dict[str, Any]) -> Tool

Validate and convert items before registration.

Source code in src/llmling_agent/tools/manager.py

def _validate_item(self, item: Tool | ToolType | dict[str, Any]) -> Tool:
    """Validate and convert items before registration."""
    match item:
        case Tool():
            return item
        case str():
            if item.startswith("crewai_tools"):
                obj = import_class(item)()
                return Tool.from_crewai_tool(obj)
            if item.startswith("langchain"):
                obj = import_class(item)()
                return Tool.from_langchain_tool(obj)
            return Tool.from_callable(item)
        case Callable():
            return Tool.from_callable(item)
        case {"callable": callable_item, **config} if callable(callable_item):
            valid_keys = {f.name for f in fields(Tool)} - {"callable"}
            tool_config = {k: v for k, v in config.items() if k in valid_keys}
            return Tool.from_callable(callable_item, **tool_config)  # type: ignore
        case _:
            typ = type(item)
            msg = f"Item must be Tool or callable. Got {typ}"
            raise ToolError(msg)

add_provider

add_provider(provider: ResourceProvider | ResourceCallable, owner: str | None = None)

Add a resource provider or tool callable.

Parameters:

Name	Type	Description	Default
provider	ResourceProvider | ResourceCallable	Either a ResourceProvider instance or a callable returning tools. Callables are automatically wrapped.	required
owner	str | None	Optional owner for the provider	None

Source code in src/llmling_agent/tools/manager.py

def add_provider(
    self,
    provider: ResourceProvider | ResourceCallable,
    owner: str | None = None,
):
    """Add a resource provider or tool callable.

    Args:
        provider: Either a ResourceProvider instance or a callable
                 returning tools. Callables are automatically wrapped.
        owner: Optional owner for the provider
    """
    from llmling_agent.resource_providers.base import ResourceProvider

    if not isinstance(provider, ResourceProvider):
        # Wrap old-style callable in ResourceProvider
        prov: ResourceProvider = CallableResourceProvider(
            name=provider.__name__,
            tool_callable=provider,
        )
    else:
        prov = provider
    if owner:
        prov.owner = owner
    self.providers.append(prov)

disable_tool

disable_tool(tool_name: str)

Disable a tool.

Source code in src/llmling_agent/tools/manager.py

def disable_tool(self, tool_name: str):
    """Disable a tool."""
    if tool_name not in self:
        msg = f"Tool not found: {tool_name}"
        raise ToolError(msg)
    tool_info = self[tool_name]
    tool_info.enabled = False
    self.events.changed(tool_name, tool_info)
    logger.debug("Disabled tool: %s", tool_name)

enable_tool

enable_tool(tool_name: str)

Enable a previously disabled tool.

Source code in src/llmling_agent/tools/manager.py

def enable_tool(self, tool_name: str):
    """Enable a previously disabled tool."""
    if tool_name not in self:
        msg = f"Tool not found: {tool_name}"
        raise ToolError(msg)
    tool_info = self[tool_name]
    tool_info.enabled = True
    self.events.changed(tool_name, tool_info)
    logger.debug("Enabled tool: %s", tool_name)

get_tool_names async

get_tool_names(state: ToolState = 'all') -> set[str]

Get tool names based on state.

Source code in src/llmling_agent/tools/manager.py

async def get_tool_names(self, state: ToolState = "all") -> set[str]:
    """Get tool names based on state."""
    return {t.name for t in await self.get_tools() if t.matches_filter(state)}

get_tools async

get_tools(state: ToolState = 'all', names: str | list[str] | None = None) -> list[Tool]

Get tool objects based on filters.

Source code in src/llmling_agent/tools/manager.py

async def get_tools(
    self,
    state: ToolState = "all",
    names: str | list[str] | None = None,
) -> list[Tool]:
    """Get tool objects based on filters."""
    tools = [t for t in self.values() if t.matches_filter(state)]
    match names:
        case str():
            tools = [t for t in tools if t.name == names]
        case list():
            tools = [t for t in tools if t.name in names]
    # Get tools from providers
    for provider in self.providers:
        try:
            provider_tools = await provider.get_tools()
            tools.extend(t for t in provider_tools if t.matches_filter(state))
        except Exception:
            logger.exception("Failed to get tools from provider: %r", provider)
            continue
    # Sort by priority if any have non-default priority
    if any(t.priority != 100 for t in tools):  # noqa: PLR2004
        tools.sort(key=lambda t: t.priority)

    return tools

is_tool_enabled

is_tool_enabled(tool_name: str) -> bool

Check if a tool is currently enabled.

Source code in src/llmling_agent/tools/manager.py

def is_tool_enabled(self, tool_name: str) -> bool:
    """Check if a tool is currently enabled."""
    return self[tool_name].enabled if tool_name in self else False

list_tools async

list_tools() -> dict[str, bool]

Get a mapping of all tools and their enabled status.

Source code in src/llmling_agent/tools/manager.py

async def list_tools(self) -> dict[str, bool]:
    """Get a mapping of all tools and their enabled status."""
    return {tool.name: tool.enabled for tool in await self.get_tools()}

register_tool

register_tool(
    tool: ToolType | Tool,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    enabled: bool = True,
    source: ToolSource = "runtime",
    priority: int = 100,
    requires_confirmation: bool = False,
    requires_capability: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Tool

Register a new tool with custom settings.

Parameters:

Name	Type	Description	Default
tool	ToolType | Tool	Tool to register (callable, LLMCallableTool, or import path)	required
enabled	bool	Whether tool is initially enabled	True
name_override	str | None	Optional name override for the tool	None
description_override	str | None	Optional description override for the tool	None
source	ToolSource	Tool source (runtime/agent/builtin/dynamic)	'runtime'
priority	int	Execution priority (lower = higher priority)	100
requires_confirmation	bool	Whether tool needs confirmation	False
requires_capability	str | None	Optional capability needed to use tool	None
metadata	dict[str, str] | None	Additional tool metadata	None

Returns:

Type	Description
Tool	Created Tool instance

Source code in src/llmling_agent/tools/manager.py

def register_tool(
    self,
    tool: ToolType | Tool,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    enabled: bool = True,
    source: ToolSource = "runtime",
    priority: int = 100,
    requires_confirmation: bool = False,
    requires_capability: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Tool:
    """Register a new tool with custom settings.

    Args:
        tool: Tool to register (callable, LLMCallableTool, or import path)
        enabled: Whether tool is initially enabled
        name_override: Optional name override for the tool
        description_override: Optional description override for the tool
        source: Tool source (runtime/agent/builtin/dynamic)
        priority: Execution priority (lower = higher priority)
        requires_confirmation: Whether tool needs confirmation
        requires_capability: Optional capability needed to use tool
        metadata: Additional tool metadata

    Returns:
        Created Tool instance
    """
    # First convert to basic Tool
    match tool:
        case LLMCallableTool():
            llm_tool = tool
            llm_tool.name = name_override or llm_tool.name
            llm_tool.description = description_override or llm_tool.description
        case Tool():
            llm_tool = tool.callable
            llm_tool.description = description_override or llm_tool.description
            llm_tool.name = name_override or llm_tool.name
        case _:
            llm_tool = LLMCallableTool.from_callable(
                tool,
                name_override=name_override,
                description_override=description_override,
            )

    if llm_tool.description and len(llm_tool.description) > MAX_LEN_DESCRIPTION:
        msg = f"Too long description for {tool}"
        raise ToolError(msg)
    tool_info = Tool(
        llm_tool,
        enabled=enabled,
        source=source,
        priority=priority,
        requires_confirmation=requires_confirmation,
        requires_capability=requires_capability,
        metadata=metadata or {},
    )
    # Register the tool
    self.register(tool_info.name, tool_info)
    return tool_info

register_worker

register_worker(
    worker: MessageNode[Any, Any],
    *,
    name: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False,
    share_context: bool = False,
    parent: AnyAgent[Any, Any] | None = None,
) -> Tool

Register an agent as a worker tool.

Parameters:

Name	Type	Description	Default
worker	MessageNode[Any, Any]	Agent to register as worker	required
name	str | None	Optional name override for the worker tool	None
reset_history_on_run	bool	Whether to clear history before each run	True
pass_message_history	bool	Whether to pass parent's message history	False
share_context	bool	Whether to pass parent's context/deps	False
parent	AnyAgent[Any, Any] | None	Optional parent agent for history/context sharing	None

Source code in src/llmling_agent/tools/manager.py

def register_worker(
    self,
    worker: MessageNode[Any, Any],
    *,
    name: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False,
    share_context: bool = False,
    parent: AnyAgent[Any, Any] | None = None,
) -> Tool:
    """Register an agent as a worker tool.

    Args:
        worker: Agent to register as worker
        name: Optional name override for the worker tool
        reset_history_on_run: Whether to clear history before each run
        pass_message_history: Whether to pass parent's message history
        share_context: Whether to pass parent's context/deps
        parent: Optional parent agent for history/context sharing
    """
    from llmling_agent import Agent, BaseTeam, StructuredAgent

    match worker:
        case BaseTeam():
            tool = worker.to_tool(name=name)
        case Agent() | StructuredAgent():
            tool = worker.to_tool(
                parent=parent,
                name=name,
                reset_history_on_run=reset_history_on_run,
                pass_message_history=pass_message_history,
                share_context=share_context,
            )
        case _:
            msg = f"Unsupported worker type: {type(worker)}"
            raise ValueError(msg)
    msg = "Registering worker %s as tool %s"
    logger.debug(msg, worker.name, tool.name)
    return self.register_tool(tool, source="agent", metadata={"agent": worker.name})

remove_provider

remove_provider(provider: ResourceProvider | ResourceCallable | ProviderName)

Remove a resource provider.

Source code in src/llmling_agent/tools/manager.py

def remove_provider(
    self, provider: ResourceProvider | ResourceCallable | ProviderName
):
    """Remove a resource provider."""
    from llmling_agent.resource_providers.base import ResourceProvider

    match provider:
        case ResourceProvider():
            self.providers.remove(provider)
        case Callable():
            # Find and remove wrapped callable
            for p in self.providers:
                if (
                    isinstance(p, CallableResourceProvider)
                    and p.tool_callable == provider
                ):
                    self.providers.remove(p)
        case str():
            for p in self.providers:
                if p.name == provider:
                    self.providers.remove(p)
        case _:
            msg = f"Invalid provider type: {type(provider)}"
            raise ValueError(msg)

reset

reset()

Reset tool states.

Source code in src/llmling_agent/tools/manager.py

def reset(self):
    """Reset tool states."""
    old_tools = {i.name: i.enabled for i in self._items.values()}
    self.reset_states()
    new_tools = {i.name: i.enabled for i in self._items.values()}

    event = self.ToolStateReset(old_tools, new_tools)
    self.tool_states_reset.emit(event)

reset_states

reset_states()

Reset all tools to their default enabled states.

Source code in src/llmling_agent/tools/manager.py

def reset_states(self):
    """Reset all tools to their default enabled states."""
    for info in self.values():
        info.enabled = True

temporary_tools

temporary_tools(
    tools: ToolType | Tool | Sequence[ToolType | Tool], *, exclusive: bool = False
) -> Iterator[list[Tool]]

Temporarily register tools.

Parameters:

Name	Type	Description	Default
tools	ToolType | Tool | Sequence[ToolType | Tool]	Tool(s) to register	required
exclusive	bool	Whether to temporarily disable all other tools	False

Yields:

Type	Description
list[Tool]	List of registered tool infos

Example

with tool_manager.temporary_tools([tool1, tool2], exclusive=True) as tools:
    # Only tool1 and tool2 are available
    await agent.run(prompt)
# Original tool states are restored

Source code in src/llmling_agent/tools/manager.py

@contextmanager
def temporary_tools(
    self,
    tools: ToolType | Tool | Sequence[ToolType | Tool],
    *,
    exclusive: bool = False,
) -> Iterator[list[Tool]]:
    """Temporarily register tools.

    Args:
        tools: Tool(s) to register
        exclusive: Whether to temporarily disable all other tools

    Yields:
        List of registered tool infos

    Example:
        ```python
        with tool_manager.temporary_tools([tool1, tool2], exclusive=True) as tools:
            # Only tool1 and tool2 are available
            await agent.run(prompt)
        # Original tool states are restored
        ```
    """
    # Normalize inputs to lists
    tools_list: list[ToolType | Tool] = (
        [tools] if not isinstance(tools, Sequence) else list(tools)
    )

    # Store original tool states if exclusive
    original_states: dict[str, bool] = {}
    if exclusive:
        original_states = {name: tool.enabled for name, tool in self.items()}
        # Disable all existing tools
        for t in self.values():
            t.enabled = False

    # Register all tools
    registered_tools: list[Tool] = []
    try:
        for tool in tools_list:
            tool_info = self.register_tool(tool)
            registered_tools.append(tool_info)
        yield registered_tools

    finally:
        # Remove temporary tools
        for tool_info in registered_tools:
            del self[tool_info.name]

        # Restore original tool states if exclusive
        if exclusive:
            for name_, was_enabled in original_states.items():
                if t := self.get(name_):
                    t.enabled = was_enabled

tool

tool(
    name: str | None = None,
    *,
    description: str | None = None,
    enabled: bool = True,
    source: ToolSource = "runtime",
    priority: int = 100,
    requires_confirmation: bool = False,
    requires_capability: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Callable[[AnyCallable], AnyCallable]

Decorator to register a function as a tool.

Parameters:

Name	Type	Description	Default
name	str | None	Optional override for tool name (defaults to function name)	None
description	str | None	Optional description override	None
enabled	bool	Whether tool is initially enabled	True
source	ToolSource	Tool source type	'runtime'
priority	int	Execution priority (lower = higher)	100
requires_confirmation	bool	Whether tool needs confirmation	False
requires_capability	str | None	Optional required capability	None
metadata	dict[str, str] | None	Additional tool metadata	None

Returns:

Type	Description
Callable[[AnyCallable], AnyCallable]	Decorator function that registers the tool

Example

@tool_manager.register( name="search_docs", description="Search documentation", requires_confirmation=True ) async def search(query: str) -> str: '''Search the docs.''' return "Results..."

Source code in src/llmling_agent/tools/manager.py

def tool(
    self,
    name: str | None = None,
    *,
    description: str | None = None,
    enabled: bool = True,
    source: ToolSource = "runtime",
    priority: int = 100,
    requires_confirmation: bool = False,
    requires_capability: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Callable[[AnyCallable], AnyCallable]:
    """Decorator to register a function as a tool.

    Args:
        name: Optional override for tool name (defaults to function name)
        description: Optional description override
        enabled: Whether tool is initially enabled
        source: Tool source type
        priority: Execution priority (lower = higher)
        requires_confirmation: Whether tool needs confirmation
        requires_capability: Optional required capability
        metadata: Additional tool metadata

    Returns:
        Decorator function that registers the tool

    Example:
        @tool_manager.register(
            name="search_docs",
            description="Search documentation",
            requires_confirmation=True
        )
        async def search(query: str) -> str:
            '''Search the docs.'''
            return "Results..."
    """

    def decorator(func: AnyCallable) -> AnyCallable:
        self.register_tool(
            func,
            name_override=name,
            description_override=description,
            enabled=enabled,
            source=source,
            priority=priority,
            requires_confirmation=requires_confirmation,
            requires_capability=requires_capability,
            metadata=metadata,
        )
        return func

    return decorator