Skip to content

tools

Class info

Classes

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

            🛈 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/tools/skills.py 
             45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
            class SkillsRegistry(BaseRegistry[str, Skill]):
    """Registry for Claude Code Skills with auto-discovery."""

    # Default skill discovery paths (can be overridden by subclasses)
    DEFAULT_SKILL_PATHS: ClassVar = [
        "~/.claude/skills",  # Global user skills
        ".claude/skills",  # Project-local skills (walks up tree)
    ]

    def __init__(self, skills_dirs: list[Path] | None = None) -> None:
        """Initialize with custom skill directories or auto-detect."""
        super().__init__()
        self.skills_dirs = skills_dirs or self._get_default_skills_dirs()

    def _get_default_skills_dirs(self) -> list[Path]:
        """Get skill directories from class attribute paths."""
        dirs = []

        for path_pattern in self.DEFAULT_SKILL_PATHS:
            if path_pattern.startswith("~/"):
                # Expand home directory
                resolved_path = Path.home() / path_pattern[2:]
                if resolved_path.exists():
                    dirs.append(resolved_path)
            elif not path_pattern.startswith("/"):
                # Relative path - walk up directory tree
                cwd = Path.cwd()
                for parent in [cwd, *list(cwd.parents)]:
                    candidate = parent / path_pattern
                    if candidate.exists():
                        dirs.append(candidate)
                        break
            else:
                # Absolute path
                absolute_path = Path(path_pattern)
                if absolute_path.exists():
                    dirs.append(absolute_path)

        return dirs

    async def discover_skills(self) -> None:
        """Scan filesystem and register all found skills."""
        for skills_dir in self.skills_dirs:
            if not skills_dir.exists():
                continue

            for skill_dir in skills_dir.iterdir():
                if not skill_dir.is_dir():
                    continue

                skill_file = skill_dir / "SKILL.md"
                if not skill_file.exists():
                    continue

                try:
                    skill = self._parse_skill(skill_dir, skills_dir)
                    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}: {e}")

    def _parse_skill(self, skill_dir: Path, source_dir: Path) -> Skill:
        """Parse a SKILL.md file and extract metadata."""
        skill_file = skill_dir / "SKILL.md"
        content = skill_file.read_text(encoding="utf-8")

        # 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=skill_dir,
            source=source_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: list[Path] | None = None) -> None

            Initialize with custom skill directories or auto-detect.

            Source code in src/llmling_agent/tools/skills.py 
            54
55
56
57
            def __init__(self, skills_dirs: list[Path] | None = None) -> None:
    """Initialize with custom skill directories or auto-detect."""
    super().__init__()
    self.skills_dirs = skills_dirs or self._get_default_skills_dirs()

            discover_skills async 

            discover_skills() -> None

            Scan filesystem and register all found skills.

            Source code in src/llmling_agent/tools/skills.py 
             85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
            async def discover_skills(self) -> None:
    """Scan filesystem and register all found skills."""
    for skills_dir in self.skills_dirs:
        if not skills_dir.exists():
            continue

        for skill_dir in skills_dir.iterdir():
            if not skill_dir.is_dir():
                continue

            skill_file = skill_dir / "SKILL.md"
            if not skill_file.exists():
                continue

            try:
                skill = self._parse_skill(skill_dir, skills_dir)
                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}: {e}")

            get_skill_instructions 

            get_skill_instructions(skill_name: str) -> str

            Lazy load full instructions for a skill.

            Source code in src/llmling_agent/tools/skills.py 
            168
169
170
171
            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()

            Tool dataclass

            Information about a registered tool.

            Source code in src/llmling_agent/tools/base.py 
             42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
            @dataclass
