undefined

undefined

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from Python Worker!")

{
  "name": "my-python-worker",
  "main": "src/entry.py",
  "compatibility_date": "2025-12-01",
  "compatibility_flags": ["python_workers"]
}

uv run pywrangler dev

undefined

uv run pywrangler deploy

undefined

**New Approach** (pywrangler):
```toml

```bash
uv tool install workers-py
uv run pywrangler deploy  # Now works!

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Access bindings via self.env
        value = await self.env.MY_KV.get("key")

        # Parse request
        url = request.url
        method = request.method

        return Response(f"Method: {method}, URL: {url}")

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # D1 Database
        result = await self.env.DB.prepare("SELECT * FROM users").all()

        # KV Storage
        value = await self.env.MY_KV.get("key")
        await self.env.MY_KV.put("key", "value")

        # R2 Object Storage
        obj = await self.env.MY_BUCKET.get("file.txt")

        # Workers AI
        response = await self.env.AI.run("@cf/meta/llama-2-7b-chat-int8", {
            "prompt": "Hello!"
        })

        return Response("OK")

from workers import WorkerEntrypoint, Response
import json

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Parse JSON body
        if request.method == "POST":
            body = await request.json()
            return Response(
                json.dumps({"received": body}),
                headers={"Content-Type": "application/json"}
            )

        # Query parameters
        url = URL(request.url)
        name = url.searchParams.get("name", "World")

        return Response(f"Hello, {name}!")

from workers import handler

@handler
async def on_scheduled(event, env, ctx):
    # Run on cron schedule
    print(f"Cron triggered at {event.scheduledTime}")

    # Do work...
    await env.MY_KV.put("last_run", str(event.scheduledTime))

{
  "triggers": {
    "crons": ["*/5 * * * *"]  // Every 5 minutes
  }
}

await step.do("my step", async () => {
  // Inline callback
  return result;
});

@step.do("my step")
async def my_step():
    # Named function with decorator
    return result

result = await my_step()

import asyncio

@step.do("step_a")
async def step_a():
    return "A"

@step.do("step_b")
async def step_b():
    return "B"

**Why This Works**: JavaScript promises from workflow steps are proxied as Python awaitables, allowing standard asyncio concurrency primitives.

**Source**: [Python Workflows Blog](https://blog.cloudflare.com/python-workflows/)

from workers import WorkflowEntrypoint, WorkerEntrypoint, Response

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        # Step 1
        @step.do("fetch data")
        async def fetch_data():
            response = await fetch("https://api.example.com/data")
            return await response.json()

        data = await fetch_data()

        # Step 2: Sleep
        await step.sleep("wait", "10 seconds")

        # Step 3: Process
        @step.do("process data")
        async def process_data():
            return {"processed": True, "count": len(data)}

        result = await process_data()
        return result


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Create workflow instance
        instance = await self.env.MY_WORKFLOW.create()
        return Response(f"Workflow started: {instance.id}")

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do("step_a")
        async def step_a():
            return "A done"

        @step.do("step_b")
        async def step_b():
            return "B done"

        # step_c waits for both step_a and step_b
        @step.do("step_c", depends=[step_a, step_b], concurrent=True)
        async def step_c(result_a, result_b):
            return f"C received: {result_a}, {result_b}"

        return await step_c()

{
  "compatibility_flags": ["python_workers", "python_workflows"],
  "compatibility_date": "2025-12-01",
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow"
    }
  ]
}

[project]
name = "my-python-worker"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "beautifulsoup4",
    "httpx"
]

[dependency-groups]
dev = [
    "workers-py",
    "workers-runtime-sdk"
]

undefined

undefined

from js import fetch, console, Response as JSResponse

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Use JavaScript fetch
        response = await fetch("https://api.example.com")
        data = await response.json()

        # Console logging
        console.log("Fetched data:", data)

        # Return JavaScript Response
        return JSResponse.new("Hello!")

from js import Object
from pyodide.ffi import to_js

**Source**: [GitHub Issue #3322](https://github.com/cloudflare/workerd/issues/3322) (Pyodide maintainer clarification)

---

undefined

undefined

undefined

undefined

{
  "compatibility_flags": ["python_workers"]
}

undefined

undefined

@step.do("process")
async def process():
    # Convert datetime to string
    return {"timestamp": datetime.now().isoformat()}

undefined

Run both workers in a single wrangler instance to enable proper RPC communication.

undefined

**Workarounds**:
- Avoid HTMLRewriter on notebook content (stream directly)
- Pre-process notebooks to extract data URLs to external files
- Use `text/plain` content-type to bypass parser

import random

Only call PRNG functions inside request handlers, not at module level.

---

from fastapi import FastAPI
from workers import WorkerEntrypoint

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello from FastAPI"}

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Route through FastAPI
        return await app(request)

{
  "workers-py": "1.7.0",
  "workers-runtime-sdk": "0.3.1",
  "wrangler": "4.58.0"
}