`context.planner()` – NodePlanner for Typed DAG Planning¶

context.planner() returns a NodePlanner: a node-bound façade over PlannerService that generates and executes typed DAG plans with an LLM.

Today, planning is static: the planner proposes a DAG from the available graphs/skills, validates it against typed inputs/outputs, then you execute it. In the future, this will evolve toward dynamic planning (re-planning based on intermediate execution signals, tool feedback, and typed state).

Core ideas¶

Typed DAG planning¶

Each plan is a directed acyclic graph (DAG) of steps.
Each step references an action (action_ref) plus inputs.
The planner uses typed input/output constraints to ensure the plan is executable.

Flow constraints (`flow_ids`)¶

flow_ids is a list of flow identifiers used to constrain the search space. This can be specified when defining the graph/graph function in the decorator @graph(flow_id=...) or @graph_fn(flow_id=...). Any graphs/actions registered under the same flow list can be combined into a single valid DAG (subject to type compatibility).

Node-bound execution context¶

NodePlanner auto-fills execution metadata from the active NodeContext (identity/session/app/agent), while still allowing overrides via keyword arguments.

1. Planning Types and Dataclasses¶

These dataclasses define the planning protocol and validation surface.

PlanStep

Represents a single step in a plan, referencing an action and its inputs.

Attributes:

Name	Type	Description
`id`	`str`	Unique identifier for the plan step.
`action`	`str`	short, human/LLM-readable name for the action
`action_ref`	`str`	Reference to the associated action specification. This is a canonicial identifier used in AetherGraph.
`inputs`	`Dict[str, Any]`	Input parameters for the action.

Methods:

Name	Description
`from_dict`	Creates a PlanStep instance from a dictionary.
`to_dict`	Serializes the PlanStep instance to a dictionary.

Notes

this class is usually not accessed unless you are building or inspecting plans directly.

CandidatePlan

CandidatePlan represents a plan consisting of multiple steps and additional metadata.

Attributes:

Name	Type	Description
`steps`	`list[PlanStep]`	A list of steps that make up the plan.
`extras`	`dict[str, Any]`	Additional metadata or information associated with the plan.

Methods:

Name	Description
`from_dict`	dict[str, Any]) -> CandidatePlan: Creates an instance of CandidatePlan from a dictionary representation.
`to_dict`	Converts the CandidatePlan instance into a dictionary representation.

ValidationIssue

Represents a validation issue encountered in a planning process.

Attributes:

Name	Type	Description
`kind`	`str`	The type of validation issue. Possible values include 'missing_input', 'unknown_action', 'type_mismatch', 'cycle', etc.
`step_id`	`str`	The identifier of the plan step where the issue occurred.
`field`	`str`	The name of the input, output, or other field related to the issue.
`message`	`str`	A human-readable message describing the issue.
`details`	`dict[str, Any]`	Additional details about the issue, provided as a dictionary. Defaults to an empty dictionary.

ValidationResult

ValidationResult represents the result of validating a plan, including its validity, any structural errors, and missing user-provided values.

Attributes:

Name	Type	Description
`ok`	`bool`	Indicates whether the plan is valid (True) or invalid (False).
`issues`	`list[ValidationIssue]`	A list of validation issues found in the plan.
`has_structural_errors`	`bool`	Indicates if the plan has structural errors.
`missing_user_bindings`	`dict[str, list[str]]`	A dictionary mapping keys to lists of locations where user-provided values are missing.

Methods: is_partial_ok() -> bool: Checks if the plan is structurally valid but requires user-provided values before execution. summary() -> str: Generates a summary string describing the validity of the plan and any issues found. If the plan is valid, the summary states "Plan is valid." If invalid, the summary lists the issues with their details.

PlanningEvent

Represents a lightweight event emitted during the planning process. This class is intended for use in logging, debugging, and updating UI progress (e.g., updating a spinner or status text). It provides optional structured payloads for additional context.

Attributes:

Name	Type	Description
`phase`	`PlanningPhase`	The current phase of the planning process.
`iteration`	`int`	The iteration number within the planning phase.
`message`	`str \| None`	An optional message describing the event.
`raw_llm_output`	`dict[str, Any] \| None`	An optional dictionary containing raw output from the language model.
`plan_dict`	`dict[str, Any] \| None`	An optional dictionary representing the plan structure.
`validation`	`ValidationResult \| None`	An optional validation result associated with the event.

