Building Agentic Workflows

This guide walks through the steps of building an agentic workflow in Microbus - from defining tasks and wiring them into a graph, to running and controlling flows via the Foreman.

Step 1: Define the Tasks

Each step of a workflow is a task endpoint. Use the coding agent to add tasks to your microservice:

Add a task “VerifyCredit” that receives a creditScore int and returns creditVerified bool. It should approve scores of 550 or higher.

A task endpoint receives ctx context.Context and flow *workflow.Flow as the first two parameters, followed by input arguments sourced from the workflow’s shared state. Output arguments are written back to the state automatically.

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, creditScore int) (creditVerified bool, err error) {
	return creditScore >= 550, nil
}

Tasks are standalone and have no knowledge of the workflow graph. They can be tested individually using the generated Executor in the API package:

t.Run("good_score", func(t *testing.T) {
	assert := testarossa.For(t)
	creditVerified, err := exec.VerifyCredit(ctx, 750)
	assert.Expect(creditVerified, true, err, nil)
})

Step 2: Define the Workflow Graph

A workflow graph endpoint returns a *workflow.Graph that wires tasks together with transitions. Use the coding agent to create it:

Add a workflow “CreditApproval” that defines the graph for the credit approval process. Inputs: applicant Applicant. Outputs: approved bool.

Tasks are registered under local names with AddTask, then wired together by name with transitions:

func (svc *Service) CreditApproval(ctx context.Context) (graph *workflow.Graph, err error) {
	graph = workflow.NewGraph("CreditApproval")
	graph.SetEndpoint("Submit", myserviceapi.SubmitApplication.URL())
	graph.SetEndpoint("Verify", myserviceapi.VerifyCredit.URL())
	graph.SetEndpoint("Decide", myserviceapi.Decision.URL())
	graph.AddTransition("Submit", "Verify")
	graph.AddTransition("Verify", "Decide")
	graph.AddTransition("Decide", workflow.END)
	return graph, nil
}

Names are local to the graph and arbitrary, but must be internally consistent: AddTransition, flow.Goto, and SetFanIn all reference a task by the name it was given in AddTask — not by its endpoint URL. Newly scaffolded graphs use PascalCase names (Submit, VerifyCredit) by convention. The same task can be registered twice under different names to place it at distinct positions in the topology with their own transitions.

Conditional Transitions

Route to different tasks based on state values:

graph.AddTransitionWhen("Verify", "ManualReview", "!creditVerified")
graph.AddTransitionWhen("Verify", "Decide", "creditVerified")

When predicates are evaluated independently and all matching ones fire, so multiple When transitions from one node fan out in parallel. When you want exactly one branch to run, reach for Switch below instead.

Switch Transitions

AddTransitionSwitch is first-match-wins routing: the Foreman evaluates the branches in registration order and follows the first whose expression matches; the rest are skipped. Because only one branch ever runs, no SetFanIn is needed.

graph.AddTransitionSwitch("Router", "HighValue", "amount >= 10000")
graph.AddTransitionSwitch("Router", "MidValue",  "amount >= 1000")
graph.AddTransitionSwitch("Router", "LowValue",  "true") // default catch-all

Switch is the safest routing primitive — the validator guarantees the branches are mutually exclusive, so two can’t accidentally both fire and leak a parallel branch. Use it for any routing where exactly one branch should run; the two mutually-exclusive When predicates above (creditVerified / !creditVerified) are better written as a Switch. Reserve When for genuine parallel fan-out. A node that uses Switch must declare every success-path transition as Switch.

Fan-Out and Fan-In

Any task with two or more outgoing transitions is a fan-out source - the Foreman runs every outgoing branch in parallel. Every fan-out must be fanned back in at a node marked with SetFanIn. The Foreman waits at that node until all sibling branches arrive, merges their state, then continues.

