Interact with a Workflow

Last updated Sep 17, 2026 Note

You must add both python_workflows and python_workers compatibility flags to your Wrangler config file.

Also, Python Workflows requires compatibility_date = "2025-08-01", or later, to be set in your Wrangler config file.

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-10-27",
    "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-10-27"
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.

  • create(options=None)* options - an optional dictionary of options to pass to the workflow instance. Should contain the same keys as the WorkflowInstanceCreateOptions type.
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.

  • 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:

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.

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

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(type, payload)* type - the type of event to send to the workflow instance. * payload - the payload to send to the 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.

Was this helpful?

YesNo

On this page

Interact with a Workflow - 图1Docs

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/workflows/python/bindings/#page","headline":"Interact with a Workflow","description":"Trigger and manage Workflows from Python Workers using FFI bindings to Cloudflare resources.","url":"https://developers.cloudflare.com/workflows/python/bindings/","inLanguage":"en","image":"https://developers.cloudflare.com/workflows/python/bindings/og.png?v=6cff989783879f23","dateModified":"2026-09-17","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"}}

description: Define directed acyclic graph (DAG) Workflows in Python with declarative step dependencies. title: DAG Workflows image: https://developers.cloudflare.com/workflows/python/dag/og.png?v=677abb6a36dab47f

Skip to content

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/workflows/llms.txt
Use this file to discover all available pages before exploring further.