Agent plans

planning=True on agent() gives an agent six tools for structuring complex work: create_plan, start_step, mark_step_done, mark_step_failed, get_plan, and get_ready_steps. The agent uses them to organize a task into named steps with dependency tracking, then works through each step using its regular tools.

Plans are guidance, not a rigid execution graph. The LLM decides when to plan, which step to work on next, and when to replan. The framework tracks state and surfaces results — it does not drive execution.

Basic setup

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

@task
async def search_web(query: str) -> str:
    """Search the web and return results."""
    ...

@task
async def write_report(topic: str, content: str) -> str:
    """Write a formatted report."""
    ...

@workflow
async def research(ctx: ExecutionContext):
    analyst = await agent(
        "You are a market research analyst. "
        "For complex research tasks, create a plan to organize your work.",
        model="openai/gpt-4o",
        tools=[search_web, write_report],
        planning=True,
        max_tool_calls=30,
    )
    return await analyst(
        f"Research the competitive landscape for {ctx.input['product']}."
    )

With planning=True, the agent assesses the task, creates a plan if it judges the task complex enough, calls start_step before each step, does the work with its available tools, and calls mark_step_done with a result before moving on. For simple tasks, it may skip the plan entirely and respond directly.

The six planning tools

Step lifecycle

Each step moves through one of two paths:

pending → in_progress → completed
                      → failed

mark_step_done accepts steps in either pending or in_progress status — the agent can complete a step without calling start_step first. mark_step_failed works the same way. Trying to complete or fail an already-completed step is a no-op; trying to complete a failed step returns an error directing the agent to replan.

Plan structure

The agent calls create_plan with a JSON array of step objects:

# The LLM emits this as a tool call argument
create_plan(steps='[
  {"name": "gather-data", "description": "Search for competitor pricing data."},
  {"name": "analyze", "description": "Analyze pricing trends.", "depends_on": ["gather-data"]},
  {"name": "report", "description": "Write the final report.", "depends_on": ["analyze"]}
]')

Steps are goals, not individual tool calls. “Research competitor pricing” is the right granularity. “Call search_web” is too fine-grained. Plans must have at least 2 steps and no more than max_plan_steps (default: 20).

Step names must be lowercase, may contain hyphens and underscores, and cannot exceed 64 characters.

Status reminder

After each tool call, the agent receives a one-line reminder:

[Plan: 2/5 done. Active: "analyze". Ready: "validate" (from gather-data: ...).]

When a ready step has dependency results, they are included inline so the agent has context without calling get_plan. The reminder keeps the agent on track during long tool sequences.

Plan continuation

If the LLM returns no content and no tool calls while steps are still incomplete, the framework injects a continuation prompt with the current plan summary. This catches the case where smaller local models stop emitting any output before the plan is finished.

Configuration

analyst = await agent(
    "You are a market research analyst.",
    model="openai/gpt-4o",
    tools=[search_web, write_report],
    planning=True,
    max_plan_steps=15,
    strict_dependencies=True,
    approve_plan=True,
    max_tool_calls=30,
)

Dependency enforcement

By default (strict_dependencies=False), starting a step with unmet dependencies produces a warning but the step still transitions to in_progress:

{
  "name": "analyze",
  "status": "in_progress",
  "warning": "Step 'analyze' has unsatisfied dependencies: ['gather-data']. Proceeding anyway."
}

With strict_dependencies=True, the same call returns an error and the step stays pending:

{
  "error": "Step 'analyze' has unsatisfied dependencies: ['gather-data']. Complete them first."
}

Use strict_dependencies=True when steps genuinely cannot proceed without their dependencies’ output — for example, when an analysis step requires raw data from a prior search step. Use the default when the dependency ordering is advisory and the agent may have enough context to proceed.

Plan approval

When approve_plan=True, calling create_plan pauses the workflow and waits for a human decision before the plan activates. This uses Flux’s standard pause/resume mechanism — the execution suspends and the agent is in a paused state until the workflow is resumed.

# Approve the plan as submitted
workflow.resume(execution_id, plan_dict)

# Resume with a modified plan
workflow.resume(execution_id, {
    "steps": [
        {"name": "gather-data", "description": "Revised research scope."},
        {"name": "report", "description": "Write the report.", "depends_on": ["gather-data"]},
    ]
})

# Reject the plan
workflow.resume(execution_id, {"rejected": True})

When rejected, create_plan returns {"error": "Plan was rejected during review."} and the agent must reconsider its approach — typically by calling create_plan again with a revised plan.

The modified-plan path re-runs validation (minimum 2 steps, maximum max_plan_steps, no circular dependencies) before activating. If the resumed input contains a steps key, those steps replace the original. Any other dict (including the original plan_dict) activates the plan as submitted.

