Skip to content

resource_providers

Class info

Classes

Name Children Inherits
AggregatingResourceProvider
llmling_agent.resource_providers.aggregating
Provider that combines resources from multiple providers.
MCPResourceProvider
llmling_agent.resource_providers.mcp_provider
Resource provider for a single MCP server.
    PlanProvider
    llmling_agent.resource_providers.plan_provider
    Provides plan-related tools for agent planning and task management.
      ResourceProvider
      llmling_agent.resource_providers.base
      Base class for resource providers.
      StaticResourceProvider
      llmling_agent.resource_providers.static
      Provider for pre-configured tools, prompts and resources.

        🛈 DocStrings

        Resource provider implementations.

        AggregatingResourceProvider

        Bases: ResourceProvider

        Provider that combines resources from multiple providers.

        Source code in src/llmling_agent/resource_providers/aggregating.py 
        18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
        class AggregatingResourceProvider(ResourceProvider):
    """Provider that combines resources from multiple providers."""

    def __init__(self, providers: list[ResourceProvider], name: str = "aggregating") -> None:
        """Initialize provider with list of providers to aggregate.

        Args:
            providers: Resource providers to aggregate (stores reference to list)
            name: Name for this provider
        """
        super().__init__(name=name)
        # Store reference to the providers list for dynamic updates
        self.providers = providers

    async def get_tools(self) -> list[Tool]:
        """Get tools from all providers."""
        return [t for provider in self.providers for t in await provider.get_tools()]

    async def get_prompts(self) -> list[BasePrompt]:
        """Get prompts from all providers."""
        return [p for provider in self.providers for p in await provider.get_prompts()]

    async def get_resources(self) -> list[ResourceInfo]:
        """Get resources from all providers."""
        return [r for provider in self.providers for r in await provider.get_resources()]

    async def get_request_parts(
        self, name: str, arguments: dict[str, str] | None = None
    ) -> list[ModelRequestPart]:
        """Try to get prompt from first provider that has it."""
        for provider in self.providers:
            try:
                return await provider.get_request_parts(name, arguments)
            except KeyError:
                continue
        msg = f"Prompt {name!r} not found in any provider"
        raise KeyError(msg)

        __init__ 

        __init__(providers: list[ResourceProvider], name: str = 'aggregating') -> None

        Initialize provider with list of providers to aggregate.

        Parameters:

        Name Type Description Default
        providers list[ResourceProvider]

        Resource providers to aggregate (stores reference to list)

        		 required
        name str

        Name for this provider

        		 'aggregating'
        Source code in src/llmling_agent/resource_providers/aggregating.py 
        21
22
23
24
25
26
27
28
29
30
        def __init__(self, providers: list[ResourceProvider], name: str = "aggregating") -> None:
    """Initialize provider with list of providers to aggregate.

    Args:
        providers: Resource providers to aggregate (stores reference to list)
        name: Name for this provider
    """
    super().__init__(name=name)
    # Store reference to the providers list for dynamic updates
    self.providers = providers

        get_prompts async 

        get_prompts() -> list[BasePrompt]

        Get prompts from all providers.

        Source code in src/llmling_agent/resource_providers/aggregating.py 
        36
37
38
        async def get_prompts(self) -> list[BasePrompt]:
    """Get prompts from all providers."""
    return [p for provider in self.providers for p in await provider.get_prompts()]

        get_request_parts async 

        get_request_parts(name: str, arguments: dict[str, str] | None = None) -> list[ModelRequestPart]

        Try to get prompt from first provider that has it.

        Source code in src/llmling_agent/resource_providers/aggregating.py 
        44
45
46
47
48
49
50
51
52
53
54
        async def get_request_parts(
    self, name: str, arguments: dict[str, str] | None = None
) -> list[ModelRequestPart]:
    """Try to get prompt from first provider that has it."""
    for provider in self.providers:
        try:
            return await provider.get_request_parts(name, arguments)
        except KeyError:
            continue
    msg = f"Prompt {name!r} not found in any provider"
    raise KeyError(msg)

        get_resources async 

        get_resources() -> list[ResourceInfo]

        Get resources from all providers.

        Source code in src/llmling_agent/resource_providers/aggregating.py 
        40
41
42
        async def get_resources(self) -> list[ResourceInfo]:
    """Get resources from all providers."""
    return [r for provider in self.providers for r in await provider.get_resources()]

        get_tools async 

        get_tools() -> list[Tool]

        Get tools from all providers.

        Source code in src/llmling_agent/resource_providers/aggregating.py 
        32
