ワークフローガイド

プロジェクトのワークフローでは、工程に置いたエージェントが工程ごとに 1 回実行されます。ここでは、その工程で動くエージェントを SDK の workflow runner（agenticstar_platform.workflow）で実装する方法を解説します。

対象

マーケットプレイス版 3.1 以降のワークフローで使えます。SDK は v3.0.1 以降が必要です（実行結果の形式が v3.0.1 で変わり、プラットフォームは新しい形式だけを受け付けます）。最新版（v3.0.15）を使ってください。チャットで動くエージェントは、これまでどおり run_marketplace_agent を使います（デプロイガイド）。

チャット用 runner との違い

項目	チャット（run_marketplace_agent）	ワークフロー（workflow.run）
起動の単位	会話のメッセージ 1 件	ワークフローの工程 1 回
入口	async def my_agent(emitter, message)	async def run(context)
入力	最後のユーザーメッセージ	設定値（パラメータ）と入力ファイル（前の工程の成果物など）
出力	関数の戻り値（completion_success の本文）	契約で宣言した成果物ファイル
インストール	agenticstar-platform[runner]	agenticstar-platform（core だけ。config.toml や DB 設定は不要）
runner が使う環境変数	EXECUTION_ID 等 + DB_* + WEBHOOK_URL	プラットフォームが注入する ASTER_RUNTIME_* と EXECUTION_ID
結果	completion_success / completion_failure	「どう終わったか」を表す outcome（下記）

仕組み

runner は親プロセスとして動き、エージェントの関数を子プロセスで実行します。エージェント側に書くのは関数の中身だけです。

1. 注入された環境変数（ASTER_RUNTIME_PROTOCOL / ASTER_RUNTIME_BASE_URL / ASTER_RUNTIME_TOKEN / EXECUTION_ID）を検証し、プラットフォームの runtime API から実行の文脈を取得します。不正な場合は、エージェントを呼ばずに終了します
2. 開始の記録（context.manifest 監査）を保存します。保存を確認できなければエージェントを起動しません
3. 入力ファイルを取得して digest とサイズを検証し、/workspace/input（読み取り専用）に置いて、作業用に /workspace/work へコピーします。from_parameters を指定した設定ファイルは、パラメータから生成します
4. エージェントの関数を、/workspace/work をカレントディレクトリとする子プロセスで実行します
5. 終了後、宣言した成果物と編集したファイルを回収・検証して保存し、実行結果（completion）を保存します

エントリポイント

エントリポイントは async def run(context) です。外部システムへの読み書きは、前後を監査で記録します（外部操作の監査）。ローカルのファイルの読み書きは監査に記録しません。

my_agent/main.pyPython

async def run(context):
    instructions = (context.work_dir / "instructions.md").read_text(encoding="utf-8")

    inv = await context.audit.tool_invoked("get_product", effect_kind="read", target="product-api")
    product = await product_api.get(context.parameters["product_code"])  # あなたの外部 API 呼び出し
    await context.audit.tool_effect(inv, status="ok", count=1)

    reply = await create_reply(product, instructions)  # あなたの処理
    (context.output_dir / "reply.md").write_text(reply, encoding="utf-8")

必要なインストールは core だけです（pip install agenticstar-platform==3.0.15）。イメージの起動コマンドで run() にエントリポイントを渡します。エージェントの関数は /workspace/work をカレントディレクトリとする子プロセスで実行されるため、エントリポイントのモジュールは、パッケージとしてインストールするか、PYTHONPATH にアプリケーションのディレクトリを入れて、どこからでも import できるようにしてください。

Dockerfile（末尾）Dockerfile

WORKDIR /app
COPY . .
# 子プロセスは /workspace/work で動くので、/app のコードを import できるようにする
ENV PYTHONPATH=/app

CMD ["python", "-c", "from agenticstar_platform.workflow import run; run('my_agent.main:run')"]

CMD は上のように exec 形式（["python", "-c", ...]）で書き、runner をコンテナの PID 1 にしてください。シェル形式（CMD python -c ...・sh -c "..."）や init プロセスを前に置くと、そのプロセスの環境変数に runner の token が残り、エージェントが実行するコマンドから読めてしまいます（v3.0.13 から、PID 1 が token を持っていると runner が警告をログに出します）。

run() の終了コードは次のとおりです。

