tools

Class info

Classes

Name	Children	Inherits
SkillsRegistry llmling_agent.skills.registry Registry for Claude Code Skills with auto-discovery.		BaseRegistry
Tool llmling_agent.tools.base Information about a registered tool.		Generic
ToolCallInfo llmling_agent.tools.tool_call_info Information about an executed tool call.		Schema
ToolError llmling_agent.tools.manager Base exception for tool-related errors.		LLMLingError
ToolManager llmling_agent.tools.manager Manages tool registration, enabling/disabling and access.		object

🛈 DocStrings

Tool implementations and related classes / functions.

SkillsRegistry

Bases: BaseRegistry[str, Skill]

Registry for Claude Code Skills with auto-discovery.

Source code in src/llmling_agent/skills/registry.py

class SkillsRegistry(BaseRegistry[str, Skill]):
    """Registry for Claude Code Skills with auto-discovery."""

    DEFAULT_SKILL_PATHS: ClassVar = ["~/.claude/skills/", ".claude/skills/"]

    def __init__(self, skills_dirs: Sequence[JoinablePathLike] | None = None) -> None:
        """Initialize with custom skill directories or auto-detect."""
        super().__init__()
        if skills_dirs:
            self.skills_dirs = [UPath(i) for i in skills_dirs or []]
        else:
            self.skills_dirs = [UPath(i) for i in self.DEFAULT_SKILL_PATHS or []]

    async def discover_skills(self) -> None:
        """Scan filesystem and register all found skills.

        Args:
            filesystem: Optional async filesystem to use. If None, will use upath_to_fs()
                       to get appropriate filesystem for each skills directory.
        """
        for skills_dir in self.skills_dirs:
            await self.register_skills_from_path(skills_dir)

    async def register_skills_from_path(
        self,
        skills_dir: UPath | AbstractFileSystem,
        **storage_options: Any,
    ) -> None:
        """Register skills from a given path.

        Args:
            skills_dir: Path to the directory containing skills.
            storage_options: Additional options to pass to the filesystem.
        """
        if isinstance(skills_dir, AbstractFileSystem):
            fs = skills_dir
            if not isinstance(fs, AsyncFileSystem):
                fs = AsyncFileSystemWrapper(fs)
        else:
            fs = upath_to_fs(skills_dir, **storage_options)

        try:
            # List entries in skills directory
            entries = await fs._ls(fs.root_marker, detail=True)
        except FileNotFoundError:
            logger.warning("Skills directory not found", path=skills_dir)
            return
        # Filter for directories that might contain skills
        skill_dirs = [entry for entry in entries if entry.get("type") == "directory"]
        if not skill_dirs:
            logger.info("No skills found", skills_dir=skills_dir)
            return
        logger.info("Found skills", skills=skill_dirs, skills_dir=skills_dir)
        for skill_entry in skill_dirs:
            skill_name = skill_entry["name"].lstrip("./")
            skill_dir_path = skills_dir / skill_name
            try:
                await fs._cat(f"{skill_name}/SKILL.md")
            except FileNotFoundError:
                continue

            try:
                skill = self._parse_skill(skill_dir_path)
                self.register(skill.name, skill, replace=True)
            except Exception as e:  # noqa: BLE001
                # Log but don't fail discovery for one bad skill
                print(f"Warning: Failed to parse skill at {skill_dir_path}: {e}")

    def _parse_skill(self, skill_dir: JoinablePathLike) -> Skill:
        """Parse a SKILL.md file and extract metadata."""
        skill_file = UPath(skill_dir) / "SKILL.md"
        content = skill_file.read_text()

        # Extract YAML frontmatter
        frontmatter_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
        if not frontmatter_match:
            msg = f"No YAML frontmatter found in {skill_file}"
            raise ToolError(msg)
        import yamling

        try:
            metadata = yamling.load_yaml(frontmatter_match.group(1))
        except yamling.YAMLError as e:
            msg = f"Invalid YAML frontmatter in {skill_file}: {e}"
            raise ToolError(msg) from e

        # Validate required fields
        if not isinstance(metadata, dict):
            msg = f"YAML frontmatter must be a dictionary in {skill_file}"
            raise ToolError(msg)

        name = metadata.get("name")
        description = metadata.get("description")

        if not name:
            msg = f"Missing 'name' field in {skill_file}"
            raise ToolError(msg)
        if not description:
            msg = f"Missing 'description' field in {skill_file}"
            raise ToolError(msg)

        # Validate limits
        if len(name) > SKILL_NAME_LIMIT:
            msg = f"{skill_file}: Skill name exceeds {SKILL_NAME_LIMIT} chars"
            raise ToolError(msg)
        if len(description) > SKILL_DESCRIPTION_LIMIT:
            msg = f"{skill_file}: Skill description exceeds {SKILL_DESCRIPTION_LIMIT} chars"
            raise ToolError(msg)

        return Skill(name=name, description=description, skill_path=UPath(skill_dir))

    @property
    def _error_class(self) -> type[ToolError]:
        """Error class to use for this registry."""
        return ToolError

    def _validate_item(self, item: Any) -> Skill:
        """Validate and possibly transform item before registration."""
        if not isinstance(item, Skill):
            msg = f"Expected Skill instance, got {type(item)}"
            raise ToolError(msg)
        return item

    def get_skill_instructions(self, skill_name: str) -> str:
        """Lazy load full instructions for a skill."""
        skill = self.get(skill_name)
        return skill.load_instructions()

