Java DSL cheatsheet

Quarkus Flow uses the CNCF Workflow Java Fluent DSL to define workflows in code. This page shows:

  • how to define a workflow class,

  • the complete set of chainable task providers from the DSL (the ones you pass directly into workflow(…​).tasks(…​)), and

  • how to shape data with transformations: inputFrom, outputAs, and exportAs.

Source of truth for the DSL classes:

  • Builder: io.quarkiverse.flow.dsl.FlowWorkflowBuilder

  • Shortcuts (task providers): io.quarkiverse.flow.dsl.FlowDSL (import statically)

  • Transformations: io.quarkiverse.flow.dsl.spi.FuncTransformations and io.quarkiverse.flow.dsl.spi.FuncTaskTransformations

  • HTTP / OpenAPI steps: io.quarkiverse.flow.dsl.FuncCallHttpStep, io.quarkiverse.flow.dsl.FuncCallOpenAPIStep

Define a workflow class (required)

Workflows are discovered at build time from CDI beans that extend io.quarkiverse.flow.Flow and override descriptor().

package org.acme;

import static io.quarkiverse.flow.dsl.FlowDSL.set;

import java.util.Map;

import jakarta.enterprise.context.ApplicationScoped;

import io.quarkiverse.flow.Flow;
import io.quarkiverse.flow.dsl.FlowWorkflowBuilder;
import io.serverlessworkflow.api.types.Workflow;

@ApplicationScoped
public class HelloWorkflow extends Flow {
    @Override
    public Workflow descriptor() {
        return FlowWorkflowBuilder.workflow("hello")
                // setting the workflow context with a map carrying a message.
                // it can be translated to JSON as { "message": "hello world!" }
                .tasks(set(Map.of("message", "hello world!")))
                .build();
    }
}

Setup (imports)

import io.serverlessworkflow.api.types.Workflow;
import io.quarkiverse.flow.dsl.FlowWorkflowBuilder;

// Static imports recommended for brevity:
import static io.quarkiverse.flow.dsl.FlowDSL.*;
import static io.quarkiverse.flow.dsl.FlowWorkflowBuilder.workflow;

Complete task providers (chainable)

All helpers below return a chainable provider/step (e.g., FuncTaskConfigurer, FuncCallStep, EmitStep, ListenStep, ConsumeStep, FuncCallHttpStep, FuncCallOpenAPIStep) so you can pass them straight into .tasks(…​):

Method Returns What it does Example

set(String expr)

FuncTaskConfigurer

Set/merge into data context using jq-style JSON expression.

set("{ message: \"hello\" }")

set(Map<String,Object> map)

FuncTaskConfigurer

Set/merge from a map (useful in Java without inline JSON).

set(Map.of("flag", true, "count", 0))

function(Function<T,R> fn)

FuncCallStep

Call a Java Function (input type inferred).

function(ai::answer).exportAs("answer")

function(Function<T,R> fn, Class<T> inClass)

FuncCallStep

Call a Java Function with explicit input type.

function(pricing::quote, QuoteRequest.class)

function(String name, Function<T,R> fn)

FuncCallStep

Named Java function call (input inferred).

function("quote", pricing::quote)

function(String name, Function<T,R> fn, Class<T> inClass)

FuncCallStep

Named + explicit input type.

function("classify", nlp::classify, Text.class)

