WorkflowBuilder Class

A builder class for constructing workflows.

This class provides a fluent API for defining workflow graphs by connecting executors with edges and configuring execution parameters. Call build to create an immutable Workflow instance.

Initialize the WorkflowBuilder with an empty list of edges and no starting executor.

Constructor

WorkflowBuilder()

Parameters

Name	Description
max_iterations	Maximum number of iterations for workflow convergence. Default is 100. Default value: 100
name	Optional human-readable name for the workflow. Default value: None
description	Optional description of what the workflow does. Default value: None

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class UpperCaseExecutor(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(text.upper())


   class ReverseExecutor(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(text[::-1])


   # Build a workflow
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: UpperCaseExecutor(id="upper"), name="UpperCase")
       .register_executor(lambda: ReverseExecutor(id="reverse"), name="Reverse")
       .add_edge("UpperCase", "Reverse")
       .set_start_executor("UpperCase")
       .build()
   )

   # Run the workflow
   events = await workflow.run("hello")
   print(events.get_outputs())  # ['OLLEH']

Methods

__init__	Initialize the WorkflowBuilder with an empty list of edges and no starting executor.
__new__
add_agent	Add an agent to the workflow by wrapping it in an AgentExecutor. This method creates an AgentExecutor that wraps the agent with the given parameters and ensures that subsequent uses of the same agent instance in other builder methods (like add_edge, set_start_executor, etc.) will reuse the same wrapped executor. Note: Agents adapt their behavior based on how the workflow is executed: run_stream(): Agents emit incremental AgentRunUpdateEvent events as tokens are produced run(): Agents emit a single AgentRunEvent containing the complete response
add_chain	Add a chain of executors to the workflow. The output of each executor in the chain will be sent to the next executor in the chain. The input types of each executor must be compatible with the output types of the previous executor. Cycles in the chain are not allowed, meaning an executor cannot appear more than once in the chain. Note: If executor instances are provided, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
add_edge	Add a directed edge between two executors. The output types of the source and the input types of the target must be compatible. Messages sent by the source executor will be routed to the target executor. Note: If instances are provided for both source and target, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
add_fan_in_edges	Add multiple edges from sources to a single target executor. The edges will be grouped together for synchronized processing, meaning the target executor will only be executed once all source executors have completed. The target executor will receive a list of messages aggregated from all source executors. Thus the input types of the target executor must be compatible with a list of the output types of the source executors. Note: If instances are provided for sources and target, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
add_fan_out_edges	Add multiple edges to the workflow where messages from the source will be sent to all targets. The output types of the source and the input types of the targets must be compatible. Messages from the source will be broadcast to all target executors concurrently. Note: If instances are provided for source and targets, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
add_multi_selection_edge_group	Add an edge group that represents a multi-selection execution model. The output types of the source and the input types of the targets must be compatible. Messages from the source executor will be sent to multiple target executors based on the provided selection function. The selection function should take a message and a list of target executor IDs, and return a list of executor IDs indicating which target executors should receive the message. Note: If instances are provided for source and targets, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
add_switch_case_edge_group	Add an edge group that represents a switch-case statement. The output types of the source and the input types of the targets must be compatible. Messages from the source executor will be sent to one of the target executors based on the provided conditions. Think of this as a switch statement where each target executor corresponds to a case. Each condition function will be evaluated in order, and the first one that returns True will determine which target executor receives the message. The default case (if provided) will receive messages that fall through all conditions (i.e., no condition matched). Note: If instances are provided for source and case targets, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.
build	Build and return the constructed workflow. This method performs validation before building the workflow to ensure: A starting executor has been set All edges connect valid executors The graph is properly connected Type compatibility between connected executors
register_agent	Register an agent factory function for lazy initialization. This method allows you to register a factory function that creates an agent. The agent will be instantiated and wrapped in an AgentExecutor only when the workflow is built, enabling deferred initialization and potentially reducing startup time.
register_executor	Register an executor factory function for lazy initialization. This method allows you to register a factory function that creates an executor. The executor will be instantiated only when the workflow is built, enabling deferred initialization and potentially reducing startup time.
set_max_iterations	Set the maximum number of iterations for the workflow. When a workflow contains cycles, this limit prevents infinite loops by capping the total number of executor invocations. The default is 100 iterations.
set_start_executor	Set the starting executor for the workflow. The start executor is the entry point for the workflow. When the workflow is executed, the initial message will be sent to this executor.
with_checkpointing	Enable checkpointing with the specified storage. Checkpointing allows workflows to save their state periodically, enabling pause/resume functionality and recovery from failures. The checkpoint storage implementation determines where checkpoints are persisted.