33
34
        async def get_tools(self) -> list[Tool]:
    """Get tools from all providers."""
    return [t for provider in self.providers for t in await provider.get_tools()]

        MCPResourceProvider

        Bases: ResourceProvider

        Resource provider for a single MCP server.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
         28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
        class MCPResourceProvider(ResourceProvider):
    """Resource provider for a single MCP server."""

    def __init__(
        self,
        server: MCPServerConfig | str,
        name: str = "mcp",
        owner: str | None = None,
        source: Literal["pool", "node"] = "node",
        sampling_callback: ClientSamplingHandler[Any] | None = None,
        accessible_roots: list[str] | None = None,
    ) -> None:
        from llmling_agent.mcp_server import MCPClient

        super().__init__(name, owner=owner)
        self.server = BaseMCPServerConfig.from_string(server) if isinstance(server, str) else server
        self.source = source
        self.exit_stack = AsyncExitStack()
        self._accessible_roots = accessible_roots
        self._sampling_callback = sampling_callback

        # Tool caching
        self._tools_cache: list[Tool] | None = None
        self._saved_enabled_states: dict[str, bool] = {}

        # Prompt caching
        self._prompts_cache: list[MCPClientPrompt] | None = None

        # Resource caching
        self._resources_cache: list[ResourceInfo] | None = None

        self.client = MCPClient(
            config=self.server,
            sampling_callback=self._sampling_callback,
            accessible_roots=self._accessible_roots,
            tool_change_callback=self._on_tools_changed,
            prompt_change_callback=self._on_prompts_changed,
            resource_change_callback=self._on_resources_changed,
        )

    def __repr__(self) -> str:
        return f"MCPResourceProvider({self.server!r}, source={self.source!r})"

    async def __aenter__(self) -> Self:
        try:
            await self.exit_stack.enter_async_context(self.client)
        except Exception as e:
            # Clean up in case of error
            await self.__aexit__(type(e), e, e.__traceback__)
            msg = "Failed to initialize MCP manager"
            raise RuntimeError(msg) from e

        return self

    async def __aexit__(self, *args: object) -> None:
        try:
            try:
                # Clean up exit stack (which includes MCP clients)
                await self.exit_stack.aclose()
            except RuntimeError as e:
                if "different task" in str(e):
                    # Handle task context mismatch
                    current_task = asyncio.current_task()
                    if current_task:
                        loop = asyncio.get_running_loop()
                        await loop.create_task(self.exit_stack.aclose())
                else:
                    raise

        except Exception as e:
            msg = "Error during MCP manager cleanup"
            logger.exception(msg, exc_info=e)
            raise RuntimeError(msg) from e

    async def _on_tools_changed(self) -> None:
        """Callback when tools change on the MCP server."""
        logger.info("MCP tool list changed, refreshing provider cache")
        self._saved_enabled_states = {t.name: t.enabled for t in self._tools_cache or []}
        self._tools_cache = None

    async def _on_prompts_changed(self) -> None:
        """Callback when prompts change on the MCP server."""
        logger.info("MCP prompt list changed, refreshing provider cache")
        self._prompts_cache = None

    async def _on_resources_changed(self) -> None:
        """Callback when resources change on the MCP server."""
        logger.info("MCP resource list changed, refreshing provider cache")
        self._resources_cache = None

    async def refresh_tools_cache(self) -> None:
        """Refresh the tools cache by fetching from client."""
        try:
            # Get fresh tools from client
            mcp_tools = await self.client.list_tools()
            all_tools: list[Tool] = []

            for tool in mcp_tools:
                try:
                    tool_info = self.client.convert_tool(tool)
                    all_tools.append(tool_info)
                except Exception:
                    logger.exception("Failed to create MCP tool", name=tool.name)
                    continue

            # Restore enabled states from saved states
            for tool_info in all_tools:
                if tool_info.name in self._saved_enabled_states:
                    tool_info.enabled = self._saved_enabled_states[tool_info.name]

            self._tools_cache = all_tools
            logger.debug("Refreshed MCP tools cache", num_tools=len(all_tools))
        except Exception:
            logger.exception("Failed to refresh MCP tools cache")
            self._tools_cache = []

    async def get_tools(self) -> list[Tool]:
        """Get cached tools, refreshing if necessary."""
        if self._tools_cache is None:
            await self.refresh_tools_cache()

        return self._tools_cache or []

    async def refresh_prompts_cache(self) -> None:
        """Refresh the prompts cache by fetching from client."""
        from llmling_agent.prompts.prompts import MCPClientPrompt

        try:
            result = await self.client.list_prompts()
            all_prompts: list[MCPClientPrompt] = []

            for prompt in result:
                try:
                    converted = MCPClientPrompt.from_fastmcp(self.client, prompt)
                    all_prompts.append(converted)
                except Exception:
                    logger.exception("Failed to convert prompt", name=prompt.name)
                    continue

            self._prompts_cache = all_prompts
            logger.debug("Refreshed MCP prompts cache", num_prompts=len(all_prompts))
        except Exception:
            logger.exception("Failed to refresh MCP prompts cache")
            self._prompts_cache = []

    async def get_prompts(self) -> list[MCPClientPrompt]:  # type: ignore
        """Get cached prompts, refreshing if necessary."""
        if self._prompts_cache is None:
            await self.refresh_prompts_cache()

        return self._prompts_cache or []

    async def refresh_resources_cache(self) -> None:
        """Refresh the resources cache by fetching from client."""
        try:
            result = await self.client.list_resources()
            all_resources: list[ResourceInfo] = []

            for resource in result:
                try:
                    converted = await ResourceInfo.from_mcp_resource(resource)
                    all_resources.append(converted)
                except Exception:
                    logger.exception("Failed to convert resource", name=resource.name)
                    continue

            self._resources_cache = all_resources
            logger.debug("Refreshed MCP resources cache", num_resources=len(all_resources))
        except Exception:
            logger.exception("Failed to refresh MCP resources cache")
            self._resources_cache = []

    async def get_resources(self) -> list[ResourceInfo]:
        """Get cached resources, refreshing if necessary."""
        if self._resources_cache is None:
            await self.refresh_resources_cache()

        return self._resources_cache or []

        get_prompts async 

        get_prompts() -> list[MCPClientPrompt]

        Get cached prompts, refreshing if necessary.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
        173
174
175
176
177
178
        async def get_prompts(self) -> list[MCPClientPrompt]:  # type: ignore
    """Get cached prompts, refreshing if necessary."""
    if self._prompts_cache is None:
        await self.refresh_prompts_cache()

    return self._prompts_cache or []

        get_resources async 

        get_resources() -> list[ResourceInfo]

        Get cached resources, refreshing if necessary.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
        200
201
202
203
204
205
        async def get_resources(self) -> list[ResourceInfo]:
    """Get cached resources, refreshing if necessary."""
    if self._resources_cache is None:
        await self.refresh_resources_cache()

    return self._resources_cache or []

        get_tools async 

        get_tools() -> list[Tool]

        Get cached tools, refreshing if necessary.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
        144
145
146
147
148
149
        async def get_tools(self) -> list[Tool]:
    """Get cached tools, refreshing if necessary."""
    if self._tools_cache is None:
        await self.refresh_tools_cache()

    return self._tools_cache or []

        refresh_prompts_cache async 

        refresh_prompts_cache() -> None

        Refresh the prompts cache by fetching from client.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
        151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
        async def refresh_prompts_cache(self) -> None:
    """Refresh the prompts cache by fetching from client."""
    from llmling_agent.prompts.prompts import MCPClientPrompt

    try:
        result = await self.client.list_prompts()
        all_prompts: list[MCPClientPrompt] = []

        for prompt in result:
            try:
                converted = MCPClientPrompt.from_fastmcp(self.client, prompt)
                all_prompts.append(converted)
            except Exception:
                logger.exception("Failed to convert prompt", name=prompt.name)
                continue

        self._prompts_cache = all_prompts
        logger.debug("Refreshed MCP prompts cache", num_prompts=len(all_prompts))
    except Exception:
        logger.exception("Failed to refresh MCP prompts cache")
        self._prompts_cache = []

        refresh_resources_cache async 

        refresh_resources_cache() -> None

        Refresh the resources cache by fetching from client.

        Source code in src/llmling_agent/resource_providers/mcp_provider.py 
        180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
        async def refresh_resources_cache(self) -> None:
    """Refresh the resources cache by fetching from client."""
    try:
        result = await self.client.list_resources()
        all_resources: list[ResourceInfo] = []

        for resource in result:
            try:
                converted = await ResourceInfo.from_mcp_resource(resource)
                all_resources.append(converted)
            except Exception:
                logger.exception("Failed to convert resource", name=resource.name)
                continue

        self._resources_cache = all_resources
        logger.debug("Refreshed MCP resources cache", num_resources=len(all_resources))
    except Exception:
        logger.exception("Failed to refresh MCP resources cache")
        self._resources_cache = []

        refresh_tools_cache async 

        refresh_tools_cache() -> None

        Refresh the tools cache by fetching from client.

        Source code in src/llmling_agent/resource_providers/mcp_provider.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