• 0: 実行結果を保存した（既に終端していた場合を含む）。エージェントの例外や、エントリポイントを import できなかった場合も、結果（failed/agent_error）を保存できれば 0 です。取消の後にプラットフォームが結果の保存を断った場合も 0 です（v3.0.13+。プラットフォームが取消として閉じます）
• 2: 実行結果を保存できなかった
• 3: 環境変数または実行の文脈が不正（エージェントは呼ばれていない）

そのほかの障害では、例外のまま 0 以外で終わることがあります（SDK は終了コードを保証しません）。結果を保存しないまま 0 以外で終わった実行はプラットフォームが failed/agent_exit_nonzero、0 で終わったのに結果が無い実行は回収の猶予のあと failed/completion_missing として閉じます。どちらも効果不明として人の確認に回ります。

環境変数と秘密情報

エージェントのプロセスには、次の環境変数が渡ります。runner が runtime API に使う token（ASTER_RUNTIME_TOKEN）は渡されません。v3.0.13 からは、エージェントのプロセスから runner の /proc/<pid>/environ を読んでも見えません（Linux。起動は exec 形式にしてください）。

• 管理画面のエージェント登録で設定した環境変数（LLM の API キーなどの秘密情報はここに登録します。チャットのエージェントと同じです）。契約の fixed_env_names に載った名前が渡ります（固定の環境変数）
• 契約の env_bindings で、パラメータの値から作る環境変数
• 契約の connection_slots に置いた接続先の access token（接続先の token）
• EXECUTION_ID などの実行の相関

EXECUTION_ID・ASTER_*・MP_*・PATH・PYTHON*・LD_* などの予約された名前は、管理画面の環境変数にも env_bindings にも使えません。

ワークフローの契約

ワークフローで使うには、管理画面のエージェント登録で「ワークフローで使う」を有効にし、契約を登録します。契約は JSON で保存され、YAML から読み込むこともできます。契約で、利用者が変えられる設定、入力ファイル、成果物、接続先を定めます。

契約の例JSON

{
  "contract_version": 1,
  "runtime_protocol": "mp-workflow/1",
  "parameters_schema": {
    "type": "object",
    "additionalProperties": false,
    "properties": {
      "product_code": {"type": "string", "maxLength": 64},
      "language": {"type": "string", "enum": ["ja", "en"]}
    },
    "required": ["product_code", "language"]
  },
  "defaults": {"product_code": "demo", "language": "ja"},
  "env_bindings": {"PRODUCT_CODE": "product_code", "REPLY_LANGUAGE": "language"},
  "files": [
    {"name": "instructions", "path": "instructions.md", "format": "markdown", "editable": true, "required": true},
    {"name": "settings", "path": "settings/agent.yaml", "format": "yaml", "from_parameters": true}
  ],
  "outputs": [
    {"name": "reply", "path": "reply.md", "format": "markdown", "required": true}
  ],
  "connection_slots": [],
  "effect_mode": "read_only",
  "limits": {"execution_timeout_seconds": 900}
}

キー	内容
runtime_protocol	"mp-workflow/1"
parameters_schema	利用者が工程・プロジェクトごとに変えられる設定の JSON Schema。type は object にし、additionalProperties は入れ子のオブジェクトも含めて false にします。使えるキーワードは type / properties / required / additionalProperties / items / enum / minLength / maxLength / minimum / maximum / minItems / maxItems / title / description / default。配列には items が必須です。入れ子は 8 段、配列は 100 要素、schema は 64KiB、パラメータの値（JSON）は 64KiB まで
defaults	パラメータの既定値。工程の設定、プロジェクトの設定の順に上書きされます
env_bindings	環境変数名 → パラメータ名。パラメータの値を環境変数で受け取りたいときに書きます。名前は英大文字で始まる英大文字・数字・_（64 文字まで）。文字列・数値・真偽値のパラメータだけで、値は文字列にして渡します（真偽値は "true" / "false"）。秘密情報らしい名前（TOKEN / SECRET / API_KEY などを含む名前）は使えません
files	入力ファイル。name / path（相対パス）/ format（markdown / yaml / json / text / binary）/ editable / required / from_parameters（パラメータ全体から生成する yaml・json）/ schema（yaml・json の検証）
outputs	成果物。name / path（/workspace/output 基準の相対パス）/ format / required / schema。source: "work" を指定すると、/workspace/work 基準の path のファイル（編集したファイル）を成果物として回収します。宣言していないファイルは回収されません
connection_slots	接続先の枠。{name, allowed_provider_ids, required}（下記）
fixed_env_names	工程に渡す固定の環境変数の名前（下記）。管理画面がエージェントの env の名前から作るため、契約に書く値ではありません
effect_mode	read_only（読み取りのみ）/ idempotent_write（冪等な書き込み）/ non_idempotent_write（非冪等な書き込み）。失敗したときに自動で再実行されうるのは、効果が不明でない read_only と idempotent_write だけです。外部へ書き込むエージェントは、書き込みに合った値を宣言してください
limits.execution_timeout_seconds	実行時間の上限（秒）。0 は制限しない（最大 86400 秒 = 24 時間）。指定しなければ 900 秒です

