diff --git a/.changeset/moody-seals-smash.md b/.changeset/moody-seals-smash.md
new file mode 100644
index 000000000000..0445665d00a4
--- /dev/null
+++ b/.changeset/moody-seals-smash.md
@@ -0,0 +1,5 @@
+---
+"wrangler": minor
+---
+
+add `wrangler workflows ...` commands
diff --git a/packages/wrangler/package.json b/packages/wrangler/package.json
index b15a93128ef6..8e66ff370b4e 100644
--- a/packages/wrangler/package.json
+++ b/packages/wrangler/package.json
@@ -75,7 +75,9 @@
 		"@esbuild-plugins/node-modules-polyfill": "^0.2.2",
 		"blake3-wasm": "^2.1.5",
 		"chokidar": "^3.5.3",
+		"date-fns": "^4.1.0",
 		"esbuild": "0.17.19",
+		"itty-time": "^1.0.6",
 		"miniflare": "workspace:*",
 		"nanoid": "^3.3.3",
 		"path-to-regexp": "^6.3.0",
diff --git a/packages/wrangler/src/__tests__/core/command-registration.test.ts b/packages/wrangler/src/__tests__/core/command-registration.test.ts
index 12f24a560120..538b38761cc4 100644
--- a/packages/wrangler/src/__tests__/core/command-registration.test.ts
+++ b/packages/wrangler/src/__tests__/core/command-registration.test.ts
@@ -191,6 +191,7 @@ describe("Command Registration", () => {
 			  wrangler pubsub                          📮 Manage Pub/Sub brokers [private beta]
 			  wrangler dispatch-namespace              🏗️  Manage dispatch namespaces
 			  wrangler ai                              🤖 Manage AI models
+			  wrangler workflows                       🔁 Manage Workflows [open-beta]
 			  wrangler login                           🔓 Login to Cloudflare
 			  wrangler logout                          🚪 Logout from Cloudflare
 			  wrangler whoami                          🕵️  Retrieve your user information
diff --git a/packages/wrangler/src/__tests__/index.test.ts b/packages/wrangler/src/__tests__/index.test.ts
index 49c9b2aaed2d..ac55c7326f60 100644
--- a/packages/wrangler/src/__tests__/index.test.ts
+++ b/packages/wrangler/src/__tests__/index.test.ts
@@ -61,6 +61,7 @@ describe("wrangler", () => {
 				  wrangler pubsub                 📮 Manage Pub/Sub brokers [private beta]
 				  wrangler dispatch-namespace     🏗️  Manage dispatch namespaces
 				  wrangler ai                     🤖 Manage AI models
+				  wrangler workflows              🔁 Manage Workflows [open-beta]
 				  wrangler login                  🔓 Login to Cloudflare
 				  wrangler logout                 🚪 Logout from Cloudflare
 				  wrangler whoami                 🕵️  Retrieve your user information
@@ -117,6 +118,7 @@ describe("wrangler", () => {
 				  wrangler pubsub                 📮 Manage Pub/Sub brokers [private beta]
 				  wrangler dispatch-namespace     🏗️  Manage dispatch namespaces
 				  wrangler ai                     🤖 Manage AI models
+				  wrangler workflows              🔁 Manage Workflows [open-beta]
 				  wrangler login                  🔓 Login to Cloudflare
 				  wrangler logout                 🚪 Logout from Cloudflare
 				  wrangler whoami                 🕵️  Retrieve your user information
diff --git a/packages/wrangler/src/__tests__/versions/versions.help.test.ts b/packages/wrangler/src/__tests__/versions/versions.help.test.ts
index 859c2c5886f7..1986212e95d6 100644
--- a/packages/wrangler/src/__tests__/versions/versions.help.test.ts
+++ b/packages/wrangler/src/__tests__/versions/versions.help.test.ts
@@ -37,6 +37,7 @@ describe("versions --help", () => {
 			  wrangler pubsub                    📮 Manage Pub/Sub brokers [private beta]
 			  wrangler dispatch-namespace        🏗️  Manage dispatch namespaces
 			  wrangler ai                        🤖 Manage AI models
+			  wrangler workflows                 🔁 Manage Workflows [open-beta]
 			  wrangler login                     🔓 Login to Cloudflare
 			  wrangler logout                    🚪 Logout from Cloudflare
 			  wrangler whoami                    🕵️  Retrieve your user information
diff --git a/packages/wrangler/src/core/teams.d.ts b/packages/wrangler/src/core/teams.d.ts
index 82499b0fab84..204e2312f745 100644
--- a/packages/wrangler/src/core/teams.d.ts
+++ b/packages/wrangler/src/core/teams.d.ts
@@ -14,4 +14,5 @@ export type Teams =
 	| "Product: AI"
 	| "Product: Hyperdrive"
 	| "Product: Vectorize"
+	| "Product: Workflows"
 	| "Product: Cloudchamber";
diff --git a/packages/wrangler/src/index.ts b/packages/wrangler/src/index.ts
index b6b1422beed3..458eb44894ad 100644
--- a/packages/wrangler/src/index.ts
+++ b/packages/wrangler/src/index.ts
@@ -43,6 +43,7 @@ import { generateHandler, generateOptions } from "./generate";
 import { hyperdrive } from "./hyperdrive/index";
 import { initHandler, initOptions } from "./init";
 import "./kv";
+import "./workflows";
 import { logBuildFailure, logger, LOGGER_LEVELS } from "./logger";
 import * as metrics from "./metrics";
 import { mTlsCertificateCommands } from "./mtls-certificate/cli";
@@ -608,6 +609,9 @@ export function createCLIParser(argv: string[]) {
 		return ai(aiYargs.command(subHelp));
 	});
 
+	// workflows
+	register.registerNamespace("workflows");
+
 	// pipelines
 	wrangler.command("pipelines", false, (pipelinesYargs) => {
 		return pipelines(pipelinesYargs.command(subHelp));
diff --git a/packages/wrangler/src/workflows/commands/delete.ts b/packages/wrangler/src/workflows/commands/delete.ts
new file mode 100644
index 000000000000..c5b1e00aa7c9
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/delete.ts
@@ -0,0 +1,33 @@
+import { fetchResult } from "../../cfetch";
+import { defineCommand } from "../../core";
+import { logger } from "../../logger";
+import { requireAuth } from "../../user";
+
+defineCommand({
+	command: "wrangler workflows delete",
+	metadata: {
+		description:
+			"Delete workflow - when deleting a workflow, it will also delete it's own instances",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+	},
+	positionalArgs: ["name"],
+
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		await fetchResult(`/accounts/${accountId}/workflows/${args.name}`, {
+			method: "DELETE",
+		});
+
+		logger.info(`Workflow "${args.name}" was successfully removed`);
+	},
+});
diff --git a/packages/wrangler/src/workflows/commands/describe.ts b/packages/wrangler/src/workflows/commands/describe.ts
new file mode 100644
index 000000000000..da7d33564543
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/describe.ts
@@ -0,0 +1,59 @@
+import { logRaw } from "@cloudflare/cli";
+import { white } from "@cloudflare/cli/colors";
+import { fetchResult } from "../../cfetch";
+import { defineCommand } from "../../core";
+import { requireAuth } from "../../user";
+import formatLabelledValues from "../../utils/render-labelled-values";
+import type { Version, Workflow } from "../types";
+
+defineCommand({
+	command: "wrangler workflows describe",
+	metadata: {
+		description: "Describe Workflow resource",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+	},
+	positionalArgs: ["name"],
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		const workflow = await fetchResult<Workflow>(
+			`/accounts/${accountId}/workflows/${args.name}`
+		);
+
+		const versions = await fetchResult<Version[]>(
+			`/accounts/${accountId}/workflows/${args.name}/versions`
+		);
+
+		const latestVersion = versions[0];
+
+		logRaw(
+			formatLabelledValues({
+				Name: workflow.name,
+				Id: workflow.id,
+				"Script Name": workflow.script_name,
+				"Class Name": workflow.class_name,
+				"Created On": workflow.created_on,
+				"Modified On": workflow.modified_on,
+			})
+		);
+		logRaw(white("Latest Version:"));
+		logRaw(
+			formatLabelledValues(
+				{
+					Id: latestVersion.id,
+					"Created On": workflow.created_on,
+					"Modified On": workflow.modified_on,
+				},
+				{ indentationCount: 2 }
+			)
+		);
+	},
+});
diff --git a/packages/wrangler/src/workflows/commands/instances/describe.ts b/packages/wrangler/src/workflows/commands/instances/describe.ts
new file mode 100644
index 000000000000..f7327e0661a3
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/instances/describe.ts
@@ -0,0 +1,279 @@
+import { logRaw } from "@cloudflare/cli";
+import { red, white } from "@cloudflare/cli/colors";
+import {
+	addMilliseconds,
+	formatDistanceStrict,
+	formatDistanceToNowStrict,
+} from "date-fns";
+import { ms } from "itty-time";
+import { fetchResult } from "../../../cfetch";
+import { defineCommand } from "../../../core";
+import { logger } from "../../../logger";
+import { requireAuth } from "../../../user";
+import formatLabelledValues from "../../../utils/render-labelled-values";
+import {
+	emojifyInstanceStatus,
+	emojifyInstanceTriggerName,
+	emojifyStepType,
+} from "../../utils";
+import type {
+	Instance,
+	InstanceSleepLog,
+	InstanceStatusAndLogs,
+	InstanceStepLog,
+	InstanceTerminateLog,
+} from "../../types";
+
+const command = defineCommand({
+	command: "wrangler workflows instances describe",
+
+	metadata: {
+		description:
+			"Describe a workflow instance - see its logs, retries and errors",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+
+	positionalArgs: ["name", "id"],
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+		id: {
+			describe:
+				"ID of the instance - instead of an UUID you can type 'latest' to get the latest instance and describe it",
+			type: "string",
+			demandOption: true,
+		},
+		"step-output": {
+			describe:
+				"Don't output the step output since it might clutter the terminal",
+			type: "boolean",
+			default: true,
+		},
+		"truncate-output-limit": {
+			describe: "Truncate step output after x characters",
+			type: "number",
+			default: 5000,
+		},
+	},
+
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		let id = args.id;
+
+		if (id == "latest") {
+			const instances = (
+				await fetchResult<Instance[]>(
+					`/accounts/${accountId}/workflows/${args.name}/instances`
+				)
+			).sort((a, b) => b.created_on.localeCompare(a.created_on));
+
+			if (instances.length == 0) {
+				logger.error(
+					`There are no deployed instances in workflow "${args.name}"`
+				);
+				return;
+			}
+
+			id = instances[0].id;
+		}
+
+		const instance = await fetchResult<InstanceStatusAndLogs>(
+			`/accounts/${accountId}/workflows/${args.name}/instances/${id}`
+		);
+
+		const formattedInstance: Record<string, string> = {
+			"Workflow Name": args.name,
+			"Instance Id": id,
+			"Version Id": instance.versionId,
+			Status: emojifyInstanceStatus(instance.status),
+			Trigger: emojifyInstanceTriggerName(instance.trigger.source),
+			Queued: new Date(instance.queued).toLocaleString(),
+		};
+
+		if (instance.success != null) {
+			formattedInstance.Success = instance.success ? "✅ Yes" : "❌ No";
+		}
+
+		// date related stuff, if the workflow is still running assume duration until now
+		if (instance.start != undefined) {
+			formattedInstance.Start = new Date(instance.start).toLocaleString();
+		}
+
+		if (instance.end != undefined) {
+			formattedInstance.End = new Date(instance.end).toLocaleString();
+		}
+
+		if (instance.start != null && instance.end != null) {
+			formattedInstance.Duration = formatDistanceStrict(
+				new Date(instance.end),
+				new Date(instance.start)
+			);
+		} else if (instance.start != null) {
+			// Convert current date to UTC
+			formattedInstance.Duration = formatDistanceStrict(
+				new Date(instance.start),
+				new Date(new Date().toUTCString().slice(0, -4))
+			);
+		}
+
+		const lastSuccessfulStepName = getLastSuccessfulStep(instance);
+		if (lastSuccessfulStepName != null) {
+			formattedInstance["Last Successful Step"] = lastSuccessfulStepName;
+		}
+
+		// display the error if the instance errored out
+		if (instance.error != null) {
+			formattedInstance.Error = red(
+				`${instance.error.name}: ${instance.error.message}`
+			);
+		}
+
+		logRaw(formatLabelledValues(formattedInstance));
+		logRaw(white("Steps:"));
+
+		instance.steps.forEach(logStep.bind(false, args));
+	},
+});
+
+function logStep(
+	args: typeof command.args,
+	step: InstanceStepLog | InstanceSleepLog | InstanceTerminateLog
+) {
+	logRaw("");
+	const formattedStep: Record<string, string> = {};
+
+	if (step.type == "sleep" || step.type == "step") {
+		formattedStep.Name = step.name;
+		formattedStep.Type = emojifyStepType(step.type);
+
+		// date related stuff, if the step is still running assume duration until now
+		if (step.start != undefined) {
+			formattedStep.Start = new Date(step.start).toLocaleString();
+		}
+
+		if (step.end != undefined) {
+			formattedStep.End = new Date(step.end).toLocaleString();
+		}
+
+		if (step.start != null && step.end != null) {
+			formattedStep.Duration = formatDistanceStrict(
+				new Date(step.end),
+				new Date(step.start)
+			);
+		} else if (step.start != null) {
+			// Convert current date to UTC
+			formattedStep.Duration = formatDistanceStrict(
+				new Date(step.start),
+				new Date(new Date().toUTCString().slice(0, -4))
+			);
+		}
+	} else if (step.type == "termination") {
+		formattedStep.Type = emojifyStepType(step.type);
+		formattedStep.Trigger = step.trigger.source;
+	}
+
+	if (step.type == "step") {
+		if (step.success !== null) {
+			formattedStep.Success = step.success ? "✅ Yes" : "❌ No";
+		} else {
+			formattedStep.Success = "▶ Running";
+		}
+
+		if (step.success === null) {
+			const latestAttempt = step.attempts.at(-1);
+			let delay = step.config.retries.delay;
+			if (latestAttempt !== undefined && latestAttempt.success === false) {
+				// SAFETY: It's okay because end date must always exist in the API, otherwise it's okay to fail
+				// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+				const endDate = new Date(latestAttempt.end!);
+				if (typeof delay === "string") {
+					delay = ms(delay);
+				}
+				const retryDate = addMilliseconds(endDate, delay);
+				formattedStep["Retries At"] =
+					`${retryDate.toLocaleString()} (in ${formatDistanceToNowStrict(retryDate)} from now)`;
+			}
+		}
+		if (step.output !== undefined && args.stepOutput) {
+			let output: string;
+			try {
+				output = JSON.stringify(step.output);
+			} catch {
+				output = step.output as string;
+			}
+			formattedStep.Output =
+				output.length > args.truncateOutputLimit
+					? output.substring(0, args.truncateOutputLimit) +
+						"[...output truncated]"
+					: output;
+		}
+	}
+
+	logRaw(formatLabelledValues(formattedStep, { indentationCount: 2 }));
+
+	if (step.type == "step") {
+		const prettyAttempts = step.attempts.map((val) => {
+			const attempt: Record<string, string> = {};
+
+			attempt.Start = new Date(val.start).toLocaleString();
+			attempt.End = val.end == null ? "" : new Date(val.end).toLocaleString();
+
+			if (val.start != null && val.end != null) {
+				attempt.Duration = formatDistanceStrict(
+					new Date(val.end),
+					new Date(val.start)
+				);
+			} else if (val.start != null) {
+				// Converting datetimes into UTC is very cool in JS
+				attempt.Duration = formatDistanceStrict(
+					new Date(val.start),
+					new Date(new Date().toUTCString().slice(0, -4))
+				);
+			}
+
+			attempt.State =
+				val.success == null
+					? "🔄 Working"
+					: val.success
+						? "✅ Success"
+						: "❌ Error";
+
+			// This is actually safe to do while logger.table only considers the first element as keys.
+			// Because if there's an error, the first row will always be an error
+			if (val.error != null) {
+				attempt.Error = red(`${val.error.name}: ${val.error.message}`);
+			}
+			return attempt;
+		});
+
+		logger.table(prettyAttempts);
+	}
+}
+
+function getLastSuccessfulStep(logs: InstanceStatusAndLogs): string | null {
+	let lastSuccessfulStepName: string | null = null;
+
+	for (const step of logs.steps) {
+		switch (step.type) {
+			case "step":
+				if (step.success == true) {
+					lastSuccessfulStepName = step.name;
+				}
+				break;
+			case "sleep":
+				if (step.end != null) {
+					lastSuccessfulStepName = step.name;
+				}
+				break;
+			case "termination":
+				break;
+		}
+	}
+
+	return lastSuccessfulStepName;
+}
diff --git a/packages/wrangler/src/workflows/commands/instances/list.ts b/packages/wrangler/src/workflows/commands/instances/list.ts
new file mode 100644
index 000000000000..8ad1166d41c7
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/instances/list.ts
@@ -0,0 +1,94 @@
+import { fetchResult } from "../../../cfetch";
+import { defineCommand } from "../../../core";
+import { logger } from "../../../logger";
+import { requireAuth } from "../../../user";
+import { emojifyInstanceStatus, validateStatus } from "../../utils";
+import type { Instance } from "../../types";
+
+defineCommand({
+	command: "wrangler workflows instances list",
+
+	metadata: {
+		description: "Instance related commands (list, describe, terminate...)",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+
+	positionalArgs: ["name"],
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+		reverse: {
+			describe: "Reverse order of the instances table",
+			type: "boolean",
+			default: false,
+		},
+		status: {
+			describe:
+				"Filters list by instance status (can be one of: queued, running, paused, errored, terminated, complete)",
+			type: "string",
+		},
+		page: {
+			describe:
+				'Show a sepecific page from the listing, can configure page size using "per-page"',
+			type: "number",
+			default: 1,
+		},
+		"per-page": {
+			describe: "Configure the maximum number of instances to show per page",
+			type: "number",
+		},
+	},
+
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		const URLParams = new URLSearchParams();
+
+		if (args.status !== undefined) {
+			const validatedStatus = validateStatus(args.status);
+			URLParams.set("status", validatedStatus);
+		}
+		if (args.perPage !== undefined) {
+			URLParams.set("per_page", args.perPage.toString());
+		}
+
+		URLParams.set("page", args.page.toString());
+
+		const instances = await fetchResult<Instance[]>(
+			`/accounts/${accountId}/workflows/${args.name}/instances`,
+			undefined,
+			URLParams
+		);
+
+		if (instances.length === 0) {
+			logger.warn(
+				`There are no instances in workflow "${args.name}". You can trigger it with "wrangler workflows trigger ${args.name}"`
+			);
+			return;
+		}
+
+		logger.info(
+			`Showing ${instances.length} instance${instances.length > 1 ? "s" : ""} from page ${args.page}:`
+		);
+
+		const prettierInstances = instances
+			.sort((a, b) =>
+				args.reverse
+					? a.modified_on.localeCompare(b.modified_on)
+					: b.modified_on.localeCompare(a.modified_on)
+			)
+			.map((instance) => ({
+				Id: instance.id,
+				Version: instance.version_id,
+				Created: new Date(instance.created_on).toLocaleString(),
+				Modified: new Date(instance.modified_on).toLocaleString(),
+				Status: emojifyInstanceStatus(instance.status),
+			}));
+
+		logger.table(prettierInstances);
+	},
+});
diff --git a/packages/wrangler/src/workflows/commands/instances/terminate.ts b/packages/wrangler/src/workflows/commands/instances/terminate.ts
new file mode 100644
index 000000000000..c82c460b43ee
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/instances/terminate.ts
@@ -0,0 +1,64 @@
+import { fetchResult } from "../../../cfetch";
+import { defineCommand } from "../../../core";
+import { logger } from "../../../logger";
+import { requireAuth } from "../../../user";
+import type { Instance } from "../../types";
+
+defineCommand({
+	command: "wrangler workflows instances terminate",
+
+	metadata: {
+		description: "Terminate a workflow instance",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+
+	positionalArgs: ["name", "id"],
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+		id: {
+			describe:
+				"ID of the instance - instead of an UUID you can type 'latest' to get the latest instance and describe it",
+			type: "string",
+			demandOption: true,
+		},
+	},
+
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		let id = args.id;
+
+		if (id == "latest") {
+			const instances = (
+				await fetchResult<Instance[]>(
+					`/accounts/${accountId}/workflows/${args.name}/instances`
+				)
+			).sort((a, b) => b.created_on.localeCompare(a.created_on));
+
+			if (instances.length == 0) {
+				logger.error(
+					`There are no deployed instances in workflow "${args.name}"`
+				);
+				return;
+			}
+
+			id = instances[0].id;
+		}
+
+		await fetchResult(
+			`/accounts/${accountId}/workflows/${args.name}/instances/${id}`,
+			{
+				method: "DELETE",
+			}
+		);
+
+		logger.info(
+			`🥷 The instance "${id}" from ${args.name} was terminated successfully`
+		);
+	},
+});
diff --git a/packages/wrangler/src/workflows/commands/list.ts b/packages/wrangler/src/workflows/commands/list.ts
new file mode 100644
index 000000000000..2f33d4e9552f
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/list.ts
@@ -0,0 +1,63 @@
+import { fetchResult } from "../../cfetch";
+import { defineCommand } from "../../core";
+import { logger } from "../../logger";
+import { requireAuth } from "../../user";
+import type { Workflow } from "../types";
+
+defineCommand({
+	command: "wrangler workflows list",
+	metadata: {
+		description: "List Workflows associated to account",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+	args: {
+		page: {
+			describe:
+				'Show a sepecific page from the listing, can configure page size using "per-page"',
+			type: "number",
+			default: 1,
+		},
+		"per-page": {
+			describe: "Configure the maximum number of workflows to show per page",
+			type: "number",
+		},
+	},
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+
+		const URLParams = new URLSearchParams();
+
+		if (args.perPage !== undefined) {
+			URLParams.set("per_page", args.perPage.toString());
+		}
+
+		URLParams.set("page", args.page.toString());
+
+		const workflows = await fetchResult<Workflow[]>(
+			`/accounts/${accountId}/workflows`,
+			undefined,
+			URLParams
+		);
+
+		if (workflows.length === 0) {
+			logger.warn("There are no deployed Workflows in this account");
+		} else {
+			// TODO(lduarte): can we improve this message once pagination is deployed
+			logger.info(
+				`Showing last ${workflows.length} workflow${workflows.length > 1 ? "s" : ""}:`
+			);
+			// sort by name and make the table head prettier by changing the keys
+			const prettierWorkflows = workflows
+				.sort((a, b) => a.name.localeCompare(b.name))
+				.map((workflow) => ({
+					Name: workflow.name,
+					"Script name": workflow.script_name,
+					"Class name": workflow.class_name,
+					Created: new Date(workflow.created_on).toLocaleString(),
+					Modified: new Date(workflow.modified_on).toLocaleString(),
+				}));
+			logger.table(prettierWorkflows);
+		}
+	},
+});
diff --git a/packages/wrangler/src/workflows/commands/trigger.ts b/packages/wrangler/src/workflows/commands/trigger.ts
new file mode 100644
index 000000000000..6c3e6abcfac2
--- /dev/null
+++ b/packages/wrangler/src/workflows/commands/trigger.ts
@@ -0,0 +1,56 @@
+import { fetchResult } from "../../cfetch";
+import { defineCommand } from "../../core";
+import { requireAuth } from "../../user";
+import type { InstanceWithoutDates } from "../types";
+
+defineCommand({
+	command: "wrangler workflows trigger",
+
+	metadata: {
+		description:
+			"Trigger a workflow, creating a new instance. Can optionally take a JSON string to pass a parameter into the workflow instance",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+
+	args: {
+		name: {
+			describe: "Name of the workflow",
+			type: "string",
+			demandOption: true,
+		},
+		params: {
+			describe: "Params for the workflow instance, encoded as a JSON string",
+			type: "string",
+			default: "",
+		},
+	},
+	positionalArgs: ["name", "params"],
+
+	async handler(args, { config, logger }) {
+		const accountId = await requireAuth(config);
+
+		if (args.params.length != 0) {
+			try {
+				JSON.parse(args.params);
+			} catch (e) {
+				logger.error(
+					`Error while parsing instance parameters: "${args.params}" with ${e}' `
+				);
+				return;
+			}
+		}
+
+		const response = await fetchResult<InstanceWithoutDates>(
+			`/accounts/${accountId}/workflows/${args.name}/instances`,
+			{
+				method: "POST",
+				body: args.params.length != 0 ? args.params : undefined,
+			}
+		);
+
+		logger.info(
+			`🚀 Workflow instance "${response.id}" has been queued successfully`
+		);
+	},
+});
diff --git a/packages/wrangler/src/workflows/index.ts b/packages/wrangler/src/workflows/index.ts
new file mode 100644
index 000000000000..5a20e510d89c
--- /dev/null
+++ b/packages/wrangler/src/workflows/index.ts
@@ -0,0 +1,26 @@
+import { defineNamespace } from "../core";
+import "./commands/list";
+import "./commands/describe";
+import "./commands/delete";
+import "./commands/trigger";
+import "./commands/instances/list";
+import "./commands/instances/describe";
+import "./commands/instances/terminate";
+
+defineNamespace({
+	command: "wrangler workflows",
+	metadata: {
+		description: "🔁 Manage Workflows",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+});
+
+defineNamespace({
+	command: "wrangler workflows instances",
+	metadata: {
+		description: "Manage Workflow instances",
+		owner: "Product: Workflows",
+		status: "open-beta",
+	},
+});
diff --git a/packages/wrangler/src/workflows/utils.ts b/packages/wrangler/src/workflows/utils.ts
new file mode 100644
index 000000000000..f4d119041af3
--- /dev/null
+++ b/packages/wrangler/src/workflows/utils.ts
@@ -0,0 +1,70 @@
+import { UserError } from "../errors";
+import type { InstanceStatus, InstanceTriggerName } from "./types";
+
+export const emojifyInstanceStatus = (status: InstanceStatus) => {
+	switch (status) {
+		case "complete":
+			return "✅ Completed";
+		case "errored":
+			return "❌ Errored";
+		case "unknown":
+			return "❓ Unknown";
+		case "paused":
+			return "⏸️ Paused";
+		case "queued":
+			return "⌛ Queued";
+		case "running":
+			return "▶ Running";
+		case "terminated":
+			return "🚫 Terminated";
+	}
+};
+
+export const emojifyInstanceTriggerName = (status: InstanceTriggerName) => {
+	switch (status) {
+		case "api":
+			return "🌎 API";
+		case "binding":
+			return "🔗 Binding";
+		case "cron":
+			return "⌛ Cron";
+		case "event":
+			return "📩 Event";
+		default:
+			return "❓ Unknown";
+	}
+};
+
+export const emojifyStepType = (type: string) => {
+	switch (type) {
+		case "step":
+			return "🎯 Step";
+		case "sleep":
+			return "💤 Sleeping";
+		case "termination":
+			return "🚫 Termination";
+		default:
+			return "❓ Unknown";
+	}
+};
+
+export const validateStatus = (status: string): InstanceStatus => {
+	switch (status) {
+		case "complete":
+			return "complete";
+		case "errored":
+			return "errored";
+		case "paused":
+			return "paused";
+		case "queued":
+			return "queued";
+		case "running":
+			return "running";
+		case "terminated":
+			return "terminated";
+		default:
+			throw new UserError(
+				`Looks like you have provided a invalid status "${status}". Valid statuses are: queued, running, paused, errored, terminated, complete`
+			);
+	}
+};
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 0943c9e8fb5a..75784919de0b 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -1647,9 +1647,15 @@ importers:
       chokidar:
         specifier: ^3.5.3
         version: 3.5.3
+      date-fns:
+        specifier: ^4.1.0
+        version: 4.1.0
       esbuild:
         specifier: 0.17.19
         version: 0.17.19
+      itty-time:
+        specifier: ^1.0.6
+        version: 1.0.6
       miniflare:
         specifier: workspace:*
         version: link:../miniflare
@@ -4593,6 +4599,9 @@ packages:
     resolution: {integrity: sha512-fnULvOpxnC5/Vg3NCiWelDsLiUc9bRwAPs/+LfTLNvetFCtCTN+yQz15C/fs4AwX1R9K5GLtLfn8QW+dWisaAw==}
     engines: {node: '>=0.11'}
 
+  date-fns@4.1.0:
+    resolution: {integrity: sha512-Ukq0owbQXxa/U3EGtsdVBkR1w7KOQ5gIBqdH2hkvknzZPYvBxb/aa6E8L7tmjFtkwZBu3UXBbjIgPo/Ez4xaNg==}
+
   date-time@3.1.0:
     resolution: {integrity: sha512-uqCUKXE5q1PNBXjPqvwhwJf9SwMoAHBgWJ6DcrnS5o+W2JOiIILl0JEdVD8SGujrNS02GGxgwAg2PN2zONgtjg==}
     engines: {node: '>=6'}
@@ -5918,6 +5927,9 @@ packages:
   itty-router@4.0.17:
     resolution: {integrity: sha512-Fu/GO3MFX6Hwd+QF1/BFjoFpPQKh2Bu4DtBdse5kECcUwldNBrZqgwq0IariLrP67iADGlzkHIlOWwJ8F4SJ4A==}
 
+  itty-time@1.0.6:
+    resolution: {integrity: sha512-+P8IZaLLBtFv8hCkIjcymZOp4UJ+xW6bSlQsXGqrkmJh7vSiMFSlNne0mCYagEE0N7HDNR5jJBRxwN0oYv61Rw==}
+
   jackspeak@2.3.6:
     resolution: {integrity: sha512-N3yCS/NegsOBokc8GAdM8UcmfsKiSS8cipheD/nivzr700H+nsMOxJjQnvwOcRYVuFkdH0wGUvW2WbXGmrZGbQ==}
     engines: {node: '>=14'}
@@ -11548,6 +11560,8 @@ snapshots:
     dependencies:
       '@babel/runtime': 7.22.5
 
+  date-fns@4.1.0: {}
+
   date-time@3.1.0:
     dependencies:
       time-zone: 1.0.0
@@ -13113,6 +13127,8 @@ snapshots:
 
   itty-router@4.0.17: {}
 
+  itty-time@1.0.6: {}
+
   jackspeak@2.3.6:
     dependencies:
       '@isaacs/cliui': 8.0.2