142
        async def refresh_tools_cache(self) -> None:
    """Refresh the tools cache by fetching from client."""
    try:
        # Get fresh tools from client
        mcp_tools = await self.client.list_tools()
        all_tools: list[Tool] = []

        for tool in mcp_tools:
            try:
                tool_info = self.client.convert_tool(tool)
                all_tools.append(tool_info)
            except Exception:
                logger.exception("Failed to create MCP tool", name=tool.name)
                continue

        # Restore enabled states from saved states
        for tool_info in all_tools:
            if tool_info.name in self._saved_enabled_states:
                tool_info.enabled = self._saved_enabled_states[tool_info.name]

        self._tools_cache = all_tools
        logger.debug("Refreshed MCP tools cache", num_tools=len(all_tools))
    except Exception:
        logger.exception("Failed to refresh MCP tools cache")
        self._tools_cache = []

        PlanProvider

        Bases: ResourceProvider

        Provides plan-related tools for agent planning and task management.

        This provider creates tools for managing agent plans and tasks, emitting domain events that can be handled by protocol adapters.

        Source code in src/llmling_agent/resource_providers/plan_provider.py 
         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
        class PlanProvider(ResourceProvider):
    """Provides plan-related tools for agent planning and task management.

    This provider creates tools for managing agent plans and tasks,
    emitting domain events that can be handled by protocol adapters.
    """

    def __init__(self) -> None:
        """Initialize plan provider."""
        super().__init__(name="plan")
        self._current_plan: list[PlanEntry] = []

    async def get_tools(self) -> list[Tool]:
        """Get plan management tools."""
        return [
            self.create_tool(self.get_plan, category="read"),
            self.create_tool(self.add_plan_entry, category="other"),
            self.create_tool(self.update_plan_entry, category="edit"),
            self.create_tool(self.remove_plan_entry, category="delete"),
        ]

    async def get_plan(self, agent_ctx: AgentContext) -> str:
        """Get the current plan formatted as markdown.

        Args:
            agent_ctx: Agent execution context

        Returns:
            Markdown-formatted plan with all entries and their status
        """
        if not self._current_plan:
            return "## Plan\n\n*No plan entries yet.*"

        lines = ["## Plan", ""]
        status_icons = {
            "pending": "⬚",
            "in_progress": "◐",
            "completed": "✓",
        }
        priority_labels = {
            "high": "🔴",
            "medium": "🟡",
            "low": "🟢",
        }
        for i, entry in enumerate(self._current_plan):
            icon = status_icons.get(entry.status, "?")
            priority = priority_labels.get(entry.priority, "")
            lines.append(f"{i}. {icon} {priority} {entry.content} *({entry.status})*")

        return "\n".join(lines)

    async def add_plan_entry(
        self,
        agent_ctx: AgentContext,
        content: str,
        priority: PlanEntryPriority = "medium",
        index: int | None = None,
    ) -> str:
        """Add a new plan entry.

        Args:
            agent_ctx: Agent execution context
            content: Description of what this task aims to accomplish
            priority: Relative importance (high/medium/low)
            index: Optional position to insert at (default: append to end)

        Returns:
            Success message indicating entry was added
        """
        entry = PlanEntry(content=content, priority=priority, status="pending")
        if index is None:
            self._current_plan.append(entry)
            entry_index = len(self._current_plan) - 1
        else:
            if index < 0 or index > len(self._current_plan):
                return f"Error: Index {index} out of range (0-{len(self._current_plan)})"
            self._current_plan.insert(index, entry)
            entry_index = index

        await self._emit_plan_update(agent_ctx)

        return f"Added plan entry at index {entry_index}: {content!r} (priority={priority!r})"

    async def update_plan_entry(
        self,
        agent_ctx: AgentContext,
        index: int,
        content: str | None = None,
        status: PlanEntryStatus | None = None,
        priority: PlanEntryPriority | None = None,
    ) -> str:
        """Update an existing plan entry.

        Args:
            agent_ctx: Agent execution context
            index: Position of entry to update (0-based)
            content: New task description
            status: New execution status
            priority: New priority level

        Returns:
            Success message indicating what was updated
        """
        if index < 0 or index >= len(self._current_plan):
            return f"Error: Index {index} out of range (0-{len(self._current_plan) - 1})"

        entry = self._current_plan[index]
        updates = []

        if content is not None:
            entry.content = content
            updates.append(f"content to {content!r}")

        if status is not None:
            entry.status = status
            updates.append(f"status to {status!r}")

        if priority is not None:
            entry.priority = priority
            updates.append(f"priority to {priority!r}")

        if not updates:
            return "No changes specified"

        await self._emit_plan_update(agent_ctx)
        return f"Updated entry {index}: {', '.join(updates)}"

    async def remove_plan_entry(self, agent_ctx: AgentContext, index: int) -> str:
        """Remove a plan entry.

        Args:
            agent_ctx: Agent execution context
            index: Position of entry to remove (0-based)

        Returns:
            Success message indicating entry was removed
        """
        if index < 0 or index >= len(self._current_plan):
            return f"Error: Index {index} out of range (0-{len(self._current_plan) - 1})"
        removed_entry = self._current_plan.pop(index)
        await self._emit_plan_update(agent_ctx)
        if self._current_plan:
            return f"Removed entry {index}: {removed_entry.content!r}, remaining entries reindexed"
        return f"Removed entry {index}: {removed_entry.content!r}, plan is now empty"

    async def _emit_plan_update(self, agent_ctx: AgentContext) -> None:
        """Emit plan update event."""
        await agent_ctx.events.plan_updated(self._current_plan)

        __init__ 

        __init__() -> None

        Initialize plan provider.

        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        56