接続先を使う場合は、connection_slots に枠を置きます。name は英字で始まる英数字・_・-（64 文字まで）、allowed_provider_ids は管理画面に登録した OAuth プロバイダーの ID です。どの接続先を使うかは、利用者がプロジェクトで選びます。

接続先の枠の例JSON

"connection_slots": [
  {"name": "gitlab", "allowed_provider_ids": ["<OAuth プロバイダーの ID>"], "required": true}
]

固定の環境変数

利用者ごとに変わらない設定値（接続先の URL、テナント ID、外部サービスの API キーなど）は、管理画面のエージェント登録の env に入れます。工程に渡るのは、そのうち契約の fixed_env_names に載った名前だけです。

• fixed_env_names は管理画面がエージェントの env の名前から作ります。契約に書く値ではありません（値も契約には持たず、起動のたびに管理画面の env から読みます）
• そのため env を足す・消す・名前を変えると契約が変わり、contract_digest も変わります。認定し直してください
• 名前は英大文字で始まる英大文字・数字・_（64 文字まで）です。ASTER_ / MP_ などの予約名と、env_bindings と同じ名前は使えません。env_bindings と違って TOKEN や API_KEY を含む名前も使えます（値を管理するのが管理者だからです）
• secret として登録した env は Kubernetes の Secret 参照で渡り、値は実行の情報に残りません

工程には基盤のデータベースの接続情報は渡りません。 接続先の endpoint（MCP サーバーの URL など）が必要な場合は、この固定の環境変数で渡してください。

契約のほかに、管理者が認定の情報を登録します。ワークフローで起動できるのは、certification の次の 5 つがすべてそろったエージェントだけです。

• sdk_version — 試験した SDK の版
• adapter_version — 試験した adapter の版
• tests_ref — 試験結果の参照
• image_digest — 認定したイメージの digest（sha256:<64 桁>）
• supported_policies — 対応している基本ルール（open・guarded・strict から、試験で確かめたものの配列）。プロジェクトの基本ルールがここに無いと、起動せず denied/policy_unenforceable になります

1 つでも欠けると、プロジェクトの画面で「利用不可 (未認定)」となり、工程に置けません（設定ガイド）。実行には認定した digest のイメージが使われるので、イメージを更新したら認定し直してください。digest は、たとえば次のコマンドで確認できます。

curl

docker buildx imagetools inspect <レジストリ>/<イメージ>:<タグ>

ワークスペースとファイル

場所	属性	用途
/workspace/input	context.input_dir	入力ファイル（読み取り専用）
/workspace/work	context.work_dir	入力の作業用コピー。editable のファイルはここで編集し、outputs の source: "work" で回収します
/workspace/output	context.output_dir	宣言した成果物を書く場所

• 上限は 1 ファイル 20MiB、1 実行で合計 100MiB・100 件です。作業ディレクトリから回収するファイル（source: "work"）は 1MiB までです。path は 1〜255 文字です
• 回収するときは、symlink・ハードリンク・ディレクトリの外を指すパス・上限を超えるファイルを拒否します。編集されていない任意（required でない）の作業ファイルは保存しません
• JSON / YAML は厳格に読みます。重複したキーは拒否します。YAML は 1 つの document だけを受け付け、anchor / alias・明示したタグ（!!str を含む）を拒否し、yes / no や日付を暗黙に変換しません。先頭が 0 の数（01 など）はエラーになります。文字列として扱いたい値は引用符で囲んでください

