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

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

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

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

Prefect MCPサーバーは、Claude Codeからflow runやログを読み取り専用で調べるためのベータ版のMCPサーバーです。Claude Codeには/plugin install prefectかclaude mcp addで繋ぎ、デプロイの作成や更新はできません。

前提条件

始める前に、次の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では効かない場面があります。詳しくは後述の「よくあるつまずき」を参照してください。

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

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

実行すると、Prefect UIへのリンクと各taskの状態(Completed)がターミナルに流れます。自己ホストのOSS版で試すなら、別のターミナルでprefect server startを動かしておくと、http://localhost:4200でUIが開きます。

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

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

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

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

スケジュールの登録先は、.serve()のほかにもUI・prefect.yaml・prefect deployの対話形式があります。コードに書く方法とCLIで足す方法は、次のように役割が分かれます。

CLIでの登録は次のとおりです。

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

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

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

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

Prefect Cloudへの登録は、クイックスタートではuvx prefect-cloud deploy 01_getting_started.py:main --name my_first_deployment --from https://github.com/PrefectHQ/quickstartの形です。スケジュールはuvx prefect-cloud schedule main/my_first_deployment "0 8 * * *"で付けます。

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

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

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

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

まずプラグインの導入です。

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

プラグインには診断・ドキュメント参照・リリースノート向けのワークフローガイドも含まれます。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

APIキーを使わず自己ホストのサーバーに繋ぐなら、PREFECT_API_URL=http://your-server:4200/apiを渡し、ベーシック認証を使っているときはPREFECT_API_AUTH_STRINGも渡します。環境変数を省略すると、ローカルの~/.prefect/profiles.tomlにあるアクティブなプロファイルの認証情報をそのまま引き継ぎます。環境変数を指定した場合は、プロファイルより環境変数が優先されます。

手動追加で設定ファイルに何が書かれるか

claude mcp addの書式はv2.1.289の--helpに例つきで出ます。stdioサーバーの追加はclaude mcp add my-server -e API_KEY=xxx -- npx my-mcp-serverの形で、--より前がclaudeのオプション、後ろがサーバーの起動コマンドです。

プロジェクトスコープ(--scope project)で実行すると、プロジェクト直下の.mcp.jsonに次の項目が書かれます。

claude mcp add --scope project prefect \
  -e PREFECT_API_URL=http://localhost:4200/api \
  -- uvx --from prefect-mcp prefect-mcp-server

{
  "mcpServers": {
    "prefect": {
      "type": "stdio",
      "command": "uvx",
      "args": ["--from", "prefect-mcp", "prefect-mcp-server"],
      "env": { "PREFECT_API_URL": "http://localhost:4200/api" }
    }
  }
}

-eの値はenvにそのまま入ります。Prefect CloudのAPIキーをPREFECT_API_KEYとして同じ形で渡すと、キーも平文で.mcp.jsonに残ります。リポジトリで共有するファイルなので、キーを渡すときはスコープを既定のlocalのままにするか、プロファイルから引き継ぐ構成を選びます。

Prefect Horizonにデプロイした自前のMCPサーバーへ繋ぐなら、HTTP転送で追加します。claude mcp add prefect --transport http https://your-server-name.fastmcp.app/mcpを同じ要領で実行すると、.mcp.jsonには"type": "http"と"url"だけの項目が入ります。

接続後にできること

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

claude.aiのコネクタ(claude.ai/directory/prefect)はClaude側の接続で、プラグインやclaude mcp addはClaude Code側の設定です。コネクタとプラグインは、どちらもPrefect CloudのOAuthでホスト型の読み取り専用サーバーに繋ぎます。claude mcp addで手元に起動するサーバーは、プロファイルか環境変数の認証情報を使います。同じプラグインはCowork向けにも使え、その場合は組織が管理するPrefect CloudのOAuthになります。

MCPツールは読み取り専用で、作成・更新はprefectのCLIかSDKで行います。MCPサーバーにはドキュメントの参照機能(docs proxy)があるため、「prefectコマンドでデプロイのスケジュールを更新して」と頼むと、Claude Codeが現行のコマンドを調べて組み立てます。

似た名前の機能に、Prefect Cloud自身が持つAIログ要約があります。アカウント管理者が「Settings > Account Settings > Controls」の「Marvin AI」を有効にすると、フローランの「Logs」タブで要約を生成できます。automationの通知テンプレートに{{ flow_run_log_summary }}を書けば、失敗の一行要約を通知に載せることも可能です。Claude CodeとMCPサーバーを繋ぐ話とは別の仕組みなので、監視・デバッグの手段としては区別して使います。

よくあるつまずき

• 同期taskのtimeoutが効かないことがある。既定のThreadPoolTaskRunnerで.submit()したtaskは、time.sleep()のようなブロッキング処理をtimeout_secondsで中断できません。中断を確実にしたい場合は、ProcessPoolTaskRunnerを使うか、taskを非同期化してawaitベースのI/Oにします。
• 環境変数とプロファイルの優先順位を取り違える。プロファイルと環境変数が食い違っていると、意図しない接続先に繋がることがあります。手元のサーバーも併用するなら、MCPサーバー名をprefect-localのようにプラグインと別名にし、作業前にget_identityで接続先を確かめます。
• Claude Codeのターミナルアクセスは、MCPの読み取り専用制限の外にあります。MCPサーバー自体は読み取り専用でも、Claude Codeがターミナルコマンドを実行できる設定であれば、prefect deployment deleteのような破壊的な操作をMCPを経由せずに実行できてしまいます。自律的にエージェントを動かす場合は、APIキーに紐づくPrefect側のロールも合わせて確認します。サービスアカウントによるAPI認証はTeam・Pro・Enterpriseプランで使えます。see_*権限だけの読み取り専用に絞れるのはProとEnterpriseです。

まとめ

Prefectを使うなら、まず手元でflowを動かして確かめ、スケジュールはコードに置くかCLIで足すかを変更の頻度で決めます。MCPサーバーを繋ぐ場合は、読み取り専用でもターミナルが使えるかぎり削除まで走れるため、Prefect側のロールで権限を絞るかどうかを先に決めておきます。