Replanning

If results change the plan mid-execution, the agent calls create_plan again with updated steps. Completed steps and their results are preserved automatically — they are restored into the new plan if a step with the same name exists. Only pending and in-progress steps are replaced.

If completed steps are not present in the new plan, the agent receives a warning:

{
  "warning": "Completed steps dropped (not in new plan): ['old-step']. Their results are lost."
}

There is no separate replan tool. Replanning is the same create_plan call — the framework detects the existing plan and merges completed state.

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

@task
async def search_database(query: str) -> str:
    """Search internal records."""
    ...

@task
async def search_web(query: str) -> str:
    """Search public sources."""
    ...

@task
async def analyze_data(data: str, focus: str) -> str:
    """Analyze gathered data."""
    ...

@task
async def generate_report(title: str, sections: str) -> str:
    """Produce a formatted report."""
    ...

@workflow
async def quarterly_review(ctx: ExecutionContext):
    analyst = await agent(
        "You are a business analyst. For multi-step tasks, create a plan first. "
        "Call start_step before each step, mark steps done when finished, "
        "and call mark_step_failed if a step cannot be completed. "
        "If a tool fails, retry it or adjust your plan.",
        model="ollama/qwen3",
        tools=[search_database, search_web, analyze_data, generate_report],
        planning=True,
        max_tool_calls=30,
    )
    return await analyst(
        "Prepare a quarterly review report. Gather internal data from the database "
        "and market data from the web, analyze both together, then generate a report. "
        "The analysis step should depend on both data-gathering steps. "
        "The report should depend on analysis."
    )

Plan persistence with long-term memory

Without a long_term_memory provider, the plan is ephemeral — it exists only for the duration of the agent() call. Providing a LongTermMemory instance causes the plan to be saved after every state change and restored when the agent is next initialized.

from flux.tasks.ai.memory import long_term_memory, in_memory, sqlite

# Plan lives as long as the process (in-memory, not cross-restart)
mem = long_term_memory(provider=in_memory(), agent="analyst", scope="analyst")

# Plan survives process restarts (database-backed)
mem = long_term_memory(
    provider=sqlite("agent_memory.db"),
    agent="analyst",
    scope="analyst",
)

analyst = await agent(
    "...",
    model="openai/gpt-4o",
    planning=True,
    long_term_memory=mem,
)

Planning with sub-agents

Planning composes with sub-agents. A manager with planning=True can create a plan where each step delegates to a specialist agent.

@workflow
async def managed_research(ctx: ExecutionContext):
    researcher = await agent(
        "You are a research specialist. Use search_web to gather information.",
        model="ollama/llama3.2",
        name="researcher",
        description="Deep research using web sources.",
        tools=[search_web],
    )

    manager = await agent(
        "You are a project manager. Plan complex tasks and delegate to your team.",
        model="openai/gpt-4o",
        agents=[researcher],
        planning=True,
        max_tool_calls=30,
    )

    return await manager(ctx.input["task"])

The manager’s delegate tool and the planning tools are both available in the same tool set. The plan steps can mix direct tool calls and delegation as needed.

Accessing plan data in code

AgentPlan and AgentStep are importable dataclasses useful for testing or programmatic inspection:

from flux.tasks.ai import AgentPlan, AgentStep

# Build a plan in code
step = AgentStep(
    name="gather-data",
    description="Collect source material.",
    depends_on=[],
    status="pending",  # "pending" | "in_progress" | "completed" | "failed"
    result=None,       # any value, stored by mark_step_done
    error=None,        # str, stored by mark_step_failed
)
plan = AgentPlan(steps=[step])

# Query by status
plan.completed_steps()
plan.pending_steps()
plan.in_progress_steps()
plan.failed_steps()

# Readiness checks
plan.ready_steps()                    # steps with all deps completed
plan.dependencies_satisfied(step)     # True/False
plan.dependency_results(step)         # {dep_name: result, ...}

# Active step
plan.active_step()                    # the in_progress step, or None

# Serialization
plan.to_dict()
AgentPlan.from_dict(data)

These classes are also what the framework stores and updates internally during a planning run, so they reflect the exact live state of any plan.

When to use planning

Planning works well when:

• The task has three or more distinct steps where one step’s output feeds into the next.
• The task is long enough that losing track of progress would cause the agent to repeat or skip work.
• Step failures matter — failed steps block dependents, which gives the agent a clear signal to replan rather than silently proceeding.

Planning adds overhead. For simple tasks — a single tool call, a quick question — it makes the agent do extra work for no benefit. The planning preamble in the system prompt guides the LLM to skip create_plan for simple cases, but the judgment is ultimately the model’s.