plan_provider

Class info

Classes

Name	Children	Inherits
AgentContext llmling_agent.agent.context Runtime context for agent execution.		NodeContext Generic
PlanEntry llmling_agent.resource_providers.plan_provider A single entry in the execution plan.		object
PlanProvider llmling_agent.resource_providers.plan_provider Provides plan-related tools for agent planning and task management.		ResourceProvider
PlanUpdateEvent llmling_agent.resource_providers.plan_provider Event indicating plan state has changed.		object
ResourceProvider llmling_agent.resource_providers.base Base class for resource providers.	StaticResourceProvider AggregatingResourceProvider MCPResourceProvider PlanProvider SkillsResourceProvider	object

🛈 DocStrings

Plan provider for agent planning and task management.

PlanEntry dataclass

A single entry in the execution plan.

Represents a task or goal that the assistant intends to accomplish as part of fulfilling the user's request.

Source code in src/llmling_agent/resource_providers/plan_provider.py

@dataclass(kw_only=True)
class PlanEntry:
    """A single entry in the execution plan.

    Represents a task or goal that the assistant intends to accomplish
    as part of fulfilling the user's request.
    """

    content: str
    """Human-readable description of what this task aims to accomplish."""

    priority: PlanEntryPriority
    """The relative importance of this task."""

    status: PlanEntryStatus
    """Current execution status of this task."""

content instance-attribute

content: str

Human-readable description of what this task aims to accomplish.

priority instance-attribute

priority: PlanEntryPriority

The relative importance of this task.

status instance-attribute

status: PlanEntryStatus

Current execution status of this task.

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

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

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

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_tools async

get_tools() -> list[Tool]

Get plan management tools.

Source code in src/llmling_agent/resource_providers/plan_provider.py

async def get_tools(self) -> list[Tool]:
    """Get plan management tools."""
    return [
        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

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

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

PlanUpdateEvent dataclass

Event indicating plan state has changed.

Source code in src/llmling_agent/resource_providers/plan_provider.py

@dataclass(kw_only=True)
class PlanUpdateEvent:
    """Event indicating plan state has changed."""

    entries: list[PlanEntry]
    """Current plan entries."""
    event_kind: Literal["plan_update"] = "plan_update"
    """Event type identifier."""

entries instance-attribute

entries: list[PlanEntry]

Current plan entries.

event_kind class-attribute instance-attribute

event_kind: Literal['plan_update'] = 'plan_update'

Event type identifier.