Skip to content
Cloudflare Docs

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",
  "compatibility_date": "2026-02-13",
  "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",
    }
  ]
}

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

Python
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.

Under the hood, the Workflow binding is a Javascript object that is exposed to the Python script via JsProxy ↗. This means that the values returned by its methods are also JsProxy objects, and need to be converted back into Python objects using python_from_rpc.

create

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

  • create(options=None)* options - an optional dictionary of options to pass to the workflow instance. Should contain the same keys as the WorkflowInstanceCreateOptions type.
Python
from js import Object
from pyodide.ffi import to_js
from workers import WorkerEntrypoint, Response




class Default(WorkerEntrypoint):
    async def fetch(self, request):
        event = {"foo": "bar"}
        options = to_js({"params": event}, dict_converter=Object.fromEntries)
        await self.env.MY_WORKFLOW.create(options)
        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.

  • create_batch(batch)* batch - list of WorkflowInstanceCreateOptions to pass when creating an instance, including a user-provided ID and payload parameters.

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

Python
from pyodide.ffi import to_js
from js import Object
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
        listOfInstances = [
            to_js({ "id": "id-abc123", "params": { "hello": "world-0" } }, dict_converter=Object.fromEntries),
            to_js({ "id": "id-def456", "params": { "hello": "world-1" } }, dict_converter=Object.fromEntries),
            to_js({ "id": "id-ghi789", "params": { "hello": "world-2" } }, dict_converter=Object.fromEntries)
        ]
        await self.env.MY_WORKFLOW.create_batch(listOfInstances)
        return Response.json({"status": "success"})

get

Get a workflow instance by ID.

  • get(id)* id - the ID of the workflow instance to get.

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

Python
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.

  • send_event(options)* type - the type of event to send to the workflow instance. * payload - the payload to send to the workflow instance.
Python
from pyodide.ffi import to_js
from js import Object
from workers import WorkerEntrypoint, Response


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_WORKFLOW.send_event(to_js({ "type": "my-event-type", "payload": { "foo": "bar" } }, dict_converter=Object.fromEntries))
        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.