init

Initialize the WorkflowBuilder with an empty list of edges and no starting executor.

__init__(max_iterations: int = DEFAULT_MAX_ITERATIONS, name: str | None = None, description: str | None = None)

Parameters

Name	Description
max_iterations	int Maximum number of iterations for workflow convergence. Default is 100. Default value: DEFAULT_MAX_ITERATIONS
name	str \| None Optional human-readable name for the workflow. Default value: None
description	str \| None Optional description of what the workflow does. Default value: None

new

__new__(**kwargs)

add_agent

Add an agent to the workflow by wrapping it in an AgentExecutor.

This method creates an AgentExecutor that wraps the agent with the given parameters and ensures that subsequent uses of the same agent instance in other builder methods (like add_edge, set_start_executor, etc.) will reuse the same wrapped executor.

Note: Agents adapt their behavior based on how the workflow is executed:

run_stream(): Agents emit incremental AgentRunUpdateEvent events as tokens are produced
run(): Agents emit a single AgentRunEvent containing the complete response

add_agent(agent: AgentProtocol, agent_thread: Any | None = None, output_response: bool = False, id: str | None = None) -> Self

Parameters

Name	Description
agent Required	AgentProtocol The agent to add to the workflow.
agent_thread	Any \| None The thread to use for running the agent. If None, a new thread will be created. Default value: None
output_response	bool Whether to yield an AgentRunResponse as a workflow output when the agent completes. Default value: False
id	str \| None A unique identifier for the executor. If None, the agent's name will be used if available. Default value: None

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Exceptions

Type	Description
ValueError	If the provided id or agent name conflicts with an existing executor.

Examples


   from agent_framework import WorkflowBuilder
   from agent_framework_anthropic import AnthropicAgent

   # Create an agent
   agent = AnthropicAgent(name="writer", model="claude-3-5-sonnet-20241022")

   # Add the agent to a workflow
   workflow = WorkflowBuilder().add_agent(agent, output_response=True).set_start_executor(agent).build()

add_chain

Add a chain of executors to the workflow.

The output of each executor in the chain will be sent to the next executor in the chain. The input types of each executor must be compatible with the output types of the previous executor.

Cycles in the chain are not allowed, meaning an executor cannot appear more than once in the chain.

Note: If executor instances are provided, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.

add_chain(executors: Sequence[Executor | AgentProtocol | str]) -> Self

Parameters

Name	Description
executors Required	Sequence[Executor \| AgentProtocol \| str] A list of executors or registered names of the executor factories to chain together.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class Step1(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(text.upper())


   class Step2(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(text[::-1])


   class Step3(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(f"Final: {text}")


   # Chain executors in sequence
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: Step1(id="step1"), name="step1")
       .register_executor(lambda: Step2(id="step2"), name="step2")
       .register_executor(lambda: Step3(id="step3"), name="step3")
       .add_chain(["step1", "step2", "step3"])
       .set_start_executor("step1")
       .build()
   )

add_edge

Add a directed edge between two executors.

The output types of the source and the input types of the target must be compatible. Messages sent by the source executor will be routed to the target executor.

Note: If instances are provided for both source and target, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.

add_edge(source: Executor | AgentProtocol | str, target: Executor | AgentProtocol | str, condition: Callable[[Any], bool] | None = None) -> Self

Parameters

Name	Description
source Required	Executor \| AgentProtocol \| str
target Required	Executor \| AgentProtocol \| str
condition	Callable[[Any], bool] \| None Default value: None

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class ProcessorA(Executor):
       @handler
       async def process(self, data: str, ctx: WorkflowContext[int]) -> None:
           await ctx.send_message(len(data))


   class ProcessorB(Executor):
       @handler
       async def process(self, count: int, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(f"Processed {count} characters")


   # Connect executors with an edge
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: ProcessorA(id="a"), name="ProcessorA")
       .register_executor(lambda: ProcessorB(id="b"), name="ProcessorB")
       .add_edge("ProcessorA", "ProcessorB")
       .set_start_executor("ProcessorA")
       .build()
   )


   # With a condition
   def only_large_numbers(msg: int) -> bool:
       return msg > 100


   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: ProcessorA(id="a"), name="ProcessorA")
       .register_executor(lambda: ProcessorB(id="b"), name="ProcessorB")
       .add_edge("ProcessorA", "ProcessorB", condition=only_large_numbers)
       .set_start_executor("ProcessorA")
       .build()
   )

add_fan_in_edges

Add multiple edges from sources to a single target executor.

The edges will be grouped together for synchronized processing, meaning the target executor will only be executed once all source executors have completed.