// Fan-out: submit fans out to three parallel verifications
graph.AddTransition("Submit", "VerifyCredit")
graph.AddTransition("Submit", "VerifyIdentity")
graph.AddTransition("Submit", "VerifyEmployment")
// Fan-in: all three converge on decide
graph.SetFanIn("Decide")
graph.AddTransition("VerifyCredit", "Decide")
graph.AddTransition("VerifyIdentity", "Decide")
graph.AddTransition("VerifyEmployment", "Decide")

SetFanIn marks Decide as the node where the parallel branches fan back in. Graph validation enforces the pairing: every fan-out must be fanned back in before a branch can reach workflow.END, otherwise the graph fails to validate.

When parallel branches write to the same state field, wire a reducer explicitly at graph-build time with graph.SetReducer(field, reducer) — ReducerAdd to sum, ReducerAppend to append, ReducerUnion to dedupe, and so on. A field with no registered reducer replaces (last write wins). Tasks must write only the delta their branch produces, not the full accumulated value, or fan-in produces duplicates. The full reducer vocabulary, the delta rule with worked examples, and the edge cases are covered separately.

graph.SetReducer("failures", workflow.ReducerAppend) // each branch appends its own failures

Dynamic Fan-Out

When the number of parallel branches depends on runtime data, use forEach to iterate over a state array:

// Spawn one VerifyEmployment task per employer in the "employers" array.
// Each instance receives the current element as "employerName" in state.
graph.AddTransitionForEach("Submit", "VerifyEmployment", "employers", "employerName")
graph.SetFanIn("Decide")
graph.AddTransition("VerifyEmployment", "Decide")

A forEach transition counts as a fan-out source, so the dynamically-spawned branches must converge on a node marked with SetFanIn - same rule as static fan-out. If the array is empty, no tasks are spawned for that transition. When a forEach transition is the only outgoing transition from a task, an empty array causes the flow to complete at that point - downstream tasks (including the fan-in target) are never reached.

Goto Transitions

A goto transition is only taken when the task explicitly calls flow.Goto(). This enables loops and dynamic routing:

graph.AddTransitionGoto("Review", "RequestMoreInfo")
graph.AddTransition("RequestMoreInfo", "Review") // Loop back
graph.AddTransition("Review", "Decide")          // Normal exit

func (svc *Service) ReviewCredit(ctx context.Context, flow *workflow.Flow, creditScore int, attempts int) (err error) {
	if creditScore < 580 && attempts < 3 {
		flow.Goto("RequestMoreInfo")
	}
	return nil
}

Error Transitions

An error transition is taken when the source task returns an error, diverting the flow to a handler rather than failing the whole flow. The error is serialized into the handler’s state under the key onErr so it can inspect what happened.

graph.AddTransitionOnError("VerifyCredit", "HandleVerifyError")

AddTransitionOnTimeout is a specialized error transition that matches only when the task fails with HTTP status 408 - the time-budget expired before the task returned. When a task has both, the timeout transition wins on a timeout and the plain error transition catches everything else.

graph.AddTransitionOnTimeout("VerifyCredit", "FallbackProvider")
graph.AddTransitionOnError("VerifyCredit", "HandleVerifyError")

To retry the same task on timeout rather than divert to a handler, call flow.Retry(initialDelay, multiplier, maxIntervalDelay, giveUpAfter) in the task body, gated on the 408 status — if errors.StatusCode(err) == http.StatusRequestTimeout && flow.Retry(...). The bound is wall-clock: giveUpAfter is the elapsed time since the step was first created, and giveUpAfter <= 0 retries forever (zero delays make it immediate).

Subgraphs

To invoke another workflow as a child, a task calls flow.Subgraph(url, in, &out) from its body — there is no static subgraph node in the graph. The calling task is registered like any other with AddTask / AddTransition. A subgraph invokes a whole workflow, never a bare task; it is usually reached through the typed NewSubgraph client, whose call shape is shown under Dynamic Subgraph in Control Signals.

Time Budgets

A task’s execution timeout is declared on the task endpoint, not the graph, via sub.TimeBudget. The foreman’s TimeBudget config is the hard ceiling on every dispatch; the effective deadline is the smaller of the two.

Step 3: Add the Foreman to Your App

The Foreman core service must be included in any application that runs workflows.