57
58
59
        def __init__(self) -> None:
    """Initialize plan provider."""
    super().__init__(name="plan")
    self._current_plan: list[PlanEntry] = []

        add_plan_entry async 

        add_plan_entry(
    agent_ctx: AgentContext,
    content: str,
    priority: PlanEntryPriority = "medium",
    index: int | None = None,
) -> str

        Add a new plan entry.

        Parameters:

        Name Type Description Default
        agent_ctx AgentContext

        Agent execution context

        		 required
        content str

        Description of what this task aims to accomplish

        		 required
        priority PlanEntryPriority

        Relative importance (high/medium/low)

        		 'medium'
        index int | None

        Optional position to insert at (default: append to end)

        		 None

        Returns:

        Type Description
        str

        Success message indicating entry was added
        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        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
        async def add_plan_entry(
    self,
    agent_ctx: AgentContext,
    content: str,
    priority: PlanEntryPriority = "medium",
    index: int | None = None,
) -> str:
    """Add a new plan entry.

    Args:
        agent_ctx: Agent execution context
        content: Description of what this task aims to accomplish
        priority: Relative importance (high/medium/low)
        index: Optional position to insert at (default: append to end)

    Returns:
        Success message indicating entry was added
    """
    entry = PlanEntry(content=content, priority=priority, status="pending")
    if index is None:
        self._current_plan.append(entry)
        entry_index = len(self._current_plan) - 1
    else:
        if index < 0 or index > len(self._current_plan):
            return f"Error: Index {index} out of range (0-{len(self._current_plan)})"
        self._current_plan.insert(index, entry)
        entry_index = index

    await self._emit_plan_update(agent_ctx)

    return f"Added plan entry at index {entry_index}: {content!r} (priority={priority!r})"

        get_plan async 

        get_plan(agent_ctx: AgentContext) -> str

        Get the current plan formatted as markdown.

        Parameters:

        Name Type Description Default
        agent_ctx AgentContext

        Agent execution context

        		 required

        Returns:

        Type Description
        str

        Markdown-formatted plan with all entries and their status
        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        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
        async def get_plan(self, agent_ctx: AgentContext) -> str:
    """Get the current plan formatted as markdown.

    Args:
        agent_ctx: Agent execution context

    Returns:
        Markdown-formatted plan with all entries and their status
    """
    if not self._current_plan:
        return "## Plan\n\n*No plan entries yet.*"

    lines = ["## Plan", ""]
    status_icons = {
        "pending": "⬚",
        "in_progress": "◐",
        "completed": "✓",
    }
    priority_labels = {
        "high": "🔴",
        "medium": "🟡",
        "low": "🟢",
    }
    for i, entry in enumerate(self._current_plan):
        icon = status_icons.get(entry.status, "?")
        priority = priority_labels.get(entry.priority, "")
        lines.append(f"{i}. {icon} {priority} {entry.content} *({entry.status})*")

    return "\n".join(lines)

        get_tools async 

        get_tools() -> list[Tool]

        Get plan management tools.

        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        61
62
63
64
65
66
67
68
        async def get_tools(self) -> list[Tool]:
    """Get plan management tools."""
    return [
        self.create_tool(self.get_plan, category="read"),
        self.create_tool(self.add_plan_entry, category="other"),
        self.create_tool(self.update_plan_entry, category="edit"),
        self.create_tool(self.remove_plan_entry, category="delete"),
    ]

        remove_plan_entry async 

        remove_plan_entry(agent_ctx: AgentContext, index: int) -> str

        Remove a plan entry.

        Parameters:

        Name Type Description Default
        agent_ctx AgentContext

        Agent execution context

        		 required
        index int

        Position of entry to remove (0-based)

        		 required

        Returns:

        Type Description
        str

        Success message indicating entry was removed
        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
        async def remove_plan_entry(self, agent_ctx: AgentContext, index: int) -> str:
    """Remove a plan entry.

    Args:
        agent_ctx: Agent execution context
        index: Position of entry to remove (0-based)

    Returns:
        Success message indicating entry was removed
    """
    if index < 0 or index >= len(self._current_plan):
        return f"Error: Index {index} out of range (0-{len(self._current_plan) - 1})"
    removed_entry = self._current_plan.pop(index)
    await self._emit_plan_update(agent_ctx)
    if self._current_plan:
        return f"Removed entry {index}: {removed_entry.content!r}, remaining entries reindexed"
    return f"Removed entry {index}: {removed_entry.content!r}, plan is now empty"

        update_plan_entry async 

        update_plan_entry(
    agent_ctx: AgentContext,
    index: int,
    content: str | None = None,
    status: PlanEntryStatus | None = None,
    priority: PlanEntryPriority | None = None,
) -> str

        Update an existing plan entry.

        Parameters:

        Name Type Description Default
        agent_ctx AgentContext

        Agent execution context

        		 required
        index int

        Position of entry to update (0-based)

        		 required
        content str | None

        New task description

        		 None
        status PlanEntryStatus | None

        New execution status

        		 None
        priority PlanEntryPriority | None

        New priority level

        		 None

        Returns:

        Type Description
        str

        Success message indicating what was updated
        Source code in src/llmling_agent/resource_providers/plan_provider.py 
        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
        async def update_plan_entry(
    self,
    agent_ctx: AgentContext,
    index: int,
    content: str | None = None,
    status: PlanEntryStatus | None = None,
    priority: PlanEntryPriority | None = None,
) -> str:
    """Update an existing plan entry.

    Args:
        agent_ctx: Agent execution context
        index: Position of entry to update (0-based)
        content: New task description
        status: New execution status
        priority: New priority level

    Returns:
        Success message indicating what was updated
    """
    if index < 0 or index >= len(self._current_plan):
        return f"Error: Index {index} out of range (0-{len(self._current_plan) - 1})"

    entry = self._current_plan[index]
    updates = []

    if content is not None:
        entry.content = content
        updates.append(f"content to {content!r}")

    if status is not None:
        entry.status = status
        updates.append(f"status to {status!r}")

    if priority is not None:
        entry.priority = priority
        updates.append(f"priority to {priority!r}")

    if not updates:
        return "No changes specified"

    await self._emit_plan_update(agent_ctx)
    return f"Updated entry {index}: {', '.join(updates)}"

        ResourceProvider

        Base class for resource providers.

        Provides tools, prompts, and other resources to agents. Default implementations return empty lists - override as needed.

        Source code in src/llmling_agent/resource_providers/base.py 
         28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
        class ResourceProvider:
    """Base class for resource providers.

    Provides tools, prompts, and other resources to agents.
    Default implementations return empty lists - override as needed.
    """

    def __init__(self, name: str, owner: str | None = None) -> None:
        """Initialize the resource provider."""
        self.name = name
        self.owner = owner
        self.log = logger.bind(name=self.name, owner=self.owner)

    async def __aenter__(self) -> Self:
        """Async context entry if required."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Async context cleanup if required."""

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name!r})"

    async def get_tools(self) -> list[Tool]:
        """Get available tools. Override to provide tools."""
        return []

    async def get_tool(self, tool_name: str) -> Tool:
        """Get specific tool."""
        tools = await self.get_tools()
        for tool in tools:
            if tool.name == tool_name:
                return tool
        msg = f"Tool {tool_name!r} not found"
        raise ValueError(msg)

    async def get_prompts(self) -> list[BasePrompt]:
        """Get available prompts. Override to provide prompts."""
        return []

    async def get_resources(self) -> list[ResourceInfo]:
        """Get available resources. Override to provide resources."""
        return []

    async def get_skills(self) -> list[Skill]:
        """Get available skills. Override to provide skills."""
        return []

    async def get_skill_instructions(self, skill_name: str) -> str:
        """Get full instructions for a specific skill.

        Args:
            skill_name: Name of the skill to get instructions for

        Returns:
            The full skill instructions for execution

        Raises:
            KeyError: If skill not found
        """
        msg = f"Skill {skill_name!r} not found"
        raise KeyError(msg)

    async def get_request_parts(
        self, name: str, arguments: dict[str, str] | None = None
    ) -> list[ModelRequestPart]:
        """Get a prompt formatted with arguments.

        Args:
            name: Name of the prompt to format
            arguments: Optional arguments for prompt formatting

        Returns:
            Single chat message with merged content

        Raises:
            KeyError: If prompt not found
            ValueError: If formatting fails
        """
        prompts = await self.get_prompts()
        prompt = next((p for p in prompts if p.name == name), None)
        if not prompt:
            msg = f"Prompt {name!r} not found"
            raise KeyError(msg)

        messages = await prompt.format(arguments or {})
        if not messages:
            msg = f"Prompt {name!r} produced no messages"
            raise ValueError(msg)

        return [p for prompt_msg in messages for p in prompt_msg.to_pydantic_parts()]

    def create_tool(
        self,
        fn: Callable[..., Any],
        read_only: bool | None = None,
        destructive: bool | None = None,
        idempotent: bool | None = None,
        open_world: bool | None = None,
        requires_confirmation: bool = False,
        metadata: dict[str, Any] | None = None,
        category: ToolKind | None = None,
        name_override: str | None = None,
        description_override: str | None = None,
        schema_override: OpenAIFunctionDefinition | None = None,
    ) -> Tool:
        """Create a tool from a function.

        Args:
            fn: Function to create a tool from
            read_only: Whether the tool is read-only
            destructive: Whether the tool is destructive
            idempotent: Whether the tool is idempotent
            open_world: Whether the tool is open-world
            requires_confirmation: Whether the tool requires confirmation
            metadata: Metadata for the tool
            category: Category of the tool
            name_override: Override the name of the tool
            description_override: Override the description of the tool
            schema_override: Override the schema of the tool

        Returns:
            Tool created from the function
        """
        return Tool.from_callable(
            fn=fn,
            category=category,
            source=self.name,
            requires_confirmation=requires_confirmation,
            metadata=metadata,
            name_override=name_override,
            description_override=description_override,
            schema_override=schema_override,
            hints=ToolHints(
                read_only=read_only,
                destructive=destructive,
                idempotent=idempotent,
                open_world=open_world,
            ),
        )

        __aenter__ async 

        __aenter__() -> Self

        Async context entry if required.

        Source code in src/llmling_agent/resource_providers/base.py 
        41