The target executor will receive a list of messages aggregated from all source executors. Thus the input types of the target executor must be compatible with a list of the output types of the source executors.

Note: If instances are provided for sources and target, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.

add_fan_in_edges(sources: Sequence[Executor | AgentProtocol | str], target: Executor | AgentProtocol | str) -> Self

Parameters

Name	Description
sources Required	Sequence[Executor \| AgentProtocol \| str] A list of source executors or registered names of the source factories for the edges.
target Required	Executor \| AgentProtocol \| str The target executor or registered name of the target factory for the edges.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class Producer(Executor):
       @handler
       async def produce(self, seed: int, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(f"result_{seed}")


   class Aggregator(Executor):
       @handler
       async def aggregate(self, results: list[str], ctx: WorkflowContext[Never, str]) -> None:
           combined = ", ".join(results)
           await ctx.yield_output(f"Combined: {combined}")


   # Collect results from multiple producers
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: Producer(id="prod_1"), name="Producer1")
       .register_executor(lambda: Producer(id="prod_2"), name="Producer2")
       .register_executor(lambda: Aggregator(id="agg"), name="Aggregator")
       .add_fan_in_edges(["Producer1", "Producer2"], "Aggregator")
       .set_start_executor("Producer1")
       .build()
   )

add_fan_out_edges

Add multiple edges to the workflow where messages from the source will be sent to all targets.

The output types of the source and the input types of the targets must be compatible. Messages from the source will be broadcast to all target executors concurrently.

Note: If instances are provided for source and targets, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.

add_fan_out_edges(source: Executor | AgentProtocol | str, targets: Sequence[Executor | AgentProtocol | str]) -> Self

Parameters

Name	Description
source Required	Executor \| AgentProtocol \| str The source executor or registered name of the source factory for the edges.
targets Required	Sequence[Executor \| AgentProtocol \| str] A list of target executors or registered names of the target factories for the edges.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class DataSource(Executor):
       @handler
       async def generate(self, count: int, ctx: WorkflowContext[str]) -> None:
           for i in range(count):
               await ctx.send_message(f"data_{i}")


   class ValidatorA(Executor):
       @handler
       async def validate(self, data: str, ctx: WorkflowContext) -> None:
           print(f"ValidatorA: {data}")


   class ValidatorB(Executor):
       @handler
       async def validate(self, data: str, ctx: WorkflowContext) -> None:
           print(f"ValidatorB: {data}")


   # Broadcast to multiple validators
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: DataSource(id="source"), name="DataSource")
       .register_executor(lambda: ValidatorA(id="val_a"), name="ValidatorA")
       .register_executor(lambda: ValidatorB(id="val_b"), name="ValidatorB")
       .add_fan_out_edges("DataSource", ["ValidatorA", "ValidatorB"])
       .set_start_executor("DataSource")
       .build()
   )

add_multi_selection_edge_group

Add an edge group that represents a multi-selection execution model.

The output types of the source and the input types of the targets must be compatible. Messages from the source executor will be sent to multiple target executors based on the provided selection function.

The selection function should take a message and a list of target executor IDs, and return a list of executor IDs indicating which target executors should receive the message.

add_multi_selection_edge_group(source: Executor | AgentProtocol | str, targets: Sequence[Executor | AgentProtocol | str], selection_func: Callable[[Any, list[str]], list[str]]) -> Self

Parameters

Name	Description
source Required	Executor \| AgentProtocol \| str The source executor or registered name of the source factory for the edge group.
targets Required	Sequence[Executor \| AgentProtocol \| str] A list of target executors or registered names of the target factories for the edges.
selection_func Required	Callable[[Any, list[str]], list[str]] A function that selects target executors for messages. Takes (message, list[executor_id]) and returns list[executor_id].

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler
   from dataclasses import dataclass


   @dataclass
   class Task:
       priority: str
       data: str


   class TaskDispatcher(Executor):
       @handler
       async def dispatch(self, text: str, ctx: WorkflowContext[Task]) -> None:
           priority = "high" if len(text) > 10 else "low"
           await ctx.send_message(Task(priority=priority, data=text))


   class WorkerA(Executor):
       @handler
       async def process(self, task: Task, ctx: WorkflowContext) -> None:
           print(f"WorkerA processing: {task.data}")


   class WorkerB(Executor):
       @handler
       async def process(self, task: Task, ctx: WorkflowContext) -> None:
           print(f"WorkerB processing: {task.data}")


   # Select workers based on task priority
   def select_workers(task: Task, available: list[str]) -> list[str]:
       if task.priority == "high":
           return available  # Send to all workers
       return [available[0]]  # Send to first worker only


   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: TaskDispatcher(id="dispatcher"), name="TaskDispatcher")
       .register_executor(lambda: WorkerA(id="worker_a"), name="WorkerA")
       .register_executor(lambda: WorkerB(id="worker_b"), name="WorkerB")
       .add_multi_selection_edge_group(
           "TaskDispatcher",
           ["WorkerA", "WorkerB"],
           selection_func=select_workers,
       )
       .set_start_executor("TaskDispatcher")
       .build()
   )

