cloudflare-python-workers

Cloudflare Python Workers

ステータス：ベータ（python_workers 互換性フラグが必要です） ランタイム：Pyodide（WebAssembly にコンパイルされた Python 3.12 以降） パッケージバージョン：[email protected]、[email protected]、[email protected] 最終確認日：2026-01-21

クイックスタート（5 分）

1. 前提条件

以下がインストールされていることを確認してください。

• uv - Python パッケージマネージャー
• Node.js - Wrangler に必要です

2. プロジェクトを初期化する

# Create project directory
mkdir my-python-worker && cd my-python-worker

# Initialize Python project
uv init

# Install pywrangler
uv tool install workers-py

# Initialize Worker configuration
uv run pywrangler init

3. エントリーポイントを作成する

src/entry.py を作成します。

from workers import WorkerEntrypoint, Response

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

4. wrangler.jsonc を設定する

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

5. ローカルで実行する

uv run pywrangler dev
# Visit http://localhost:8787

6. デプロイする

uv run pywrangler deploy

---

2025 年 12 月以前の Workers からの移行

2025 年 12 月より前に Python Worker を作成した場合、使用できるのは組み込みパッケージに限られていました。pywrangler（2025 年 12 月）により、外部パッケージを使ってデプロイできるようになりました。

以前の方法（現在は不要です）：

# Limited to built-in packages only
# Could only use httpx, aiohttp, beautifulsoup4, etc.
# Error: "You cannot yet deploy Python Workers that depend on
# packages defined in requirements.txt [code: 10021]"

新しい方法（pywrangler）：

# pyproject.toml
[project]
dependencies = ["fastapi", "any-pyodide-compatible-package"]

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

これまでの経緯：

• 2024 年 4 月～2025 年 12 月：パッケージのデプロイは完全に不可能
• 2025 年 12 月 8 日：Pywrangler がリリースされ、パッケージのデプロイが可能に
• 2026 年 1 月：パッケージを全面的にサポートするオープンベータ

詳細：パッケージデプロイに関する問題の履歴

---

基本概念

WorkerEntrypoint クラスパターン

2025 年 8 月以降、Python Workers ではクラスベースのパターン（グローバルハンドラーではありません）を使用します。

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}")

バインディングへのアクセス

すべての Cloudflare バインディングには self.env 経由でアクセスします。

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")

サポートされるバインディング：

• D1（SQL データベース）
• KV（キー・バリュー型ストレージ）
• R2（オブジェクトストレージ）
• Workers AI
• Vectorize
• Durable Objects
• Queues
• Analytics Engine

詳細については、Cloudflare バインディングのドキュメントを参照してください。

リクエスト／レスポンスの処理

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}!")

スケジュールされたハンドラー（Cron）

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))

wrangler.jsonc で設定します。

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

---

Python ワークフロー

Python Workflows を使うと、自動リトライと状態の永続化に対応した、耐障害性のある複数ステップの自動化を実現できます。

なぜデコレーターパターンなのか？

Python Workflows では @step.do() デコレーターパターンを使用します。これは、Python では匿名コールバックを簡単にサポートできないためです（インラインのアロー関数を使用できる JavaScript／TypeScript とは異なります）。これは言語の根本的な違いであり、Cloudflare の実装上の制限ではありません。

JavaScript のパターン（そのままでは置き換えられません）：

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

Python のパターン（必須）：

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

result = await my_step()

出典：Python Workflows のブログ

asyncio.gather による並行処理

Pyodide は JavaScript の Promise（thenable）を捕捉し、Python の awaitable としてプロキシします。これにより、標準的な Python の非同期パターンを使って Promise.all と同等の動作を実現できます。

import asyncio

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

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

# Concurrent execution (like Promise.all)
results = await asyncio.gather(step_a(), step_b())
# results = ["A", "B"]

これが機能する理由：ワークフローのステップから返される JavaScript の Promise は Python の awaitable としてプロキシされるため、標準の asyncio 並行処理プリミティブを使用できます。

出典：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}")

DAG の依存関係

並列実行のためのステップ依存関係を定義します。

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"
    }
  ]
}

---

パッケージ管理

pyproject.toml の設定

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

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

サポートされるパッケージ

Python Workers は以下をサポートします。

• PyPI の純粋な Python パッケージ
• Pyodide パッケージ（WebAssembly 用にビルド済み）

Pyodide パッケージ一覧を参照してください。

HTTP クライアント

非同期 HTTP ライブラリのみ使用できます。

# ✅ WORKS - httpx (async)
import httpx

async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com")

# ✅ WORKS - aiohttp
import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.get("https://api.example.com") as response:
        data = await response.json()