context で使えるもの

属性・メソッド	内容
execution_id / run_id / project_id / root_execution_id / entry_ref / operation_key / attempt	実行の相関。operation_key は外部操作の冪等キーの元に使えます
parameters	解決済みの設定値（読み取り専用）
connections	接続 slot 名 → 接続先の ID
effect_mode	契約の effect_mode
input_dir / work_dir / output_dir	ワークスペースのパス
deadlines / deadline_remaining()	期限と、エージェントが止められるまでの残り秒数（不明なら None）
audit.tool_invoked() / tool_effect() / tool_denied()	外部操作の監査（次の節）
is_cancel_requested(max_age=5.0) / ensure_can_act()	取消・期限の確認
skip(reason)	処理対象が無いときに工程を skip する（no_target / not_applicable / nothing_to_do）
policy_unavailable()	外部操作の前に、操作の可否を確かめられなかったことを伝える
log(message)	runner のログに書く（2000 文字まで）
connection_token(slot, min_ttl=300.0)	接続先の access token を取り直す（v3.0.5+）
gate_decision	確認依頼の決定を受けて起動した実行なら、その決定（読み取り専用。それ以外は None。v3.0.9+）

gate（人の確認）を決めることや、任意のデータベース接続は context からはできません。決まった確認の内容は gate_decision で読めます（確認の決定を受けた実行）。

外部操作の監査

外部システムへの読み書きは、前後を監査に記録します。

1. 操作の前に tool_invoked(tool, effect_kind=..., action_ref=..., target=...) を呼びます。この呼び出しは記録が保存されてから戻ります。保存できないとき（取消・期限の後を含む）は AuditNotPersisted を送出するので、その操作には進まないでください
2. 操作の後に、tool_invoked の戻り値を渡して tool_effect(invoked, status=..., count=...) を呼びます。status は ok / failed / unknown です。結果が分からないときは unknown にします。書き込みの効果は「不明」として記録され（had_mutating_success が null になりうる）、その実行が成功しなかった場合は自動で再実行されず、人の確認に回ります。書き込みの tool_invoked に対応する tool_effect が無い場合も、効果は不明になります（読み取りには不要です）
3. 判定の結果として操作を実行しなかった場合は tool_denied(tool, reason=...) を呼びます。1 回でも呼ぶと、関数が正常に戻っても実行全体の結果は denied になります（監査の保存失敗・取消・期限など、判定の順で上にあるものが優先されます）。拒否された実行のその後の扱いは、ワークフローの設定に従います

監査の記録の保存に 1 件でも失敗した実行は failed/audit_persist_failed になります。取消・期限・ワークフローの実行の終了の後に断られた監査は保存の失敗には数えず、結果は取消・期限の理由になります（v3.0.12 までは、runner が停止をまだ知らないうちに断られた監査は failed/audit_persist_failed になりました）。

監査の metadata（**extra）に渡せるのは、文字列・整数・有限の小数・真偽値だけです。それ以外（NaN / Infinity・リスト・辞書・None）は SDK が捨て、捨てたキーの名前を警告に出します（v3.0.13+。値はログに出しません）。

監査を記録しても、操作を許すかどうかが自動で判定されるわけではありません。エージェント内の操作の統制は、エージェント自身が実装して従います（SDK の監査 API や ポリシーモジュール を import しただけでは、操作は止まりません）。可否を確かめられないまま外部操作を始めないでください（処理しない・確かめられないとき）。

effect_kind は、読み取りが read / list / query / fetch、書き込みが write / create / update / delete / send / mutate です。監査の metadata には文字列・数値・真偽値だけが残ります。本文・token・引数の値は入れないでください。

外部への送信（冪等な書き込み）Python

async def run(context):
    await context.ensure_can_act()  # 取消・期限の後は新しい外部操作を始めない
    action_ref = f"send:{context.operation_key[:16]}"  # 同じ工程の再送で変わらない冪等キー
    body = (context.work_dir / "reply.md").read_text(encoding="utf-8")

    invoked = await context.audit.tool_invoked("send_mail", effect_kind="send", action_ref=action_ref, target="mail")
    try:
        receipt = await mail_client.send(idempotency_key=action_ref, body=body)  # あなたの送信処理
    except Exception as e:
        # 結果が分からない失敗を「送っていない」として扱わない
        await context.audit.tool_effect(invoked, status="unknown", action_ref=action_ref, error=type(e).__name__)
        raise
    await context.audit.tool_effect(invoked, status="ok", count=1, action_ref=action_ref)
    # receipt.md は契約の outputs に宣言しておく（宣言していないファイルは回収されない）
    (context.output_dir / "receipt.md").write_text(f"# receipt\n\n{receipt}\n", encoding="utf-8")

