INTEGRITY Documentation

← Cloudflare Workflows / workflows / python

Interact with a Workflow

The Python Workers platform leverages FFI ↗ to access bindings to Cloudflare resources. Refer to the bindings documentation for more information.

From the configuration perspective, enabling Python Workflows requires adding the python_workflows compatibility flag to your Wrangler configuration file.

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "workflows-starter",
	"main": "src/index.py",
	// Set this to today's date
	"compatibility_date": "2026-08-28",
	"compatibility_flags": ["python_workflows", "python_workers"],
	"workflows": [
		{
			// name of your workflow
			"name": "workflows-starter",
			// binding name env.MY_WORKFLOW
			"binding": "MY_WORKFLOW",
			// this is class that extends the Workflow class in src/index.py
			"class_name": "MyWorkflow",
		}
	]
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "workflows-starter"
main = "src/index.py"
# Set this to today's date
compatibility_date = "2026-08-28"
compatibility_flags = [ "python_workflows", "python_workers" ]

[[workflows]]
name = "workflows-starter"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"

And this is how you use the payload in your workflow:

from workers import WorkflowEntrypoint

class DemoWorkflowClass(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do('step-name')
        async def first_step():
            payload = event["payload"]
            return payload

Workflow

The Workflow binding gives you access to the Workflow class. All its methods are available on the binding.

create

Create (trigger) a new instance of a given Workflow.

from workers import WorkerEntrypoint, Response


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        event = {"foo": "bar"}
        await self.env.MY_WORKFLOW.create(params=event)
        return Response.json({"status": "success"})

The create method returns a WorkflowInstance object, which can be used to query the status of the workflow instance. Note that this is a Javascript object, and not a Python object.

create_batch

Create (trigger) a batch of new workflow instances, up to 100 instances at a time. This is useful if you need to create multiple instances at once within the instance creation limit.

Each element of the batch list is expected to include both id and params properties:

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
			# Create a new batch of 3 Workflow instances, each with its own ID and pass params to the Workflow instances
        instances = [
            {"id": "id-abc123", "params": {"hello": "world-0"}},
            {"id": "id-def456", "params": {"hello": "world-1"}},
            {"id": "id-ghi789", "params": {"hello": "world-2"}},
        ]
        await self.env.MY_WORKFLOW.create_batch(instances)
        return Response.json({"status": "success"})

get

Get a workflow instance by ID.

Returns a WorkflowInstance object, which can be used to query the status of the workflow instance.

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        instance = await self.env.MY_WORKFLOW.get("abc-123")

        # FFI methods available for WorkflowInstance
        await instance.status()
        await instance.pause()
        await instance.resume()
        await instance.restart()
        await instance.terminate()
        return Response.json({"status": "success"})

send_event

Send an event to a workflow instance.

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_WORKFLOW.send_event(type="my-event-type", payload={"foo": "bar"})
        return Response.json({"status": "success"})

REST API (HTTP)

Refer to the Workflows REST API documentation.

Command line (CLI)

Refer to the CLI quick start to learn more about how to manage and trigger Workflows via the command-line.