# ❌ DOES NOT WORK - requests (sync)
import requests  # Will fail!

新しいパッケージのリクエスト

以下で新しいパッケージのサポートをリクエストしてください。 https://github.com/cloudflare/workerd/discussions/categories/python-packages

---

FFI（外部関数インターフェース）

Pyodide の FFI を使って、Python から JavaScript API にアクセスできます。

JavaScript グローバル

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!")

型変換

重要：to_py()はJavaScriptオブジェクトのメソッドであり、単独の関数ではありません。関数なのはto_js()のみです。

from js import Object
from pyodide.ffi import to_js

# ❌ WRONG - ImportError!
from pyodide.ffi import to_py
python_data = to_py(js_data)

# ✅ CORRECT - to_py() is a method
async def fetch(self, request):
    data = await request.json()  # Returns JS object
    python_data = data.to_py()   # Convert to Python dict

# Convert Python dict to JavaScript object
python_dict = {"name": "test", "count": 42}
js_object = to_js(python_dict, dict_converter=Object.fromEntries)

# Use in Response
return Response(to_js({"status": "ok"}))

出典：GitHub Issue #3322（Pyodideメンテナーによる説明）

---

既知の問題の予防

このスキルにより、11件の文書化された問題を防げます。

問題 #1：レガシーハンドラーパターン

エラー：TypeError: on_fetch is not defined

理由：2025年8月にハンドラーパターンが変更されました。

# ❌ OLD (deprecated)
@handler
async def on_fetch(request):
    return Response("Hello")

# ✅ NEW (current)
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello")

問題 #2：同期HTTPライブラリ

エラー：RuntimeError: cannot use blocking call in async context

理由：Python Workersは非同期処理のみで動作します。同期ライブラリはイベントループをブロックします。

# ❌ FAILS
import requests
response = requests.get("https://api.example.com")

# ✅ WORKS
import httpx
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com")

問題 #3：ネイティブ／コンパイル済みパッケージ

エラー：ModuleNotFoundError: No module named 'numpy'（または同様のエラー）

理由：動作するのは純粋なPythonパッケージのみです。ネイティブC拡張機能はサポートされていません。

解決策：Pyodide互換の代替パッケージを使用するか、Pyodideパッケージを確認してください。

問題 #4：互換性フラグの不足

エラー：Error: Python Workers require the python_workers compatibility flag

修正：wrangler.jsoncに追加します。

{
  "compatibility_flags": ["python_workers"]
}

Workflowsの場合は、"python_workflows"も追加してください。

問題 #5：ワークフローのステップ外でのI/O

エラー：ワークフローの状態が正しく永続化されない

理由：耐久性を確保するには、すべてのI/Oを@step.doの内部で実行する必要があります。

# ❌ BAD - fetch outside step
response = await fetch("https://api.example.com")
@step.do("use data")
async def use_data():
    return await response.json()  # response may be stale on retry

# ✅ GOOD - fetch inside step
@step.do("fetch and use")
async def fetch_and_use():
    response = await fetch("https://api.example.com")
    return await response.json()

問題 #6：型のシリアル化エラー

エラー：TypeError: Object of type X is not JSON serializable

理由：ワークフローのステップから返す値は、JSONシリアル化可能である必要があります。

修正：返す前に複雑なオブジェクトを変換します。

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

問題 #7：コールドスタートのパフォーマンス

注記：Python WorkersはJavaScriptよりコールドスタートに時間がかかります。Wasmメモリスナップショット（2025年12月）により、FastAPIやPydanticなどの重量級パッケージの読み込み時間は、以前の約10秒から約1秒になりましたが、それでもJavaScript Workers（約50ms）より約2倍遅くなります。

パフォーマンスの数値（2025年12月時点）：

• スナップショット導入前：FastAPI／Pydanticの読み込みに約10秒
• スナップショット導入後：約1秒（10倍の改善）
• JavaScriptの場合：約50ms

軽減策：

• トップレベルでのインポートを最小限にする
• 重量級パッケージには遅延読み込みを使用する
• レイテンシが重要な処理にはJavaScript Workersを検討する
• Wasmスナップショットにより、コールドスタートが自動的に改善されます（設定不要）

出典：Python Workers Redux Blog | InfoQ Coverage

問題 #8：パッケージのインストール失敗

エラー：Failed to install package X

原因：

• パッケージにネイティブ依存関係がある
• パッケージがPyodideエコシステムに含まれていない
• バンドル中にネットワークの問題が発生した

修正：パッケージの互換性を確認し、代替パッケージを使用するか、サポートを依頼してください。

問題 #9：開発レジストリによるJSからPythonへのRPCの不具合