__init__

__init__(skills_dirs: Sequence[JoinablePathLike] | None = None) -> None

Initialize with custom skill directories or auto-detect.

Source code in src/llmling_agent/skills/registry.py

def __init__(self, skills_dirs: Sequence[JoinablePathLike] | None = None) -> None:
    """Initialize with custom skill directories or auto-detect."""
    super().__init__()
    if skills_dirs:
        self.skills_dirs = [UPath(i) for i in skills_dirs or []]
    else:
        self.skills_dirs = [UPath(i) for i in self.DEFAULT_SKILL_PATHS or []]

discover_skills async

discover_skills() -> None

Scan filesystem and register all found skills.

Parameters:

Name	Type	Description	Default
filesystem		Optional async filesystem to use. If None, will use upath_to_fs() to get appropriate filesystem for each skills directory.	required

Source code in src/llmling_agent/skills/registry.py

async def discover_skills(self) -> None:
    """Scan filesystem and register all found skills.

    Args:
        filesystem: Optional async filesystem to use. If None, will use upath_to_fs()
                   to get appropriate filesystem for each skills directory.
    """
    for skills_dir in self.skills_dirs:
        await self.register_skills_from_path(skills_dir)

get_skill_instructions

get_skill_instructions(skill_name: str) -> str

Lazy load full instructions for a skill.

Source code in src/llmling_agent/skills/registry.py

def get_skill_instructions(self, skill_name: str) -> str:
    """Lazy load full instructions for a skill."""
    skill = self.get(skill_name)
    return skill.load_instructions()

register_skills_from_path async

register_skills_from_path(skills_dir: UPath | AbstractFileSystem, **storage_options: Any) -> None

Register skills from a given path.

Parameters:

Name	Type	Description	Default
skills_dir	UPath | AbstractFileSystem	Path to the directory containing skills.	required
storage_options	Any	Additional options to pass to the filesystem.	{}

Source code in src/llmling_agent/skills/registry.py

async def register_skills_from_path(
    self,
    skills_dir: UPath | AbstractFileSystem,
    **storage_options: Any,
) -> None:
    """Register skills from a given path.

    Args:
        skills_dir: Path to the directory containing skills.
        storage_options: Additional options to pass to the filesystem.
    """
    if isinstance(skills_dir, AbstractFileSystem):
        fs = skills_dir
        if not isinstance(fs, AsyncFileSystem):
            fs = AsyncFileSystemWrapper(fs)
    else:
        fs = upath_to_fs(skills_dir, **storage_options)

    try:
        # List entries in skills directory
        entries = await fs._ls(fs.root_marker, detail=True)
    except FileNotFoundError:
        logger.warning("Skills directory not found", path=skills_dir)
        return
    # Filter for directories that might contain skills
    skill_dirs = [entry for entry in entries if entry.get("type") == "directory"]
    if not skill_dirs:
        logger.info("No skills found", skills_dir=skills_dir)
        return
    logger.info("Found skills", skills=skill_dirs, skills_dir=skills_dir)
    for skill_entry in skill_dirs:
        skill_name = skill_entry["name"].lstrip("./")
        skill_dir_path = skills_dir / skill_name
        try:
            await fs._cat(f"{skill_name}/SKILL.md")
        except FileNotFoundError:
            continue

        try:
            skill = self._parse_skill(skill_dir_path)
            self.register(skill.name, skill, replace=True)
        except Exception as e:  # noqa: BLE001
            # Log but don't fail discovery for one bad skill
            print(f"Warning: Failed to parse skill at {skill_dir_path}: {e}")

