Claude CodeでPrefectのワークフローを書く — 定義からスケジュール登録まで

Prefectとは — Python関数を本番ワークフローに変えるオーケストレーションツール

Prefectは、Pythonの関数に@flowと@taskという2つのデコレータを足すだけで、データパイプラインを本番運用できる形に変えるオープンソースのオーケストレーションツールです。YAMLや専用のDSLを書く必要はありません。状態管理・失敗時のリトライ・実行状況のモニタリングは、Prefect側が自動で担います。

Claude Codeでこの種のコードを書く場面は珍しくありません。定期実行するデータ処理、複数のAPIを呼ぶバッチ、失敗したら再試行したい社内スクリプトは、どれもPrefectのflowとtaskにそのまま乗ります。本記事では、Claude Codeでflow・taskを書くところから、デプロイとスケジュール登録、Prefect MCPサーバーを使った監視までを順に扱います。

前提条件

始める前に、次の3つを確認します。

• Python 3.10以降(Prefectの動作要件)
• Prefect本体のインストール(pip install -U prefectまたはuv add prefect)
• 実行先の確保。Prefect Cloudのアカウント、または自分のマシンやサーバーで動かすPrefectサーバーのどちらか

pip install -U prefect
prefect version

prefect versionの出力にバージョン番号とPython・OSの情報が表示されれば、インストールは完了です。

Claude Codeでflowとtaskを書く

PrefectのAPIはシンプルです。ワークフローの入口になる関数に@flowを、そこから呼ぶ個々の処理に@taskを付けるだけです。Claude Codeへの指示も「この関数をflowにして、内部の処理をtaskに分けて」で伝わります。

from prefect import flow, task
import random
 
@task
def get_customer_ids() -> list[str]:
    # データベースやAPIから顧客IDを取得
    return [f"customer{n}" for n in random.choices(range(100), k=10)]
 
@task
def process_customer(customer_id: str) -> str:
    # 顧客ごとの処理
    return f"Processed {customer_id}"
 
@flow
def main() -> list[str]:
    customer_ids = get_customer_ids()
    results = process_customer.map(customer_ids)
    return results
 
 
if __name__ == "__main__":
    main()

process_customer.map(customer_ids)が、taskをリストの要素数だけ並列展開する書き方です。1件ずつforループを回さなくても、Prefectが依存関係のグラフを内部で組み立てます。各taskの完了・失敗も個別に追跡されます。

flowはクラスのインスタンスメソッド・staticmethod・ジェネレータ関数にも付けられます。既存のクラスベースの処理をそのままflow化したいときも、書き換える範囲は最小限で済みます。この書き方はMongoDB AtlasのクラスタをClaude Codeに任せるときと同様に、外部サービスへの呼び出しをtask単位に切り出すと再利用しやすくなります。

本番運用を見据えるなら、@taskのキーワード引数も合わせて指示します。retriesとretry_delay_secondsで失敗時の自動再試行を設定でき、cache_key_fnとcache_policyで同じ入力に対する再計算を省けます。timeout_secondsは同期taskでは要注意です。既定のThreadPoolTaskRunner上で.submit()したtaskは、time.sleep()のようなブロッキング処理をtimeoutで中断できません。詳しくは後述の「よくあるつまずき」で扱います。

ローカルで実行して結果を確認する

git clone https://github.com/PrefectHQ/quickstart && cd quickstart
python 01_getting_started.py

実行すると、Prefect UIへのリンクと各taskの状態(Completed)がターミナルに流れます。ローカルのSQLiteサーバーだけで完結するので、Prefect Cloudのアカウントを作る前でも動作を確認できます。

デプロイとスケジュールを登録する

flowをローカルで動かすところまで進んだら、次は「いつ、どこで動かすか」を決めます。もっとも手早いのは.serve()にcron式を渡す方法です。

if __name__ == "__main__":
    main.serve(
        name="my-first-deployment",
        cron="0 8 * * *",  # 毎日8:00に実行
    )

.serve()を呼ぶプロセスを起動したままにしておくと、指定した時刻に自動でflow runが作られます。cronの代わりにinterval(秒数間隔)やrruleも指定できます。

すでに動いているデプロイにスケジュールを足したいだけなら、CLIのほうが手早く済みます。

prefect deployment schedule create my-flow/my-deployment --cron "0 9 * * *" --timezone "Asia/Tokyo"

--replaceを付けると、既存のスケジュールを残さず置き換えます。付けなければ追加登録になり、複数のスケジュールが同時に有効になります。スケジュール変更をチームのレビューに乗せたい場合は、Claude CodeでPRを作成する手順と組み合わせると、flow定義の差分もコードレビューの対象にできます。

Prefect CloudとOSS版の使い分け早見表

実行先の選び方は、インフラをどこまで自分で持ちたいかで決まります。