42
43
        async def __aenter__(self) -> Self:
    """Async context entry if required."""
    return self

        __aexit__ async 

        __aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

        Async context cleanup if required.

        Source code in src/llmling_agent/resource_providers/base.py 
        45
46
47
48
49
50
51
        async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context cleanup if required."""

        __init__ 

        __init__(name: str, owner: str | None = None) -> None

        Initialize the resource provider.

        Source code in src/llmling_agent/resource_providers/base.py 
        35
36
37
38
39
        def __init__(self, name: str, owner: str | None = None) -> None:
    """Initialize the resource provider."""
    self.name = name
    self.owner = owner
    self.log = logger.bind(name=self.name, owner=self.owner)

        create_tool 

        create_tool(
    fn: Callable[..., Any],
    read_only: bool | None = None,
    destructive: bool | None = None,
    idempotent: bool | None = None,
    open_world: bool | None = None,
    requires_confirmation: bool = False,
    metadata: dict[str, Any] | None = None,
    category: ToolKind | None = None,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: OpenAIFunctionDefinition | None = None,
) -> Tool

        Create a tool from a function.

        Parameters:

        Name Type Description Default
        fn Callable[..., Any]

        Function to create a tool from

        		 required
        read_only bool | None

        Whether the tool is read-only

        		 None
        destructive bool | None

        Whether the tool is destructive

        		 None
        idempotent bool | None

        Whether the tool is idempotent

        		 None
        open_world bool | None

        Whether the tool is open-world

        		 None
        requires_confirmation bool

        Whether the tool requires confirmation

        		 False
        metadata dict[str, Any] | None

        Metadata for the tool

        		 None
        category ToolKind | None

        Category of the tool

        		 None
        name_override str | None

        Override the name of the tool

        		 None
        description_override str | None

        Override the description of the tool

        		 None
        schema_override OpenAIFunctionDefinition | None

        Override the schema of the tool

        		 None

        Returns:

        Type Description
        Tool

        Tool created from the function
        Source code in src/llmling_agent/resource_providers/base.py 
        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
        def create_tool(
    self,
    fn: Callable[..., Any],
    read_only: bool | None = None,
    destructive: bool | None = None,
    idempotent: bool | None = None,
    open_world: bool | None = None,
    requires_confirmation: bool = False,
    metadata: dict[str, Any] | None = None,
    category: ToolKind | None = None,
    name_override: str | None = None,
    description_override: str | None = None,
    schema_override: OpenAIFunctionDefinition | None = None,
) -> Tool:
    """Create a tool from a function.

    Args:
        fn: Function to create a tool from
        read_only: Whether the tool is read-only
        destructive: Whether the tool is destructive
        idempotent: Whether the tool is idempotent
        open_world: Whether the tool is open-world
        requires_confirmation: Whether the tool requires confirmation
        metadata: Metadata for the tool
        category: Category of the tool
        name_override: Override the name of the tool
        description_override: Override the description of the tool
        schema_override: Override the schema of the tool

    Returns:
        Tool created from the function
    """
    return Tool.from_callable(
        fn=fn,
        category=category,
        source=self.name,
        requires_confirmation=requires_confirmation,
        metadata=metadata,
        name_override=name_override,
        description_override=description_override,
        schema_override=schema_override,
        hints=ToolHints(
            read_only=read_only,
            destructive=destructive,
            idempotent=idempotent,
            open_world=open_world,
        ),
    )

        get_prompts async 

        get_prompts() -> list[BasePrompt]

        Get available prompts. Override to provide prompts.

        Source code in src/llmling_agent/resource_providers/base.py 
        69
70
71
        async def get_prompts(self) -> list[BasePrompt]:
    """Get available prompts. Override to provide prompts."""
    return []

        get_request_parts async 

        get_request_parts(name: str, arguments: dict[str, str] | None = None) -> list[ModelRequestPart]

        Get a prompt formatted with arguments.

        Parameters:

        Name Type Description Default
        name str

        Name of the prompt to format

        		 required
        arguments dict[str, str] | None

        Optional arguments for prompt formatting

        		 None

        Returns:

        Type Description
        list[ModelRequestPart]

        Single chat message with merged content

        Raises:

        Type Description
        KeyError

        If prompt not found
        ValueError

        If formatting fails
        Source code in src/llmling_agent/resource_providers/base.py 
         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
        async def get_request_parts(
    self, name: str, arguments: dict[str, str] | None = None
) -> list[ModelRequestPart]:
    """Get a prompt formatted with arguments.

    Args:
        name: Name of the prompt to format
        arguments: Optional arguments for prompt formatting

    Returns:
        Single chat message with merged content

    Raises:
        KeyError: If prompt not found
        ValueError: If formatting fails
    """
    prompts = await self.get_prompts()
    prompt = next((p for p in prompts if p.name == name), None)
    if not prompt:
        msg = f"Prompt {name!r} not found"
        raise KeyError(msg)

    messages = await prompt.format(arguments or {})
    if not messages:
        msg = f"Prompt {name!r} produced no messages"
        raise ValueError(msg)

    return [p for prompt_msg in messages for p in prompt_msg.to_pydantic_parts()]

        get_resources async 

        get_resources() -> list[ResourceInfo]

        Get available resources. Override to provide resources.

        Source code in src/llmling_agent/resource_providers/base.py 
        73
74
75
        async def get_resources(self) -> list[ResourceInfo]:
    """Get available resources. Override to provide resources."""
    return []

        get_skill_instructions async 

        get_skill_instructions(skill_name: str) -> str

        Get full instructions for a specific skill.

        Parameters:

        Name Type Description Default
        skill_name str

        Name of the skill to get instructions for

        		 required

        Returns:

        Type Description
        str

        The full skill instructions for execution

        Raises:

        Type Description
        KeyError

        If skill not found
        Source code in src/llmling_agent/resource_providers/base.py 
        81
82
83
84
85
86
87
88
89
90
91
92
93
94
        async def get_skill_instructions(self, skill_name: str) -> str:
    """Get full instructions for a specific skill.

    Args:
        skill_name: Name of the skill to get instructions for

    Returns:
        The full skill instructions for execution

    Raises:
        KeyError: If skill not found
    """
    msg = f"Skill {skill_name!r} not found"
    raise KeyError(msg)

        get_skills async 

        get_skills() -> list[Skill]

        Get available skills. Override to provide skills.

        Source code in src/llmling_agent/resource_providers/base.py 
        77
78
79
        async def get_skills(self) -> list[Skill]:
    """Get available skills. Override to provide skills."""
    return []

        get_tool async 

        get_tool(tool_name: str) -> Tool

        Get specific tool.

        Source code in src/llmling_agent/resource_providers/base.py 
        60
61
62
63
64
65
66
67
        async def get_tool(self, tool_name: str) -> Tool:
    """Get specific tool."""
    tools = await self.get_tools()
    for tool in tools:
        if tool.name == tool_name:
            return tool
    msg = f"Tool {tool_name!r} not found"
    raise ValueError(msg)

        get_tools async 

        get_tools() -> list[Tool]

        Get available tools. Override to provide tools.

        Source code in src/llmling_agent/resource_providers/base.py 
        56
57
58
        async def get_tools(self) -> list[Tool]:
    """Get available tools. Override to provide tools."""
    return []

        StaticResourceProvider

        Bases: ResourceProvider

        Provider for pre-configured tools, prompts and resources.

        Allows creating a provider that serves a fixed set of resources passed during initialization. Useful for converting static configurations to the common ResourceProvider interface.

        Source code in src/llmling_agent/resource_providers/static.py 
         21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
        class StaticResourceProvider(ResourceProvider):
    """Provider for pre-configured tools, prompts and resources.

    Allows creating a provider that serves a fixed set of resources
    passed during initialization. Useful for converting static configurations
    to the common ResourceProvider interface.
    """

    def __init__(
        self,
        name: str = "static",
        tools: Sequence[Tool] | None = None,
        prompts: Sequence[BasePrompt] | None = None,
        resources: Sequence[ResourceInfo] | None = None,
    ) -> None:
        """Initialize provider with static resources.

        Args:
            name: Name of the provider
            tools: Optional list of tools to serve
            prompts: Optional list of prompts to serve
            resources: Optional list of resources to serve
        """
        super().__init__(name=name)
        self._tools = list(tools) if tools else []
        self._prompts = list(prompts) if prompts else []
        self._resources = list(resources) if resources else []

    async def get_tools(self) -> list[Tool]:
        """Get pre-configured tools."""
        return self._tools

    async def get_prompts(self) -> list[BasePrompt]:
        """Get pre-configured prompts."""
        return self._prompts

    async def get_resources(self) -> list[ResourceInfo]:
        """Get pre-configured resources."""
        return self._resources

    def add_tool(self, tool: Tool) -> None:
        """Add a tool to this provider.

        Args:
            tool: Tool to add
        """
        self._tools.append(tool)

    def remove_tool(self, name: str) -> bool:
        """Remove a tool by name.

        Args:
            name: Name of tool to remove

        Returns:
            True if tool was found and removed, False otherwise
        """
        for i, tool in enumerate(self._tools):
            if tool.name == name:
                self._tools.pop(i)
                return True
        return False

    def add_prompt(self, prompt: BasePrompt) -> None:
        """Add a prompt to this provider.

        Args:
            prompt: Prompt to add
        """
        self._prompts.append(prompt)

    def remove_prompt(self, name: str) -> bool:
        """Remove a prompt by name.

        Args:
            name: Name of prompt to remove

        Returns:
            True if prompt was found and removed, False otherwise
        """
        for i, prompt in enumerate(self._prompts):
            if prompt.name == name:
                self._prompts.pop(i)
                return True
        return False

    def add_resource(self, resource: ResourceInfo) -> None:
        """Add a resource to this provider.

        Args:
            resource: Resource to add
        """
        self._resources.append(resource)

    def remove_resource(self, name: str) -> bool:
        """Remove a resource by name.

        Args:
            name: Name of resource to remove

        Returns:
            True if resource was found and removed, False otherwise
        """
        for i, resource in enumerate(self._resources):
            if resource.name == name:
                self._resources.pop(i)
                return True
        return False

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

        Args:
            tool: Tool to register (callable, Tool instance, or import path)
            name_override: Optional name override for the tool
            description_override: Optional description override for the tool
            enabled: Whether tool is initially enabled
            source: Tool source (runtime/agent/builtin/dynamic)
            requires_confirmation: Whether tool needs confirmation
            metadata: Additional tool metadata

        Returns:
            Created Tool instance
        """
        from llmling_agent.tools.base import Tool as ToolClass

        match tool:
            case ToolClass():
                tool.description = description_override or tool.description
                tool.name = name_override or tool.name
                tool.source = source
                tool.metadata = tool.metadata | (metadata or {})
                tool.enabled = enabled

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

        self.add_tool(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
        from llmling_agent.agent.acp_agent import ACPAgent

        match worker:
            case BaseTeam():
                tool = worker.to_tool(name=name)
            case ACPAgent():
                tool = worker.to_tool(name=name, description=worker.description)
            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)

        self.add_tool(tool)
        return tool

    @overload
    def tool(self, func: Callable[..., Any]) -> Callable[..., Any]: ...

    @overload
    def tool(
        self,
        *,
        name: str | None = None,
        description: str | None = None,
        enabled: bool = True,
        requires_confirmation: bool = False,
        metadata: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]: ...

    def tool(
        self,
        func: Callable[..., Any] | None = None,
        *,
        name: str | None = None,
        description: str | None = None,
        enabled: bool = True,
        requires_confirmation: bool = False,
        metadata: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> Callable[..., Any] | Callable[[Callable[..., Any]], Callable[..., Any]]:
        """Decorator to register a function as a tool.

        Can be used with or without parameters:

        ```python
        # Without parameters
        @provider.tool
        def my_function(x: int) -> str:
            return str(x)

        # With parameters
        @provider.tool(name="custom_name", description="Custom description")
        def another_function(y: str) -> str:
            return y.upper()
        ```

        Args:
            func: Function to register (when used without parentheses)
            name: Override for tool name
            description: Override for tool description
            enabled: Whether tool is initially enabled
            requires_confirmation: Whether execution needs confirmation
            metadata: Additional tool metadata
            **kwargs: Additional arguments passed to Tool.from_callable
        """
        from llmling_agent.tools.base import Tool

        def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
            tool = Tool.from_callable(
                f,
                name_override=name,
                description_override=description,
                enabled=enabled,
                requires_confirmation=requires_confirmation,
                metadata=metadata or {},
                **kwargs,
            )
            self.add_tool(tool)
            return f

        if func is None:
            # Called with arguments: @provider.tool(...)
            return decorator
        # Called without arguments: @provider.tool
        return decorator(func)

        __init__ 

        __init__(
    name: str = "static",
    tools: Sequence[Tool] | None = None,
    prompts: Sequence[BasePrompt] | None = None,
    resources: Sequence[ResourceInfo] | None = None,
) -> None

        Initialize provider with static resources.

        Parameters:

        Name Type Description Default
        name str

        Name of the provider

        		 'static'
        tools Sequence[Tool] | None

        Optional list of tools to serve

        		 None
        prompts Sequence[BasePrompt] | None

        Optional list of prompts to serve

        		 None
        resources Sequence[ResourceInfo] | None

        Optional list of resources to serve

        		 None
        Source code in src/llmling_agent/resource_providers/static.py 
        29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
        def __init__(
    self,
    name: str = "static",
    tools: Sequence[Tool] | None = None,
    prompts: Sequence[BasePrompt] | None = None,
    resources: Sequence[ResourceInfo] | None = None,
) -> None:
    """Initialize provider with static resources.

    Args:
        name: Name of the provider
        tools: Optional list of tools to serve
        prompts: Optional list of prompts to serve
        resources: Optional list of resources to serve
    """
    super().__init__(name=name)
    self._tools = list(tools) if tools else []
    self._prompts = list(prompts) if prompts else []
    self._resources = list(resources) if resources else []

        add_prompt 

        add_prompt(prompt: BasePrompt) -> None

        Add a prompt to this provider.

        Parameters:

        Name Type Description Default
        prompt BasePrompt

        Prompt to add

        		 required
        Source code in src/llmling_agent/resource_providers/static.py 
        84
85
86
87
88
89
90
        def add_prompt(self, prompt: BasePrompt) -> None:
    """Add a prompt to this provider.

    Args:
        prompt: Prompt to add
    """
    self._prompts.append(prompt)

        add_resource 

        add_resource(resource: ResourceInfo) -> None

        Add a resource to this provider.

        Parameters:

        Name Type Description Default
        resource ResourceInfo

        Resource to add

        		 required
        Source code in src/llmling_agent/resource_providers/static.py 
        107
108
109
110
111
112
113
        def add_resource(self, resource: ResourceInfo) -> None:
    """Add a resource to this provider.

    Args:
        resource: Resource to add
    """
    self._resources.append(resource)

        add_tool 

        add_tool(tool: Tool) -> None

        Add a tool to this provider.

        Parameters:

        Name Type Description Default
        tool Tool

        Tool to add

        		 required
        Source code in src/llmling_agent/resource_providers/static.py 
        61
62
63
64
65
66
67
        def add_tool(self, tool: Tool) -> None:
    """Add a tool to this provider.

    Args:
        tool: Tool to add
    """
    self._tools.append(tool)

        get_prompts async 

        get_prompts() -> list[BasePrompt]

        Get pre-configured prompts.

        Source code in src/llmling_agent/resource_providers/static.py 
        53
54
55
        async def get_prompts(self) -> list[BasePrompt]:
    """Get pre-configured prompts."""
    return self._prompts

        get_resources async 

        get_resources() -> list[ResourceInfo]

        Get pre-configured resources.

        Source code in src/llmling_agent/resource_providers/static.py 
        57
58
59
        async def get_resources(self) -> list[ResourceInfo]:
    """Get pre-configured resources."""
    return self._resources

        get_tools async 

        get_tools() -> list[Tool]

        Get pre-configured tools.

        Source code in src/llmling_agent/resource_providers/static.py 
        49
50
51
        async def get_tools(self) -> list[Tool]:
    """Get pre-configured tools."""
    return self._tools

        register_tool 

        register_tool(
    tool: ToolType | Tool,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    enabled: bool = True,
    source: ToolSource = "dynamic",
    requires_confirmation: bool = False,
    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, Tool instance, or import path)

        		 required
        name_override str | None

        Optional name override for the tool

        		 None
        description_override str | None

        Optional description override for the tool

        		 None
        enabled bool

        Whether tool is initially enabled

        		 True
        source ToolSource

        Tool source (runtime/agent/builtin/dynamic)

        		 'dynamic'
        requires_confirmation bool

        Whether tool needs confirmation

        		 False
        metadata dict[str, str] | None

        Additional tool metadata

        		 None

        Returns:

        Type Description
        Tool

        Created Tool instance
        Source code in src/llmling_agent/resource_providers/static.py 
        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
        def register_tool(
    self,
    tool: ToolType | Tool,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    enabled: bool = True,
    source: ToolSource = "dynamic",
    requires_confirmation: bool = False,
    metadata: dict[str, str] | None = None,
) -> Tool:
    """Register a new tool with custom settings.

    Args:
        tool: Tool to register (callable, Tool instance, or import path)
        name_override: Optional name override for the tool
        description_override: Optional description override for the tool
        enabled: Whether tool is initially enabled
        source: Tool source (runtime/agent/builtin/dynamic)
        requires_confirmation: Whether tool needs confirmation
        metadata: Additional tool metadata

    Returns:
        Created Tool instance
    """
    from llmling_agent.tools.base import Tool as ToolClass

    match tool:
        case ToolClass():
            tool.description = description_override or tool.description
            tool.name = name_override or tool.name
            tool.source = source
            tool.metadata = tool.metadata | (metadata or {})
            tool.enabled = enabled

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

    self.add_tool(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/resource_providers/static.py 
        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
        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
    from llmling_agent.agent.acp_agent import ACPAgent

    match worker:
        case BaseTeam():
            tool = worker.to_tool(name=name)
        case ACPAgent():
            tool = worker.to_tool(name=name, description=worker.description)
        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)

    self.add_tool(tool)
    return tool

        remove_prompt 

        remove_prompt(name: str) -> bool

        Remove a prompt by name.

        Parameters:

        Name Type Description Default
        name str

        Name of prompt to remove

        		 required

        Returns:

        Type Description
        bool

        True if prompt was found and removed, False otherwise
        Source code in src/llmling_agent/resource_providers/static.py 
         92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
        def remove_prompt(self, name: str) -> bool:
    """Remove a prompt by name.

    Args:
        name: Name of prompt to remove

    Returns:
        True if prompt was found and removed, False otherwise
    """
    for i, prompt in enumerate(self._prompts):
        if prompt.name == name:
            self._prompts.pop(i)
            return True
    return False

        remove_resource 

        remove_resource(name: str) -> bool

        Remove a resource by name.

        Parameters:

        Name Type Description Default
        name str

        Name of resource to remove

        		 required

        Returns:

        Type Description
        bool

        True if resource was found and removed, False otherwise
        Source code in src/llmling_agent/resource_providers/static.py 
        115
116
117
118
119
120
121
122
123
124
125
126
127
128
        def remove_resource(self, name: str) -> bool:
    """Remove a resource by name.

    Args:
        name: Name of resource to remove

    Returns:
        True if resource was found and removed, False otherwise
    """
    for i, resource in enumerate(self._resources):
        if resource.name == name:
            self._resources.pop(i)
            return True
    return False

        remove_tool 

        remove_tool(name: str) -> bool

        Remove a tool by name.

        Parameters:

        Name Type Description Default
        name str

        Name of tool to remove

        		 required

        Returns:

        Type Description
        bool

        True if tool was found and removed, False otherwise
        Source code in src/llmling_agent/resource_providers/static.py 
        69
70
71
72
73
74
75
76
77
78
79
80
81
82
        def remove_tool(self, name: str) -> bool:
    """Remove a tool by name.

    Args:
        name: Name of tool to remove

    Returns:
        True if tool was found and removed, False otherwise
    """
    for i, tool in enumerate(self._tools):
        if tool.name == name:
            self._tools.pop(i)
            return True
    return False

        tool 

        tool(func: Callable[..., Any]) -> Callable[..., Any]

        tool(
    *,
    name: str | None = None,
    description: str | None = None,
    enabled: bool = True,
    requires_confirmation: bool = False,
    metadata: dict[str, str] | None = None,
    **kwargs: Any
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

        tool(
    func: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    enabled: bool = True,
    requires_confirmation: bool = False,
    metadata: dict[str, str] | None = None,
    **kwargs: Any
) -> Callable[..., Any] | Callable[[Callable[..., Any]], Callable[..., Any]]

        Decorator to register a function as a tool.

        Can be used with or without parameters:

        # Without parameters
@provider.tool
def my_function(x: int) -> str:
    return str(x)

# With parameters
@provider.tool(name="custom_name", description="Custom description")
def another_function(y: str) -> str:
    return y.upper()

        Parameters:

        Name Type Description Default
        func Callable[..., Any] | None

        Function to register (when used without parentheses)

        		 None
        name str | None

        Override for tool name

        		 None
        description str | None

        Override for tool description

        		 None
        enabled bool

        Whether tool is initially enabled

        		 True
        requires_confirmation bool

        Whether execution needs confirmation

        		 False
        metadata dict[str, str] | None

        Additional tool metadata

        		 None
        **kwargs Any

        Additional arguments passed to Tool.from_callable

        		 {}
        Source code in src/llmling_agent/resource_providers/static.py 
        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
        def tool(
    self,
    func: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    enabled: bool = True,
    requires_confirmation: bool = False,
    metadata: dict[str, str] | None = None,
    **kwargs: Any,
) -> Callable[..., Any] | Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to register a function as a tool.

    Can be used with or without parameters:

    ```python
    # Without parameters
    @provider.tool
    def my_function(x: int) -> str:
        return str(x)

    # With parameters
    @provider.tool(name="custom_name", description="Custom description")
    def another_function(y: str) -> str:
        return y.upper()
    ```

    Args:
        func: Function to register (when used without parentheses)
        name: Override for tool name
        description: Override for tool description
        enabled: Whether tool is initially enabled
        requires_confirmation: Whether execution needs confirmation
        metadata: Additional tool metadata
        **kwargs: Additional arguments passed to Tool.from_callable
    """
    from llmling_agent.tools.base import Tool

    def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
        tool = Tool.from_callable(
            f,
            name_override=name,
            description_override=description,
            enabled=enabled,
            requires_confirmation=requires_confirmation,
            metadata=metadata or {},
            **kwargs,
        )
        self.add_tool(tool)
        return f

    if func is None:
        # Called with arguments: @provider.tool(...)
        return decorator
    # Called without arguments: @provider.tool
    return decorator(func)