diff --git a/asyncapi-call-plan.md b/asyncapi-call-plan.md
new file mode 100644
index 000000000..75ee298c3
--- /dev/null
+++ b/asyncapi-call-plan.md
@@ -0,0 +1,488 @@
+# AsyncAPI Call Implementation Plan
+
+## Top-Level Overview
+
+Implement the `call: asyncapi` task type in the `impl/asyncapi` module, as defined
+by the [DSL reference specification](https://github.com/open-workflow-specification/specification/blob/main/dsl-reference.md#asyncapi-call).
+
+The module skeleton already exists (`AsyncAPIExecutor`, `AsyncAPIExecutorBuilder`,
+`AsyncAPIReader`) but contains no real logic — `accept()` returns `false` and
+`init()` returns `null`.
+
+An AsyncAPI call supports two operations:
+- **Publish**: send a message to a channel using an operation from an AsyncAPI document.
+- **Subscribe**: consume one or more messages from a channel, optionally running a
+  per-message task list (`foreach`) and stopping according to a consumption policy
+  (`amount`, `while`, or `until`), with an optional `for` timeout.
+
+The implementation follows the same `CallableTaskBuilder` / `CallableTask` SPI pattern
+used by the HTTP, gRPC, and OpenAPI call executors.
+
+**Modules touched**: `impl/asyncapi` only. No changes to `impl/core`, `types`, `api`,
+or any other module.
+
+**Non-goals**:
+- Implementing a new messaging protocol transport from scratch. The executor will
+  delegate to a pluggable `AsyncApiChannelProvider` SPI, similar to how
+  `HttpClientResolver` delegates to a pluggable JAX-RS `Client`.
+- Supporting all 19 AsyncAPI protocols in a single PR. The SPI design enables providers
+  to be added independently.
+
+---
+
+## Architecture Overview
+
+```
+CallAsyncAPI (task type)
+    ↓
+AsyncAPIExecutorBuilder.accept(CallAsyncAPI.class) → true
+AsyncAPIExecutorBuilder.init(task, definition, position) → CallableTaskFactory
+    ↓ reads AsyncApiArguments:
+      - document (ExternalResource → AsyncAPI spec)
+      - operation (operation id, v3.0.0)
+      - channel (channel name, v2.6.0)
+      - server / protocol (target server selection)
+      - message (optional, payload + headers → publish)
+      - subscription (optional, consume policy + foreach → subscribe)
+      - authentication (optional)
+    ↓
+AsyncAPIExecutor (implements CallableTask)
+    ↓ dispatch
+    ├── if message present → PUBLISH path → AsyncApiChannelProvider.publish(...)
+    └── if subscription present → SUBSCRIBE path → AsyncApiChannelProvider.subscribe(...)
+                                    ↓ per-message foreach task list
+                                    ↓ consumption policy (amount / while / until)
+                                    ↓ optional timeout (consume.for)
+```
+
+---
+
+## Sub-Tasks
+
+### Sub-Task 0 — AsyncAPI document model and parsing
+
+**Status**: [ ] pending
+
+**Intent**
+The executor needs to parse AsyncAPI documents (both v2.6.0 and v3.0.0) to extract
+server URLs, channel names, and operation details. Following the OpenAPI pattern
+(`UnifiedOpenAPI` + `UnifiedOpenAPIReader` + `JacksonUnifiedOpenAPIReader`), we define
+a lightweight unified model and a Jackson-based reader — no external parser library needed.
+
+**Expected Outcomes**
+- A `UnifiedAsyncAPI` record (or set of records) capturing the fields needed by the
+  executor: servers (name → url + protocol + variables), channels, and operations.
+- The model supports both AsyncAPI v2.6.0 (`channels.{name}.publish`/`subscribe`) and
+  v3.0.0 (`operations.{name}` referencing a channel).
+- A `UnifiedAsyncAPIReader` interface with a `read(ExternalResourceHandler)` method,
+  mirroring `UnifiedOpenAPIReader`.
+- A Jackson-based implementation that deserializes JSON/YAML into `UnifiedAsyncAPI`
+  using `WorkflowFormat.fromFileName(handler.name()).mapper()`.
+- The existing `AsyncAPIReader` is replaced or repurposed as the reader implementation.
+
+**Todo List**
+1. Define `UnifiedAsyncAPI` as Java records:
+   ```
+   UnifiedAsyncAPI(String asyncapi, Map<String, Server> servers,
+                   Map<String, Channel> channels, Map<String, Operation> operations)
+   Server(String url, String protocol, Map<String, ServerVariable> variables)
+   ServerVariable(String defaultValue, List<String> enumValues, String description)
+   Channel(String address, Map<String, Operation> operations)  // v2.6.0 compat
+   Operation(String action, ChannelRef channel, List<ServerRef> servers)  // v3.0.0
+   ```
+   Use `@JsonIgnoreProperties(ignoreUnknown = true)` on each record to tolerate
+   unneeded fields.
+2. Replace `AsyncAPIReader` with a `UnifiedAsyncAPIReader` interface:
+   ```java
+   public interface UnifiedAsyncAPIReader {
+       String UNIFIED_ASYNC_API_READER = "UnifiedAsyncAPIReader";
+       UnifiedAsyncAPI read(ExternalResourceHandler handler) throws IOException;
+   }
+   ```
+3. Implement `JacksonUnifiedAsyncAPIReader` (or inline into `AsyncAPIReader`):
+   ```java
+   ObjectMapper mapper = WorkflowFormat.fromFileName(handler.name()).mapper();
+   try (InputStream is = handler.open()) {
+       return mapper.readValue(is, UnifiedAsyncAPI.class);
+   }
+   ```
+4. Add a helper method to resolve the target server URL:
+   - If `args.getServer().getName()` is set → find server by name in the parsed document.
+   - Else if `args.getProtocol()` is set → find the first server matching that protocol.
+   - Else → use the first server in the document.
+   - Substitute `args.getServer().getVariables()` into the server URL template
+     (replace `{varName}` with the provided or default value).
+5. Add a helper method to resolve the operation/channel:
+   - v3.0.0 (`asyncapi` field starts with `3.`): look up `operations[operationId]`.
+   - v2.6.0 (`asyncapi` field starts with `2.`): look up `channels[channelName]`,
+     then select `publish` or `subscribe` based on whether `message` or `subscription`
+     is configured.
+
+**Relevant Context**
+- Pattern to follow: [`UnifiedOpenAPI`](impl/openapi/src/main/java/io/serverlessworkflow/impl/executors/openapi/UnifiedOpenAPI.java)
+  and [`JacksonUnifiedOpenAPIReader`](impl/openapi-jackson/src/main/java/io/serverlessworkflow/impl/executors/openapi/jackson/JacksonUnifiedOpenAPIReader.java).
+- The existing [`AsyncAPIReader`](impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java)
+  currently reads into `String.class` — it needs to be replaced.
+- `WorkflowFormat.fromFileName(name).mapper()` selects JSON or YAML ObjectMapper.
+
+---
+
+### Sub-Task 1 — Define the `AsyncApiChannelProvider` SPI
+
+**Status**: [ ] pending
+
+**Intent**
+The executor needs to publish or subscribe to messages without being coupled to any
+specific messaging protocol (Kafka, MQTT, AMQP, …). A pluggable SPI interface lets
+external modules provide protocol-specific transport implementations, mirroring how
+`HttpClientResolver` resolves the JAX-RS `Client` from an `additionalObject`.
+
+**Expected Outcomes**
+- A new `AsyncApiChannelProvider` interface in `impl/asyncapi` with two methods:
+  `publish(...)` and `subscribe(...)`.
+- `publish(...)` sends a single message and returns `CompletableFuture<Void>`.
+- `subscribe(...)` accepts a `Consumer<AsyncApiInboundMessage>` callback and returns
+  an `AsyncApiSubscriptionHandle` for lifecycle management.
+- The provider is transport-only — consumption policy, foreach, and timeout are handled
+  by the executor.
+
+**Todo List**
+1. Create `AsyncApiChannelInfo` — a value object carrying the resolved server URL,
+   channel name, operation name, protocol, and authentication token (already resolved
+   before calling the provider):
+   ```java
+   public record AsyncApiChannelInfo(
+       URI serverUri, String channel, String operation,
+       String protocol, Optional<String> authToken) {}
+   ```
+2. Create `AsyncApiInboundMessage` — the message model delivered by the provider to
+   the executor's callback, matching the spec's inbound message structure:
+   ```java
+   public record AsyncApiInboundMessage(
+       Map<String, Object> payload,
+       Map<String, Object> headers,
+       Optional<String> correlationId) {}
+   ```
+3. Create `AsyncApiSubscriptionHandle` — a handle returned by `subscribe` with:
+   ```java
+   public interface AsyncApiSubscriptionHandle {
+       void unsubscribe();
+       CompletableFuture<Void> closed();  // completes on error or clean shutdown
+   }
+   ```
+4. Create `AsyncApiChannelProvider` interface:
+   ```java
+   public interface AsyncApiChannelProvider {
+       String ASYNC_API_CHANNEL_PROVIDER = "asyncApiChannelProvider";
+
+       CompletableFuture<Void> publish(
+           AsyncApiChannelInfo info,
+           Map<String, Object> payload,
+           Map<String, Object> headers);
+
+       AsyncApiSubscriptionHandle subscribe(
+           AsyncApiChannelInfo info,
+           Consumer<AsyncApiInboundMessage> messageConsumer);
+   }
+   ```
+
+**Relevant Context**
+- Pattern to follow: [`HttpClientResolver`](impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpClientResolver.java)
+  uses `HTTP_CLIENT_PROVIDER = "httpClientProvider"` constant and looks up via
+  `application.additionalObject(key, workflowContext, taskContext)`.
+- The provider key `"asyncApiChannelProvider"` allows users to register a custom provider
+  via `WorkflowApplication.builder().withAdditionalObject(...)`.
+- The spec defines inbound messages with `payload`, `headers`, and `correlationId` fields
+  (see [AsyncAPI Inbound Message](https://github.com/open-workflow-specification/specification/blob/main/dsl-reference.md#asyncapi-inbound-message)).
+  There is no generated `AsyncApiInboundMessage` type in the `types` module, so we
+  define our own in `impl/asyncapi`.
+
+---
+
+### Sub-Task 2 — Implement `AsyncAPIExecutorBuilder`
+
+**Status**: [ ] pending
+
+**Intent**
+Wire the `CallAsyncAPI` task configuration into the executor. The builder is responsible
+for resolving all compile-time artifacts (document parsing, expression compilation,
+auth policy resolution) so that the executor itself only handles per-invocation work.
+
+**Expected Outcomes**
+- `accept(CallAsyncAPI.class)` returns `true`.
+- `init(...)` reads the `AsyncApiArguments` and returns a `CallableTaskFactory` that
+  creates an `AsyncAPIExecutor` pre-loaded with resolved resolvers and policy objects.
+- The AsyncAPI document is loaded at runtime via `ResourceLoader.load(...)` (not
+  `loadStatic`), because it may require auth context and expression-based URIs.
+- The operation name, channel, server name, and protocol are extracted from the
+  `AsyncApiArguments`.
+- Authentication policy is resolved via `AuthProviderFactory`.
+- The outbound message payload and headers (if publish) are compiled into expression
+  resolvers.
+- The subscription `filter` expression (if subscribe) is compiled into a predicate.
+- The subscription `foreach.do` task list (if subscribe) is compiled into a
+  `TaskExecutor<?>` via `TaskExecutorHelper.createExecutorList(...)`.
+- The consumption policy variant is detected from the
+  `AsyncApiMessageConsumptionPolicyUnion` union type.
+- The `consume.for` timeout (if present) is resolved via `WorkflowUtils.fromTimeoutAfter()`.
+
+**Todo List**
+1. Change `accept(...)` to return `clazz.equals(CallAsyncAPI.class)`.
+2. In `init(...)`:
+   a. Read `args.getDocument()` and store the `ExternalResource` for runtime loading.
+   b. Extract `operation`, `channel`, `server`, `protocol` fields.
+   c. Resolve `authentication` via
+      `definition.application().authProviderFactory().getAuth(definition, args.getAuthentication(), ...)`.
+   d. If `args.getMessage() != null` (publish path):
+      - Build expression resolvers for `message.getPayload().getAdditionalProperties()`
+        and `message.getHeaders().getAdditionalProperties()`.
+   e. If `args.getSubscription() != null` (subscribe path):
+      - Compile `subscription.getFilter()` into `Optional<WorkflowPredicate>` via
+        `application.expressionFactory().buildPredicate(...)`.
+      - If `subscription.getForeach() != null` and `subscription.getForeach().getDo() != null`:
+        compile the task list via `TaskExecutorHelper.createExecutorList(position, foreach.getDo(), definition)`.
+      - Extract `foreach.getItem()` (default `"item"`) and `foreach.getAt()` (default `"index"`).
+      - Detect consumption policy from `subscription.getConsume()` union:
+        ```java
+        AsyncApiMessageConsumptionPolicyUnion consume = subscription.getConsume();
+        AsyncApiMessageConsumptionPolicyAmount amount = consume.getAsyncApiMessageConsumptionPolicyAmount();
+        AsyncApiMessageConsumptionPolicyWhile whilePolicy = consume.getAsyncApiMessageConsumptionPolicyWhile();
+        AsyncApiMessageConsumptionPolicyUntil untilPolicy = consume.getAsyncApiMessageConsumptionPolicyUntil();
+        ```
+      - If `consume.get().getFor() != null` (the `TimeoutAfter` field on the base
+        `AsyncApiMessageConsumptionPolicy`), resolve the timeout duration via
+        `WorkflowUtils.fromTimeoutAfter(application, consume.get().getFor())`.
+3. Return a `CallableTaskFactory` lambda (`() -> new AsyncAPIExecutor(...)`) that
+   constructs the executor with all the above pre-resolved artifacts.
+
+**Relevant Context**
+- Pattern: [`CallableTaskHttpExecutorBuilder.init(...)`](impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/CallableTaskHttpExecutorBuilder.java)
+  and [`OpenAPIExecutorBuilder`](impl/openapi/src/main/java/io/serverlessworkflow/impl/executors/openapi/OpenAPIExecutorBuilder.java).
+- Expression compilation: `application.expressionFactory().buildPredicate(...)`.
+- Task list compilation: [`ForExecutor`](impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java)
+  uses `TaskExecutorHelper.createExecutorList(position, task.getDo(), definition)`.
+- Auth resolution: [`HttpExecutorBuilder.buildRequestExecutor()`](impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java)
+  calls `definition.application().authProviderFactory().getAuth(definition, policy, method)`.
+- Timeout resolution: `WorkflowUtils.fromTimeoutAfter(application, timeoutAfter)` converts
+  a `TimeoutAfter` to a `WorkflowValueResolver<Duration>`.
+- Union navigation: `AsyncApiMessageConsumptionPolicyUnion` has typed getters for each
+  variant (`getAsyncApiMessageConsumptionPolicyAmount()`, etc.); non-active variants
+  return `null`.
+
+---
+
+### Sub-Task 3 — Implement `AsyncAPIExecutor` — Publish path
+
+**Status**: [ ] pending
+
+**Intent**
+Implement the publish (send message) path of the executor. When `message` is present in
+the `AsyncApiArguments`, the executor resolves the target server URL from the loaded
+AsyncAPI document, builds the `AsyncApiChannelInfo`, resolves the auth token, and
+delegates to the `AsyncApiChannelProvider`.
+
+**Expected Outcomes**
+- When called with `message` configured, `apply(...)` sends the message and returns
+  `CompletableFuture<WorkflowModel>` that completes with the task input as output
+  (publish is fire-and-forget per spec).
+- Server URL is resolved from the loaded AsyncAPI document by matching
+  `args.server.name` or `args.protocol`, with server variable substitution applied.
+- Auth token (if any) is resolved from the `AuthProvider`.
+- The `AsyncApiChannelProvider` is looked up from
+  `WorkflowApplication.additionalObject("asyncApiChannelProvider", workflowContext, taskContext)`.
+
+**Todo List**
+1. In `apply(...)`, detect publish vs subscribe based on whether the message payload
+   resolver was set during `init()`.
+2. Load the AsyncAPI document via `resourceLoader.load(documentResource, reader::read,
+   workflowContext, taskContext, input)` — uses the runtime-context variant to support
+   auth and expression-based URIs.
+3. Parse the document using `UnifiedAsyncAPIReader` to get `UnifiedAsyncAPI`.
+4. Resolve the target server:
+   a. If `server.name` is set → find server by name in `unifiedAsyncAPI.servers()`.
+   b. Else if `protocol` is set → find first server matching that protocol.
+   c. Else → use the first server.
+   d. Substitute server variables into the URL template (`{varName}` → value from
+      `server.variables` or the server variable's default value).
+5. Resolve the operation/channel:
+   a. v3.0.0 → look up `operations[operationId]` to get the channel name.
+   b. v2.6.0 → use `channel` directly from `args.getChannel()`.
+6. Resolve auth token via `authProvider.apply(workflowContext, taskContext, input)`.
+7. Build `AsyncApiChannelInfo` with resolved server URI, operation, channel, protocol,
+   and auth token.
+8. Look up `AsyncApiChannelProvider` from
+   `application.additionalObject("asyncApiChannelProvider", workflowContext, taskContext)`.
+9. Resolve payload and headers maps from the expression resolvers.
+10. Call `provider.publish(channelInfo, payload, headers)`.
+11. Return `future.thenApply(v -> input)` — publish returns the task input as output.
+
+**Relevant Context**
+- Resource loading at runtime: see how [`OpenAPIExecutor`](impl/openapi/src/main/java/io/serverlessworkflow/impl/executors/openapi/OpenAPIExecutor.java)
+  loads the OpenAPI document using `resourceLoader.load(...)` with full auth context.
+- Server variable substitution: AsyncAPI server URLs use `{varName}` templates, similar
+  to OpenAPI's server variables. Replace each `{varName}` with the value from
+  `args.getServer().getVariables()`, falling back to the variable's `defaultValue` from
+  the parsed document.
+- Provider lookup: follows [`HttpClientResolver.client()`](impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpClientResolver.java)
+  pattern with `application.additionalObject(key, workflowContext, taskContext)`.
+
+---
+
+### Sub-Task 4 — Implement `AsyncAPIExecutor` — Subscribe path
+
+**Status**: [ ] pending
+
+**Intent**
+Implement the subscribe (consume messages) path. When `subscription` is present,
+the executor subscribes to the channel, processes each arriving message through the
+optional `foreach` task list, applies the `filter` expression, and terminates according
+to the consumption policy (`amount`, `while`, or `until`), with an optional `for` timeout.
+
+**Expected Outcomes**
+- When called with `subscription` configured, `apply(...)` returns a
+  `CompletableFuture<WorkflowModel>` that resolves once the consumption policy is met
+  or the timeout expires.
+- Each received message is wrapped as a `WorkflowModel` (from `AsyncApiInboundMessage`
+  fields: payload, headers, correlationId).
+- `filter` expression is evaluated on each message; non-matching messages are skipped
+  and do not count toward the consumption policy.
+- When `foreach` is set, messages are processed sequentially (FIFO) — the `foreach.do`
+  task list for message N must complete before message N+1 is processed.
+- Consumption policies:
+  - `amount` — completes after N filtered (and processed) messages.
+  - `while` — evaluated after each filtered message; completes when expression is false.
+  - `until` — evaluated after each filtered message; completes when expression is true.
+- The `consume.for` timeout applies `CompletableFuture.orTimeout()` — if the timeout
+  expires before the consumption policy is satisfied, the future completes with whatever
+  messages have been collected so far (partial result, not an error).
+- The subscription is unregistered on completion, timeout, or cancellation.
+- The accumulated list of consumed messages (as `WorkflowModelCollection`) is the task
+  output.
+
+**Todo List**
+1. In `apply(...)`, when subscription is configured:
+   a. Create a `CompletableFuture<WorkflowModel>` as the task's result future.
+   b. Create a thread-safe list (or `JacksonModelCollection`) to accumulate messages.
+   c. Load and parse the AsyncAPI document (same as publish path).
+   d. Resolve server, channel, auth (same as publish path — extract shared helper).
+   e. Subscribe via `provider.subscribe(channelInfo, message -> enqueueMessage(message))`.
+   f. Register the handle for cleanup:
+      `handle.closed().whenComplete((v, ex) -> { if (ex != null) future.completeExceptionally(ex); })`.
+2. In `enqueueMessage(AsyncApiInboundMessage message)`:
+   a. Convert `AsyncApiInboundMessage` to a `WorkflowModel` (a map with `payload`,
+      `headers`, and `correlationId` keys).
+   b. Evaluate `filter` on the message model; skip if predicate returns false.
+   c. If `foreach` is configured:
+      - Set `taskContext.variables().put(itemVar, messageModel)` (default var name: `"item"`).
+      - Set `taskContext.variables().put(atVar, index)` (default var name: `"index"`).
+      - Run `TaskExecutorHelper.processTaskList(foreachExecutor, workflow, Optional.of(taskContext), messageModel)`.
+      - Wait for the foreach future to complete before processing the next message (FIFO).
+   d. Add the (possibly transformed) result to the collection.
+   e. Evaluate consumption policy:
+      - `amount`: if `collection.size() >= amount` → `future.complete(collection)`.
+      - `while`: evaluate expression; if false → `future.complete(collection)`.
+      - `until`: evaluate expression; if true → `future.complete(collection)`.
+   f. On completion, call `handle.unsubscribe()`.
+3. Apply the `consume.for` timeout:
+   ```java
+   if (timeoutDuration != null) {
+       Duration duration = timeoutResolver.apply(workflowContext, taskContext, input);
+       ScheduledFuture<?> timeoutTask = scheduler.schedule(
+           () -> { if (!future.isDone()) future.complete(collection); },
+           duration.toMillis(), TimeUnit.MILLISECONDS);
+       future.whenComplete((r, ex) -> timeoutTask.cancel(false));
+   }
+   ```
+   Note: we use a scheduled task instead of `CompletableFuture.orTimeout()` because
+   timeout here should produce a **partial result** (the messages collected so far),
+   not an exception.
+4. Ensure cleanup on all exit paths:
+   `future.whenComplete((r, ex) -> handle.unsubscribe())`.
+
+**Relevant Context**
+- **Not `ListenExecutor`**: Although `ListenExecutor` handles similar subscription
+  semantics, it extends `RegularTaskExecutor<ListenTask>` — a different class hierarchy
+  from `CallableTask`. The subscribe logic must be implemented entirely within
+  `CallableTask.apply()`, returning a `CompletableFuture<WorkflowModel>`. Use
+  `ListenExecutor` as **inspiration for the consumption loop** (how it accumulates
+  messages and evaluates predicates), not as a base class or reusable component.
+- `foreach` iteration variables: same pattern as
+  [`ForExecutor`](impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java)
+  — `taskContext.variables().put(each, item)` and `taskContext.variables().put(at, index)`.
+- `foreach` task list processing:
+  [`ListenExecutor.processCe()`](impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java)
+  calls `TaskExecutorHelper.processTaskList(executor, workflow, Optional.of(taskContext), node)`.
+- FIFO requirement: the spec states "consumed messages should be stored in a FIFO queue
+  while awaiting iteration" — the `foreach.do` for message N must complete before N+1
+  starts. Use a serial chain of `CompletableFuture.thenCompose(...)` calls, not parallel
+  execution.
+- Output and Export: `SubscriptionIterator` has `output` and `export` fields, but these
+  are handled per-iteration inside the `foreach.do` task list processing by the framework
+  (`AbstractTaskExecutor.apply()` applies output/export processors automatically). The
+  **task-level** output/export for the entire `call: asyncapi` task is handled by the
+  `CallTaskExecutor` that wraps this `CallableTask`.
+- Timeout: unlike the task-level timeout (which throws `WorkflowException` via
+  `AbstractTaskExecutor`), the `consume.for` timeout is a **graceful** termination —
+  it completes the future with the partial collection, not an error.
+
+---
+
+### Sub-Task 5 — Write tests
+
+**Status**: [ ] pending
+
+**Intent**
+Verify both the publish and subscribe paths against a test double of the
+`AsyncApiChannelProvider`, without requiring a live broker. Follow the existing
+test pattern used by the HTTP and OpenAPI executors.
+
+**Expected Outcomes**
+- A test verifying **publish**: the provider's `publish(...)` is called with the correct
+  channel info, payload, and headers resolved from the workflow input.
+- A test verifying **subscribe with `amount: 1`**: the future completes after the first
+  matching message is delivered, with that message as the output array.
+- A test verifying **subscribe with a `foreach.do` task list**: each message is processed
+  by the inner task list before being added to the output.
+- A test verifying **subscribe with `filter`**: non-matching messages are skipped and do
+  not count toward the consumption policy.
+- A test verifying **subscribe with `consume.for` timeout**: the future completes with
+  a partial result when the timeout expires before enough messages arrive.
+- A test verifying **server variable substitution**: the provider receives a server URI
+  with variables correctly replaced.
+
+**Todo List**
+1. Create a test `AsyncApiChannelProviderStub` implementing `AsyncApiChannelProvider`
+   that:
+   - Records `publish(...)` calls for assertion.
+   - Implements `subscribe(...)` by storing the `Consumer<AsyncApiInboundMessage>`
+     callback and exposing a `deliver(AsyncApiInboundMessage)` method for tests to
+     inject messages on demand.
+   - Returns an `AsyncApiSubscriptionHandle` that tracks unsubscribe calls.
+2. Create a minimal AsyncAPI v3.0.0 document fixture (JSON or YAML) with:
+   - A server with a variable in the URL (e.g., `{environment}`).
+   - An operation referencing a channel.
+3. Register the stub via
+   `WorkflowApplication.builder().withAdditionalObject("asyncApiChannelProvider", stub)`.
+4. Write YAML workflow fixture files with `call: asyncapi` tasks for each scenario:
+   - `asyncapi-publish.yaml` — publish with payload and headers.
+   - `asyncapi-subscribe-amount.yaml` — subscribe with `consume.amount: 1`.
+   - `asyncapi-subscribe-foreach.yaml` — subscribe with `foreach.do` task list.
+   - `asyncapi-subscribe-filter.yaml` — subscribe with `filter` expression.
+   - `asyncapi-subscribe-timeout.yaml` — subscribe with `consume.for` and insufficient
+     messages (verifies partial result).
+5. Assert that:
+   - The stub `publish(...)` was called with expected `AsyncApiChannelInfo` (including
+     substituted server URL), payload, and headers.
+   - The subscribe future completes with the expected collected output array.
+   - `filter` correctly skips non-matching messages.
+   - Timeout produces a partial result, not an exception.
+   - The subscription handle's `unsubscribe()` is called on completion.
+
+**Relevant Context**
+- HTTP test location for pattern reference:
+  `impl/http/src/test/java/io/serverlessworkflow/impl/`
+- OpenAPI test location:
+  `impl/openapi/src/test/java/io/serverlessworkflow/impl/`
+- Test infrastructure: JUnit 5, AssertJ, Awaitility.
+- `WorkflowApplication.builder().withAdditionalObject("asyncApiChannelProvider", stub)` is the
+  registration hook.
\ No newline at end of file
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java
index 76a0dc57d..28f7ec072 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java
@@ -46,6 +46,7 @@ public abstract class BaseTaskItemListBuilder<SELF extends BaseTaskItemListBuild
   protected final String TYPE_HTTP = "http";
   protected final String TYPE_OPENAPI = "openapi";
   protected final String TYPE_GRPC = "grpc";
+  protected final String TYPE_ASYNCAPI = "asyncapi";
   protected final String TYPE_WORKFLOW = "workflow";
 
   private final List<TaskItem> list;
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java
index 329e486e7..365322618 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java
@@ -136,7 +136,11 @@ private SELF appendDo(Consumer<DBuilder> configurer) {
     configurer.accept(doBuilder);
 
     final List<TaskItem> newItems = doBuilder.build().getDo();
-    if (newItems == null || newItems.isEmpty()) return self();
+    if (newItems == null || newItems.isEmpty()) {
+      throw new IllegalStateException(
+          "Task list must contain at least one task. "
+              + "Use .tasks(d -> d.set(...)) or similar to define tasks.");
+    }
 
     final List<TaskItem> merged =
         new ArrayList<>(this.workflow.getDo() != null ? this.workflow.getDo() : List.of());
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java
new file mode 100644
index 000000000..0ff475bed
--- /dev/null
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.api.types.CallAsyncAPI;
+import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent;
+
+public class CallAsyncAPITaskBuilder extends TaskBaseBuilder<CallAsyncAPITaskBuilder>
+    implements CallAsyncAPITaskFluent<CallAsyncAPITaskBuilder> {
+
+  CallAsyncAPITaskBuilder() {
+    final CallAsyncAPI callAsyncAPI = new CallAsyncAPI();
+    callAsyncAPI.setWith(new AsyncApiArguments());
+    super.setTask(callAsyncAPI);
+  }
+
+  @Override
+  public CallAsyncAPITaskBuilder self() {
+    return this;
+  }
+}
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java
index 4199811e3..d28279477 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java
@@ -104,6 +104,12 @@ public DoTaskBuilder openapi(String name, Consumer<CallOpenAPITaskBuilder> items
     return this;
   }
 
+  @Override
+  public DoTaskBuilder asyncapi(String name, Consumer<CallAsyncAPITaskBuilder> itemsConfigurer) {
+    this.listBuilder().asyncapi(name, itemsConfigurer);
+    return this;
+  }
+
   @Override
   public DoTaskBuilder grpc(String name, Consumer<CallGrpcTaskBuilder> itemsConfigurer) {
     this.listBuilder().grpc(name, itemsConfigurer);
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java
index d227718f3..a1a962d58 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java
@@ -154,6 +154,22 @@ public TaskItemListBuilder openapi(
     return addTaskItem(new TaskItem(name, task));
   }
 
+  @Override
+  public TaskItemListBuilder asyncapi(
+      String name, Consumer<CallAsyncAPITaskBuilder> itemsConfigurer) {
+    name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_ASYNCAPI);
+
+    final CallAsyncAPITaskBuilder callAsyncAPIBuilder = new CallAsyncAPITaskBuilder();
+    itemsConfigurer.accept(callAsyncAPIBuilder);
+
+    final CallTask callTask = new CallTask();
+    callTask.setCallAsyncAPI(callAsyncAPIBuilder.build());
+    final Task task = new Task();
+    task.setCallTask(callTask);
+
+    return addTaskItem(new TaskItem(name, task));
+  }
+
   @Override
   public TaskItemListBuilder grpc(String name, Consumer<CallGrpcTaskBuilder> itemsConfigurer) {
     name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_GRPC);
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java
new file mode 100644
index 000000000..adbbe78bc
--- /dev/null
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec.configurers;
+
+import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder;
+import java.util.function.Consumer;
+
+@FunctionalInterface
+public interface CallAsyncAPIConfigurer extends Consumer<CallAsyncAPITaskBuilder> {}
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java
new file mode 100644
index 000000000..029982777
--- /dev/null
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec.dsl;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder;
+import io.serverlessworkflow.fluent.spec.SubscriptionIteratorBuilder;
+import io.serverlessworkflow.fluent.spec.TaskItemListBuilder;
+import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
+import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer;
+import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public final class CallAsyncAPISpec implements CallAsyncAPIConfigurer {
+
+  private final List<Consumer<CallAsyncAPITaskFluent<?>>> steps = new ArrayList<>();
+
+  public CallAsyncAPISpec document(String uri) {
+    steps.add(b -> b.document(uri));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(String uri, AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.document(uri, authenticationConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(URI uri) {
+    steps.add(b -> b.document(uri));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(URI uri, AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.document(uri, authenticationConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec channel(String channel) {
+    steps.add(b -> b.channel(channel));
+    return this;
+  }
+
+  public CallAsyncAPISpec operation(String operation) {
+    steps.add(b -> b.operation(operation));
+    return this;
+  }
+
+  public CallAsyncAPISpec server(String name) {
+    steps.add(b -> b.server(name));
+    return this;
+  }
+
+  public CallAsyncAPISpec server(String name, Map<String, Object> variables) {
+    steps.add(b -> b.server(name, variables));
+    return this;
+  }
+
+  public CallAsyncAPISpec protocol(AsyncApiArguments.AsyncApiProtocol protocol) {
+    steps.add(b -> b.protocol(protocol));
+    return this;
+  }
+
+  public CallAsyncAPISpec message(Map<String, Object> payload) {
+    steps.add(b -> b.message(payload));
+    return this;
+  }
+
+  public CallAsyncAPISpec message(Map<String, Object> payload, Map<String, Object> headers) {
+    steps.add(b -> b.message(payload, headers));
+    return this;
+  }
+
+  public CallAsyncAPISpec payload(Map<String, Object> payload) {
+    steps.add(b -> b.payload(payload));
+    return this;
+  }
+
+  public CallAsyncAPISpec headers(Map<String, Object> headers) {
+    steps.add(b -> b.headers(headers));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeAmount(int amount) {
+    steps.add(b -> b.consumeAmount(amount));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeWhile(String expression) {
+    steps.add(b -> b.consumeWhile(expression));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeUntil(String expression) {
+    steps.add(b -> b.consumeUntil(expression));
+    return this;
+  }
+
+  public CallAsyncAPISpec filter(String filterExpression) {
+    steps.add(b -> b.filter(filterExpression));
+    return this;
+  }
+
+  public CallAsyncAPISpec subscription(
+      Consumer<SubscriptionIteratorBuilder<TaskItemListBuilder>> foreachConfigurer) {
+    steps.add(b -> b.subscription(foreachConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec authentication(AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.authentication(authenticationConfigurer));
+    return this;
+  }
+
+  @Override
+  public void accept(CallAsyncAPITaskBuilder builder) {
+    for (var s : steps) {
+      s.accept(builder);
+    }
+  }
+}
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java
index c88f9a2d1..671714fc2 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java
@@ -28,6 +28,7 @@
 import io.serverlessworkflow.fluent.spec.TimeoutBuilder;
 import io.serverlessworkflow.fluent.spec.TryTaskBuilder;
 import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
+import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallGrpcConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallHttpConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallOpenAPIConfigurer;
@@ -114,6 +115,28 @@ public static CallGrpcSpec grpc() {
     return new CallGrpcSpec();
   }
 
+  /**
+   * Create a new AsyncAPI call specification to be used with {@link #call(CallAsyncAPIConfigurer)}.
+   *
+   * <p>Typical usage:
+   *
+   * <pre>{@code
+   * tasks(
+   *   call(
+   *     asyncapi()
+   *       .document("http://acme.org/asyncapi.yaml")
+   *       .operation("greet")
+   *       .message(Map.of("greeting", "hello"))
+   *   )
+   * );
+   * }</pre>
+   *
+   * @return a new {@link CallAsyncAPISpec} instance
+   */
+  public static CallAsyncAPISpec asyncapi() {
+    return new CallAsyncAPISpec();
+  }
+
   public static WorkflowSpec workflow(String namespace, String name, String version) {
     return new WorkflowSpec().namespace(namespace).name(name).version(version);
   }
@@ -760,6 +783,27 @@ public static TasksConfigurer call(String name, CallOpenAPIConfigurer configurer
     return list -> list.openapi(name, configurer);
   }
 
+  /**
+   * Create a {@link TasksConfigurer} that adds an AsyncAPI call task.
+   *
+   * @param configurer AsyncAPI configurer
+   * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task
+   */
+  public static TasksConfigurer call(CallAsyncAPIConfigurer configurer) {
+    return list -> list.asyncapi(configurer);
+  }
+
+  /**
+   * Create a {@link TasksConfigurer} that adds an AsyncAPI call task with an explicit name.
+   *
+   * @param name the task name
+   * @param configurer AsyncAPI configurer
+   * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task
+   */
+  public static TasksConfigurer call(String name, CallAsyncAPIConfigurer configurer) {
+    return list -> list.asyncapi(name, configurer);
+  }
+
   public static TasksConfigurer call(CallGrpcConfigurer configurer) {
     return list -> list.grpc(configurer);
   }
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java
new file mode 100644
index 000000000..0d186f7de
--- /dev/null
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec.spi;
+
+import io.serverlessworkflow.fluent.spec.TaskBaseBuilder;
+import java.util.function.Consumer;
+
+public interface CallAsyncAPIFluent<SELF extends TaskBaseBuilder<SELF>, LIST> {
+
+  LIST asyncapi(String name, Consumer<SELF> itemsConfigurer);
+
+  default LIST asyncapi(Consumer<SELF> itemsConfigurer) {
+    return this.asyncapi(null, itemsConfigurer);
+  }
+}
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java
new file mode 100644
index 000000000..a8c5aa447
--- /dev/null
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java
@@ -0,0 +1,244 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec.spi;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyAmount;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUnion;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUntil;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyWhile;
+import io.serverlessworkflow.api.types.AsyncApiMessageHeaders;
+import io.serverlessworkflow.api.types.AsyncApiMessagePayload;
+import io.serverlessworkflow.api.types.AsyncApiOutboundMessage;
+import io.serverlessworkflow.api.types.AsyncApiServer;
+import io.serverlessworkflow.api.types.AsyncApiSubscription;
+import io.serverlessworkflow.api.types.CallAsyncAPI;
+import io.serverlessworkflow.api.types.Endpoint;
+import io.serverlessworkflow.api.types.EndpointConfiguration;
+import io.serverlessworkflow.api.types.EndpointUri;
+import io.serverlessworkflow.api.types.ExternalResource;
+import io.serverlessworkflow.api.types.ReferenceableAuthenticationPolicy;
+import io.serverlessworkflow.api.types.UriTemplate;
+import io.serverlessworkflow.fluent.spec.ReferenceableAuthenticationPolicyBuilder;
+import io.serverlessworkflow.fluent.spec.SubscriptionIteratorBuilder;
+import io.serverlessworkflow.fluent.spec.TaskBaseBuilder;
+import io.serverlessworkflow.fluent.spec.TaskItemListBuilder;
+import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
+import java.net.URI;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public interface CallAsyncAPITaskFluent<SELF extends TaskBaseBuilder<SELF>> {
+
+  default CallAsyncAPI build() {
+    return ((CallAsyncAPI) this.self().getTask());
+  }
+
+  SELF self();
+
+  default SELF document(String uri) {
+    ((CallAsyncAPI) this.self().getTask())
+        .getWith()
+        .setDocument(new ExternalResource().withEndpoint(EndpointUtil.fromString(uri)));
+    return self();
+  }
+
+  default SELF document(URI uri) {
+    ((CallAsyncAPI) this.self().getTask())
+        .getWith()
+        .withDocument(
+            new ExternalResource()
+                .withEndpoint(
+                    new Endpoint().withUriTemplate(new UriTemplate().withLiteralUri(uri))));
+    return self();
+  }
+
+  default SELF document(String uri, AuthenticationConfigurer authenticationConfigurer) {
+    final ReferenceableAuthenticationPolicyBuilder policy =
+        new ReferenceableAuthenticationPolicyBuilder();
+    authenticationConfigurer.accept(policy);
+    ReferenceableAuthenticationPolicy auth = policy.build();
+    ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(auth);
+    ((CallAsyncAPI) this.self().getTask())
+        .getWith()
+        .setDocument(new ExternalResource().withEndpoint(EndpointUtil.fromString(uri, auth)));
+    return self();
+  }
+
+  default SELF document(URI uri, AuthenticationConfigurer authenticationConfigurer) {
+    final ReferenceableAuthenticationPolicyBuilder policy =
+        new ReferenceableAuthenticationPolicyBuilder();
+    authenticationConfigurer.accept(policy);
+    ReferenceableAuthenticationPolicy auth = policy.build();
+    ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(auth);
+    ((CallAsyncAPI) this.self().getTask())
+        .getWith()
+        .setDocument(
+            new ExternalResource()
+                .withEndpoint(
+                    new Endpoint()
+                        .withEndpointConfiguration(
+                            new EndpointConfiguration()
+                                .withUri(
+                                    new EndpointUri()
+                                        .withLiteralEndpointURI(
+                                            new UriTemplate().withLiteralUri(uri)))
+                                .withAuthentication(auth))));
+    return self();
+  }
+
+  default SELF channel(String channel) {
+    ((CallAsyncAPI) this.self().getTask()).getWith().setChannel(channel);
+    return self();
+  }
+
+  default SELF operation(String operation) {
+    ((CallAsyncAPI) this.self().getTask()).getWith().setOperation(operation);
+    return self();
+  }
+
+  default SELF server(String name) {
+    ((CallAsyncAPI) this.self().getTask()).getWith().setServer(new AsyncApiServer(name));
+    return self();
+  }
+
+  default SELF server(String name, Map<String, Object> variables) {
+    AsyncApiServer server = new AsyncApiServer(name);
+    io.serverlessworkflow.api.types.AsyncApiServerVariables vars =
+        new io.serverlessworkflow.api.types.AsyncApiServerVariables();
+    variables.forEach(vars::withAdditionalProperty);
+    server.setVariables(vars);
+    ((CallAsyncAPI) this.self().getTask()).getWith().setServer(server);
+    return self();
+  }
+
+  default SELF protocol(AsyncApiArguments.AsyncApiProtocol protocol) {
+    ((CallAsyncAPI) this.self().getTask()).getWith().setProtocol(protocol);
+    return self();
+  }
+
+  default SELF message(Map<String, Object> payload) {
+    AsyncApiOutboundMessage msg = ensureMessage();
+    AsyncApiMessagePayload p = new AsyncApiMessagePayload();
+    payload.forEach(p::withAdditionalProperty);
+    msg.setPayload(p);
+    return self();
+  }
+
+  default SELF message(Map<String, Object> payload, Map<String, Object> headers) {
+    AsyncApiOutboundMessage msg = ensureMessage();
+    AsyncApiMessagePayload p = new AsyncApiMessagePayload();
+    payload.forEach(p::withAdditionalProperty);
+    msg.setPayload(p);
+    AsyncApiMessageHeaders h = new AsyncApiMessageHeaders();
+    headers.forEach(h::withAdditionalProperty);
+    msg.setHeaders(h);
+    return self();
+  }
+
+  default SELF payload(Map<String, Object> payload) {
+    AsyncApiOutboundMessage msg = ensureMessage();
+    AsyncApiMessagePayload p = new AsyncApiMessagePayload();
+    payload.forEach(p::withAdditionalProperty);
+    msg.setPayload(p);
+    return self();
+  }
+
+  default SELF headers(Map<String, Object> headers) {
+    AsyncApiOutboundMessage msg = ensureMessage();
+    AsyncApiMessageHeaders h = new AsyncApiMessageHeaders();
+    headers.forEach(h::withAdditionalProperty);
+    msg.setHeaders(h);
+    return self();
+  }
+
+  private AsyncApiOutboundMessage ensureMessage() {
+    AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith();
+    if (args.getMessage() == null) {
+      args.setMessage(new AsyncApiOutboundMessage());
+    }
+    return args.getMessage();
+  }
+
+  default SELF subscription(
+      Consumer<SubscriptionIteratorBuilder<TaskItemListBuilder>> foreachConfigurer) {
+    AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith();
+    if (args.getSubscription() == null) {
+      args.setSubscription(
+          new AsyncApiSubscription(
+              new AsyncApiMessageConsumptionPolicyUnion()
+                  .withAsyncApiMessageConsumptionPolicyAmount(
+                      new AsyncApiMessageConsumptionPolicyAmount(1))));
+    }
+    SubscriptionIteratorBuilder<TaskItemListBuilder> builder =
+        new SubscriptionIteratorBuilder<>(new TaskItemListBuilder(0));
+    foreachConfigurer.accept(builder);
+    args.getSubscription().setForeach(builder.build());
+    return self();
+  }
+
+  default SELF consumeAmount(int amount) {
+    ensureSubscription()
+        .setConsume(
+            new AsyncApiMessageConsumptionPolicyUnion()
+                .withAsyncApiMessageConsumptionPolicyAmount(
+                    new AsyncApiMessageConsumptionPolicyAmount(amount)));
+    return self();
+  }
+
+  default SELF consumeWhile(String expression) {
+    ensureSubscription()
+        .setConsume(
+            new AsyncApiMessageConsumptionPolicyUnion()
+                .withAsyncApiMessageConsumptionPolicyWhile(
+                    new AsyncApiMessageConsumptionPolicyWhile().withWhile(expression)));
+    return self();
+  }
+
+  default SELF consumeUntil(String expression) {
+    ensureSubscription()
+        .setConsume(
+            new AsyncApiMessageConsumptionPolicyUnion()
+                .withAsyncApiMessageConsumptionPolicyUntil(
+                    new AsyncApiMessageConsumptionPolicyUntil().withUntil(expression)));
+    return self();
+  }
+
+  default SELF filter(String filterExpression) {
+    ensureSubscription().setFilter(filterExpression);
+    return self();
+  }
+
+  private AsyncApiSubscription ensureSubscription() {
+    AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith();
+    if (args.getSubscription() == null) {
+      args.setSubscription(
+          new AsyncApiSubscription(
+              new AsyncApiMessageConsumptionPolicyUnion()
+                  .withAsyncApiMessageConsumptionPolicyAmount(
+                      new AsyncApiMessageConsumptionPolicyAmount(1))));
+    }
+    return args.getSubscription();
+  }
+
+  default SELF authentication(AuthenticationConfigurer authenticationConfigurer) {
+    final ReferenceableAuthenticationPolicyBuilder policy =
+        new ReferenceableAuthenticationPolicyBuilder();
+    authenticationConfigurer.accept(policy);
+    ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(policy.build());
+    return self();
+  }
+}
diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java
index 37c5f461b..249e3a1c3 100644
--- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java
+++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java
@@ -15,6 +15,7 @@
  */
 package io.serverlessworkflow.fluent.spec.spi;
 
+import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder;
 import io.serverlessworkflow.fluent.spec.CallGrpcTaskBuilder;
 import io.serverlessworkflow.fluent.spec.CallHttpTaskBuilder;
 import io.serverlessworkflow.fluent.spec.CallOpenAPITaskBuilder;
@@ -49,5 +50,6 @@ public interface DoFluent<T>
         WaitFluent<WaitTaskBuilder, T>,
         RaiseFluent<RaiseTaskBuilder, T>,
         CallOpenAPIFluent<CallOpenAPITaskBuilder, T>,
+        CallAsyncAPIFluent<CallAsyncAPITaskBuilder, T>,
         CallGrpcFluent<CallGrpcTaskBuilder, T>,
         WorkflowFluent<WorkflowTaskBuilder, T> {}
diff --git a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java
index 4ba41c203..d12290c42 100644
--- a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java
+++ b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java
@@ -33,6 +33,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.serverlessworkflow.api.types.AuthenticationPolicyUnion;
@@ -112,6 +113,17 @@ void testUseAuthenticationsBasic() {
     assertNotNull(union.getBasicAuthenticationPolicy(), "BasicAuthenticationPolicy should be set");
   }
 
+  @Test
+  void testEmptyTasksThrows() {
+    assertThrows(IllegalStateException.class, () -> WorkflowBuilder.workflow().tasks().build());
+  }
+
+  @Test
+  void testEmptyTasksConsumerThrows() {
+    assertThrows(
+        IllegalStateException.class, () -> WorkflowBuilder.workflow().tasks(d -> {}).build());
+  }
+
   @Test
   void testDoTaskSetAndForEach() {
     Workflow wf =
diff --git a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java
new file mode 100644
index 000000000..960b41047
--- /dev/null
+++ b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java
@@ -0,0 +1,210 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.fluent.spec.dsl;
+
+import static io.serverlessworkflow.fluent.spec.dsl.DSL.asyncapi;
+import static io.serverlessworkflow.fluent.spec.dsl.DSL.basic;
+import static io.serverlessworkflow.fluent.spec.dsl.DSL.call;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.spec.WorkflowBuilder;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+public class CallAsyncApiDslTest {
+
+  @Test
+  void when_call_asyncapi_publish_with_message() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("greet")
+                        .message(
+                            Map.of("greeting", "${ .name }"),
+                            Map.of("content-type", "application/json"))))
+            .build();
+
+    var taskItem = wf.getDo().get(0);
+    var callAsyncAPI = taskItem.getTask().getCallTask().getCallAsyncAPI();
+    assertThat(callAsyncAPI).isNotNull();
+
+    var with = callAsyncAPI.getWith();
+    assertThat(with).isNotNull();
+    assertThat(with.getDocument()).isNotNull();
+    assertThat(with.getOperation()).isEqualTo("greet");
+
+    assertThat(with.getMessage()).isNotNull();
+    assertThat(with.getMessage().getPayload()).isNotNull();
+    assertThat(with.getMessage().getPayload().getAdditionalProperties())
+        .containsEntry("greeting", "${ .name }");
+    assertThat(with.getMessage().getHeaders()).isNotNull();
+    assertThat(with.getMessage().getHeaders().getAdditionalProperties())
+        .containsEntry("content-type", "application/json");
+  }
+
+  @Test
+  void when_call_asyncapi_subscribe_with_amount() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("receive")
+                        .consumeAmount(5)))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getSubscription()).isNotNull();
+    assertThat(with.getSubscription().getConsume()).isNotNull();
+    assertThat(
+            with.getSubscription()
+                .getConsume()
+                .getAsyncApiMessageConsumptionPolicyAmount()
+                .getAmount())
+        .isEqualTo(5);
+  }
+
+  @Test
+  void when_call_asyncapi_subscribe_with_until() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("receive")
+                        .consumeUntil("${ (. | length) >= 2 }")))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getSubscription()).isNotNull();
+    assertThat(
+            with.getSubscription()
+                .getConsume()
+                .getAsyncApiMessageConsumptionPolicyUntil()
+                .getUntil())
+        .isEqualTo("${ (. | length) >= 2 }");
+  }
+
+  @Test
+  void when_call_asyncapi_with_channel_and_protocol() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .channel("greetings")
+                        .protocol(AsyncApiArguments.AsyncApiProtocol.KAFKA)
+                        .message(Map.of("hello", "world"))))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getChannel()).isEqualTo("greetings");
+    assertThat(with.getProtocol()).isEqualTo(AsyncApiArguments.AsyncApiProtocol.KAFKA);
+  }
+
+  @Test
+  void when_call_asyncapi_with_explicit_name() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    "myAsyncCall",
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("greet")
+                        .message(Map.of("greeting", "hello"))))
+            .build();
+
+    assertThat(wf.getDo()).hasSize(1);
+    assertThat(wf.getDo().get(0).getName()).isEqualTo("myAsyncCall");
+    assertThat(wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI()).isNotNull();
+  }
+
+  @Test
+  void when_call_asyncapi_with_server() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("greet")
+                        .server("production")
+                        .message(Map.of("greeting", "hello"))))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getServer()).isNotNull();
+    assertThat(with.getServer().getName()).isEqualTo("production");
+  }
+
+  @Test
+  void when_call_asyncapi_with_basic_auth_on_document() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml", basic("alice", "secret"))
+                        .operation("greet")
+                        .message(Map.of("greeting", "hello"))))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getAuthentication()).isNotNull();
+    assertThat(with.getAuthentication().getAuthenticationPolicy()).isNotNull();
+    assertThat(
+            with.getAuthentication()
+                .getAuthenticationPolicy()
+                .getBasicAuthenticationPolicy()
+                .getBasic()
+                .getBasicAuthenticationProperties()
+                .getUsername())
+        .isEqualTo("alice");
+  }
+
+  @Test
+  void when_call_asyncapi_with_filter() {
+    Workflow wf =
+        WorkflowBuilder.workflow("f", "ns", "1")
+            .tasks(
+                call(
+                    asyncapi()
+                        .document("https://example.com/asyncapi.yaml")
+                        .operation("receive")
+                        .filter("${ .payload.roomId == \"room-1\" }")
+                        .consumeAmount(2)))
+            .build();
+
+    var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith();
+    assertThat(with.getSubscription()).isNotNull();
+    assertThat(with.getSubscription().getFilter()).isEqualTo("${ .payload.roomId == \"room-1\" }");
+    assertThat(
+            with.getSubscription()
+                .getConsume()
+                .getAsyncApiMessageConsumptionPolicyAmount()
+                .getAmount())
+        .isEqualTo(2);
+  }
+}
diff --git a/impl/asyncapi/pom.xml b/impl/asyncapi/pom.xml
new file mode 100644
index 000000000..b95d6171a
--- /dev/null
+++ b/impl/asyncapi/pom.xml
@@ -0,0 +1,23 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.serverlessworkflow</groupId>
+        <artifactId>serverlessworkflow-impl</artifactId>
+        <version>8.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>serverlessworkflow-impl-asyncapi</artifactId>
+    <name>Serverless Workflow :: Impl :: AsyncAPI</name>
+    <dependencies>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-impl-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java
new file mode 100644
index 000000000..49cf16394
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java
@@ -0,0 +1,329 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments.AsyncApiProtocol;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyAmount;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUnion;
+import io.serverlessworkflow.api.types.AsyncApiServer;
+import io.serverlessworkflow.api.types.ExternalResource;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowModelCollection;
+import io.serverlessworkflow.impl.WorkflowModelFactory;
+import io.serverlessworkflow.impl.WorkflowPredicate;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.auth.AuthProvider;
+import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.TaskExecutor;
+import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
+import java.net.URI;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+class AsyncAPIExecutor implements CallableTask {
+
+  record PublishConfig(
+      WorkflowValueResolver<Map<String, Object>> payloadResolver,
+      WorkflowValueResolver<Map<String, Object>> headersResolver) {}
+
+  record SubscribeConfig(
+      Optional<WorkflowPredicate> filterPredicate,
+      AsyncApiMessageConsumptionPolicyUnion consumePolicy,
+      Optional<WorkflowValueResolver<Duration>> consumeTimeout,
+      Optional<WorkflowPredicate> whilePredicate,
+      Optional<WorkflowPredicate> untilPredicate,
+      TaskExecutor<?> foreachExecutor,
+      String foreachItem,
+      String foreachAt) {}
+
+  private final ExternalResource document;
+  private final String operationName;
+  private final String channelName;
+  private final AsyncApiServer serverConfig;
+  private final AsyncApiProtocol protocolConfig;
+  private final Optional<AuthProvider> authProvider;
+  private final PublishConfig publishConfig;
+  private final SubscribeConfig subscribeConfig;
+
+  AsyncAPIExecutor(
+      ExternalResource document,
+      String operationName,
+      String channelName,
+      AsyncApiServer serverConfig,
+      AsyncApiProtocol protocolConfig,
+      Optional<AuthProvider> authProvider,
+      PublishConfig publishConfig,
+      SubscribeConfig subscribeConfig) {
+    this.document = document;
+    this.operationName = operationName;
+    this.channelName = channelName;
+    this.serverConfig = serverConfig;
+    this.protocolConfig = protocolConfig;
+    this.authProvider = authProvider;
+    this.publishConfig = publishConfig;
+    this.subscribeConfig = subscribeConfig;
+  }
+
+  @Override
+  public CompletableFuture<WorkflowModel> apply(
+      WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
+    return CompletableFuture.supplyAsync(
+            () -> loadAndResolve(workflowContext, taskContext, input),
+            workflowContext.definition().application().executorService())
+        .thenCompose(
+            channelInfo -> {
+              AsyncApiChannelProvider provider = lookupProvider(workflowContext, taskContext);
+              if (publishConfig != null) {
+                return doPublish(provider, channelInfo, workflowContext, taskContext, input);
+              } else {
+                return doSubscribe(provider, channelInfo, workflowContext, taskContext, input);
+              }
+            });
+  }
+
+  private AsyncApiChannelInfo loadAndResolve(
+      WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
+    UnifiedAsyncAPI asyncApi =
+        workflowContext
+            .definition()
+            .resourceLoader()
+            .load(document, AsyncAPIReader::read, workflowContext, taskContext, input);
+
+    UnifiedAsyncAPI.Server server = resolveServer(asyncApi);
+    String resolvedChannel = resolveChannel(asyncApi);
+    String url = substituteVariables(server.effectiveUrl(), server);
+    URI serverUri = URI.create(server.protocol() + "://" + url);
+
+    Optional<String> authToken =
+        authProvider.map(
+            auth -> auth.content(workflowContext, taskContext, input, serverUri).join());
+
+    return new AsyncApiChannelInfo(
+        serverUri,
+        resolvedChannel,
+        operationName != null ? operationName : channelName,
+        server.protocol(),
+        authToken);
+  }
+
+  private UnifiedAsyncAPI.Server resolveServer(UnifiedAsyncAPI asyncApi) {
+    if (asyncApi.servers() == null || asyncApi.servers().isEmpty()) {
+      throw new IllegalArgumentException("AsyncAPI document has no servers defined");
+    }
+    if (serverConfig != null && serverConfig.getName() != null) {
+      UnifiedAsyncAPI.Server server = asyncApi.servers().get(serverConfig.getName());
+      if (server != null) {
+        return server;
+      }
+      throw new IllegalArgumentException(
+          "Server '" + serverConfig.getName() + "' not found in AsyncAPI document");
+    }
+    if (protocolConfig != null) {
+      String proto = protocolConfig.value();
+      return asyncApi.servers().values().stream()
+          .filter(s -> proto.equals(s.protocol()))
+          .findFirst()
+          .orElseThrow(
+              () ->
+                  new IllegalArgumentException(
+                      "No server with protocol '" + proto + "' in AsyncAPI document"));
+    }
+    return asyncApi.servers().values().iterator().next();
+  }
+
+  private String resolveChannel(UnifiedAsyncAPI asyncApi) {
+    if (channelName != null) {
+      return channelName;
+    }
+    if (operationName != null && asyncApi.isV3() && asyncApi.operations() != null) {
+      UnifiedAsyncAPI.Operation op = asyncApi.operations().get(operationName);
+      if (op != null && op.channel() != null) {
+        String name = op.channel().channelName();
+        if (asyncApi.channels() != null && asyncApi.channels().containsKey(name)) {
+          UnifiedAsyncAPI.Channel ch = asyncApi.channels().get(name);
+          return ch.address() != null ? ch.address() : name;
+        }
+        return name;
+      }
+    }
+    throw new IllegalArgumentException(
+        "Cannot resolve channel: provide 'channel' (v2) or 'operation' (v3)");
+  }
+
+  private String substituteVariables(String url, UnifiedAsyncAPI.Server docServer) {
+    if (serverConfig != null
+        && serverConfig.getVariables() != null
+        && serverConfig.getVariables().getAdditionalProperties() != null) {
+      for (Map.Entry<String, Object> entry :
+          serverConfig.getVariables().getAdditionalProperties().entrySet()) {
+        url = url.replace("{" + entry.getKey() + "}", entry.getValue().toString());
+      }
+    }
+    if (docServer.variables() != null) {
+      for (Map.Entry<String, UnifiedAsyncAPI.ServerVariable> entry :
+          docServer.variables().entrySet()) {
+        if (entry.getValue().defaultValue() != null) {
+          url = url.replace("{" + entry.getKey() + "}", entry.getValue().defaultValue());
+        }
+      }
+    }
+    return url;
+  }
+
+  private CompletableFuture<WorkflowModel> doPublish(
+      AsyncApiChannelProvider provider,
+      AsyncApiChannelInfo channelInfo,
+      WorkflowContext workflowContext,
+      TaskContext taskContext,
+      WorkflowModel input) {
+    Map<String, Object> payload =
+        publishConfig.payloadResolver() != null
+            ? publishConfig.payloadResolver().apply(workflowContext, taskContext, input)
+            : Collections.emptyMap();
+    Map<String, Object> headers =
+        publishConfig.headersResolver() != null
+            ? publishConfig.headersResolver().apply(workflowContext, taskContext, input)
+            : Collections.emptyMap();
+    return provider.publish(channelInfo, payload, headers).thenApply(v -> input);
+  }
+
+  private CompletableFuture<WorkflowModel> doSubscribe(
+      AsyncApiChannelProvider provider,
+      AsyncApiChannelInfo channelInfo,
+      WorkflowContext workflowContext,
+      TaskContext taskContext,
+      WorkflowModel input) {
+    WorkflowModelFactory factory = workflowContext.definition().application().modelFactory();
+    WorkflowModelCollection collection = factory.createCollection();
+    CompletableFuture<WorkflowModel> result = new CompletableFuture<>();
+
+    AsyncApiSubscriptionHandle handle =
+        provider.subscribe(
+            channelInfo,
+            msg -> {
+              synchronized (collection) {
+                if (result.isDone()) {
+                  return;
+                }
+                WorkflowModel messageModel = toWorkflowModel(factory, msg);
+                if (subscribeConfig.filterPredicate().isPresent()
+                    && !subscribeConfig
+                        .filterPredicate()
+                        .get()
+                        .test(workflowContext, taskContext, messageModel)) {
+                  return;
+                }
+                WorkflowModel processedModel =
+                    processMessage(messageModel, collection, workflowContext, taskContext);
+                collection.add(processedModel);
+                if (isConsumptionPolicySatisfied(workflowContext, taskContext, collection)) {
+                  result.complete(collection);
+                }
+              }
+            });
+
+    result.whenComplete((r, ex) -> handle.unsubscribe());
+
+    subscribeConfig
+        .consumeTimeout()
+        .ifPresent(
+            resolver -> {
+              Duration duration = resolver.apply(workflowContext, taskContext, input);
+              CompletableFuture.delayedExecutor(duration.toMillis(), TimeUnit.MILLISECONDS)
+                  .execute(
+                      () -> {
+                        synchronized (collection) {
+                          if (!result.isDone()) {
+                            result.complete(collection);
+                          }
+                        }
+                      });
+            });
+
+    return result;
+  }
+
+  private WorkflowModel processMessage(
+      WorkflowModel messageModel,
+      WorkflowModelCollection collection,
+      WorkflowContext workflowContext,
+      TaskContext taskContext) {
+    if (subscribeConfig.foreachExecutor() != null) {
+      taskContext.variables().put(subscribeConfig.foreachItem(), messageModel);
+      taskContext.variables().put(subscribeConfig.foreachAt(), collection.size());
+      return TaskExecutorHelper.processTaskList(
+              subscribeConfig.foreachExecutor(),
+              workflowContext,
+              Optional.of(taskContext),
+              messageModel)
+          .join();
+    }
+    return messageModel;
+  }
+
+  private WorkflowModel toWorkflowModel(WorkflowModelFactory factory, AsyncApiInboundMessage msg) {
+    Map<String, Object> map = new HashMap<>();
+    map.put("payload", msg.payload());
+    map.put("headers", msg.headers());
+    msg.correlationId().ifPresent(id -> map.put("correlationId", id));
+    return factory.from(map);
+  }
+
+  private boolean isConsumptionPolicySatisfied(
+      WorkflowContext workflowContext,
+      TaskContext taskContext,
+      WorkflowModelCollection collection) {
+    AsyncApiMessageConsumptionPolicyUnion policy = subscribeConfig.consumePolicy();
+    if (policy == null) {
+      return false;
+    }
+    AsyncApiMessageConsumptionPolicyAmount amount =
+        policy.getAsyncApiMessageConsumptionPolicyAmount();
+    if (amount != null) {
+      return collection.size() >= amount.getAmount();
+    }
+    if (subscribeConfig.whilePredicate().isPresent()) {
+      return !subscribeConfig.whilePredicate().get().test(workflowContext, taskContext, collection);
+    }
+    if (subscribeConfig.untilPredicate().isPresent()) {
+      return subscribeConfig.untilPredicate().get().test(workflowContext, taskContext, collection);
+    }
+    return false;
+  }
+
+  private AsyncApiChannelProvider lookupProvider(
+      WorkflowContext workflowContext, TaskContext taskContext) {
+    return workflowContext
+        .definition()
+        .application()
+        .<AsyncApiChannelProvider>additionalObject(
+            AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER, workflowContext, taskContext)
+        .orElseThrow(
+            () ->
+                new IllegalStateException(
+                    "Missing AsyncApiChannelProvider. Register one via"
+                        + " WorkflowApplication.builder().withAdditionalObject(\""
+                        + AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER
+                        + "\", provider)"));
+  }
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java
new file mode 100644
index 000000000..6243b8938
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java
@@ -0,0 +1,155 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.api.types.AsyncApiArguments.AsyncApiProtocol;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUnion;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUntil;
+import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyWhile;
+import io.serverlessworkflow.api.types.AsyncApiOutboundMessage;
+import io.serverlessworkflow.api.types.AsyncApiServer;
+import io.serverlessworkflow.api.types.AsyncApiSubscription;
+import io.serverlessworkflow.api.types.CallAsyncAPI;
+import io.serverlessworkflow.api.types.ExternalResource;
+import io.serverlessworkflow.api.types.SubscriptionIterator;
+import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
+import io.serverlessworkflow.impl.WorkflowPredicate;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.auth.AuthProvider;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
+import io.serverlessworkflow.impl.executors.CallableTaskFactory;
+import io.serverlessworkflow.impl.executors.TaskExecutor;
+import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+
+public class AsyncAPIExecutorBuilder implements CallableTaskBuilder<CallAsyncAPI> {
+
+  static final String DEFAULT_INDEX = "index";
+  static final String DEFAULT_ITEM = "item";
+
+  @Override
+  public boolean accept(Class<? extends TaskBase> clazz) {
+    return CallAsyncAPI.class.equals(clazz);
+  }
+
+  @Override
+  public CallableTaskFactory init(
+      CallAsyncAPI task, WorkflowDefinition definition, WorkflowMutablePosition position) {
+    AsyncApiArguments args = task.getWith();
+    WorkflowApplication application = definition.application();
+
+    ExternalResource document = args.getDocument();
+    String operationName = args.getOperation();
+    String channelName = args.getChannel();
+    AsyncApiServer serverConfig = args.getServer();
+    AsyncApiProtocol protocolConfig = args.getProtocol();
+
+    Optional<AuthProvider> authProvider =
+        args.getAuthentication() != null
+            ? application.authProviderFactory().getAuth(definition, args.getAuthentication(), null)
+            : Optional.empty();
+
+    AsyncAPIExecutor.PublishConfig publishConfig =
+        Optional.ofNullable(args.getMessage())
+            .map(msg -> buildPublishConfig(application, msg))
+            .orElse(null);
+
+    AsyncAPIExecutor.SubscribeConfig subscribeConfig =
+        Optional.ofNullable(args.getSubscription())
+            .map(sub -> buildSubscribeConfig(application, sub, position, definition))
+            .orElse(null);
+
+    return () ->
+        new AsyncAPIExecutor(
+            document,
+            operationName,
+            channelName,
+            serverConfig,
+            protocolConfig,
+            authProvider,
+            publishConfig,
+            subscribeConfig);
+  }
+
+  private static AsyncAPIExecutor.PublishConfig buildPublishConfig(
+      WorkflowApplication application, AsyncApiOutboundMessage message) {
+    WorkflowValueResolver<Map<String, Object>> payloadResolver =
+        Optional.ofNullable(message.getPayload())
+            .map(p -> p.getAdditionalProperties())
+            .map(props -> WorkflowUtils.buildMapResolver(application, props))
+            .orElse(null);
+    WorkflowValueResolver<Map<String, Object>> headersResolver =
+        Optional.ofNullable(message.getHeaders())
+            .map(h -> h.getAdditionalProperties())
+            .map(props -> WorkflowUtils.buildMapResolver(application, props))
+            .orElse(null);
+    return new AsyncAPIExecutor.PublishConfig(payloadResolver, headersResolver);
+  }
+
+  private static AsyncAPIExecutor.SubscribeConfig buildSubscribeConfig(
+      WorkflowApplication application,
+      AsyncApiSubscription subscription,
+      WorkflowMutablePosition position,
+      WorkflowDefinition definition) {
+    Optional<WorkflowPredicate> filterPredicate =
+        Optional.ofNullable(subscription.getFilter())
+            .map(f -> WorkflowUtils.buildPredicate(application, f));
+
+    AsyncApiMessageConsumptionPolicyUnion consumePolicy = subscription.getConsume();
+
+    Optional<WorkflowValueResolver<Duration>> consumeTimeout =
+        Optional.ofNullable(consumePolicy.get().getFor())
+            .map(t -> WorkflowUtils.fromTimeoutAfter(application, t));
+
+    Optional<WorkflowPredicate> whilePredicate =
+        Optional.ofNullable(consumePolicy.getAsyncApiMessageConsumptionPolicyWhile())
+            .map(AsyncApiMessageConsumptionPolicyWhile::getWhile)
+            .map(expr -> WorkflowUtils.buildPredicate(application, expr));
+
+    Optional<WorkflowPredicate> untilPredicate =
+        Optional.ofNullable(consumePolicy.getAsyncApiMessageConsumptionPolicyUntil())
+            .map(AsyncApiMessageConsumptionPolicyUntil::getUntil)
+            .map(expr -> WorkflowUtils.buildPredicate(application, expr));
+
+    SubscriptionIterator foreach = subscription.getForeach();
+    TaskExecutor<?> foreachExecutor =
+        Optional.ofNullable(foreach)
+            .map(SubscriptionIterator::getDo)
+            .filter(tasks -> !tasks.isEmpty())
+            .map(tasks -> TaskExecutorHelper.createExecutorList(position, tasks, definition))
+            .orElse(null);
+    String foreachItem =
+        foreach != null && foreach.getItem() != null ? foreach.getItem() : DEFAULT_ITEM;
+    String foreachAt = foreach != null && foreach.getAt() != null ? foreach.getAt() : DEFAULT_INDEX;
+
+    return new AsyncAPIExecutor.SubscribeConfig(
+        filterPredicate,
+        consumePolicy,
+        consumeTimeout,
+        whilePredicate,
+        untilPredicate,
+        foreachExecutor,
+        foreachItem,
+        foreachAt);
+  }
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java
new file mode 100644
index 000000000..4c6aa9e2d
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import io.serverlessworkflow.api.WorkflowFormat;
+import io.serverlessworkflow.impl.resources.ExternalResourceHandler;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+
+public class AsyncAPIReader {
+
+  private AsyncAPIReader() {}
+
+  public static UnifiedAsyncAPI read(ExternalResourceHandler handler) {
+    try (InputStream is = handler.open()) {
+      return WorkflowFormat.fromFileName(handler.name())
+          .mapper()
+          .readValue(is, UnifiedAsyncAPI.class);
+    } catch (IOException e) {
+      throw new UncheckedIOException("Error reading AsyncAPI document " + handler.name(), e);
+    }
+  }
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelInfo.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelInfo.java
new file mode 100644
index 000000000..0dffb8546
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelInfo.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import java.net.URI;
+import java.util.Optional;
+
+public record AsyncApiChannelInfo(
+    URI serverUri, String channel, String operation, String protocol, Optional<String> authToken) {}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelProvider.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelProvider.java
new file mode 100644
index 000000000..4e890135d
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelProvider.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+public interface AsyncApiChannelProvider {
+  String ASYNC_API_CHANNEL_PROVIDER = "asyncApiChannelProvider";
+
+  CompletableFuture<Void> publish(
+      AsyncApiChannelInfo info, Map<String, Object> payload, Map<String, Object> headers);
+
+  AsyncApiSubscriptionHandle subscribe(
+      AsyncApiChannelInfo info, Consumer<AsyncApiInboundMessage> messageConsumer);
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiInboundMessage.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiInboundMessage.java
new file mode 100644
index 000000000..ddc45d6f4
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiInboundMessage.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+public record AsyncApiInboundMessage(
+    Map<String, Object> payload, Map<String, Object> headers, Optional<String> correlationId) {
+
+  public AsyncApiInboundMessage {
+    payload = payload != null ? payload : Collections.emptyMap();
+    headers = headers != null ? headers : Collections.emptyMap();
+    correlationId = correlationId != null ? correlationId : Optional.empty();
+  }
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiSubscriptionHandle.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiSubscriptionHandle.java
new file mode 100644
index 000000000..6af8b994d
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiSubscriptionHandle.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface AsyncApiSubscriptionHandle {
+  void unsubscribe();
+
+  CompletableFuture<Void> closed();
+}
diff --git a/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/UnifiedAsyncAPI.java b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/UnifiedAsyncAPI.java
new file mode 100644
index 000000000..218e34cd9
--- /dev/null
+++ b/impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/UnifiedAsyncAPI.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.asyncapi;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import java.util.Map;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public record UnifiedAsyncAPI(
+    String asyncapi,
+    Map<String, Server> servers,
+    Map<String, Channel> channels,
+    Map<String, Operation> operations) {
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public record Server(
+      String url,
+      String host,
+      String pathname,
+      String protocol,
+      Map<String, ServerVariable> variables) {
+    public String effectiveUrl() {
+      if (host != null) {
+        return host + (pathname != null ? pathname : "");
+      }
+      return url;
+    }
+  }
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public record ServerVariable(
+      @JsonProperty("default") String defaultValue,
+      @JsonProperty("enum") List<String> enumValues,
+      String description) {}
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public record Channel(String address, Map<String, Object> messages) {}
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public record Operation(String action, OperationChannel channel) {}
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public record OperationChannel(@JsonProperty("$ref") String ref) {
+    public String channelName() {
+      if (ref != null && ref.startsWith("#/channels/")) {
+        return ref.substring("#/channels/".length());
+      }
+      return ref;
+    }
+  }
+
+  public boolean isV3() {
+    return asyncapi != null && asyncapi.startsWith("3.");
+  }
+}
diff --git a/impl/asyncapi/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.CallableTaskBuilder b/impl/asyncapi/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.CallableTaskBuilder
new file mode 100644
index 000000000..d8747b25d
--- /dev/null
+++ b/impl/asyncapi/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.CallableTaskBuilder
@@ -0,0 +1 @@
+io.serverlessworkflow.impl.executors.asyncapi.AsyncAPIExecutorBuilder
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
index d125572cd..e393dcda0 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
@@ -114,6 +114,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory;
   private final ScheduledExecutorService schedulerExecutorService;
   private final Set<String> allowedCommands;
+  private final Map<Class<?>, ServiceLoader<?>> servicesLoaded = new ConcurrentHashMap<>();
 
   private WorkflowApplication(Builder builder) {
     this.taskFactory = builder.taskFactory;
@@ -708,4 +709,22 @@ public WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory() {
   public Set<String> allowedCommands() {
     return allowedCommands;
   }
+
+  @SuppressWarnings("unchecked")
+  public <T extends Comparable<?>> List<T> serviceLoadedClasses(Class<T> clazz) {
+    ServiceLoader<?> serviceLoader = servicesLoaded.computeIfAbsent(clazz, ServiceLoader::load);
+    return (List<T>) serviceLoader.stream().map(ServiceLoader.Provider::get).sorted().toList();
+  }
+
+  public <T extends Comparable<?>> T serviceLoadedClass(Class<T> serviceClass) {
+    ServiceLoader<?> serviceLoader =
+        servicesLoaded.computeIfAbsent(serviceClass, ServiceLoader::load);
+    return (T)
+        serviceLoader.stream()
+            .map(ServiceLoader.Provider::get)
+            .sorted()
+            .findFirst()
+            .orElseThrow(
+                () -> new IllegalStateException("No " + serviceClass + " implementation found"));
+  }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java
index e1e8fb6e2..3404ac1d2 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java
@@ -16,13 +16,13 @@
 package io.serverlessworkflow.impl.auth;
 
 import static io.serverlessworkflow.impl.WorkflowUtils.checkSecret;
-import static io.serverlessworkflow.impl.WorkflowUtils.loadFirst;
 import static io.serverlessworkflow.impl.WorkflowUtils.secret;
 
 import io.serverlessworkflow.api.types.OAuth2AuthenticationData;
 import io.serverlessworkflow.api.types.SecretBasedAuthenticationPolicy;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
@@ -35,14 +35,6 @@ public abstract class CommonOAuthProvider implements AuthProvider {
 
   private final WorkflowValueResolver<AccessTokenProvider> tokenProvider;
 
-  private static JWTConverter jwtConverter =
-      loadFirst(JWTConverter.class)
-          .orElseThrow(() -> new IllegalStateException("No JWTConverter implementation found"));
-
-  private static AccessTokenProviderFactory accessTokenProviderFactory =
-      loadFirst(AccessTokenProviderFactory.class)
-          .orElseThrow(() -> new IllegalStateException("No JWTConverter implementation found"));
-
   protected CommonOAuthProvider(WorkflowValueResolver<AccessTokenProvider> tokenProvider) {
     this.tokenProvider = tokenProvider;
   }
@@ -67,35 +59,42 @@ protected static OAuth2AuthenticationData fillFromMap(
   }
 
   protected static WorkflowValueResolver<AccessTokenProvider> accessToken(
+      WorkflowApplication app,
       Workflow workflow,
       OAuth2AuthenticationData authenticationData,
       SecretBasedAuthenticationPolicy secret,
       AuthRequestBuilder<?> builder) {
     if (authenticationData != null) {
-      return build(authenticationData, builder);
+      return build(authenticationData, builder, app);
     } else if (secret != null) {
-      return build(checkSecret(workflow, secret), builder);
+      return build(checkSecret(workflow, secret), builder, app);
     }
     throw new IllegalStateException("Both policy and secret are null");
   }
 
   private static WorkflowValueResolver<AccessTokenProvider> build(
-      OAuth2AuthenticationData authenticationData, AuthRequestBuilder authBuilder) {
+      OAuth2AuthenticationData authenticationData,
+      AuthRequestBuilder authBuilder,
+      WorkflowApplication app) {
     AccessTokenProvider tokenProvider =
-        accessTokenProviderFactory.build(
-            authBuilder.apply(authenticationData), authenticationData.getIssuers(), jwtConverter);
+        app.serviceLoadedClass(AccessTokenProviderFactory.class)
+            .build(
+                authBuilder.apply(authenticationData),
+                authenticationData.getIssuers(),
+                app.serviceLoadedClass(JWTConverter.class));
     return (w, t, m) -> tokenProvider;
   }
 
   private static WorkflowValueResolver<AccessTokenProvider> build(
-      String secretName, AuthRequestBuilder authBuilder) {
+      String secretName, AuthRequestBuilder authBuilder, WorkflowApplication app) {
     return (w, t, m) -> {
       Map<String, Object> secret = secret(w, secretName);
       String issuers = (String) secret.get("issuers");
-      return accessTokenProviderFactory.build(
-          authBuilder.apply(secret),
-          issuers != null ? Arrays.asList(issuers.split(",")) : null,
-          jwtConverter);
+      return app.serviceLoadedClass(AccessTokenProviderFactory.class)
+          .build(
+              authBuilder.apply(secret),
+              issuers != null ? Arrays.asList(issuers.split(",")) : null,
+              app.serviceLoadedClass(JWTConverter.class));
     };
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java
index 6bce3d814..16e0e1ea7 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java
@@ -24,6 +24,7 @@ public OAuth2AuthProvider(
       WorkflowApplication application, Workflow workflow, OAuthPolicyData policyData) {
     super(
         accessToken(
+            application,
             workflow,
             policyData.data(),
             policyData.secret(),
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java
index 80dd4138b..425e3c082 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java
@@ -24,6 +24,7 @@ public OpenIdAuthProvider(
       WorkflowApplication application, Workflow workflow, OAuthPolicyData policyData) {
     super(
         accessToken(
+            application,
             workflow,
             policyData.data(),
             policyData.secret(),
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
index 55363ac99..85a198d9b 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
@@ -18,6 +18,7 @@
 import io.serverlessworkflow.api.types.CallTask;
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallTaskExecutor.CallTaskExecutorBuilder;
@@ -32,9 +33,7 @@
 import io.serverlessworkflow.impl.executors.SwitchExecutor.SwitchExecutorBuilder;
 import io.serverlessworkflow.impl.executors.TryExecutor.TryExecutorBuilder;
 import io.serverlessworkflow.impl.executors.WaitExecutor.WaitExecutorBuilder;
-import java.util.Collection;
-import java.util.ServiceLoader;
-import java.util.ServiceLoader.Provider;
+import java.util.List;
 
 public class DefaultTaskExecutorFactory implements TaskExecutorFactory {
 
@@ -46,9 +45,6 @@ public static TaskExecutorFactory get() {
 
   protected DefaultTaskExecutorFactory() {}
 
-  private Collection<CallableTaskBuilder> callTasks =
-      ServiceLoader.load(CallableTaskBuilder.class).stream().map(Provider::get).sorted().toList();
-
   @Override
   public TaskExecutorBuilder<? extends TaskBase> getTaskExecutor(
       WorkflowMutablePosition position, Task task, WorkflowDefinition definition) {
@@ -57,7 +53,10 @@ public TaskExecutorBuilder<? extends TaskBase> getTaskExecutor(
       TaskBase taskBase = (TaskBase) callTask.get();
       if (taskBase != null) {
         return new CallTaskExecutorBuilder(
-            position, taskBase, definition, findCallTask(taskBase.getClass()));
+            position,
+            taskBase,
+            definition,
+            findCallTask(taskBase.getClass(), definition.application()));
       }
     } else if (task.getSwitchTask() != null) {
       return new SwitchExecutorBuilder(position, task.getSwitchTask(), definition);
@@ -86,7 +85,9 @@ public TaskExecutorBuilder<? extends TaskBase> getTaskExecutor(
   }
 
   @SuppressWarnings("unchecked")
-  private <T extends TaskBase> CallableTaskBuilder<T> findCallTask(Class<T> clazz) {
+  private <T extends TaskBase> CallableTaskBuilder<T> findCallTask(
+      Class<T> clazz, WorkflowApplication app) {
+    List<CallableTaskBuilder> callTasks = app.serviceLoadedClasses(CallableTaskBuilder.class);
     return (CallableTaskBuilder<T>)
         callTasks.stream()
             .filter(s -> s.accept(clazz))
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java
index f34253471..1d42b9b5d 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java
@@ -41,16 +41,10 @@
 import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
-import java.util.ServiceLoader;
 import java.util.concurrent.CompletableFuture;
 
 public class EmitExecutor extends RegularTaskExecutor<EmitTask> {
 
-  private static final Collection<EmittedEventDecorator> emittedDecorators =
-      ServiceLoader.load(EmittedEventDecorator.class).stream()
-          .map(ServiceLoader.Provider::get)
-          .sorted()
-          .toList();
   private final EventPropertiesBuilder props;
 
   public static class EmitExecutorBuilder
@@ -139,7 +133,11 @@ private CloudEvent buildCloudEvent(WorkflowContext workflow, TaskContext taskCon
         .additionalFilter()
         .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
         .ifPresent(value -> value.forEach((k, v) -> addExtension(ceBuilder, k, v)));
-    emittedDecorators.forEach(d -> d.decorate(ceBuilder, workflow, taskContext));
+    workflow
+        .definition()
+        .application()
+        .serviceLoadedClasses(EmittedEventDecorator.class)
+        .forEach(d -> d.decorate(ceBuilder, workflow, taskContext));
     return ceBuilder.build();
   }
 
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java
index 87198d540..02d2b813b 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java
@@ -27,7 +27,6 @@
 import io.serverlessworkflow.impl.scripts.ScriptRunner;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.ServiceLoader;
 
 public class RunScriptExecutorBuilder implements RunnableTaskBuilder<RunScript> {
 
@@ -65,10 +64,8 @@ public CallableTask build(RunScript taskConfiguration, WorkflowDefinition defini
                         m),
         taskConfiguration.isAwait(),
         taskConfiguration.getReturn(),
-        ServiceLoader.load(ScriptRunner.class).stream()
-            .map(ServiceLoader.Provider::get)
+        application.serviceLoadedClasses(ScriptRunner.class).stream()
             .filter(s -> s.identifier().equals(language))
-            .sorted()
             .findFirst()
             .orElseThrow(
                 () ->
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java
index c398a36d7..f1fe7861e 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java
@@ -22,17 +22,12 @@
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
-import java.util.ServiceLoader;
-import java.util.ServiceLoader.Provider;
 import java.util.concurrent.CompletableFuture;
 
 public class RunTaskExecutor extends RegularTaskExecutor<RunTask> {
 
   private final CallableTask runnable;
 
-  private static final ServiceLoader<RunnableTaskBuilder> runnables =
-      ServiceLoader.load(RunnableTaskBuilder.class);
-
   public static class RunTaskExecutorBuilder
       extends RegularTaskExecutorBuilder<RunTask, RunTaskExecutor> {
     private CallableTask runnable;
@@ -42,10 +37,8 @@ protected RunTaskExecutorBuilder(
       super(position, task, definition);
       RunTaskConfiguration config = task.getRun().get();
       this.runnable =
-          runnables.stream()
-              .map(Provider::get)
+          definition.application().serviceLoadedClasses(RunnableTaskBuilder.class).stream()
               .filter(r -> r.accept(config.getClass()))
-              .sorted()
               .findFirst()
               .map(r -> r.build(config, definition))
               .orElseThrow(
diff --git a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java
index 5c9f025bb..50693a5ee 100644
--- a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java
+++ b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java
@@ -30,13 +30,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.ServiceLoader;
 
 public class HttpExecutorBuilder {
 
   public static final String HTTP_REQUEST_DECORATOR_KEY = "HttpRequestDecorators";
   private final WorkflowDefinition definition;
-  private final List<HttpRequestDecorator> requestDecorators;
+  private List<HttpRequestDecorator> requestDecorators;
   private WorkflowValueResolver<URI> pathSupplier;
   private Object body;
   private String method = HttpMethod.GET;
@@ -47,13 +46,13 @@ public class HttpExecutorBuilder {
 
   private HttpExecutorBuilder(WorkflowDefinition definition) {
     this.definition = definition;
-    this.requestDecorators = new ArrayList<>();
+    this.requestDecorators =
+        new ArrayList<>(definition.application().serviceLoadedClasses(HttpRequestDecorator.class));
     requestDecorators.addAll(
         definition
             .application()
             .<Collection<HttpRequestDecorator>>additionalObject(HTTP_REQUEST_DECORATOR_KEY)
             .orElse(List.of()));
-    ServiceLoader.load(HttpRequestDecorator.class).forEach(requestDecorators::add);
     Collections.sort(requestDecorators);
   }
 
diff --git a/impl/pom.xml b/impl/pom.xml
index 5813c83be..45d0fe269 100644
--- a/impl/pom.xml
+++ b/impl/pom.xml
@@ -135,6 +135,11 @@
                 <artifactId>serverlessworkflow-impl-cron</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.serverlessworkflow</groupId>
+                <artifactId>serverlessworkflow-impl-asyncapi</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>net.thisptr</groupId>
                 <artifactId>jackson-jq</artifactId>
@@ -237,5 +242,6 @@
         <module>openapi-jackson</module>
         <module>a2a</module>
         <module>cron</module>
+        <module>asyncapi</module>
     </modules>
 </project>
diff --git a/impl/test/pom.xml b/impl/test/pom.xml
index 18f12414a..7656bec10 100644
--- a/impl/test/pom.xml
+++ b/impl/test/pom.xml
@@ -66,6 +66,10 @@
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-impl-cron</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-impl-asyncapi</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.glassfish.jersey.core</groupId>
             <artifactId>jersey-client</artifactId>
diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/AsyncAPITest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/AsyncAPITest.java
new file mode 100644
index 000000000..379b77a86
--- /dev/null
+++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/AsyncAPITest.java
@@ -0,0 +1,307 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.test;
+
+import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.executors.asyncapi.AsyncApiChannelInfo;
+import io.serverlessworkflow.impl.executors.asyncapi.AsyncApiChannelProvider;
+import io.serverlessworkflow.impl.executors.asyncapi.AsyncApiInboundMessage;
+import io.serverlessworkflow.impl.executors.asyncapi.AsyncApiSubscriptionHandle;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okio.Buffer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class AsyncAPITest {
+
+  private static WorkflowApplication app;
+  private static byte[] asyncApiSpec;
+  private static InMemoryAsyncApiChannelProvider channelProvider;
+
+  private MockWebServer specServer;
+
+  @BeforeAll
+  static void init() throws IOException {
+    try (InputStream is =
+        AsyncAPITest.class.getResourceAsStream("/schema/asyncapi/asyncapi.yaml")) {
+      asyncApiSpec = is.readAllBytes();
+    }
+    channelProvider = new InMemoryAsyncApiChannelProvider();
+    app =
+        WorkflowApplication.builder()
+            .withAdditionalObject(
+                AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER, (w, t) -> channelProvider)
+            .build();
+  }
+
+  @AfterAll
+  static void cleanup() {
+    app.close();
+  }
+
+  @BeforeEach
+  void setUp() throws IOException {
+    specServer = new MockWebServer();
+    specServer.start(8889);
+    channelProvider.reset();
+  }
+
+  @AfterEach
+  void tearDown() throws IOException {
+    specServer.shutdown();
+  }
+
+  @Test
+  void testPublishWithPayloadAndHeaders() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-publish.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    app.workflowDefinition(workflow).instance(Map.of("greeting", "Hello, World!")).start().get();
+
+    assertThat(channelProvider.published()).hasSize(1);
+
+    InMemoryAsyncApiChannelProvider.PublishRecord record = channelProvider.published().get(0);
+    assertThat(record.info().channel()).isEqualTo("greetings");
+    assertThat(record.info().operation()).isEqualTo("greet");
+    assertThat(record.info().protocol()).isEqualTo("kafka");
+    assertThat(record.info().serverUri().toString()).isEqualTo("kafka://127.0.0.1:9092");
+    assertThat(record.payload()).containsEntry("greeting", "Hello, World!");
+    assertThat(record.headers()).containsEntry("content-type", "application/json");
+  }
+
+  @Test
+  void testSubscribeWithAmountPolicy() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    channelProvider.preloadMessages(
+        "chat/inbox",
+        List.of(
+            new AsyncApiInboundMessage(
+                Map.of("roomId", "room-1", "message", "Hello"),
+                Map.of("sender", "alice"),
+                Optional.empty()),
+            new AsyncApiInboundMessage(
+                Map.of("roomId", "room-1", "message", "World"),
+                Map.of("sender", "bob"),
+                Optional.of("corr-123"))));
+
+    WorkflowModel result = app.workflowDefinition(workflow).instance().start().get();
+
+    Collection<WorkflowModel> messages = result.asCollection();
+    assertThat(messages).hasSize(2);
+
+    List<WorkflowModel> messageList = new ArrayList<>(messages);
+    Map<String, Object> first = messageList.get(0).asMap().orElseThrow();
+    assertThat((Map<String, Object>) first.get("payload")).containsEntry("message", "Hello");
+    assertThat((Map<String, Object>) first.get("headers")).containsEntry("sender", "alice");
+
+    Map<String, Object> second = messageList.get(1).asMap().orElseThrow();
+    assertThat((Map<String, Object>) second.get("payload")).containsEntry("message", "World");
+    assertThat(second).containsEntry("correlationId", "corr-123");
+  }
+
+  @Test
+  void testPublishWithLiteralPayload() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-publish.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    app.workflowDefinition(workflow).instance(Map.of("greeting", "Bonjour!")).start().get();
+
+    assertThat(channelProvider.published()).hasSize(1);
+    assertThat(channelProvider.published().get(0).payload()).containsEntry("greeting", "Bonjour!");
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testSubscribeWithFilter() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    channelProvider.preloadMessages(
+        "chat/inbox",
+        List.of(
+            new AsyncApiInboundMessage(
+                Map.of("roomId", "room-1", "message", "First"), Map.of(), Optional.empty()),
+            new AsyncApiInboundMessage(
+                Map.of("roomId", "room-2", "message", "Filtered out"), Map.of(), Optional.empty()),
+            new AsyncApiInboundMessage(
+                Map.of("roomId", "room-1", "message", "Second"), Map.of(), Optional.empty())));
+
+    WorkflowModel result = app.workflowDefinition(workflow).instance().start().get();
+
+    Collection<WorkflowModel> messages = result.asCollection();
+    assertThat(messages).hasSize(2);
+
+    List<WorkflowModel> messageList = new ArrayList<>(messages);
+    assertThat((Map<String, Object>) messageList.get(0).asMap().orElseThrow().get("payload"))
+        .containsEntry("message", "First");
+    assertThat((Map<String, Object>) messageList.get(1).asMap().orElseThrow().get("payload"))
+        .containsEntry("message", "Second");
+  }
+
+  @Test
+  void testSubscribeWithForeach() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    channelProvider.preloadMessages(
+        "chat/inbox",
+        List.of(
+            new AsyncApiInboundMessage(Map.of("message", "Hello"), Map.of(), Optional.empty())));
+
+    WorkflowModel result = app.workflowDefinition(workflow).instance().start().get();
+
+    Collection<WorkflowModel> messages = result.asCollection();
+    assertThat(messages).hasSize(1);
+
+    Map<String, Object> processed = messages.iterator().next().asMap().orElseThrow();
+    assertThat(processed).containsEntry("processed", true);
+    assertThat(processed).containsEntry("content", "Hello");
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testSubscribeWithUntilPolicy() throws Exception {
+    Workflow workflow =
+        readWorkflowFromClasspath("workflows-samples/asyncapi/asyncapi-subscribe-until.yaml");
+
+    specServer.enqueue(
+        new MockResponse()
+            .setBody(new Buffer().write(asyncApiSpec))
+            .setHeader("Content-Type", "application/yaml")
+            .setResponseCode(200));
+
+    channelProvider.preloadMessages(
+        "chat/inbox",
+        List.of(
+            new AsyncApiInboundMessage(Map.of("message", "One"), Map.of(), Optional.empty()),
+            new AsyncApiInboundMessage(Map.of("message", "Two"), Map.of(), Optional.empty()),
+            new AsyncApiInboundMessage(
+                Map.of("message", "Three - should not appear"), Map.of(), Optional.empty())));
+
+    WorkflowModel result = app.workflowDefinition(workflow).instance().start().get();
+
+    Collection<WorkflowModel> messages = result.asCollection();
+    assertThat(messages).hasSize(2);
+
+    List<WorkflowModel> messageList = new ArrayList<>(messages);
+    assertThat((Map<String, Object>) messageList.get(0).asMap().orElseThrow().get("payload"))
+        .containsEntry("message", "One");
+    assertThat((Map<String, Object>) messageList.get(1).asMap().orElseThrow().get("payload"))
+        .containsEntry("message", "Two");
+  }
+
+  static class InMemoryAsyncApiChannelProvider implements AsyncApiChannelProvider {
+    private final List<PublishRecord> publishedMessages = new CopyOnWriteArrayList<>();
+    private final Map<String, List<AsyncApiInboundMessage>> preloaded = new ConcurrentHashMap<>();
+
+    @Override
+    public CompletableFuture<Void> publish(
+        AsyncApiChannelInfo info, Map<String, Object> payload, Map<String, Object> headers) {
+      publishedMessages.add(new PublishRecord(info, payload, headers));
+      return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public AsyncApiSubscriptionHandle subscribe(
+        AsyncApiChannelInfo info, Consumer<AsyncApiInboundMessage> messageConsumer) {
+      CompletableFuture<Void> closed = new CompletableFuture<>();
+      List<AsyncApiInboundMessage> messages = preloaded.remove(info.channel());
+      if (messages != null) {
+        messages.forEach(messageConsumer);
+      }
+      return new AsyncApiSubscriptionHandle() {
+        @Override
+        public void unsubscribe() {
+          closed.complete(null);
+        }
+
+        @Override
+        public CompletableFuture<Void> closed() {
+          return closed;
+        }
+      };
+    }
+
+    void preloadMessages(String channel, List<AsyncApiInboundMessage> messages) {
+      preloaded.put(channel, new ArrayList<>(messages));
+    }
+
+    List<PublishRecord> published() {
+      return Collections.unmodifiableList(publishedMessages);
+    }
+
+    void reset() {
+      publishedMessages.clear();
+      preloaded.clear();
+    }
+
+    record PublishRecord(
+        AsyncApiChannelInfo info, Map<String, Object> payload, Map<String, Object> headers) {}
+  }
+}
diff --git a/impl/test/src/test/resources/schema/asyncapi/asyncapi.yaml b/impl/test/src/test/resources/schema/asyncapi/asyncapi.yaml
new file mode 100644
index 000000000..64e7130cb
--- /dev/null
+++ b/impl/test/src/test/resources/schema/asyncapi/asyncapi.yaml
@@ -0,0 +1,50 @@
+asyncapi: 3.0.0
+info:
+  title: Greeting Service
+  version: 1.0.0
+  description: A sample AsyncAPI service for testing
+
+servers:
+  greetingsServer:
+    host: 127.0.0.1:9092
+    protocol: kafka
+    variables:
+      environment:
+        default: dev
+        enum:
+          - dev
+          - staging
+          - production
+        description: Deployment environment
+
+channels:
+  greetings:
+    address: greetings
+    messages:
+      GreetingMessage:
+        payload:
+          type: object
+          properties:
+            greeting:
+              type: string
+  chatInbox:
+    address: chat/inbox
+    messages:
+      ChatMessage:
+        payload:
+          type: object
+          properties:
+            roomId:
+              type: string
+            message:
+              type: string
+
+operations:
+  greet:
+    action: send
+    channel:
+      $ref: '#/channels/greetings'
+  chatInbox:
+    action: receive
+    channel:
+      $ref: '#/channels/chatInbox'
\ No newline at end of file
diff --git a/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-publish.yaml b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-publish.yaml
new file mode 100644
index 000000000..f50a2b2f4
--- /dev/null
+++ b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-publish.yaml
@@ -0,0 +1,19 @@
+document:
+  dsl: '1.0.3'
+  namespace: test
+  name: asyncapi-publish
+  version: '0.1.0'
+do:
+  - publishGreeting:
+      call: asyncapi
+      with:
+        document:
+          endpoint: http://127.0.0.1:8889/asyncapi.yaml
+        operation: greet
+        server:
+          name: greetingsServer
+        message:
+          payload:
+            greeting: '${ .greeting }'
+          headers:
+            content-type: application/json
\ No newline at end of file
diff --git a/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml
new file mode 100644
index 000000000..f07b35935
--- /dev/null
+++ b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml
@@ -0,0 +1,17 @@
+document:
+  dsl: '1.0.3'
+  namespace: test
+  name: asyncapi-subscribe-amount
+  version: '0.1.0'
+do:
+  - subscribeToChatInbox:
+      call: asyncapi
+      with:
+        document:
+          endpoint: http://127.0.0.1:8889/asyncapi.yaml
+        operation: chatInbox
+        server:
+          name: greetingsServer
+        subscription:
+          consume:
+            amount: 2
\ No newline at end of file
diff --git a/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml
new file mode 100644
index 000000000..ecd13d7c1
--- /dev/null
+++ b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml
@@ -0,0 +1,18 @@
+document:
+  dsl: '1.0.3'
+  namespace: test
+  name: asyncapi-subscribe-filter
+  version: '0.1.0'
+do:
+  - subscribeToChatInbox:
+      call: asyncapi
+      with:
+        document:
+          endpoint: http://127.0.0.1:8889/asyncapi.yaml
+        operation: chatInbox
+        server:
+          name: greetingsServer
+        subscription:
+          filter: '${ .payload.roomId == "room-1" }'
+          consume:
+            amount: 2
\ No newline at end of file
diff --git a/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml
new file mode 100644
index 000000000..f199c3c2b
--- /dev/null
+++ b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml
@@ -0,0 +1,24 @@
+document:
+  dsl: '1.0.3'
+  namespace: test
+  name: asyncapi-subscribe-foreach
+  version: '0.1.0'
+do:
+  - subscribeToChatInbox:
+      call: asyncapi
+      with:
+        document:
+          endpoint: http://127.0.0.1:8889/asyncapi.yaml
+        operation: chatInbox
+        server:
+          name: greetingsServer
+        subscription:
+          consume:
+            amount: 1
+          foreach:
+            item: msg
+            do:
+              - transform:
+                  set:
+                    processed: true
+                    content: '${ $msg.payload.message }'
\ No newline at end of file
diff --git a/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-until.yaml b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-until.yaml
new file mode 100644
index 000000000..f30879f88
--- /dev/null
+++ b/impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-until.yaml
@@ -0,0 +1,17 @@
+document:
+  dsl: '1.0.3'
+  namespace: test
+  name: asyncapi-subscribe-until
+  version: '0.1.0'
+do:
+  - subscribeToChatInbox:
+      call: asyncapi
+      with:
+        document:
+          endpoint: http://127.0.0.1:8889/asyncapi.yaml
+        operation: chatInbox
+        server:
+          name: greetingsServer
+        subscription:
+          consume:
+            until: '${ (. | length) >= 2 }'
\ No newline at end of file