この例のように外部へ書き込むエージェントは、契約の effect_mode を idempotent_write などにします。再送に備えて、operation_key と安定した action_ref を外部の冪等キーにしてください。結果を照合できない書き込みを、ネットワークエラーだけを理由に呼び直さないでください。

処理しない・確かめられないとき

skip() と policy_unavailable() は記録するだけで、エージェントの処理は止まりません。呼んだら関数から戻ってください。

Python

async def run(context):
    targets = load_targets(context.work_dir)
    if not targets:
        await context.skip("no_target")   # 処理対象が無い
        return

    if not await can_check_permissions():
        await context.policy_unavailable()  # 外部操作の前に、操作の可否を確かめられなかった
        return
    ...

• skip(reason)（no_target / not_applicable / nothing_to_do）: 工程は skipped で終わり、ワークフローは次の工程へ進みます
• policy_unavailable(): 工程は skipped/policy_unavailable で終わり、次の工程へは進みません。プラットフォームが自動再実行の範囲で確かめ直し、自動再実行が無い・上限に達した場合は人の確認に回します。この実行で書き込みの tool_invoked を 1 件でも呼んでいると、failed/skip_after_effect になります

取消と期限

• 外部操作を始める前に await context.ensure_can_act() を呼びます。取消・期限の後は WorkflowError を送出します
• runner は、停止・回収・保存のための余裕を回収期限から取り、実行期限と「回収期限 − 余裕」の早い方でエージェントを止めます。残り時間は context.deadline_remaining() で分かります
• 止めるときは、子プロセスのグループに SIGTERM を送り、猶予（10 秒ずつ）を待っても残っていれば SIGKILL を送ります。止まったことを確認できたら、その時点で書かれていた成果物を回収します。結果は取消・期限・停止の理由に従います（実行結果）
• 取消・期限・Pod の停止を runner が知ると、以後の is_cancel_requested() は True を返し、ensure_can_act() と connection_token() は WorkflowError を送出します（v3.0.6 からは、監査など他の呼び出しの応答でも停止が伝わります）
• 取消の後は、プラットフォームが成果物と結果の保存も断ります。runner は停止として終わり、結果を保存しないまま終了コード 0 で終わります。実行はプラットフォームが取消として閉じます（v3.0.13+）

接続先の token

契約の connection_slots に置いた接続先の access token は、起動時に環境変数 ASTER_CONNECTION_<SLOT>_TOKEN（slot 名を大文字にし、- を _ に置き換えた名前。そのため a-b と a_b の slot は同時に使えません）で渡されます。この token は 1 時間ほどで切れることがあるため、長い実行で外部 API を呼ぶ前は connection_token() を使ってください（v3.0.5+）。

Python

token = await context.connection_token("gitlab")      # {access_token, token_type, expires_at}
headers = {"Authorization": f"Bearer {token['access_token']}"}

接続先の endpoint は token と一緒には渡りません。工程には基盤のデータベースの接続情報も渡らないため、接続先の URL は固定の環境変数で渡してください。

期限まで min_ttl 秒（既定 300 秒）を切るまでは手元の値を返し、切ったらプラットフォームから取り直します。期限の分からない token は 600 秒ごとに取り直します。起動時に束ねていない slot、失効・再認可待ちの接続、取消・期限の後は WorkflowError になります。token はログや例外の文言に出さないでください。

確認の決定を受けた実行

確認担当者が差し戻した後の戻り先の工程の再実行と、進めた後に続く工程の実行では、context.gate_decision にその確認依頼の決定が入ります（v3.0.9+）。それ以外の実行（ワークフローを始めたときの最初の工程、確認を経ずに起動した工程）では None です。