PlanningContext

Represents the context for planning, including user inputs, external slots, memory snippets, and other parameters that guide the planning process.

Attributes:

Name	Type	Description
`goal`	`str`	The goal or objective of the planning process.
`user_inputs`	`dict[str, Any]`	A dictionary of user-provided inputs.
`external_slots`	`dict[str, Any]`	A dictionary of external slots or parameters.
`memory_snippets`	`list[str]`	A list of memory snippets for narrow, LLM-facing context.
`artifact_snippets`	`list[str]`	A list of artifact snippets for narrow, LLM-facing context.
`flow_ids`	`list[str] \| None`	A list of flow IDs representing the tools/graphs that are allowed to be used. Defaults to None.
`instruction`	`str \| None`	A richer agent context, used by planner code but not directly dumped. Can be parsed from skills. Defaults to None.
`allow_partial_plans`	`bool`	Indicates whether the planner should accept structurally valid but missing user input plans. Defaults to True.
`preferred_external_keys`	`list[str] \| None`	A list of keys that are allowed or preferred as `${user.<key>}`. Defaults to None.

PlanResult

PlanResult is a data class that encapsulates the result of a planning operation.

Attributes:

Name	Type	Description
`plan`	`CandidatePlan \| None`	The candidate plan resulting from the planning operation. It can be None if no valid plan was generated.
`validation`	`ValidationResult \| None`	The validation result associated with the plan. It can be None if validation was not performed or not applicable.
`events`	`list[PlanningEvent]`	A list of planning events that occurred during the planning process. Defaults to an empty list.

2. Execution Types and Dataclasses¶

Execution returns structured events and results, and can be run synchronously or in the background.

ExecutionEvent

Represents an event emitted during the execution of a plan. This class is used to encapsulate information about the current state of execution, which can be useful for logging purposes or updating the UI with progress information.

Attributes:

Name	Type	Description
`phase`	`ExecutionPhase`	The current phase of execution.
`step_id`	`str \| None`	The identifier of the current step, if applicable.
`message`	`str \| None`	An optional message providing additional context about the event.
`step_outputs`	`dict[str, Any] \| None`	Outputs produced by the step, applicable for success phases.
`error`	`Exception \| None`	The exception raised during execution, applicable for failure phases.

ExecutionResult

Represents the result of an execution process.

Attributes:

Name	Type	Description
`ok`	`bool`	Indicates whether the execution was successful.
`outputs`	`dict[str, Any]`	A dictionary containing the outputs of the execution.
`errors`	`list[ExecutionEvent]`	A list of errors that occurred during execution. Defaults to an empty list.

BackgroundExecutionHandle

BackgroundExecutionHandle is a data class that represents a handle for managing the execution of a background task.

Attributes:

Name	Type	Description
`id`	`str`	A unique identifier for the execution handle.
`plan_id`	`str \| None`	The identifier of the associated plan, if any.
`task`	`Task[ExecutionResult]`	The asyncio task representing the background execution.

Methods:

Name	Description
`done`	Checks if the background task has completed.
`cancelled`	Checks if the background task has been cancelled.
`cancel`	Cancels the background task.
`wait`	Awaits the completion of the background task and returns its result.

3. Core APIs¶

Below are the primary entry points. Paths are stable (do not change imports/targets for these API docs).

Plan¶

plan() is the standard entry point for planning with high-level inputs.
plan_with_context() is a lower-level entry point when you already have a PlanningContext.
parse_inputs() extracts missing structured inputs from free-form user text.

plan(*, goal, user_inputs, external_slots, flow_id, ...)

Generate a plan based on the provided goal and context.

This method delegates the planning process to the underlying PlannerService, using the provided inputs and optional callbacks for event handling.

Examples:

Basic usage to generate a plan:

result = await node_planner.plan(goal="Optimize workflow")

Providing additional inputs and handling events:

result = await node_planner.plan(
    goal="Generate report",
    user_inputs={"date": "2023-10-01"},
    on_event=lambda event: print(f"Event: {event}")
)