prefect-clientはCLIとサーバー機能を含みません。接続先に投げるだけの軽量なランタイムに向きます。どれを選んでも、flow・taskのPythonコード自体は変わりません。

Claude CodeにPrefect MCPサーバーをつないで監視を任せる

ここまではPrefectだけで完結する話でした。Claude Code固有の価値が出るのは、Prefect MCPサーバーを繋いでflow runの監視・デバッグをClaude Codeの会話の中で行えるようにしたときです。Prefect自体もMCPサーバーとして提供されており、n8nのMCP Server Triggerでワークフローをサーバー化する方法と同じく、既存のツールをMCP経由でAIエージェントに開放する動きが広がっています。

Claude Codeへの接続方法は2通りあります。

/plugin marketplace add prefecthq/prefect-mcp-server
/plugin install prefect

マーケットプレイス経由のプラグインは、Prefectがホストする読み取り専用のMCPサーバーに繋がります。OAuthでPrefect Cloudにサインインし、アクセスさせるワークスペースを選ぶだけで使えます。自己ホストのPrefectサーバーに繋ぎたい場合や、認証情報を明示したい場合は、CLIで手動追加します。

claude mcp add prefect \
  -e PREFECT_API_URL=https://api.prefect.cloud/api/accounts/[ACCOUNT_ID]/workspaces/[WORKSPACE_ID] \
  -e PREFECT_API_KEY=your-cloud-api-key \
  -- uvx --from prefect-mcp prefect-mcp-server

環境変数を省略すると、ローカルの~/.prefect/profiles.tomlにあるアクティブなプロファイルの認証情報をそのまま引き継ぎます。環境変数を指定した場合は、プロファイルより環境変数が優先されます。

接続後にできることは、ダッシュボードの状態確認、flow run・task runの検索、実行ログの取得、失敗したflow runの原因診断です。「直近の失敗したflow runをデバッグして」「このデプロイが動かない理由を教えて」といった聞き方で、ログを横断的に調べてくれます。flow runのログを読み取り専用の範囲で追わせる考え方は、Claudeで監視ツール連携による障害対応ワークフローを設計するで扱った設計とも共通します。

接続先はClaude.ai・Claude Desktopの「Prefectコネクタ」とは別物です。claude.ai/directory/prefectから追加できるのはClaude.aiやClaude Desktop向けのホスト型接続で、Claude Codeのプラグイン・CLI追加とは設定する場所が異なります。3つの形態のどこに接続するかを混同すると、期待した場所にツールが出てきません。

デプロイの作成やスケジュール変更のような書き込み操作は、Claude Codeに「prefectコマンドを使って」と明示して初めて実行されます。MCPサーバー自体はこれらの操作を提供していないため、指示なしではCLI経由の操作は起きません。

似た名前の機能に、Prefect Cloud自身が持つAIログ要約があります。こちらはPrefect側のMarvin AIがログを要約してUIに表示する、Prefect Cloud限定の機能です。Claude CodeとMCPサーバーを繋ぐ話とは別の仕組みで、有効化にはアカウント管理者による設定が要ります。Claude Codeでの監視・デバッグと混同しないよう区別しておきます。

よくあるつまずき

• Prefect MCPサーバーに書き込みを頼んでも動かない。デプロイの作成やスケジュール変更をMCP経由で頼むと、対応するツールが存在せず失敗します。「prefectコマンドで」と付け加えるとCLI操作に切り替わります。
• 同期taskのtimeoutが効かないことがある。既定のThreadPoolTaskRunnerで.submit()したtaskは、time.sleep()のようなブロッキング処理をtimeout_secondsで中断できません。中断を確実にしたい場合は、ProcessPoolTaskRunnerを使うか、taskを非同期化してawaitベースのI/Oにします。
• 環境変数とプロファイルの優先順位を取り違える。MCPサーバーは環境変数を優先し、未設定なら~/.prefect/profiles.tomlを見ます。両方が設定されていると、意図しない接続先に繋がることがあります。
• Claude Codeのターミナルアクセスは、MCPの読み取り専用制限の外にあります。MCPサーバー自体は読み取り専用でも、Claude Codeがターミナルコマンドを実行できる設定であれば、prefect deployment deleteのような破壊的な操作をMCPを経由せずに実行できてしまいます。自律的にエージェントを動かす場合は、APIキーに紐づくPrefect側のロールも合わせて確認します。読み取り専用に絞れるのは、Prefect CloudのTeam・Pro・Enterpriseプランのサービスアカウントです。

まとめ

Claude CodeでPrefectのワークフローを書く流れは、@flowと@taskでPython関数を装飾し、ローカルで実行を確認し、.serve()かCLIでスケジュールを登録する、という3段階です。Prefect MCPサーバーを追加で繋げば、flow runの監視・デバッグもClaude Codeの会話の中で進められます。MCPサーバーは読み取り専用なので、デプロイ作成やスケジュール変更まで任せたい場合は、prefectコマンドを使うよう指示に明記します。