Tool dataclass

Information about a registered tool.

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

@dataclass
class Tool[TOutputType = Any]:
    """Information about a registered tool."""

    callable: Callable[..., TOutputType]
    """The actual tool implementation"""

    name: str
    """The name of the tool."""

    description: str = ""
    """The description of the tool."""

    schema_override: schemez.OpenAIFunctionDefinition | None = None
    """Schema override. If not set, the schema is inferred from the callable."""

    hints: ToolHints | None = None
    """Hints for the tool."""

    import_path: str | None = None
    """The import path for the tool."""

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

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

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

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

    category: ToolKind | None = None
    """The category of the tool."""

    __repr__ = dataclasses_no_defaults_repr

    def to_pydantic_ai(self) -> PydanticAiTool:
        """Convert tool to Pydantic AI tool."""
        metadata = {
            **self.metadata,
            "agent_name": self.agent_name,
            "category": self.category,
        }
        return PydanticAiTool(
            function=self.callable,
            name=self.name,
            # takes_ctx=self.takes_ctx,
            # max_retries=self.max_retries,
            description=self.description,
            requires_approval=self.requires_confirmation,
            metadata=metadata,
        )

    @property
    def schema_obj(self) -> FunctionSchema:
        """Get the OpenAI function schema for the tool."""
        return schemez.create_schema(
            self.callable,
            name_override=self.name,
            description_override=self.description,
        )

    @property
    def schema(self) -> schemez.OpenAIFunctionTool:
        """Get the OpenAI function schema for the tool."""
        schema = self.schema_obj.model_dump_openai()
        if self.schema_override:
            schema["function"] = self.schema_override
        return schema

    def matches_filter(self, state: ToolState) -> 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)

    @logfire.instrument("Executing tool {self.name} with args={args}, kwargs={kwargs}")
    async def execute(self, *args: Any, **kwargs: Any) -> Any:
        """Execute tool, handling both sync and async cases."""
        return await execute(self.callable, *args, **kwargs, use_thread=True)

    @classmethod
    def from_code(
        cls,
        code: str,
        name: str | None = None,
        description: str | None = None,
    ) -> Tool[Any]:
        """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,  # pyright: ignore[reportArgumentType]
            name_override=name,
            description_override=description,
        )

    @classmethod
    def from_callable(
        cls,
        fn: Callable[..., TOutputType | Awaitable[TOutputType]] | str,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: schemez.OpenAIFunctionDefinition | None = None,
        hints: ToolHints | None = None,
        **kwargs: Any,
    ) -> Tool[TOutputType]:
        if isinstance(fn, str):
            import_path = fn
            from llmling_agent.utils import importing

            callable_obj = importing.import_callable(fn)
            name = getattr(callable_obj, "__name__", "unknown")
            import_path = fn
        else:
            callable_obj = fn
            module = fn.__module__
            if hasattr(fn, "__qualname__"):  # Regular function
                name = fn.__name__
                import_path = f"{module}.{fn.__qualname__}"
            else:  # Instance with __call__ method
                name = fn.__class__.__name__
                import_path = f"{module}.{fn.__class__.__qualname__}"

        return cls(
            callable=callable_obj,  # pyright: ignore[reportArgumentType]
            name=name_override or name,
            description=description_override or inspect.getdoc(callable_obj) or "",
            import_path=import_path,
            schema_override=schema_override,
            hints=hints,
            **kwargs,
        )

    @classmethod
    def from_crewai_tool(
        cls,
        tool: Any,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: schemez.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Tool[Any]:
        """Allows importing crewai tools."""
        # vaidate_import("crewai_tools", "crewai")
        try:
            from crewai.tools import (  # type: ignore[import-not-found]
                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: schemez.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Tool[Any]:
        """Create a tool from a LangChain tool."""
        # vaidate_import("langchain_core", "langchain")
        try:
            from langchain_core.tools import (  # type: ignore[import-not-found]
                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: schemez.OpenAIFunctionDefinition | None = None,
        **kwargs: Any,
    ) -> Tool[Any]:
        """Create a tool from a AutoGen tool."""
        # vaidate_import("autogen_core", "autogen")
        try:
            from autogen_core import CancellationToken  # type: ignore[import-not-found]
            from autogen_core.tools import BaseTool  # type: ignore[import-not-found]
        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]

        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,
            name_override=name,
            description_override=description,
            schema_override=schema_override,
            **kwargs,
        )

    def to_mcp_tool(self) -> MCPTool:
        """Convert internal Tool to MCP Tool."""
        schema = self.schema
        from mcp.types import Tool as MCPTool, ToolAnnotations

        return MCPTool(
            name=schema["function"]["name"],
            description=schema["function"]["description"],
            inputSchema=schema["function"]["parameters"],  # pyright: ignore
            annotations=ToolAnnotations(
                title=self.name,
                readOnlyHint=self.hints.read_only if self.hints else None,
                destructiveHint=self.hints.destructive if self.hints else None,
                idempotentHint=self.hints.idempotent if self.hints else None,
                openWorldHint=self.hints.open_world if self.hints else None,
            ),
        )

agent_name class-attribute instance-attribute

agent_name: str | None = None

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

callable instance-attribute

callable: Callable[..., TOutputType]

The actual tool implementation

category class-attribute instance-attribute

category: ToolKind | None = None

The category of the tool.

description class-attribute instance-attribute

description: str = ''

The description of the tool.

enabled class-attribute instance-attribute

enabled: bool = True

Whether the tool is currently enabled

hints class-attribute instance-attribute

hints: ToolHints | None = None

Hints for the tool.

import_path class-attribute instance-attribute

import_path: str | None = None

The import path for the tool.

metadata class-attribute instance-attribute

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

Additional tool metadata

name instance-attribute

name: str

The name of the tool.

parameters property

parameters: list[ToolParameter]

Get information about tool parameters.

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.

schema_obj property

schema_obj: FunctionSchema

Get the OpenAI function schema for the tool.

schema_override class-attribute instance-attribute

schema_override: OpenAIFunctionDefinition | None = None

Schema override. If not set, the schema is inferred from the callable.

source class-attribute instance-attribute

source: ToolSource = 'dynamic'

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

@logfire.instrument("Executing tool {self.name} with args={args}, kwargs={kwargs}")
async def execute(self, *args: Any, **kwargs: Any) -> Any:
    """Execute tool, handling both sync and async cases."""
    return await execute(self.callable, *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
) -> Tool[Any]

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: schemez.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Tool[Any]:
    """Create a tool from a AutoGen tool."""
    # vaidate_import("autogen_core", "autogen")
    try:
        from autogen_core import CancellationToken  # type: ignore[import-not-found]
        from autogen_core.tools import BaseTool  # type: ignore[import-not-found]
    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]

    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,
        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) -> Tool[Any]

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,
) -> Tool[Any]:
    """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,  # pyright: ignore[reportArgumentType]
        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
) -> Tool[Any]

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: schemez.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Tool[Any]:
    """Allows importing crewai tools."""
    # vaidate_import("crewai_tools", "crewai")
    try:
        from crewai.tools import (  # type: ignore[import-not-found]
            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
) -> Tool[Any]

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: schemez.OpenAIFunctionDefinition | None = None,
    **kwargs: Any,
) -> Tool[Any]:
    """Create a tool from a LangChain tool."""
    # vaidate_import("langchain_core", "langchain")
    try:
        from langchain_core.tools import (  # type: ignore[import-not-found]
            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: ToolState) -> bool