Parameters:

Name	Type	Description	Default
`goal`	`str`	The primary objective or goal for the planning process.	required
`user_inputs`	`dict[str, Any] \| None`	Optional dictionary of user-provided inputs for the plan.	`None`
`external_slots`	`dict[str, Any] \| None`	Optional dictionary of external slot values to consider.	`None`
`flow_ids`	`list[str] \| None`	Optional list of flow identifiers to constrain the planning scope.	`None`
`instruction`	`str \| None`	Optional instruction or guidance for the planner.	`None`
`allow_partial`	`bool`	Whether to allow partial plans if the goal cannot be fully satisfied (default: True).	`True`
`preferred_external_keys`	`list[str] \| None`	Optional list of preferred external keys to prioritize.	`None`
`memory_snippets`	`list[str] \| None`	Optional list of memory snippets to include in the planning context.	`None`
`artifact_snippets`	`list[str] \| None`	Optional list of artifact snippets to include in the planning context.	`None`
`on_event`	`PlanEventsCallback \| None`	Optional callback function to handle planning events. The callback should be of type `PlanningEventCallback = Callable[[PlanningEvent], None] \| Callable[[PlanningEvent], Awaitable[None]]`.	`None`

Returns:

Name	Type	Description
`PlanResult`	`PlanResult`	The result of the planning process, including the generated plan and metadata.

Notes: You can use on_event to monitor the planning process and react to intermediate events.

plan_with_context(ctx, *, on_event)

Plan a task using the provided planning context.

This method delegates the planning process to the underlying service, allowing for the generation of a plan based on the given context and optional event callback.

Examples:

Basic usage to generate a plan:

result = await planner.plan_with_context(context)

Using an event callback to monitor planning progress:

async def on_event(event):
    print(f"Received event: {event}")

result = await planner.plan_with_context(context, on_event=on_event)

Parameters:

Name	Type	Description	Default
`ctx`	`PlanningContext`	The `PlanningContext` object containing the necessary information for planning.	required
`on_event`	`PlanEventsCallback \| None`	Optional callback function to handle planning events. Defaults to None.	`None`

Returns:

Type	Description
`PlanResult`	A `PlanResult` object containing the outcome of the planning process.

Notes

The on_event callback can be used to receive updates or intermediate results during the planning process.

parse_inputs(*, message, missing_keys, instruction)

Parse input data and handle missing keys.

This method processes the provided input message, identifies any missing keys, and optionally uses an instruction to guide the parsing process. It delegates the actual parsing logic to the service.parse_inputs method.

Examples:

Basic usage to parse inputs:

result = await node_planner.parse_inputs(
    message="Input data with key1 = 0.1 and key2 = dummy string inputs",
    missing_keys=["key1", "key2"]
)

Parsing with an additional instruction:

result = await node_planner.parse_inputs(
    message="Input data with key1 equals 0.1 and key2 is some dummy string inputs",
    missing_keys=["key1", "key2"],
    instruction="key1 is a float and key2 is a string. Make sure all values are included in a dictionary."
)

Parameters:

Name	Type	Description	Default
`message`	`str`	The input message to be parsed.	required
`missing_keys`	`list[str]`	Field names whose values we want to extract.	required
`instruction`	`str \| None`	Optional instruction to guide the parsing process.	`None`

Returns:

Type	Description
	The result of the parsing operation as returned by the `service.parse_inputs` method.

Notes: This method is asynchronous and relies on the service.parse_inputs implementation for the actual parsing logic.

Execute¶

execute() runs a validated CandidatePlan immediately.
execute_background() runs the plan asynchronously and returns a handle.

execute(plan, *, user_inputs, on_event, ...)

Execute a plan with the provided context and parameters.

This method uses the NodeContext to auto-fill execution metadata such as identity, session, agent, and application IDs. Callers can override these defaults by providing explicit keyword arguments.

Examples:

Basic usage to execute a plan:

result = await node_planner.execute(plan)

Executing with additional user inputs and event handling:

result = await node_planner.execute(
plan,
user_inputs={"key1": "value1"},
on_event=lambda event: print(f"Execution event: {event}")
)

Parameters:

