Planning
Planning gives the model a structured, self-updating task list through a small toolset — and surfaces the current plan back to the model every turn without ever invalidating the prompt cache. It can stay in memory for a single run or persist to SQLite/Postgres, break steps into subtasks with dependencies, and emit events from granular changes.
While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.
Long agentic runs drift: the model loses track of what it set out to do and what’s left. The usual fix — keep a running plan and re-inject it into the system prompt each turn — invalidates the prompt cache. The system prompt sits at the front of the request, so every plan edit changes the cached prefix and forces the whole conversation to be re-processed at full token price.
The model owns the plan through the planning toolset. The current plan is surfaced back as an ephemeral reminder appended to the tail of each request, led by the capability’s single cache breakpoint:
- The reminder is added after the durable history is persisted, so it reaches the model but is never written to
message_history. No reminders accumulate across turns. - The reminder opens with a
CachePoint, which providers attach to the content right before it: the last user prompt, or in a tool loop the latest tool result. That content is what the next request starts with, so on Anthropic, Bedrock and OpenAI each request reads the previous one’s prefix back from cache, tool results included, and only the reminder is sent uncached. On OpenRouter the breakpoint stops at the previous user message, so the assistant turns and tool results after it stay outside the explicit prefix.
As with all capability cache breakpoints, provider mapping applies: OpenAI models only receive the CachePoint when the model profile enables explicit prompt cache breakpoints, OpenRouter attaches it to the previous user message rather than to a tool result, and models without explicit breakpoints (such as Google’s) ignore it. With Anthropic automatic caching (anthropic_cache), the automatic breakpoint lands on the reminder, which the next request doesn’t contain; the Planning breakpoint right before it is the one the next request reads back. A request with no content before the reminder (an instructions-only first request) gets the reminder without a breakpoint, since there is no prefix to protect.
A capability listed before Planning that also appends a tail part each request (for example SystemReminders) puts its content between the durable history and the plan reminder. The Planning breakpoint then covers that content too, so its cache entry is only read back while that content stays the same. The earlier capability’s own breakpoint still marks the durable history.
Construct an Agent with Planning() in its capabilities. The tools are registered automatically and static usage guidance is added to the system prompt:
from pydantic_ai import Agent
from pydantic_ai_harness import Planning
agent = Agent('anthropic:claude-sonnet-5-5', capabilities=[Planning()])
result = agent.run_sync('Refactor the auth module and add tests.')
print(result.output)
| Tool | Purpose |
|---|---|
write_plan(items) | Create or replace the full plan (whole-list replacement). |
read_plan() | Read the current plan with step ids and a progress summary. |
add_task(content, active_form) | Append a single pending step. |
update_task_status(task_id, status) | Move one step between statuses by id. |
update_task_statuses(updates) | Apply several status changes in one call, validated all-or-nothing. |
remove_task(task_id) | Delete a step by id. |
Each step is a content string, an optional present-continuous active_form label, and a status (pending, in_progress, completed, cancelled). The convention — stated in the guidance and the tools’ replies — is to keep exactly one step in_progress.
All six are registered by default. tools= narrows that to an allowlist, and the built-in guidance follows it:
from pydantic_ai_harness import Planning
planning = Planning(tools=['write_plan']) # whole-plan replacement only -- one tool, no step ids to track
Naming a tool the current mode does not register raises ValueError, as does an unknown key in descriptions.
Pass enable_subtasks=True to add three more tools, the blocked status, and a hierarchical view in read_plan:
| Tool | Purpose |
|---|---|
add_subtask(parent_id, content, active_form) | Add a child step under a parent. |
set_dependency(task_id, depends_on_id) | Make one step wait for another; the dependent step is auto-blocked until its prerequisite is resolved (completed or cancelled). Self-dependencies, cycles, and duplicates are rejected. |
get_available_tasks() | List steps with no incomplete dependencies — the ones that can start now. |
parent_id, depends_on, and the blocked status are rejected by write_plan unless enable_subtasks is set: without the subtask tools nothing reconciles a dependency and no view renders the hierarchy, so storing them would be a write the plan does not reflect.
By default the plan is a fresh, isolated in-memory plan per run. Pass a store to persist it:
from pydantic_ai_harness import Planning
from pydantic_ai_harness.planning import SqlitePlanStore
planning = Planning(store=SqlitePlanStore('plan.db', session='user-123'))
Built-in stores are InMemoryPlanStore, SqlitePlanStore, PostgresPlanStore (over a caller-owned asyncpg pool), and RedisPlanStore (over a caller-owned redis.asyncio client) — so the harness needs no database driver. Any PlanStore implementation works, and store_resolver selects one per run. SqlitePlanStore keeps its database on the machine running the agent, not in the workspace. It requires a file-backed database; use InMemoryPlanStore for ephemeral plans rather than ':memory:'.
The tail reminder reads the store on every model request, so a store that raises fails the run rather than degrading — the reminder is not best-effort. That is deliberate: a plan the model can no longer see is not a state to continue running in silently. Retry and fallback policy belongs to the store, not to Planning, and PlanStore is a protocol precisely so you can wrap one:
class BestEffort:
"""Serve the last known plan when the backing store is unreachable."""
def __init__(self, inner: PlanStore) -> None:
self._inner, self._last = inner, []
async def get_items(self) -> list[PlanItem]:
try:
self._last = await self._inner.get_items()
except ConnectionError:
pass
return self._last
# ... delegate the other five methods to `self._inner`
A shared store is the whole handoff mechanism between two runs. One agent writes the plan, a second one executes it, and the plan is the only state that crosses between them:
store = SqlitePlanStore('plan.db', session='issue-403')
planner = Agent('anthropic:claude-opus-5-5', capabilities=[Planning(store=store)])
executor = Agent('anthropic:claude-sonnet-5-5', capabilities=[Planning(store=store)])
await planner.run('Investigate the issue and write a plan. Do not implement anything.')
await executor.run('Implement the plan.')
The executor starts with no message_history, so it never pays for the planner’s investigation. Its first request carries only the new prompt plus the plan reminder, which the capability rebuilds from the store. That is why the two agents can run on different models: a large-context model can do the reading and the reasoning, and a smaller one can execute against the resulting checklist.
The planner’s read-only discipline is a property of how you configure that agent (which toolsets it gets, and what its instructions say), not something the capability enforces.
Subscribe to typed plan events to react to changes made through the Planning tools:
from pydantic_ai import Agent
from pydantic_ai_harness import Planning
from pydantic_ai_harness.planning import PlanCompletedEvent
agent = Agent('anthropic:claude-sonnet-5-5', capabilities=[Planning()])
@agent.on_event(PlanCompletedEvent)
async def announce(ctx, event):
print('done:', event.item.content)
The family contains
PlanCreatedEvent, PlanUpdatedEvent, PlanStatusChangedEvent, PlanCompletedEvent, and
PlanDeletedEvent; each carries the affected item and, for updates, previous_state.
Run events come from planning tool paths, including write_plan. Direct application mutations on a
PlanStore have no run context and do not produce run events. PlanEventEmitter, EventCallback,
and store event_emitter parameters remain supported but are deprecated.
Addressing steps by mutable integer index (insert/remove/reorder) is error-prone for both the code and the model. write_plan restates the whole plan each call, so there are no indices to track. Granular edits (add_task, update_task_status, remove_task) instead reference the stable id shown by read_plan.
The plan is never injected into the system prompt or instructions. Static usage guidance goes there (cache-stable); only the mutable plan rides the ephemeral tail reminder, which lives solely in the per-request copy and is never persisted. Set inject=False to disable it. Pydantic AI maps CachePoint for models whose profiles support prompt caching; on other models it is ignored.
With a durable-execution capability attached, every plan store call is a journaled capability
operation: the plan read used to build that reminder, and each read and write the plan tools make.
Replay reuses the recorded result instead of calling the store again, so recovering a run doesn’t
append a step to a persistent store a second time. Under an engine that runs function tools in workflow code,
like DBOS, a plan tool call runs alone, without overlapping other tool calls, and emits its run
events after its last store call. So its recorded store calls keep the same order on replay,
including relative to the steps an event_stream_handler runs for those events.
Planning carries the stable default id='planning', so durable recovery works without
configuration.
Engines that run tools and capability operations in a separate worker, like Temporal, don’t
carry the run’s in-memory plan there: each call resolves its store from the run context. Pass a
persistent store or a store_resolver (such as SqlitePlanStore or PostgresPlanStore) to keep
the plan across steps.
from pydantic_ai_harness import Planning
Planning(
guidance=None, # static system-prompt guidance; None = default, '' = omit
cache_ttl='5m', # TTL for the cache breakpoint leading the plan reminder ('5m' | '1h')
store=None, # None = fresh in-memory plan per run; or a PlanStore to persist
enable_subtasks=False, # add subtask/dependency tools and the 'blocked' status
inject=True, # surface the current plan as a cache-safe tail reminder
tools=None, # None = every tool the mode registers; or an allowlist of names
descriptions=None, # optional per-tool description overrides, keyed by tool name
)
Planning works with Pydantic AI’s agent spec:
# agent.yaml
model: anthropic:claude-sonnet-5-5
capabilities:
- Planning: {}
from pydantic_ai import Agent
from pydantic_ai_harness import Planning
agent = Agent.from_file('agent.yaml', custom_capability_types=[Planning])
result = agent.run_sync('...')
print(result.output)
- Pydantic AI capabilities
- Anthropic prompt caching
- Code Mode — another prompt-cache-aware harness capability
Bases: AbstractCapability[AgentDepsT]
Structured task planning that keeps the conversation in the prompt cache.
The model owns the plan through a small toolset (write_plan, read_plan,
add_task, update_task_status, update_task_statuses, remove_task, and
— when enable_subtasks is set — add_subtask, set_dependency,
get_available_tasks); tools narrows that surface to an allowlist. The
current plan is surfaced back as an ephemeral reminder appended to the
tail of each request and never persisted. The reminder leads with a
CachePoint, which providers attach to the content right before it — the
last user prompt, tool result, or retry prompt, which the next request
starts with — so on most providers only the reminder itself is re-read each
turn. On OpenRouter the breakpoint stops at the previous user message.
By default the plan lives in memory for the duration of a single run (a
fresh, isolated plan per run). Pass a store (or store_resolver) to
persist it — e.g. SqlitePlanStore or PostgresPlanStore.
from pydantic_ai import Agent
from pydantic_ai_harness.planning import Planning
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[Planning()])
Static planning guidance for the system prompt. Cache-stable.
Three states, so opting out is something you do on purpose rather than by accident:
None(the default): use the built-in guidance.'': no guidance at all.- any other string: use it instead of the built-in guidance.
A single str | None where None meant “no guidance” would leave no way to
ask for the default explicitly, and would turn a config that resolves to
None into a silent opt-out. This matches memory, exa and
runtime_authoring, which read the same way.
Type: str | None Default: None
TTL for the cache breakpoint leading the plan reminder.
Type: Literal[‘5m’, ‘1h’] Default: '5m'
Storage backend. None keeps a fresh in-memory plan per run (the original
ephemeral behaviour). Pass a store to persist the plan across runs.
Type: PlanStore | None Default: None
Optional per-run store resolver, e.g. lambda ctx: ctx.deps.plan_store.
Type: Callable[[RunContext[AgentDepsT]], PlanStore] | None Default: None
Add the subtask/dependency tools and the blocked status when true.
Type: bool Default: False
Surface the current plan as an ephemeral tail reminder each turn.
Type: bool Default: True
Optional allowlist of tool names to register; None registers all of them.
The full surface is write_plan, read_plan, add_task, update_task_status,
update_task_statuses, remove_task, plus add_subtask, set_dependency and
get_available_tasks under enable_subtasks. tools=['write_plan'] is the smallest
useful plan surface. Naming a tool this mode does not register raises ValueError.
The built-in guidance follows the allowlist for the whole-plan/granular/subtask split;
trimming within a group is better paired with a guidance string of your own.
Type: Sequence[str] | None Default: None
Optional per-tool description overrides, keyed by tool name. Unknown names raise ValueError.
Type: dict[str, str] | None Default: None
@async
def for_run(ctx: RunContext[AgentDepsT]) -> Planning[AgentDepsT]
Return a clone with this run’s store resolved and cached (per-run isolation).
Planning[AgentDepsT]
def resolve_store(ctx: RunContext[AgentDepsT]) -> PlanStore
Return the cached run store, or resolve one for direct toolset use.
PlanStore
def get_toolset() -> AgentToolset[AgentDepsT] | None
Provide the stable planning toolset, shared with every per-run copy.
AgentToolset[AgentDepsT] | None
def get_instructions() -> AgentInstructions[AgentDepsT] | None
Provide static, cache-stable guidance on using the planning tools.
A custom guidance string is used verbatim. The default is assembled from
the tools actually registered — the granular sentence is dropped when
tools excludes them all, and the subtask/dependency workflow is added
under enable_subtasks — so the model is not told about tools it lacks.
AgentInstructions[AgentDepsT] | None
@async
def wrap_model_request(
ctx: RunContext[AgentDepsT],
*,
request_context: ModelRequestContext,
handler: WrapModelRequestHandler,
) -> ModelResponse
Append the ephemeral plan reminder to the request, leading with a cache breakpoint.
The breakpoint marks the end of the content before the reminder, so each request caches what the next one starts with, including every tool result since the last user prompt, while the reminder stays outside the cached prefix and out of message history.
@classmethod
def from_spec(
cls,
*,
backend: Literal['memory', 'sqlite'] = 'memory',
database: str = '.agent-plan.db',
session: str = 'default',
enable_subtasks: bool = False,
inject: bool = True,
guidance: str | None = None,
cache_ttl: Literal['5m', '1h'] = '5m',
tools: list[str] | None = None,
) -> Planning[AgentDepsT]
Construct a Planning capability from serializable options.
Planning[AgentDepsT]
@classmethod
def get_serialization_name(cls) -> str | None
Serialization name for agent-spec support.