Check if tool matches state filter.

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

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

to_mcp_tool

to_mcp_tool() -> Tool

Convert internal Tool to MCP Tool.

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

def to_mcp_tool(self) -> MCPTool:
    """Convert internal Tool to MCP Tool."""
    schema = self.schema
    from mcp.types import Tool as MCPTool, ToolAnnotations

    return MCPTool(
        name=schema["function"]["name"],
        description=schema["function"]["description"],
        inputSchema=schema["function"]["parameters"],  # pyright: ignore
        annotations=ToolAnnotations(
            title=self.name,
            readOnlyHint=self.hints.read_only if self.hints else None,
            destructiveHint=self.hints.destructive if self.hints else None,
            idempotentHint=self.hints.idempotent if self.hints else None,
            openWorldHint=self.hints.open_world if self.hints else None,
        ),
    )

to_pydantic_ai

to_pydantic_ai() -> Tool

Convert tool to Pydantic AI tool.

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

def to_pydantic_ai(self) -> PydanticAiTool:
    """Convert tool to Pydantic AI tool."""
    metadata = {
        **self.metadata,
        "agent_name": self.agent_name,
        "category": self.category,
    }
    return PydanticAiTool(
        function=self.callable,
        name=self.name,
        # takes_ctx=self.takes_ctx,
        # max_retries=self.max_retries,
        description=self.description,
        requires_approval=self.requires_confirmation,
        metadata=metadata,
    )

