getWorld
Access the World instance for low-level storage, queuing, and streaming operations.
Retrieves the World instance for direct access to workflow storage, queuing, and streaming backends. This function returns a World which provides low-level access to manage workflow runs, steps, events, and hooks.
Use this function when you need direct access to the underlying workflow infrastructure, such as listing all runs, querying events, or implementing custom workflow management logic.
import { getWorld } from "workflow/runtime";
const world = getWorld();API Signature
Parameters
This function does not accept any parameters.
Returns
Returns a World object:
| Name | Type | Description |
|---|---|---|
start | () => Promise<void> | A function that will be called to start any background tasks needed by the World implementation. For example, in the case of a queue backed World, this would start the queue processing. |
close | () => Promise<void> | Release any resources held by the World implementation (connection pools, listeners, etc.).
After calling close(), the World instance should not be used again.
This is important for CLI commands and short-lived processes that need to exit cleanly
without relying on process.exit(). |
getEncryptionKeyForRun | (run: string | { runId: string; deploymentId: string; workflowName: string; input: unknown; createdAt: Date; updatedAt: Date; status: "pending" | "running"; output: undefined; error: undefined; ... 4 more ...; startedAt?: Date | undefined; } | { ...; } | { ...; } | { ...; }) => Promise<...> | Retrieve the AES-256 encryption key for a specific workflow run.
The returned key is a ready-to-use 32-byte AES-256 key. The World
implementation handles all key retrieval and derivation internally
(e.g., HKDF from a deployment key). The core encryption module uses
this key directly for AES-GCM encrypt/decrypt operations.
Accepts either a full WorkflowRun object or a plain runId string:
- WorkflowRun — Used by the o11y/CLI path when the run entity is
already available. Provides deploymentId for cross-deployment key
resolution without a redundant lookup.
- string (runId) — Used by start() and step-handler in the
runtime path where the run entity may not exist yet or isn't needed.
The World assumes the current deployment for key resolution.
When not implemented, encryption is disabled — data is stored unencrypted. |
getDeploymentId | () => Promise<string> | |
queue | (queueName: __wkf_step_${string} | __wkf_workflow_${string}, message: { runId: string; traceCarrier?: Record<string, string> | undefined; requestedAt?: Date | undefined; serverErrorRetryCount?: number | undefined; } | { ...; } | { ...; }, opts?: QueueOptions | undefined) => Promise<...> | Enqueues a message to the specified queue. |
createQueueHandler | (queueNamePrefix: "__wkf_step_" | "__wkf_workflow_", handler: (message: unknown, meta: { attempt: number; queueName: __wkf_step_${string} | __wkf_workflow_${string}; messageId: string & $brand<"MessageId">; }) => Promise<...>) => (req: Request) => Promise<...> | Creates an HTTP queue handler for processing messages from a specific queue. |
runs | { get(id: string, params: GetWorkflowRunParams & { resolveData: "none"; }): Promise<WorkflowRunWithoutData>; get(id: string, params?: (GetWorkflowRunParams & { ...; }) | undefined): Promise<...>; get(id: string, params?: GetWorkflowRunParams | undefined): Promise<...>; list(params: ListWorkflowRunsParams & { ...; })... | |
steps | { get(runId: string | undefined, stepId: string, params: GetStepParams & { resolveData: "none"; }): Promise<StepWithoutData>; get(runId: string | undefined, stepId: string, params?: (GetStepParams & { ...; }) | undefined): Promise<...>; get(runId: string | undefined, stepId: string, params?: GetStepParams | undefine... | |
events | { create(runId: string | null, data: { eventType: "run_created"; eventData: { deploymentId: string; workflowName: string; input: unknown; executionContext?: Record<string, any> | undefined; }; correlationId?: string | undefined; specVersion?: number | undefined; }, params?: CreateEventParams | undefined): Promise<..... | |
hooks | { get(hookId: string, params?: GetHookParams | undefined): Promise<Hook>; getByToken(token: string, params?: GetHookParams | undefined): Promise<...>; list(params: ListHooksParams): Promise<...>; } | |
writeToStream | (name: string, runId: string, chunk: string | Uint8Array<ArrayBufferLike>) => Promise<void> | |
writeToStreamMulti | (name: string, runId: string, chunks: (string | Uint8Array<ArrayBufferLike>)[]) => Promise<void> | Write multiple chunks to a stream in a single operation. This is an optional optimization for world implementations that can batch multiple writes efficiently (e.g., single HTTP request for world-vercel). If not implemented, the caller should fall back to sequential writeToStream() calls. |
closeStream | (name: string, runId: string) => Promise<void> | |
readFromStream | (name: string, startIndex?: number | undefined) => Promise<ReadableStream<Uint8Array<ArrayBufferLike>>> | |
listStreamsByRunId | (runId: string) => Promise<string[]> |
Examples
List Workflow Runs
List all workflow runs with pagination:
import { getWorld } from "workflow/runtime";
export async function GET(req: Request) {
const url = new URL(req.url);
const cursor = url.searchParams.get("cursor") ?? undefined;
try {
const world = getWorld();
const runs = await world.runs.list({
pagination: { cursor },
});
return Response.json(runs);
} catch (error) {
return Response.json(
{ error: "Failed to list workflow runs" },
{ status: 500 }
);
}
}Cancel a Workflow Run
Cancel a running workflow:
import { getWorld } from "workflow/runtime";
export async function POST(req: Request) {
const { runId } = await req.json();
if (!runId) {
return Response.json({ error: "No runId provided" }, { status: 400 });
}
try {
const world = getWorld();
const run = await world.runs.cancel(runId);
return Response.json({ status: run.status });
} catch (error) {
return Response.json(
{ error: "Failed to cancel workflow run" },
{ status: 500 }
);
}
}