キー	内容
gate_id	確認依頼の ID
decision	send_back（差し戻し）または proceed（進めた）
decided_by	human（担当者が決めた）または system（外部条件待ちの条件が成立して自動に決まった。マーケットプレイス版では期限で決まることはありません）
note	担当者が書いた差し戻しの理由、または進めたときのコメント（前後の空白を除いたもの）。書かれていない・空白だけのとき、自動の決定では None
decided_at	決まった日時（ISO 8601。分からないときは None）

差し戻しの理由をエージェントへの指示に足すと、指摘を直した成果物を作り直せます。v3.0.8 までは理由がエージェントに届かず、エージェントは指摘を踏まえて作り直せませんでした。

Python

async def run(context):
    instructions = (context.work_dir / "instructions.md").read_text(encoding="utf-8")

    review = context.gate_decision
    if review and review["decision"] == "send_back" and review["note"]:
        instructions += f"\n\n## 修正の依頼\n\n{review['note']}\n"
    ...

• 開始の記録（context.manifest）には gate_id / decision / decided_by だけが入り、理由・コメントの本文は入りません
• gate_decision を返さないプラットフォームの版では、常に None で、ほかの動作は変わりません。使うには、この版の SDK でイメージを作り直し、認定し直してください

実行結果（outcome）

結果は「どう終わったか」を表す値で記録されます。成否はエージェントが最後までやり遂げたかで決まり、書き込みの失敗や効果を観測できなかったことでは下がりません。まず、次の表を上から順に判定します。

outcome / 理由コード	いつ
failed / audit_persist_failed	監査を保存できなかった
stopped / cancelled	取り消された
timeout / execution_timeout	期限を過ぎた
stopped / terminal_elsewhere	実行がほかで終端していた
failed / runner_stopped	Pod が停止された
denied / <reason_code>	tool_denied で拒否を記録した（最初の拒否の reason_code。既定 tool_denied）
failed / agent_error	エージェントの関数が例外を送出した、またはエントリポイントを import できなかった
failed / output_invalid	宣言した成果物が不正、または読めない
failed / skip_after_effect	書き込みの tool_invoked を呼んだ実行で policy_unavailable() を呼んだ
skipped / <理由>	skip() または policy_unavailable() を呼んだ（書き込みの成功を観測していない場合）
failed / required_output_missing	必須の成果物が無い
completed / completed	関数が正常に戻った

そのあと、runner 自身の状態で結果が置き換わることがあります。

outcome / 理由コード	いつ
failed / runner_restarted	runner が再起動し、エージェントの結果を観測できなかった
failed / artifact_save_failed	成果物の保存が一部失敗した（completed / skipped を置き換える）
failed / cleanup_uncertain	前のエージェントのプロセスを止め切れなかった
failed / input_integrity	入力ファイルの digest・サイズ・パスが一致しなかった（エージェントは起動しない）

エージェントの起動前（入力の取り込み中を含む）に取消・期限が来た場合は、エージェントを起動せずに stopped/cancelled / timeout/execution_timeout で終わります。incomplete は共通の語彙にありますが、workflow runner は出しません。

何を書き込んだかは、結果の counts / tools と had_mutating_success（true = 書き込みの成功を観測 / false = 効果が無いと確定 / null = 不明）に残ります。null で成功しなかった実行は自動では再実行されず、人の確認に回ります。

デバッグ

• エージェントの標準出力・標準エラーは Pod のログに出ます。context.log() の内容は runner のログ（Pod の標準エラー）に出ます。v3.0.13 から、run() はログの出力先が未設定なら SDK のログ（INFO 以上）を標準エラーに出します（v3.0.12 までは、推奨の起動のままだと context.log() は出ませんでした）。ロガーを自分で設定している場合はその設定に従います
• エージェントの例外は、既定ではクラス名だけが標準エラーと結果（error_class）に出ます。トレースバック全体を出すには、イメージの Dockerfile で ENV ASTER_WORKFLOW_DEBUG_TRACEBACK=1 を設定します（本文や token がログに出うるので、調査が終わったら外してください）
• サンプル（回答案・模擬送信の 2 本）は、PyPI のソース配布物（sdist）の docs/examples/ に入っています

次のステップ

SDK API リファレンス — ワークフローモジュール

run / WorkflowContext / 例外 / 結果の完全仕様

ガイドを見る

デプロイガイド

イメージのビルドと管理画面への登録

ガイドを見る