diff --git a/src/content/docs/workflows/get-started/durable-agents.mdx b/src/content/docs/workflows/get-started/durable-agents.mdx index 8ba1a7532cb..59318e794cd 100644 --- a/src/content/docs/workflows/get-started/durable-agents.mdx +++ b/src/content/docs/workflows/get-started/durable-agents.mdx @@ -611,6 +611,7 @@ Connect to your Agent via WebSocket to receive real-time progress updates. The ` ``` ```tsx +import { useState } from "react"; import { useAgent } from "agents/react"; function ResearchUI({ agentId = "default" }) { diff --git a/src/content/docs/workflows/python/bindings.mdx b/src/content/docs/workflows/python/bindings.mdx index 10a1ab24679..71e8b122be5 100644 --- a/src/content/docs/workflows/python/bindings.mdx +++ b/src/content/docs/workflows/python/bindings.mdx @@ -26,20 +26,17 @@ From the configuration perspective, enabling Python Workflows requires adding th { "$schema": "./node_modules/wrangler/config-schema.json", "name": "workflows-starter", - "main": "src/index.ts", + "main": "src/index.py", "compatibility_date": "$today", - "compatibility_flags": [ - "python_workflows", - "python_workers" - ], + "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.ts - "class_name": "MyWorkflow" + // this is class that extends the Workflow class in src/index.py + "class_name": "MyWorkflow", } ] } @@ -50,7 +47,7 @@ From the configuration perspective, enabling Python Workflows requires adding th And this is how you use the payload in your workflow: ```python -from pyodide.ffi import to_js +from workers import WorkflowEntrypoint class DemoWorkflowClass(WorkflowEntrypoint): async def run(self, event, step): @@ -110,15 +107,18 @@ Each element of the `batch` list is expected to include both `id` and `params` p ```python from pyodide.ffi import to_js from js import Object +from workers import WorkerEntrypoint, Response -# 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 env.MY_WORKFLOW.create_batch(listOfInstances); +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` @@ -130,14 +130,19 @@ Get a workflow instance by ID. Returns a [`WorkflowInstance`](/workflows/build/workers-api/#workflowinstance) object, which can be used to query the status of the workflow instance. ```python -instance = await 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() +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` @@ -150,8 +155,12 @@ Send an event to a workflow instance. ```python from pyodide.ffi import to_js from js import Object +from workers import WorkerEntrypoint, Response -await env.MY_WORKFLOW.send_event(to_js({ "type": "my-event-type", "payload": { "foo": "bar" } }, dict_converter=Object.fromEntries)) +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"}) ``` :::note diff --git a/src/content/docs/workflows/python/dag.mdx b/src/content/docs/workflows/python/dag.mdx index 5ec898fa465..8bdba984a0e 100644 --- a/src/content/docs/workflows/python/dag.mdx +++ b/src/content/docs/workflows/python/dag.mdx @@ -3,7 +3,6 @@ title: DAG Workflows pcx_content_type: concept sidebar: order: 4 - --- The Python Workflows SDK supports DAG workflows in a declarative way, using the `step.do` decorator with the `depends` parameter to define dependencies (other steps that must complete before this step can run). @@ -19,7 +18,7 @@ class PythonWorkflowStarter(WorkflowEntrypoint): except TypeError as e: print(f"Successfully caught {type(e).__name__}: {e}") - step.sleep('demo sleep', '10 seconds') + await step.sleep('demo sleep', '10 seconds') @step.do('dependency1') async def dep_1(): @@ -47,4 +46,4 @@ On this example, `dep_1` and `dep_2` are run concurrently before execution of `f Having `concurrent=True` allows the dependencies to be resolved concurrently. If one of the callables passed to `depends` has already completed, it will be skipped and its return value will be reused. -This pattern is useful for diamond shaped workflows, where a step depends on two or more other steps that can run concurrently. \ No newline at end of file +This pattern is useful for diamond shaped workflows, where a step depends on two or more other steps that can run concurrently. diff --git a/src/content/docs/workflows/python/index.mdx b/src/content/docs/workflows/python/index.mdx index 2e45b958525..023ddaad0db 100644 --- a/src/content/docs/workflows/python/index.mdx +++ b/src/content/docs/workflows/python/index.mdx @@ -46,7 +46,8 @@ class PythonWorkflowStarter(WorkflowEntrypoint): # does stuff print('executing step2') - await await_step(step_1,step_2) + await step_1() + await step_2() async def on_fetch(request, env): await env.MY_WORKFLOW.create() diff --git a/src/content/docs/workflows/python/python-workers-api.mdx b/src/content/docs/workflows/python/python-workers-api.mdx index b11f47ffc16..3573cb02df6 100644 --- a/src/content/docs/workflows/python/python-workers-api.mdx +++ b/src/content/docs/workflows/python/python-workers-api.mdx @@ -3,7 +3,6 @@ title: Python Workers API pcx_content_type: concept sidebar: order: 2 - --- This guide covers the Python Workflows SDK, with instructions on how to build and create workflows using Python. @@ -16,7 +15,7 @@ The `WorkflowEntrypoint` is the main entrypoint for a Python workflow. It extend from workers import WorkflowEntrypoint class MyWorkflow(WorkflowEntrypoint): - def run(self, event, step): + async def run(self, event, step): # steps here ``` @@ -62,6 +61,8 @@ async def run(self, event, step): * `timestamp` - a `datetime.datetime` object or seconds from the Unix epoch to sleep the workflow instance until. ```python +import datetime + async def run(self, event, step): await step.sleep_until("my-sleep-step", datetime.datetime.now() + datetime.timedelta(seconds=10)) ``` @@ -77,7 +78,6 @@ async def run(self, event, step): await step.wait_for_event("my-wait-for-event-step", "my-event-type") ``` - ### `event` parameter The `event` parameter is a dictionary that contains the payload passed to the workflow instance, along with other metadata: @@ -131,6 +131,8 @@ You can bind a step to a specific retry policy by passing a `WorkflowStepConfig` With Python Workflows, you need to make sure that your `dict` respects the [`WorkflowStepConfig`](/workflows/build/workers-api/#workflowstepconfig) type. ```python +from workers import WorkflowEntrypoint + class DemoWorkflowClass(WorkflowEntrypoint): async def run(self, event, step): @step.do('step-name', config={"retries": {"limit": 1, "delay": "10 seconds"}}) @@ -146,6 +148,8 @@ Note that `env` is a JavaScript object exposed to the Python script via [JsProxy Let's consider the previous binding called `MY_WORKFLOW`. Here's how you would create a new instance: ```python +from workers import Response, WorkerEntrypoint + class Default(WorkerEntrypoint): async def fetch(self, request): instance = await self.env.MY_WORKFLOW.create() diff --git a/src/content/docs/workflows/reference/limits.mdx b/src/content/docs/workflows/reference/limits.mdx index 3f596406755..4c63c6528b8 100644 --- a/src/content/docs/workflows/reference/limits.mdx +++ b/src/content/docs/workflows/reference/limits.mdx @@ -74,7 +74,7 @@ type Env = { export class MyWorkflow extends WorkflowEntrypoint { async run(event: WorkflowEvent, step: WorkflowStep) { - const apiResponse = await step.do("initial work", async () => { + await step.do("initial work", async () => { let resp = await fetch("https://api.cloudflare.com/client/v4/ips"); return await resp.json(); }); @@ -86,7 +86,7 @@ export class MyWorkflow extends WorkflowEntrypoint { { retries: { limit: 5, - delay: "5 second", + delay: "5 seconds", backoff: "exponential", }, timeout: "15 minutes",