add_switch_case_edge_group

Add an edge group that represents a switch-case statement.

The output types of the source and the input types of the targets must be compatible. Messages from the source executor will be sent to one of the target executors based on the provided conditions.

Think of this as a switch statement where each target executor corresponds to a case. Each condition function will be evaluated in order, and the first one that returns True will determine which target executor receives the message.

The default case (if provided) will receive messages that fall through all conditions (i.e., no condition matched).

Note: If instances are provided for source and case targets, they will be shared across all workflow instances created from the built Workflow. To avoid this, consider registering the executors and agents using register_executor and register_agent and referencing them by factory name for lazy initialization instead.

add_switch_case_edge_group(source: Executor | AgentProtocol | str, cases: Sequence[Case | Default]) -> Self

Parameters

Name	Description
source Required	Executor \| AgentProtocol \| str The source executor or registered name of the source factory for the edge group.
cases Required	Sequence[Case \| Default] A list of case objects that determine the target executor for each message.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler, Case, Default
   from dataclasses import dataclass


   @dataclass
   class Result:
       score: int


   class Evaluator(Executor):
       @handler
       async def evaluate(self, text: str, ctx: WorkflowContext[Result]) -> None:
           await ctx.send_message(Result(score=len(text)))


   class HighScoreHandler(Executor):
       @handler
       async def handle(self, result: Result, ctx: WorkflowContext) -> None:
           print(f"High score: {result.score}")


   class LowScoreHandler(Executor):
       @handler
       async def handle(self, result: Result, ctx: WorkflowContext) -> None:
           print(f"Low score: {result.score}")


   # Route based on score value
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: Evaluator(id="eval"), name="Evaluator")
       .register_executor(lambda: HighScoreHandler(id="high"), name="HighScoreHandler")
       .register_executor(lambda: LowScoreHandler(id="low"), name="LowScoreHandler")
       .add_switch_case_edge_group(
           "Evaluator",
           [
               Case(condition=lambda r: r.score > 10, target="HighScoreHandler"),
               Default(target="LowScoreHandler"),
           ],
       )
       .set_start_executor("Evaluator")
       .build()
   )

build

Build and return the constructed workflow.

This method performs validation before building the workflow to ensure:

A starting executor has been set
All edges connect valid executors
The graph is properly connected
Type compatibility between connected executors

build() -> Workflow

Returns

Type	Description
Workflow	An immutable Workflow instance ready for execution.

Exceptions

Type	Description
ValueError	If starting executor is not set.
WorkflowValidationError	If workflow validation fails (includes EdgeDuplicationError, TypeCompatibilityError, and GraphConnectivityError subclasses).

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class MyExecutor(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(text.upper())


   # Build and execute a workflow
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: MyExecutor(id="executor"), name="MyExecutor")
       .set_start_executor("MyExecutor")
       .build()
   )

   # The workflow is now immutable and ready to run
   events = await workflow.run("hello")
   print(events.get_outputs())  # ['HELLO']

   # Workflows can be reused multiple times
   events2 = await workflow.run("world")
   print(events2.get_outputs())  # ['WORLD']

register_agent

This method allows you to register a factory function that creates an agent. The agent will be instantiated and wrapped in an AgentExecutor only when the workflow is built, enabling deferred initialization and potentially reducing startup time.

register_agent(factory_func: Callable[[], AgentProtocol], name: str, agent_thread: AgentThread | None = None, output_response: bool = False) -> Self

Parameters

Name	Description
factory_func Required	Callable[[], AgentProtocol] A callable that returns an AgentProtocol instance when called.
name Required	str The name of the registered agent factory. This doesn't have to match the agent's internal name. But it must be unique within the workflow.
agent_thread	AgentThread \| None The thread to use for running the agent. If None, a new thread will be created when the agent is instantiated. Default value: None
output_response	bool Whether to yield an AgentRunResponse as a workflow output when the agent completes. Default value: False

Returns

Type	Description
Self