Name	Type	Description	Default
`plan`	`CandidatePlan`	The `CandidatePlan` object to execute.	required
`user_inputs`	`dict[str, Any] \| None`	Optional dictionary of user-provided inputs for the execution. Values referenced as "${user.}"	`None`
`on_event`	`ExecEventsCallback \| None`	Optional callback function to handle execution events.	`None`
`identity`	`RequestIdentity \| None`	Optional `RequestIdentity` to override the default identity.	`None`
`visibility`	`RunVisibility`	Visibility level for the execution (default: `RunVisibility.normal`).	`normal`
`importance`	`RunImportance`	Importance level for the execution (default: `RunImportance.normal`).	`normal`
`session_id`	`str \| None`	Optional session ID to override the default session.	`None`
`agent_id`	`str \| None`	Optional agent ID to override the default agent.	`None`
`app_id`	`str \| None`	Optional application ID to override the default application.	`None`
`tags`	`list[str] \| None`	Optional list of tags to associate with the execution.	`None`
`origin`	`RunOrigin \| None`	Optional `RunOrigin` to specify the origin of the execution.	`None`

Returns:

Name	Type	Description
`ExecutionResult`	`ExecutionResult`	The result of the execution process, including status and metadata.

Notes: Use the on_event callback to monitor execution progress and handle intermediate events.

execute_background(plan, *, user_inputs, on_event, ...)

Execute a candidate plan in the background, bound to this node. This method initiates a fire-and-forget execution of the provided plan, leveraging the node's context and metadata. Progress updates can be reported via the on_event callback, and completion is signaled through the on_complete callback.

Examples:

Basic usage to execute a plan in the background:

handle = await node_planner.execute_background(plan)

Executing with additional metadata and callbacks:

handle = await node_planner.execute_background(
    user_inputs={"param1": "value1"},
    on_event=event_callback,
    on_complete=completion_callback,
    tags=["custom:tag"],
)

Parameters:

Name	Type	Description	Default
`plan`	`CandidatePlan`	The candidate plan to execute.	required
`user_inputs`	`dict[str, Any] \| None`	Optional dictionary of user-provided inputs for the execution.	`None`
`on_event`	`ExecEventsCallback \| None`	Optional callback to handle execution progress events.	`None`
`identity`	`RequestIdentity \| None`	Optional identity to associate with the execution; defaults to the node's context identity.	`None`
`visibility`	`RunVisibility`	The visibility level of the execution (default: RunVisibility.normal).	`normal`
`importance`	`RunImportance`	The importance level of the execution (default: RunImportance.normal).	`normal`
`session_id`	`str \| None`	Optional session ID to associate with the execution; defaults to the node's context session ID.	`None`
`agent_id`	`str \| None`	Optional agent ID to associate with the execution; defaults to the node's context agent ID.	`None`
`app_id`	`str \| None`	Optional application ID to associate with the execution; defaults to the node's context app ID.	`None`
`tags`	`list[str] \| None`	Optional list of tags to associate with the execution; additional tags are derived from the node's context.	`None`
`origin`	`RunOrigin \| None`	Optional origin of the execution; defaults to `RunOrigin.agent` if an agent ID is present, otherwise `RunOrigin.user`.	`None`
`on_complete`	`Callable[[ExecutionResult], Any] \| None`	Optional callback to handle execution completion.	`None`
`exec_id`	`str \| None`	Optional explicit execution ID to associate with the execution.	`None`

Returns:

Name	Type	Description
`BackgroundExecutionHandle`	`BackgroundExecutionHandle`	A handle to monitor or interact with the background execution.

Notes

Tags automatically include identifiers for the graph, node, and run (if available) from the node's context.
The on_event callback is invoked with progress updates, while the on_complete callback is invoked upon completion.

context.planner() – NodePlanner for Typed DAG Planning¶

Core ideas¶

Typed DAG planning¶

Flow constraints (flow_ids)¶

Node-bound execution context¶

1. Planning Types and Dataclasses¶

2. Execution Types and Dataclasses¶

3. Core APIs¶

Plan¶

Execute¶

`context.planner()` – NodePlanner for Typed DAG Planning¶

Flow constraints (`flow_ids`)¶