ToolCallInfo

Bases: Schema

Information about an executed tool call.

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

class ToolCallInfo(Schema):
    """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."""

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

    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.

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

ToolError

Bases: LLMLingError

Base exception for tool-related errors.

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

class ToolError(LLMLingError):
    """Base exception for tool-related errors."""

ToolManager

Manages tool registration, enabling/disabling and access.

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

class ToolManager:
    """Manages tool registration, enabling/disabling and access."""

    def __init__(
        self,
        tools: Sequence[Tool | ToolType] | None = None,
        tool_mode: ToolMode | None = None,
    ) -> None:
        """Initialize tool manager.

        Args:
            tools: Initial tools to register
            tool_mode: Tool execution mode (None or "codemode")
        """
        super().__init__()
        self.external_providers: list[ResourceProvider] = []
        self.worker_provider = StaticResourceProvider(name="workers")
        self.builtin_provider = StaticResourceProvider(name="builtin")
        self.tool_mode = tool_mode

        # CodeModeResourceProvider gets populated with providers in providers property
        from llmling_agent.resource_providers.codemode.provider import CodeModeResourceProvider

        self._codemode_provider: CodeModeResourceProvider = CodeModeResourceProvider([])

        # Forward to provider methods
        self.tool = self.builtin_provider.tool
        self.register_tool = self.builtin_provider.register_tool
        self.register_worker = self.worker_provider.register_worker

        # Register initial tools
        for tool in tools or []:
            t = self._validate_item(tool)
            self.builtin_provider.add_tool(t)

    @property
    def providers(self) -> list[ResourceProvider]:
        """Get all providers: external + worker + builtin providers."""
        if self.tool_mode == "codemode":
            # Update the providers list with current providers
            self._codemode_provider.providers[:] = [
                *self.external_providers,
                self.worker_provider,
                self.builtin_provider,
            ]
            return [self._codemode_provider]

        return [*self.external_providers, self.worker_provider, self.builtin_provider]

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

    def add_provider(self, provider: ResourceProvider, owner: str | None = None) -> None:
        """Add an external resource provider.

        Args:
            provider: ResourceProvider instance (e.g., MCP server, custom provider)
            owner: Optional owner for the provider
        """
        if owner:
            provider.owner = owner
        self.external_providers.append(provider)

    def remove_provider(self, provider: ResourceProvider | ProviderName) -> None:
        """Remove an external resource provider."""
        from llmling_agent.resource_providers import ResourceProvider

        match provider:
            case ResourceProvider():
                self.external_providers.remove(provider)
            case str():
                for p in self.external_providers:
                    if p.name == provider:
                        self.external_providers.remove(p)
            case _ as unreachable:
                assert_never(unreachable)

    async def reset_states(self) -> None:
        """Reset all tools to their default enabled states."""
        for info in await self.get_tools():
            info.enabled = True

    def _validate_item(self, item: Tool | ToolType) -> 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():  # type: ignore[misc]
                return Tool.from_callable(item)
            case _:
                typ = type(item)
                msg = f"Item must be Tool or callable. Got {typ}"
                raise ToolError(msg)

    async def enable_tool(self, tool_name: str) -> None:
        """Enable a previously disabled tool."""
        tool_info = await self.get_tool(tool_name)
        tool_info.enabled = True
        logger.debug("Enabled tool", tool_name=tool_name)

    async def disable_tool(self, tool_name: str) -> None:
        """Disable a tool."""
        tool_info = await self.get_tool(tool_name)
        tool_info.enabled = False
        logger.debug("Disabled tool", tool_name=tool_name)

    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: list[Tool] = []
        # Get tools from providers concurrently
        provider_coroutines = [provider.get_tools() for provider in self.providers]
        results = await asyncio.gather(*provider_coroutines, return_exceptions=True)
        for provider, result in zip(self.providers, results, strict=False):
            if isinstance(result, BaseException):
                logger.warning(
                    "Failed to get tools from provider",
                    provider=provider,
                    result=result,
                )
                continue
            tools.extend(t for t in result 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]
        return tools

    async def get_tool(self, name: str) -> Tool:
        """Get a specific tool by name.

        First checks local tools, then uses concurrent provider fetching.

        Args:
            name: Name of the tool to retrieve

        Returns:
            Tool instance if found, None otherwise
        """
        all_tools = await self.get_tools()
        tool = next((tool for tool in all_tools if tool.name == name), None)
        if not tool:
            msg = f"Tool not found: {tool}"
            raise ToolError(msg)
        return tool

    async def list_prompts(self) -> list[MCPClientPrompt]:
        """Get all prompts from all providers.

        Returns:
            List of Prompt instances
        """
        from llmling_agent.mcp_server.manager import MCPManager

        all_prompts: list[MCPClientPrompt] = []

        # Get prompts from all external providers (check if they're MCP providers)
        for provider in self.external_providers:
            if isinstance(provider, MCPManager):
                try:
                    # Get prompts from MCP providers via the aggregating provider
                    agg_provider = provider.get_aggregating_provider()
                    prompts = await agg_provider.get_prompts()
                    all_prompts.extend(prompts)
                except Exception:
                    logger.exception("Failed to get prompts from provider", provider=provider)

        return all_prompts

    @asynccontextmanager
    async def temporary_tools(
        self,
        tools: ToolType | Tool | Sequence[ToolType | Tool],
        *,
        exclusive: bool = False,
    ) -> AsyncIterator[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
        tools = await self.get_tools()
        original_states: dict[str, bool] = {}
        if exclusive:
            original_states = {t.name: t.enabled for t in tools}
            # Disable all existing tools
            for t in tools:
                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:
                self.builtin_provider.remove_tool(tool_info.name)

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

providers property

providers: list[ResourceProvider]

Get all providers: external + worker + builtin providers.

__init__

__init__(tools: Sequence[Tool | ToolType] | None = None, tool_mode: ToolMode | None = None) -> None

Initialize tool manager.

Parameters:

Name	Type	Description	Default
tools	Sequence[Tool | ToolType] | None	Initial tools to register	None
tool_mode	ToolMode | None	Tool execution mode (None or "codemode")	None

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

def __init__(
    self,
    tools: Sequence[Tool | ToolType] | None = None,
    tool_mode: ToolMode | None = None,
) -> None:
    """Initialize tool manager.

    Args:
        tools: Initial tools to register
        tool_mode: Tool execution mode (None or "codemode")
    """
    super().__init__()
    self.external_providers: list[ResourceProvider] = []
    self.worker_provider = StaticResourceProvider(name="workers")
    self.builtin_provider = StaticResourceProvider(name="builtin")
    self.tool_mode = tool_mode

    # CodeModeResourceProvider gets populated with providers in providers property
    from llmling_agent.resource_providers.codemode.provider import CodeModeResourceProvider

    self._codemode_provider: CodeModeResourceProvider = CodeModeResourceProvider([])

    # Forward to provider methods
    self.tool = self.builtin_provider.tool
    self.register_tool = self.builtin_provider.register_tool
    self.register_worker = self.worker_provider.register_worker

    # Register initial tools
    for tool in tools or []:
        t = self._validate_item(tool)
        self.builtin_provider.add_tool(t)

add_provider

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

Add an external resource provider.

Parameters:

Name	Type	Description	Default
provider	ResourceProvider	ResourceProvider instance (e.g., MCP server, custom provider)	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, owner: str | None = None) -> None:
    """Add an external resource provider.

    Args:
        provider: ResourceProvider instance (e.g., MCP server, custom provider)
        owner: Optional owner for the provider
    """
    if owner:
        provider.owner = owner
    self.external_providers.append(provider)

