Agent Workflow API Reference

public interface Step<I, O> {
    String name();
    O execute(AgentContext ctx, I input);

    default AgentContext updateContext(AgentContext ctx, O output) { return ctx; }

    default Class<?> inputType()  { return Object.class; }
    default Class<?> outputType() { return Object.class; }

    static <I,O> Step<I,O> named(String name, BiFunction<AgentContext, I, O> fn);
    static <T> Step<T,T> noop();           // anonymous pass-through
    static <T> Step<T,T> noop(String name); // named pass-through — referenceable in IR
}

class ClassifierStep implements Step<String, ClassificationResult> {
    @Override
    public ClassificationResult execute(AgentContext ctx, String input) {
        return runModel(input);  // rich result carries all data
    }

    @Override
    public AgentContext updateContext(AgentContext ctx, ClassificationResult output) {
        return ctx.mutate()
            .with(CONFIDENCE, output.confidence())
            .with(DETECTED_LABEL, output.label())
            .build();
    }
}

public interface AgentStep { /* marker — no methods */ }

ClaudeStep.of("Fix the tests in {input}")
    .workingDirectory(projectPath)
    .permissionMode(PermissionMode.ACCEPT_EDITS)
    .withMcp(mcpConfig)
    .withA2a(a2aEndpoint);

Workflow.<I, O>define("name")
    .step(step)                                    // first step
    .then(step)                                    // chain
    .branch(predicate).then(a).otherwise(b)        // conditional
    .gather(step1, step2)                          // homogeneous fan-out → List<Object>
    .parallel(step1, step2)                        // enrichment fan-out → fork input unchanged, branches write to context
    .parallel(itemsSupplier, stepFactory)           // dynamic fan-out
    .repeatUntil(predicate).step(a).end()          // while-do
    .repeatUntilOutput(predicate).step(a).end()    // do-while
    .decision(chat).option("a", step).end()        // LLM routing
    .gate(gate).onPass(a).onFail(b).maxRetries(3).end()  // quality gate
    .onError(ExType.class, recovery)               // error recovery
    .run(input)                                     // execute
    .run(input, RunOptions.maxCost(5.0))           // with constraints
    .compile()                                      // → WorkflowGraph (IR only)
    .build()                                        // → Workflow (implements Step)

Workflow.sequential("name").step(a).then(b).run(input);
Workflow.loop("name").step(a).times(5).run(input);
Workflow.loop("name").step(a).until(predicate).run(input);
Workflow.parallel("name").step(a, b, c).run(input);
Workflow.supervisor("name", chat).agents(a, b).until(pred).run(input);

public interface Gate<O> {
    GateDecision evaluate(AgentContext ctx, O output);
}

.gate(new JudgeGate(jury, 0.8))
    .onPass(approveStep)          // required
    .onFail(reviseStep)           // optional
    .onTimeout(fallbackStep)      // optional
    .withReflector(reflectorStep) // Verdict → feedback text for retry
    .maxRetries(3)                // cap
    .end()

class AssessCodeQualityGate implements Gate<PrContext> {
    static final ContextKey<Assessment> QUALITY_ASSESSMENT = ContextKey.of("quality-assessment", Assessment.class);

    private Assessment assessment;

    @Override
    public GateDecision evaluate(AgentContext ctx, PrContext pr) {
        assessment = runJudge(pr);
        return assessment.score() >= threshold ? GateDecision.PASS : GateDecision.FAIL;
    }

    @Override
    public AgentContext updateContext(AgentContext ctx, PrContext pr) {
        return ctx.mutate().with(QUALITY_ASSESSMENT, assessment).build();
    }
}

// Read
ctx.get(AgentContext.ITERATION_COUNT)      // Optional<Integer>
ctx.require(AgentContext.ITERATION_COUNT)   // throws if missing

// Write (returns new context)
ctx.mutate().with(key, value).build()

AgentContext merged = base.mergeFrom(donor);

Workflow<PrContext, AssessmentResult> aiAssessment = Workflow.<PrContext, AssessmentResult>define("ai-assessment")
    .step(assessCodeQuality)    // updateContext() writes QUALITY_ASSESSMENT
    .then(assessBackport)       // updateContext() writes BACKPORT_ASSESSMENT
    .build();