app.Add(
	foreman.NewService(),
)

Configure the database connection in config.yaml:

foreman.core:
  SQLDataSourceName: root:root@tcp(127.0.0.1:3306)/my_database

In tests, the Foreman automatically uses an in-memory SQLite database - no configuration needed.

Step 4: Run a Workflow

Use the Foreman’s client to run a flow:

client := foremanapi.NewClient(svc)

// Create a flow with initial state. There is no separate start step —
// Create immediately runs the flow and returns the running flow's key.
// The trailing nil is an optional *workflow.FlowOptions (priority / fairness).
flowID, err := client.Create(ctx, myserviceapi.CreditApproval.URL(), map[string]any{
	"applicant": applicant,
}, nil)

// Wait for completion. Await returns a *workflow.FlowOutcome:
// outcome.Status is the terminal status, outcome.State the final state.
outcome, err := client.Await(ctx, flowID)

To defer the start, have the entry task call flow.Interrupt and Resume it when ready, rather than holding the flow in a created state.

Run is the synchronous equivalent — it creates, runs, and blocks until the flow stops, returning a *workflow.FlowOutcome with the terminal status and state:

outcome, err := client.Run(ctx, myserviceapi.CreditApproval.URL(), map[string]any{
	"applicant": applicant,
}, nil)

To run a flow at a non-default priority or under a fairness key, pass a *workflow.FlowOptions as the last argument to Create or Run instead of nil:

outcome, err := client.Run(ctx, myserviceapi.CreditApproval.URL(), map[string]any{
	"applicant": applicant,
}, &workflow.FlowOptions{Priority: 1})

The generated *api package also exposes a typed executor (exec.CreditApproval(ctx, applicant)). It is a testing convenience for driving a workflow end-to-end from service_test.go — production code submits the flow to the Foreman via the client as shown above.

Polling for Completion

A flow can outlast a single request’s time budget — an LLM tool-calling loop, a long human-in-the-loop pause, a slow downstream. Blocking on Await or Run would tie up the request until the flow stops, which may be longer than the caller has. Instead, launch the flow with Create and long-poll with Poll:

flowID, err := client.Create(ctx, myserviceapi.CreditApproval.URL(), map[string]any{
	"applicant": applicant,
}, nil)
// ... handle err; store flowID ...

// Later, from a status endpoint the browser (or caller) polls:
outcome, err := client.Poll(ctx, flowID)
// ... handle err ...
if !outcome.Stopped() {
	// Still running — answer "running" and let the caller re-poll immediately.
}

Poll waits up to the request’s time budget for the flow to stop, but a timeout is not an error: a still-running flow returns a non-error outcome with outcome.Stopped() == false, so the caller can answer within its own budget and re-poll with no client-side delay. A completed flow’s outcome is delivered the instant it finishes. PollAndParse additionally unmarshals the terminal state into a typed struct. This is the launch-then-poll pattern the bank support example uses to bridge an open-ended agent to a bounded HTTP request.

Control Signals

Tasks can influence the flow’s execution using control signals on the *workflow.Flow carrier.

Retry

Re-execute the current task on the next pass with exponential backoff, preserving changes made so far. Useful for polling or incremental progress. flow.Retry(initialDelay, multiplier, maxIntervalDelay, giveUpAfter) returns true while the step is within its wall-clock retry horizon and false once giveUpAfter of elapsed time is spent — the task writes the retryable condition explicitly.

func (svc *Service) PollStatus(ctx context.Context, flow *workflow.Flow, jobID string) (jobResult string, err error) {
	status, result, err := svc.checkJob(ctx, jobID)
	if err != nil {
		return "", errors.Trace(err)
	}
	if status == "running" {
		if !flow.Retry(30*time.Second, 1, 30*time.Second, time.Hour) { // 30s interval, give up after 1h
			return "", errors.New("job did not complete within poll budget")
		}
		return "", nil // retry scheduled; the foreman re-runs this task after the delay
	}
	return result, nil
}

Interrupt