disable_tool async

disable_tool(tool_name: str) -> None

Disable a tool.

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

async def disable_tool(self, tool_name: str) -> None:
    """Disable a tool."""
    tool_info = await self.get_tool(tool_name)
    tool_info.enabled = False
    logger.debug("Disabled tool", tool_name=tool_name)

enable_tool async

enable_tool(tool_name: str) -> None

Enable a previously disabled tool.

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

async def enable_tool(self, tool_name: str) -> None:
    """Enable a previously disabled tool."""
    tool_info = await self.get_tool(tool_name)
    tool_info.enabled = True
    logger.debug("Enabled tool", tool_name=tool_name)

get_tool async

get_tool(name: str) -> Tool

Get a specific tool by name.

First checks local tools, then uses concurrent provider fetching.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool to retrieve	required

Returns:

Type	Description
Tool	Tool instance if found, None otherwise

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

async def get_tool(self, name: str) -> Tool:
    """Get a specific tool by name.

    First checks local tools, then uses concurrent provider fetching.

    Args:
        name: Name of the tool to retrieve

    Returns:
        Tool instance if found, None otherwise
    """
    all_tools = await self.get_tools()
    tool = next((tool for tool in all_tools if tool.name == name), None)
    if not tool:
        msg = f"Tool not found: {tool}"
        raise ToolError(msg)
    return tool

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: list[Tool] = []
    # Get tools from providers concurrently
    provider_coroutines = [provider.get_tools() for provider in self.providers]
    results = await asyncio.gather(*provider_coroutines, return_exceptions=True)
    for provider, result in zip(self.providers, results, strict=False):
        if isinstance(result, BaseException):
            logger.warning(
                "Failed to get tools from provider",
                provider=provider,
                result=result,
            )
            continue
        tools.extend(t for t in result 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]
    return tools