Workflow.<Integer, Path>define("pr-review")
    .step(fetchPrContext)       // writes PR_CONTEXT to ctx
    .branch(skipAi)
        .then(skipStep)
        .otherwise(aiAssessment) // sub-workflow — context writes propagate back
    .then(qualityJudge)         // reads QUALITY_ASSESSMENT and BACKPORT_ASSESSMENT ✓
    .run(prNumber);

RunOptions.maxCost(5.0)
RunOptions.maxIterations(50)
RunOptions.maxDuration(Duration.ofMinutes(10))
RunOptions.unlimited()

// Chain
RunOptions.maxCost(5.0).withMaxIterations(50).withMaxDuration(Duration.ofMinutes(10))

WorkflowGraph<I, O> graph = workflow.compile();
graph.name()         // String
graph.nodes()        // List<WorkflowNode>
graph.edges()        // List<WorkflowEdge>
graph.startNode()    // String
graph.finishNode()   // String

// Typed steps declare their types
static class ClassifyStep implements Step<String, ClassificationResult> {
    @Override public Class<?> inputType()  { return String.class; }
    @Override public Class<?> outputType() { return ClassificationResult.class; }
    // ...
}

static class FormatResultStep implements Step<ClassificationResult, String> {
    @Override public Class<?> inputType()  { return ClassificationResult.class; }
    @Override public Class<?> outputType() { return String.class; }
    // ...
}

// Compile and check — no LLM calls needed
WorkflowGraph<String, Object> graph = Workflow.<String, Object>define("pipeline")
        .step(new ClassifyStep(chat))
        .then(new FormatResultStep())
        .compile();

WorkflowGraphAssert.assertTypeCompatible(graph);  // passes

.step(new ClassifyStep(chat))       // outputs ClassificationResult
.then(new IntegerConsumer())         // expects Integer — MISMATCH

// Throws:
// TypeIncompatibleException: step 'classify' outputs ClassificationResult
//   but step 'int-consumer' expects Integer

.step(new ClassifyStep(chat))                           // typed
.then(Step.named("transform", (ctx, in) -> in.toString()))  // untyped — skipped
.then(new IntegerConsumer())                             // no check — prior step untyped

class QualityGateStep implements Step<ReviewResult, Path> {
    @Override
    public Path execute(AgentContext ctx, ReviewResult result) {
        if (result.isTooRisky()) {
            // abort — caller receives the error report, not an exception
            throw new WorkflowAbortException(writeErrorReport(result));
        }
        return writeApprovalReport(result);
    }
}

@Bean StepRunner stepRunner() {
    return new LocalStepRunner();  // in-process, zero overhead
}

CheckpointManager manager = new CheckpointManager(readRepo, writeRepo);
manager.getRunState("run-1");        // inspect: what completed, what failed
manager.resetFailedSteps("run-1");   // delete FAILED records (only if transient)
executor.execute(graph, ctx, input); // retry — COMPLETED skipped, reset steps re-run

record StepTransition(
    String workflowRunId, String workflowName,
    String fromStep, String toStep,
    Instant timestamp, Duration stepDuration,
    long tokensUsed, double costUsd,
    NodeType nodeType, String label,
    String tracePath     // absolute path to JSONL trace file, or null
) {}

agent-workflow/
├── workflow-api/        # Step, AgentContext, ContextKey, Gate, RunOptions
├── workflow-core/       # WorkflowGraph, WorkflowExecutor, StepRunner, TraceRecorder
├── workflow-flows/      # Workflow DSL, ClaudeStep, ChatClientStep, AgentClientStep, A2AStep, JudgeGate
├── workflow-batch/      # CheckpointingStepRunner, JdbcTraceRecorder, JPA entities
├── workflow-temporal/   # TemporalStepRunner, StepActivityImpl
├── workflow-journal/    # JournalTraceRecorder, WorkflowStepEvent
├── workflow-tools/      # BashTool, ReadTool, WriteTool, EditTool, GlobTool, GrepTool
├── workflow-agents/     # AgentLoop
└── workflow-examples/   # Example pipelines