Pause the flow for external input. The caller reads the interrupt payload via Snapshot or Poll and resumes the flow with Resume.

func (svc *Service) VerifySSN(ctx context.Context, flow *workflow.Flow, ssn string) (ssnVerified bool, err error) {
	if ssn == "" {
		var resume map[string]any
		yield, err := flow.Interrupt(map[string]any{"request": "ssn"}, &resume)
		if yield {
			return false, nil // parked until Resume provides the SSN
		}
		if err != nil {
			return false, errors.Trace(err)
		}
		ssn, _ = resume["ssn"].(string)
	}
	ssnVerified = isValidSSN(ssn)
	return ssnVerified, nil
}

The caller resumes the flow by providing the missing data:

err = client.Resume(ctx, flowID, map[string]any{"ssn": "123-45-6789"})

Sleep

Delay the next step by a duration. Useful for rate limiting or waiting for external state to settle.

flow.Sleep(5 * time.Minute)

Dynamic Subgraph

flow.Subgraph(url, in, &out) launches a child workflow from a task body — the only way one workflow invokes another. The step parks until the child terminates, then the task is re-executed. The call returns (yield, err) and unmarshals the child’s final state into the out pointer: yield is true on the first (parking) pass, and err carries a child failure.

func (svc *Service) RunChild(ctx context.Context, flow *workflow.Flow, applicantName string) (verified bool, err error) {
    var out otherserviceapi.SomeWorkflowOut
    yield, err := flow.Subgraph(otherserviceapi.SomeWorkflow.URL(), otherserviceapi.SomeWorkflowIn{
        ApplicantName: applicantName, // only the fields you set cross into the child
    }, &out)
    if yield {
        return false, nil // first pass: parked, child running
    }
    if err != nil {
        return false, errors.Trace(err) // child failed; retry, route, or propagate
    }
    verified = out.Verified // re-entry: adopt what you want from the child's typed outputs
    return verified, nil
}

Only the explicit in map crosses into the child as its complete initial state — the parent’s state is not auto-inherited. A nil input starts the child empty; pass flow.Snapshot() to hand over the parent’s full state. The child’s final state comes back as out, not merged into the parent — the calling task adopts the fields it wants. Because a child failure surfaces as the err return, the task can flow.Retry (re-running the child), route, or propagate rather than only cascading a flow failure.

A subgraph always invokes a whole workflow — a task is only ever a node in a graph, so to call a single task on its own you declare a one-node workflow around it and invoke that. In practice you rarely write the raw call: each microservice’s generated *api package exposes a typed NewSubgraph(flow) client that wraps flow.Subgraph with typed In/Out per workflow endpoint:

verified, yield, err := otherserviceapi.NewSubgraph(flow).SomeWorkflow(ctx, applicantName)

Each method mirrors the workflow’s typed arguments with a yield bool before err. (The generated Executor client looks similar but is test-only — never call it, or foremanapi, from a task body.)

If the child workflow interrupts, the interrupt propagates up through the parent chain to the root flow. A task must set at most one control signal - setting both Subgraph and another signal (Goto, Retry, Interrupt) will fail the step.

Debugging

History

Retrieve the step-by-step execution history of a flow:

steps, err := client.History(ctx, flowID)

Snapshot returns the flow’s current outcome, and Step(stepKey) returns a single step’s full payload — state, changes, status, error, and timings — for inspecting state mid-flow.

Fork

A terminated flow (completed, failed, or cancelled) is immutable — it is never re-run in place. To recover or explore, Fork clones the flow’s prefix up to a chosen step into a new, self-contained running flow and re-executes from that step, optionally with state overrides applied to it. The original is never modified:

// Re-run from a chosen step (its key comes from History) with an edit that lets it succeed.
newFlowKey, err := client.Fork(ctx, step.StepKey, map[string]any{
	"creditScore": 800, // override a state field on the forked step
})

The fork point may be any recorded step, including one inside a subgraph; the clone re-runs from there and bubbles back up to the root. The fork inherits the origin flow’s scheduling and baggage. To recover a partially-failed fan-out, fork one failed branch at a time.