list_prompts async

list_prompts() -> list[MCPClientPrompt]

Get all prompts from all providers.

Returns:

Type	Description
list[MCPClientPrompt]	List of Prompt instances

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

async def list_prompts(self) -> list[MCPClientPrompt]:
    """Get all prompts from all providers.

    Returns:
        List of Prompt instances
    """
    from llmling_agent.mcp_server.manager import MCPManager

    all_prompts: list[MCPClientPrompt] = []

    # Get prompts from all external providers (check if they're MCP providers)
    for provider in self.external_providers:
        if isinstance(provider, MCPManager):
            try:
                # Get prompts from MCP providers via the aggregating provider
                agg_provider = provider.get_aggregating_provider()
                prompts = await agg_provider.get_prompts()
                all_prompts.extend(prompts)
            except Exception:
                logger.exception("Failed to get prompts from provider", provider=provider)

    return all_prompts

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()}

remove_provider

remove_provider(provider: ResourceProvider | ProviderName) -> None

Remove an external resource provider.

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

def remove_provider(self, provider: ResourceProvider | ProviderName) -> None:
    """Remove an external resource provider."""
    from llmling_agent.resource_providers import ResourceProvider

    match provider:
        case ResourceProvider():
            self.external_providers.remove(provider)
        case str():
            for p in self.external_providers:
                if p.name == provider:
                    self.external_providers.remove(p)
        case _ as unreachable:
            assert_never(unreachable)

reset_states async

reset_states() -> None

Reset all tools to their default enabled states.

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

async def reset_states(self) -> None:
    """Reset all tools to their default enabled states."""
    for info in await self.get_tools():
        info.enabled = True

temporary_tools async

temporary_tools(
    tools: ToolType | Tool | Sequence[ToolType | Tool], *, exclusive: bool = False
) -> AsyncIterator[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
AsyncIterator[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

@asynccontextmanager
async def temporary_tools(
    self,
    tools: ToolType | Tool | Sequence[ToolType | Tool],
    *,
    exclusive: bool = False,
) -> AsyncIterator[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
    tools = await self.get_tools()
    original_states: dict[str, bool] = {}
    if exclusive:
        original_states = {t.name: t.enabled for t in tools}
        # Disable all existing tools
        for t in tools:
            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:
            self.builtin_provider.remove_tool(tool_info.name)

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