Examples


   from agent_framework import WorkflowBuilder
   from agent_framework_anthropic import AnthropicAgent


   # Build a workflow
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: ..., name="SomeOtherExecutor")
       .register_agent(
           lambda: AnthropicAgent(name="writer", model="claude-3-5-sonnet-20241022"),
           name="WriterAgent",
           output_response=True,
       )
       .add_edge("SomeOtherExecutor", "WriterAgent")
       .set_start_executor("SomeOtherExecutor")
       .build()
   )

register_executor

This method allows you to register a factory function that creates an executor. The executor will be instantiated only when the workflow is built, enabling deferred initialization and potentially reducing startup time.

register_executor(factory_func: Callable[[], Executor], name: str | list[str]) -> Self

Parameters

Name	Description
factory_func Required	Callable[[], Executor] A callable that returns an Executor instance when called.
name Required	str \| list[str] The name(s) of the registered executor factory. This doesn't have to match the executor's ID, but it must be unique within the workflow.

Returns

Type	Description
Self

Examples

If multiple names are provided, the same factory function will be registered under each name.


   from agent_framework import WorkflowBuilder, Executor, WorkflowContext, handler


   class LoggerExecutor(Executor):
       @handler
       async def log(self, message: str, ctx: WorkflowContext) -> None:
           print(f"Log: {message}")


   # Register the same executor factory under multiple names
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: LoggerExecutor(id="logger"), name=["ExecutorA", "ExecutorB"])
       .set_start_executor("ExecutorA")
       .add_edge("ExecutorA", "ExecutorB")
       .build()

set_max_iterations

Set the maximum number of iterations for the workflow.

When a workflow contains cycles, this limit prevents infinite loops by capping the total number of executor invocations. The default is 100 iterations.

set_max_iterations(max_iterations: int) -> Self

Parameters

Name	Description
max_iterations Required	int The maximum number of iterations the workflow will run for convergence.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class StepA(Executor):
       @handler
       async def process(self, count: int, ctx: WorkflowContext[int]) -> None:
           if count < 10:
               await ctx.send_message(count + 1)


   class StepB(Executor):
       @handler
       async def process(self, count: int, ctx: WorkflowContext[int]) -> None:
           await ctx.send_message(count)


   # Set a custom iteration limit for workflow with cycles
   workflow = (
       WorkflowBuilder()
       .set_max_iterations(500)
       .register_executor(lambda: StepA(id="step_a"), name="StepA")
       .register_executor(lambda: StepB(id="step_b"), name="StepB")
       .add_edge("StepA", "StepB")
       .add_edge("StepB", "StepA")  # Cycle
       .set_start_executor("StepA")
       .build()
   )

set_start_executor

Set the starting executor for the workflow.

The start executor is the entry point for the workflow. When the workflow is executed, the initial message will be sent to this executor.

set_start_executor(executor: Executor | AgentProtocol | str) -> Self

Parameters

Name	Description
executor Required	Executor \| AgentProtocol \| str The starting executor, which can be an Executor instance, AgentProtocol instance, or the name of a registered executor factory.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


   class EntryPoint(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(text.upper())


   class Processor(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(text)


   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: EntryPoint(id="entry"), name="EntryPoint")
       .register_executor(lambda: Processor(id="proc"), name="Processor")
       .add_edge("EntryPoint", "Processor")
       .set_start_executor("EntryPoint")
       .build()
   )

with_checkpointing

Enable checkpointing with the specified storage.

Checkpointing allows workflows to save their state periodically, enabling pause/resume functionality and recovery from failures. The checkpoint storage implementation determines where checkpoints are persisted.

with_checkpointing(checkpoint_storage: CheckpointStorage) -> Self

Parameters

Name	Description
checkpoint_storage Required	CheckpointStorage The checkpoint storage implementation to use.

Returns

Type	Description
Self	The WorkflowBuilder instance for method chaining.

Examples


   from typing_extensions import Never
   from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler
   from agent_framework import FileCheckpointStorage


   class ProcessorA(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
           await ctx.send_message(text.upper())


   class ProcessorB(Executor):
       @handler
       async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
           await ctx.yield_output(text)


   # Enable checkpointing with file-based storage
   storage = FileCheckpointStorage("./checkpoints")
   workflow = (
       WorkflowBuilder()
       .register_executor(lambda: ProcessorA(id="proc_a"), name="ProcessorA")
       .register_executor(lambda: ProcessorB(id="proc_b"), name="ProcessorB")
       .add_edge("ProcessorA", "ProcessorB")
       .set_start_executor("ProcessorA")
       .with_checkpointing(storage)
       .build()
   )

   # Run with checkpoint saving
   events = await workflow.run("input")

Feedback

Was this page helpful?