withContext(JavaContextFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Call with workflow context (payload, ctx) → result.

withContext((payload, ctx) → svc.do(ctx, payload), Input.class)

withContext(String name, JavaContextFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Named context-aware call.

withContext("withCtx", svc::withCtx, Input.class)

withInstanceId(InstanceIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Call with workflow instance id (instanceId, payload) → result.

withInstanceId((id,p) → agent.run(id,p), Prompt.class)

withInstanceId(String name, InstanceIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Named (instanceId, payload) call.

withInstanceId("agent", agent::run, Prompt.class)

withFilter(JavaFilterFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Call that can see both workflow and task context (payload, wctx, tctx) → result.

withFilter((payload, wctx, tctx) → audit(payload, wctx, tctx), Event.class)

withFilter(String name, JavaFilterFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Named context+task-aware call.

withFilter("auditFilter", auditFn, Event.class)

withUniqueId(UniqueIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Call with a stable unique id (uniqueId, payload) → result, where uniqueId = "<instanceId>-<taskJsonPointer>".

withUniqueId((uid,p) → tool.call(uid,p), Prompt.class)

withUniqueId(String name, UniqueIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Named unique-id-aware call.

withUniqueId("toolCall", (uid,p) → tool.call(uid,p), Prompt.class)

agent(UniqueIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Sugar for “agent-style” calls that need a memory id (the unique id described above).

agent(criticAgent::critique, String.class)

agent(String name, UniqueIdBiFunction<T,R> fn, Class<T> inClass)

FuncCallStep

Named agent call.

agent("criticAgent", criticAgent::critique, String.class)

consume(Consumer<T> consumer, Class<T> inClass)

ConsumeStep

Fire-and-forget side-effect. No data is exported; only effects.

consume(msg → mail.send(msg), Email.class)

consume(String name, Consumer<T> consumer, Class<T> inClass)

ConsumeStep

Named side-effect.

consume("audit", audit::log, Event.class)

emit(Consumer<FuncEmitTaskBuilder> cfg)

EmitStep

Low-level emit using the CloudEvent builder.

emit(e → e.type("org.acme.created"))

emit(String name, Consumer<FuncEmitTaskBuilder> cfg)

EmitStep

Named low-level emit.

emit("publish", e → e.type("org.acme.created"))

emit(String type, Function<T,CloudEventData> bodyFn)

EmitStep

Emit event; body encoded via T → CloudEventData.

emit("org.acme.ready", DataMappers::toCloudEvent)

emit(String name, String type, Function<T,CloudEventData> bodyFn)

EmitStep

Named variant.

emit("notify","org.acme.ready", DataMappers::toCloudEvent)

emit(String type, Function<T,byte[]> serializer, Class<T> inClass)

EmitStep

Custom bytes payload from T → byte[] (no explicit ObjectMapper).

emit("org.acme.raw", Ser::toBytes, Payload.class)

emit(String name, String type, Function<T,byte[]> serializer, Class<T> inClass)

EmitStep

Named custom bytes emit.

emit("binaryOut","org.acme.raw", Ser::toBytes, Payload.class)

emitJson(String type, Class<T> inClass)

EmitStep

Emit JSON CloudEvent for POJO input (content-type application/json).

emitJson("org.acme.review.required", Review.class)

emitJson(String name, String type, Class<T> inClass)

EmitStep

Named JSON emit.

emitJson("draftReady","org.acme.newsletter.review", Review.class)

listen(FuncListenSpec spec)

ListenStep

Listen using a spec (e.g., toOne, toAny, toAll). Chain .first() on the spec to extract the first event from the collection.

listen(toOne("org.acme.done").first())

listen(String name, FuncListenSpec spec)

ListenStep

Named listen. Chain .first() to unwrap, .envelope().first() to keep the full CE.

listen("waitHuman", toOne("org.acme.review.done").first())

switchCase(Consumer<FuncSwitchTaskBuilder> switchConsumer)

FuncTaskConfigurer

Low-level switch (builder consumer).

switchCase(sw → sw.on(c → c.when(".flag").then("ok")).onDefault("fallback"))

switchCase(String taskName, Consumer<FuncSwitchTaskBuilder> switchConsumer)

FuncTaskConfigurer

Named low-level switch.

switchCase("route", sw → sw.on(…​).onDefault(…​))

switchCase(SwitchCaseConfigurer…​ cases)

FuncTaskConfigurer

Switch composed from caseOf(…​) and caseDefault(…​).

switchCase(cases(caseOf(p).then("A"), caseDefault("B")))

switchCase(String taskName, SwitchCaseConfigurer…​ cases)

FuncTaskConfigurer

Named switch composed from cases.

switchCase("route", caseOf(p).then("A"), caseDefault("B"))

switchWhen(Predicate<T> pred, String thenTask, Class<T> predClass)

FuncTaskConfigurer

Typed single-case jump.

switchWhen((HumanReview h) → h.needsRevision(), "revise", HumanReview.class)

switchWhen(String jqExpr, String thenTask)

FuncTaskConfigurer

JQ expression single-case jump.

switchWhen(".score >= 80", "send")

switchWhenOrElse(Predicate<T> pred, String thenTask, String otherwiseTask, Class<T> predClass)

FuncTaskConfigurer

Typed single-case + default task.

switchWhenOrElse(h → !ok(h), "fix", "abort", HumanReview.class)

switchWhenOrElse(Predicate<T> pred, String thenTask, FlowDirectiveEnum otherwiseDirective, Class<T> predClass)

FuncTaskConfigurer

Typed single-case + default directive (END, CONTINUE, …​).

switchWhenOrElse(h → ok(h), "ok", FlowDirectiveEnum.END, HumanReview.class)

switchWhenOrElse(String jqExpr, String thenTask, String otherwiseTask)

FuncTaskConfigurer

JQ single-case + default task.

switchWhenOrElse(".approved", "ship", "reject")

switchWhenOrElse(String jqExpr, String thenTask, FlowDirectiveEnum otherwiseDirective)

FuncTaskConfigurer

JQ single-case + directive.

switchWhenOrElse(".approved", "ship", FlowDirectiveEnum.END)

forEach(Function<T,Collection<?>> collectionFn, Consumer<FuncTaskItemListBuilder> body)

FuncTaskConfigurer

Iterate over a collection computed from the current input.

forEachOrder o) → o.items(), inner → inner.tasks(set("{ item: . }")

forEach(Collection<?> collection, Consumer<FuncTaskItemListBuilder> body)

FuncTaskConfigurer

Iterate over a constant collection.

forEach(List.of(1,2,3), inner → inner.tasks(set("{ n: . }")))

forEach(List<E> list, Consumer<FuncTaskItemListBuilder> body)

FuncTaskConfigurer

Iterate over a constant list (convenience).

forEach(List.of(user1,user2), inner → inner.tasks(set("{ user: . }")))

tasks(FuncTaskConfigurer…​ steps)

Consumer<FuncTaskItemListBuilder>

Group multiple steps/configurers (useful inside forEach).

tasks(set("{x:1}"), function(svc::op), emitJson("org.acme.done", Result.class))

http()

FuncCallHttpStep

Start a fluent HTTP call spec (method, headers, body, etc.). Pass it directly to .tasks(…​) or through call("name", http(…​)) to force a name.

http().get().endpoint("https://service/api").acceptJSON()

http(String name)

FuncCallHttpStep

Named HTTP call spec.

http("fetchUsers").get().endpoint("https://service/users")

http(String urlExpr, AuthenticationConfigurer auth)

FuncCallHttpStep

HTTP spec preconfigured with endpoint expression + auth.

http("http://service/api", auth → auth.use("my-auth")).get()

http(URI url, AuthenticationConfigurer auth)

FuncCallHttpStep

HTTP spec preconfigured with a concrete URI + auth.

http(URI.create("https://service/api"), auth → auth.use("default"))

get(String endpoint)

FuncCallHttpStep

Convenience HTTP GET spec (unnamed) for a string endpoint.

get("https://service/health")

get(String name, String endpoint)

FuncCallHttpStep

Named GET spec for a string endpoint.

get("checkHealth", "https://service/health")

get(String endpoint, AuthenticationConfigurer auth)

FuncCallHttpStep

GET with auth (unnamed).

get("https://service/users", auth → auth.use("users-auth"))

get(String name, String endpoint, AuthenticationConfigurer auth)

FuncCallHttpStep

Named GET with auth.

get("fetchUsers", "https://service/users", auth → auth.use("users-auth"))

get(URI endpoint)

FuncCallHttpStep

GET for a concrete URI (unnamed).

get(URI.create("https://service/health"))

get(String name, URI endpoint)

FuncCallHttpStep

Named GET for a concrete URI.

get("health", URI.create("https://service/health"))

get(URI endpoint, AuthenticationConfigurer auth)

FuncCallHttpStep

GET for a URI with auth (unnamed).

get(URI.create("https://service/api"), auth → auth.use("api-auth"))

get(String name, URI endpoint, AuthenticationConfigurer auth)

FuncCallHttpStep

Named GET for a URI with auth.

get("fetchApi", URI.create("https://service/api"), auth → auth.use("api-auth"))

post(Object body, String endpointExpr)

FuncCallHttpStep

POST spec (unnamed) with body and string endpoint.

post(Map.of("name","Ricardo"), "https://service/users")

post(String name, Object body, String endpointExpr)

FuncCallHttpStep

Named POST spec.

post("createUser", user, "https://service/users")

post(Object body, String endpointExpr, AuthenticationConfigurer auth)

FuncCallHttpStep

POST with body + auth (unnamed).

post(payload, "https://service/users", auth → auth.use("users-auth"))

post(String name, Object body, String endpointExpr, AuthenticationConfigurer auth)

FuncCallHttpStep

Named POST with body + auth.

post("createUser", payload, "https://service/users", auth → auth.use("users-auth"))

call(FuncCallHttpStep spec)

FuncTaskConfigurer

Attach an HTTP spec as a task (unnamed or using its internal name).

call(http().get().endpoint("https://service/ping"))

call(String name, FuncCallHttpStep spec)

FuncTaskConfigurer

Attach an HTTP spec as a named task (overrides internal name).

call("pingService", http().get().endpoint("https://service/ping"))

call(FuncCallHttpConfigurer configurer)

FuncTaskConfigurer

Low-level HTTP call based on a builder configurer.

call(b → b.get().endpoint("https://service/ping"))

call(String name, FuncCallHttpConfigurer configurer)

FuncTaskConfigurer

Named low-level HTTP call.

call("pingService", b → b.get().endpoint("https://service/ping"))

openapi()

FuncCallOpenAPIStep

Start a fluent OpenAPI call spec (document, operation, params, auth, etc.).

openapi().document("https://petstore3.swagger.io/api/v3/openapi.json").operation("findPetById")

openapi(String name)

FuncCallOpenAPIStep

Named OpenAPI spec.

openapi("findPet").document("…​").operation("findPetById")

call(FuncCallOpenAPIStep spec)

FuncTaskConfigurer

Attach an OpenAPI spec as a task (unnamed or using its internal name).

call(openapi().document("…​").operation("findPetById"))

call(String name, FuncCallOpenAPIStep spec)

FuncTaskConfigurer

Attach an OpenAPI spec as a named task (overrides internal name).

call("fetchPet", openapi().document("…​").operation("findPetById"))

call(FuncCallOpenAPIConfigurer configurer)

FuncTaskConfigurer

Low-level OpenAPI call based on a builder configurer.

call(b → b.document("…​").operation("findPetById"))

call(String name, FuncCallOpenAPIConfigurer configurer)

FuncTaskConfigurer

Named low-level OpenAPI call.

call("fetchPet", b → b.document("…​").operation("findPetById"))
Helpers that help you build specs for emit/listen/switch/HTTP/OpenAPI but are not task providers themselves:

  • Events & listen: event(…​), eventJson(…​), eventBytes(…​), to(), toOne(…​), toAny(…​), toAll(…​). Call .first() on a toOne(…​) spec to extract the first element from the event collection (the spec always delivers events as a collection). Chain .envelope().first() to keep the full CloudEvent envelope instead of just the data payload.

  • Schedule events: one(…​), allOfType(…​), on(…​). Call .first() on one(…​) to auto-extract the first CloudEvent data via inputFrom. Chain .envelope().first() to keep the full envelope.

  • Switch helpers: cases(…​), caseOf(…​), caseDefault(…​)

You pass the resulting steps/specs (e.g. listen(…​), emitJson(…​), openapi(), http(), get(…​), post(…​)) into .tasks(…​).

Data flow and transformations

Quarkus Flow follows the Serverless Workflow Specification data-flow model: every step can shape what it extracts as input, how it transforms its direct result, and what it merges back into the global context.

This section is only a quick reference. For a full explanation, context-aware Java functions, and more examples, see Data flow and context management.

inputFrom(…​) — what the task extracts from the context

Choose which slice of the global workflow context a task receives as input.

  • jq:

    function(pricing::quote, QuoteRequest.class)
  .inputFrom("$.cart.quoteRequest")

  • Java (with domain type):

    function(pricing::quote, QuoteRequest.class)
  .inputFrom((MyCheckout ctx) -> ctx.quoteRequest(), MyCheckout.class)

outputAs(…​) — transforming the direct task result

Shape the direct result returned by the task before it is merged back into the workflow context.

  • jq:

    function(nlp::classify, Text.class)
  .outputAs("{ sentiment: ., reviewed: false }")

  • Java:

    agent("investmentAnalyst", analyst::analyse, InvestmentMemo.class)
  .outputAs(memo -> Map.of("memo", memo), InvestmentMemo.class);

exportAs(…​) — merging back into the global context

Dictate how the task’s (optionally transformed) output is merged into the global workflow context for downstream tasks to use.

  • Java:

    agent("draftNewsletter", drafter::draft, Draft.class)
  .exportAs((draft, wfContext) -> {
      var current = (NewsletterContext) wfContext.currentData();
      return current.withDraft(draft);
  }, Draft.class);

End-to-end example (agentic + HITL + transforms)

Workflow w = workflow("intelligent-newsletter")
  .tasks(
    // 1) Draft Agent: Extracts the prompt, generates draft, merges back to context
    agent("draftAgent", drafterAgent::draft, String.class)
      .inputFrom("$.seedPrompt")
      .exportAs("{ draft: . }"),

    // 2) Critique Agent: Extracts only the draft, returns a composite status
    agent("criticAgent", criticAgent::critique, String.class)
      .inputFrom("$.draft")
      .exportAs(r -> Map.of("review", r, "status", r.needsRevision() ? "REVISION" : "OK")),

    // 3) Emit review request for human oversight
    emitJson("org.acme.newsletter.review.required", CriticAgentReview.class),

    // 4) Wait for human review; extract the first event from the collection
    listen("waitHumanReview", toOne("org.acme.newsletter.review.done").first()),

    // 5) Branch: revision loop or final send based on human input
    switchWhenOrElse(
      (HumanReview h) -> ReviewStatus.NEEDS_REVISION.equals(h.status()),
      "draftAgent",
      "sendNewsletter",
      HumanReview.class
    ),

    // 6) Final side-effect
    consume("sendNewsletter",
      (HumanReview reviewedDraft) ->
        mailService.send("subscribers@acme.finance.org", "Weekly Newsletter", reviewedDraft.draft()),
      HumanReview.class
    )
  )
  .build();

Tips

  • Prefer static imports: workflow, set, function, withContext, withFilter, withUniqueId, agent, emitJson, listen, switchWhenOrElse, consume, to, toOne, event, on, one, http, get, post, openapi, call.

  • Name tasks you branch to: stable task names make switchWhen* targets explicit and work well with withUniqueId/agent memory ids.

  • Keep transformations close to the step that needs them (inputFrom, exportAs, outputAs) for readability.

  • HTTP/OpenAPI: you can either:

    • use fluent specs directly as steps: tasks(http().get().endpoint("…​")), or

    • wrap them with call("name", http().get()…​) / call("name", openapi().operation("…​")) to force an explicit task name.

  • Remember: the workflow context is often JSON; your domain objects are (de)serialized via Jackson, so you can keep typed payloads in your steps while still using jq for quick projections.

See also