class Tool:
    """Information about a registered tool."""

    callable: Callable
    """The actual tool implementation"""

    name: str

    description: str = ""

    schema_override: schemez.OpenAIFunctionDefinition | None = None

    hints: ToolHints | None = None

    import_path: str | None = None

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

    source: ToolSource = "dynamic"
    """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."""

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

    @property
    def schema(self) -> schemez.OpenAIFunctionTool:
        """Get the OpenAI function schema for the tool."""
        schema = schemez.create_schema(self.callable).model_dump_openai()
        schema["function"]["name"] = self.name
        schema["function"]["description"] = self.description
        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,
    ) -> 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: schemez.OpenAIFunctionDefinition | None = None,
        hints: ToolHints | None = None,
        **kwargs: Any,
    ) -> Self:
        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,
            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,
    ) -> Self:
        """Allows importing crewai tools."""
        # vaidate_import("crewai_tools", "crewai")
        try:
            from crewai.tools import BaseTool as CrewAiBaseTool  # pyright: ignore
        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,
    ) -> Self:
        """Create a tool from a LangChain tool."""
        # vaidate_import("langchain_core", "langchain")
        try:
            from langchain_core.tools import (  # pyright: ignore
                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,
    ) -> Self:
        """Create a tool from a AutoGen tool."""
        # vaidate_import("autogen_core", "autogen")
        try:
            from autogen_core import CancellationToken  # pyright: ignore
            from autogen_core.tools import BaseTool  # pyright: ignore
        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,
        )

    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.

            cache_enabled class-attribute instance-attribute 

            cache_enabled: bool = False

            Whether to enable caching for this tool.

            callable instance-attribute 

            callable: Callable

            The actual tool implementation

            category class-attribute instance-attribute 

            category: ToolKind | None = None

            The category of the tool.

            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

            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 = '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 
            136
137
138
139
            @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 
            123
124
125
126
127
128
129
130
131
132
133
134
            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 
            259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
            @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,
) -> Self:
    """Create a tool from a AutoGen tool."""
    # vaidate_import("autogen_core", "autogen")
    try:
        from autogen_core import CancellationToken  # pyright: ignore
        from autogen_core.tools import BaseTool  # pyright: ignore
    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 
            141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
            @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 
            197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
            @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,
) -> Self:
    """Allows importing crewai tools."""
    # vaidate_import("crewai_tools", "crewai")
    try:
        from crewai.tools import BaseTool as CrewAiBaseTool  # pyright: ignore
    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 
            227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
            @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,
) -> Self:
    """Create a tool from a LangChain tool."""
    # vaidate_import("langchain_core", "langchain")
    try:
        from langchain_core.tools import (  # pyright: ignore
            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 
             96
 97
 98
 99
100
101
102
103
104
            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 
            306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
            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,
        ),
    )

            ToolCallInfo

            Bases: Schema

            Information about an executed tool call.

            Source code in src/llmling_agent/tools/tool_call_info.py 
             71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
            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 
            104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
            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 
            42