エラー：JavaScript WorkerからPython Workerを呼び出したときのNetwork connection lost 出典：GitHub Issue #11438

発生する理由：開発レジストリは、別々のターミナルで実行されているWorker間のRPC呼び出しを適切にルーティングできません。

予防策：

# ❌ Doesn't work - separate terminals
# Terminal 1: npx wrangler dev (JS worker)
# Terminal 2: npx wrangler dev (Python worker)
# Result: Network connection lost error

# ✅ Works - single wrangler instance
npx wrangler dev -c ts/wrangler.jsonc -c py/wrangler.jsonc

両方のWorkerを1つのwranglerインスタンス内で実行すると、RPC通信を適切に行えます。

問題 #10：Data URLによるHTMLRewriterのメモリ制限

エラー：TypeError: Parser error: The memory limit has been exceeded 出典：GitHub Issue #10814

発生する理由：HTML内に大きなインラインのdata: URL（10MB超）があると、パーサーのメモリ制限に達します。これはレスポンスサイズの問題ではありません。10MBのプレーンテキストは問題ありませんが、データURLを埋め込んだ10MBのHTMLではエラーになります。プロットにインライン画像を使用するPython Jupyter Notebookでよく発生します。

予防策：

# ❌ FAILS - HTMLRewriter triggered on notebook HTML with data: URLs
response = await fetch("https://origin.example.com/notebook.html")
return response  # Crashes if HTML contains large data: URLs

# ✅ WORKS - Stream directly or use text/plain
response = await fetch("https://origin.example.com/notebook.html")
headers = {"Content-Type": "text/plain"}  # Bypass parser
return Response(await response.text(), headers=headers)

回避策：

• ノートブックのコンテンツにはHTMLRewriterを使用せず、直接ストリーミングする
• ノートブックを前処理し、データURLを外部ファイルに抽出する
• text/plainのコンテンツタイプを使用してパーサーを回避する

Issue #11：初期化中に PRNG をシードできない

エラー：ユーザーエラーでデプロイが失敗します 情報源：Python Workers Redux Blog

原因：Wasm スナップショットは、リクエストハンドラーの前に PRNG を初期化することをサポートしていません。モジュールの初期化中に random.seed() などの疑似乱数生成器 API を呼び出すと、デプロイが失敗します。

防止策：

import random

# ❌ FAILS deployment - module-level PRNG call
random.seed(42)

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response(str(random.randint(1, 100)))

# ✅ WORKS - PRNG calls inside handlers
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        random.seed(42)  # Initialize inside handler
        return Response(str(random.randint(1, 100)))

PRNG 関数はモジュールレベルではなく、リクエストハンドラー内でのみ呼び出してください。

---

ベストプラクティス

必ず行うこと

• WorkerEntrypoint クラスパターンを使用する
• 非同期 HTTP クライアント（httpx、aiohttp）を使用する
• すべての I/O をワークフローのステップ内に配置する
• python_workers 互換性フラグを追加する
• すべてのバインディングで self.env を使用する
• ワークフローのステップから JSON にシリアライズ可能なデータを返す

絶対に行わないこと

• 同期 HTTP ライブラリ（requests）を使用する
• ネイティブ／コンパイル済みパッケージを使用する
• ワークフローのステップ外で I/O を実行する
• fetch にレガシーな @handler デコレーターを使用する
• JavaScript レベルのコールドスタート時間を期待する

---

フレームワークに関する注記：FastAPI

FastAPI は Python Workers で動作しますが、制限があります。

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)

制限事項：

• 非同期のみ（同期エンドポイントは使用不可）
• WSGI ミドルウェアは使用不可
• ベータ版の安定性

詳細については、Cloudflare FastAPI exampleをご覧ください。

---

公式ドキュメント

• Python Workers の概要
• Python Workers の基本
• Python Workers の仕組み
• Python パッケージ
• FFI（外部関数インターフェース）
• Python Workflows
• Pywrangler CLI
• Pyodide パッケージリスト

---

依存関係

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

注記：再現可能なビルドのため、必ずバージョンを固定してください。最新リリースについては、PyPI workers-pyをご確認ください。

---

本番環境での検証

• Cloudflare の変更履歴：2025年12月8日（Pywrangler とコールドスタートの改善）
• workers-py 1.7.0：最新の安定版（2026年1月）
• Python Workflows ベータ版：2025年8月22日
• ハンドラーパターンの変更：2025年8月14日

互換性日付のガイダンス：

• 新規プロジェクトでは 2025-12-01 を使用してください（pywrangler の改善を含む最新機能を利用できます）
• 古い本番環境の Workers に合わせる必要がある場合にのみ 2025-08-01 を使用してください