43
            class ToolError(LLMLingError):
    """Base exception for tool-related errors."""

            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 
             46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
            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 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 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", tool_name=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", tool_name=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 concurrently
        if self.providers:
            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.exception(
                        "Failed to get tools from provider", provider=provider
                    )
                    continue
                tools.extend(t for t in result if t.matches_filter(state))
        # 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(self, name: str) -> Tool | None:
        """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
        """
        # Early exit for "local" tools
        if name in self:
            return self[name]
        all_tools = await self.get_tools()
        return next((tool for tool in all_tools if tool.name == name), None)

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

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

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

        all_prompts: list[Prompt] = []

        # Get prompts from all MCP providers
        for provider in self.providers:
            if isinstance(provider, MCPManager):
                try:
                    prompts = await provider.list_prompts()
                    all_prompts.extend(prompts)
                except Exception:
                    logger.exception(
                        "Failed to get prompts from provider", provider=provider
                    )

        return all_prompts

    def register_tool(
        self,
        tool: ToolType | Tool,
        *,
        name_override: str | None = None,
        description_override: str | None = None,
        enabled: bool = True,
        source: ToolSource = "dynamic",
        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, 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 Tool():
                tool.description = description_override or tool.description
                tool.name = name_override or tool.name
                tool.source = source
                tool.metadata = tool.metadata | (metadata or {})

            case _:
                tool = Tool.from_callable(
                    tool,
                    enabled=enabled,
                    source=source,
                    name_override=name_override,
                    description_override=description_override,
                    priority=priority,
                    requires_confirmation=requires_confirmation,
                    requires_capability=requires_capability,
                    metadata=metadata or {},
                )

        # Register the tool
        self.register(tool.name, tool)
        return tool

    def register_worker(
        self,
        worker: MessageNode[Any, Any],
        *,
        name: str | None = None,
        reset_history_on_run: bool = True,
        pass_message_history: bool = False,
        parent: Agent[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
            parent: Optional parent agent for history/context sharing
        """
        from llmling_agent import Agent, BaseTeam

        match worker:
            case BaseTeam():
                tool = worker.to_tool(name=name)
            case Agent():
                tool = worker.to_tool(
                    parent=parent,
                    name=name,
                    reset_history_on_run=reset_history_on_run,
                    pass_message_history=pass_message_history,
                )
            case _:
                msg = f"Unsupported worker type: {type(worker)}"
                raise ValueError(msg)
        msg = "Registering worker as tool"
        logger.debug(msg, worker_name=worker.name, tool_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 = "dynamic",
        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

            ToolStateReset dataclass

            Emitted when tool states are reset.

            Source code in src/llmling_agent/tools/manager.py 
            59
60
61
62
63
64
65
            @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 
            69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
            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)

            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 
             92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
            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 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 
            187
188
189
190
191
192
193
194
195
            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", tool_name=tool_name)

            enable_tool 

            enable_tool(tool_name: str)

            Enable a previously disabled tool.

            Source code in src/llmling_agent/tools/manager.py 
            177
178
179
180
181
182
183
184
185
            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", tool_name=tool_name)

            get_tool async 

            get_tool(name: str) -> Tool | None

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

            Tool instance if found, None otherwise
            Source code in src/llmling_agent/tools/manager.py 
            234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
            async def get_tool(self, name: str) -> Tool | None:
    """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
    """
    # Early exit for "local" tools
    if name in self:
        return self[name]
    all_tools = await self.get_tools()
    return next((tool for tool in all_tools if tool.name == name), None)

            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 
            251
252
253
            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 
            205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
            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 concurrently
    if self.providers:
        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.exception(
                    "Failed to get tools from provider", provider=provider
                )
                continue
            tools.extend(t for t in result if t.matches_filter(state))
    # 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 
            197
198
199
            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_prompts async 

            list_prompts() -> list[Prompt]

            Get all prompts from all providers.

            Returns:

            Type Description
            list[Prompt]

            List of Prompt instances
            Source code in src/llmling_agent/tools/manager.py 
            255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
            async def list_prompts(self) -> list[Prompt]:
    """Get all prompts from all providers.

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

    all_prompts: list[Prompt] = []

    # Get prompts from all MCP providers
    for provider in self.providers:
        if isinstance(provider, MCPManager):
            try:
                prompts = await provider.list_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 
            201
202
203
            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 = "dynamic",
    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, 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)

            		 'dynamic'
            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 
            278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
            def register_tool(
    self,
    tool: ToolType | Tool,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    enabled: bool = True,
    source: ToolSource = "dynamic",
    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, 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 Tool():
            tool.description = description_override or tool.description
            tool.name = name_override or tool.name
            tool.source = source
            tool.metadata = tool.metadata | (metadata or {})

        case _:
            tool = Tool.from_callable(
                tool,
                enabled=enabled,
                source=source,
                name_override=name_override,
                description_override=description_override,
                priority=priority,
                requires_confirmation=requires_confirmation,
                requires_capability=requires_capability,
                metadata=metadata or {},
            )

    # Register the tool
    self.register(tool.name, tool)
    return tool

            register_worker 

            register_worker(
    worker: MessageNode[Any, Any],
    *,
    name: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False,
    parent: Agent[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
            parent Agent[Any, Any] | None

            Optional parent agent for history/context sharing

            		 None
            Source code in src/llmling_agent/tools/manager.py 
            332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
            def register_worker(
    self,
    worker: MessageNode[Any, Any],
    *,
    name: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False,
    parent: Agent[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
        parent: Optional parent agent for history/context sharing
    """
    from llmling_agent import Agent, BaseTeam

    match worker:
        case BaseTeam():
            tool = worker.to_tool(name=name)
        case Agent():
            tool = worker.to_tool(
                parent=parent,
                name=name,
                reset_history_on_run=reset_history_on_run,
                pass_message_history=pass_message_history,
            )
        case _:
            msg = f"Unsupported worker type: {type(worker)}"
            raise ValueError(msg)
    msg = "Registering worker as tool"
    logger.debug(msg, worker_name=worker.name, tool_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 
            118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
            def remove_provider(
    self, provider: ResourceProvider | ResourceCallable | ProviderName
):
    """Remove a resource provider."""
    from llmling_agent.resource_providers 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 
            369
370
371
372
373
374
375
376
            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 
            143
144
145
146
            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 
            378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
            @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 = "dynamic",
    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

            		 'dynamic'
            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 
            434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
            def tool(
    self,
    name: str | None = None,
    *,
    description: str | None = None,
    enabled: bool = True,
    source: ToolSource = "dynamic",
    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