@PDC-Kurashinak(kurashinak@pdc-ds.com)in

Strands AgentsのWorkflowツールを試してみた！

Last updated at 2026-07-14Posted at 2026-07-14

どうもこんにちは。

最近、AIエージェント構築を色々と試している中で、「どうやったら人間がやっているタスクを正確無比にやってくれるエージェントを構築できるのだろうか」と考えている今日この頃です。

AIエージェント構築によく使うSDKは、PythonのStrands Agentsなんですが、リファレンスを読んでいると知らないツールが結構出てきたりしてます。（「new」の文字が結構早いペースで増えていっているような...）
そんな中で、長いこと気にはなっていたけど検証ができていなかった、workflowツールの検証をしてみようと思います。

題材は、何か新しいサービスを開発する時に調査を依頼したり、自分の頭の中に浮かんでいる構成を評価してもらうための「技術選定&調査エージェント」です。

たとえば、次のような入力を渡します。

{
  "prompt": "新規B2B SaaSを、Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrockで作る案を評価して"
}

この時、想定している動作としては、以下の観点に分けてレビューし、最後に採用判断つきのMarkdownレポートを返す。ということを想定しています。

要件整理
フロントエンド
バックエンド
インフラ
セキュリティ
コスト
運用

また、今回は、「マルチエージェント構成」が前提となるので、マルチエージェント構成でエージェントデプロイしてみたいねんって方は参考にしてみてくださいませ。

この記事は誰向け？

この記事は以下のような方向けに執筆しています。

Strands Agents のマルチエージェント構成を試してみたい方
strands_tools.workflow が何をしてくれるのか知りたい方
Amazon Bedrock AgentCore Runtime 上で Strands Agents を動かしたい方
AWS CDK（Python）で AgentCore Runtime へエージェントをデプロイしたい方

デモコードの場所

実際に動かしたい方は、以下を参照ください。

この検証でやりたいこと

今回のゴールは次の3点です。

Strands Agents の workflow ツールで、複数タスクの依存関係を定義する
親エージェントからタスクごとの子エージェントを呼び出し、技術選定レビューを複数の専門タスクに分割して実行する
最終的にはレポートをレスポンスとして返す

構成としては以下のようになります。

上記のフローの中で、各タスク（「要件整理」以降）は子エージェントによって実行されるようになっています。

実装の流れ

0. 作業プロジェクトの用意

mkdir strands-agents-agent-workflows
cd strands-agents-agent-workflows

1. CDKプロジェクトの雛形を整える

1-1. 依存パッケージ

CDK側の依存関係は以下です。

requirements.txt

aws-cdk-lib==2.255.0
constructs>=10.0.0,<11.0.0
cdk-ecr-deployment>=4.0.0,<5.0.0

開発用には pytest も入れます。

requirements-dev.txt

-r requirements.txt
pytest>=8.0.0,<9.0.0

cdk-ecr-deployment は、CDK がビルドした Docker image asset を、Runtime 用に作成した専用 ECR Repository へコピーするために使っています。

Runtime コンテナ側の依存関係は以下です。

agent/requirements.txt

bedrock-agentcore>=1.3.0,<2.0.0
strands-agents>=1.0.0,<2.0.0
strands-agents-tools>=0.2.0,<1.0.0
boto3>=1.42.0

今回の主役は strands-agents-tools に含まれる workflow ツールであるため、strands-agents-toolsのパッケージは必要となります。

1-2. CDKエントリーポイント

app.py は通常の CDK エントリーポイントです。

app.py

#!/usr/bin/env python3
import os

import aws_cdk as cdk

from infra.agentcore_tech_selection_stack import AgentCoreTechSelectionWorkflowStack


app = cdk.App()

AgentCoreTechSelectionWorkflowStack(
    app,
    "AgentCoreTechSelectionWorkflowStack",
    env=cdk.Environment(
        account=os.environ.get("CDK_DEFAULT_ACCOUNT"),
        region=os.environ.get("CDK_DEFAULT_REGION", "ap-northeast-1"),
    ),
)

app.synth()

1-3. cdk.json

cdk.json は、CDK CLI が常にプロジェクト内の仮想環境を使うようにしています。

cdk.json

{
  "app": ".venv/bin/python app.py",
  "context": {
    "@aws-cdk/core:checkSecretUsage": true,
    "@aws-cdk/core:target-partitions": [
      "aws"
    ]
  }
}

.venv/bin/python を固定しておくと、仮想環境を activate していない状態でも cdk synth や cdk deploy が同じ Python 環境を使ってくれます。

1-4. 仮想環境

python3.12 -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt

AWS認証とリージョンを確認します。

aws sts get-caller-identity
export AWS_REGION=ap-northeast-1

必要であれば bootstrap します。

cdk bootstrap

2. workflow で技術選定レビューの流れを定義する

まず、今回の技術選定会議をタスクとして定義します。

実装は agent/tech_selection_agent/workflow_spec.py に置きました。

agent/
└── tech_selection_agent/
    ├── agent.py             # エージェントを実装
    ├── runtime_app.py       # エントリーポイントを実装
    └── workflow_spec.py     # ワークフロー内のタスクなどを定義

2-1. タスクの全体像

今回の workflow は次の形です。

最初に requirements_reader が評価対象の技術選定案を整理します。

その結果を受けて、6つのレビュータスク（サブエージェント）が並列に動きます。

frontend_reviewer
backend_reviewer
infra_reviewer
security_reviewer
cost_reviewer
operability_reviewer

最後に、decision_maker がレビュー結果を統合して、final_report がMarkdownレポートを作成します。

2-2. タスク（サブエージェント）定義

workflow ツールに渡すタスクは、Python の辞書として定義します。

agent/tech_selection_agent/workflow_spec.py

def build_tasks(prompt: str) -> list[dict[str, Any]]:
    proposal = prompt.strip()
    base_context = (
        "評価対象の技術選定案:\n"
        f"{proposal}\n\n"
        "日本語で、実務上の判断に使える具体性で回答してください。"
    )

    return [
        # 要件を整理するタスク（サブエージェント）
        {
            "task_id": "requirements_reader",
            "description": (
                f"{base_context}\n\n"
                "この案の目的、想定プロダクト、暗黙の前提、未確定事項、"
                "評価軸を整理してください。後続レビューが参照しやすい箇条書きにしてください。"
            ),
            "system_prompt": "あなたは要件定義と技術評価のファシリテーターです。",
            "tools": NO_TASK_TOOLS,
            "priority": 5,
        },
        # 要件整理後、フロントエンドの観点で評価を行うタスク（サブエージェント）
        {
            "task_id": "frontend_reviewer",
            "description": (
                f"{base_context}\n\n"
                "フロントエンド観点で評価してください。UI開発生産性、型安全性、"
                "認証連携、SSR/CSR、パフォーマンス、テスト、チーム運用を扱ってください。"
            ),
            # ここで「何のタスクの後に実行するのか」を定義する
            "dependencies": ["requirements_reader"],
            "system_prompt": "あなたはB2B SaaSのフロントエンドアーキテクトです。",
            "tools": NO_TASK_TOOLS,
            "priority": 4,
        },
    ]

dependencies に依存タスクを指定すると、そのタスクが完了するまで実行されません。

この例だと、frontend_reviewer は requirements_reader が終わってから実行されます。

2-3. タスク（サブエージェント）には tools を明示しておく

今回、少し気をつけたのが tools の指定です。

strands_tools.workflow は、タスクに tools を指定しない場合、親Agentが持っているツールを子Agentへ継承します。

今回の親Agentは workflow ツールを持っています。つまり、何もしないと子Agentも workflow を持つことになります。

これが必ず悪いわけではないですが、子エージェントが新たにワークフローツールを実行してしまうと、無駄なループ処理が発生してしまうため、避けた方が良いかなと思い、以下のように存在しないツール名を指定して子Agentにはツールを渡さないようにしました。

# 最初に`agent/tech_selection_agent/workflow_spec.py`に定数として定義しておく
NO_TASK_TOOLS = ["__no_task_tools__"]

# タスク（サブエージェント）の引数に`NO_TASK_TOOLS`を渡しておく
"tools": NO_TASK_TOOLS

これで、各タスクはツールなしの専門レビューAgentとして動作します。

3. 親Agentから workflow を実行する

次に、workflow ツールを持った親Agentを作成します。

実装は agent/tech_selection_agent/agent.py です。

3-1. workflow import 前に保存先を決める

strands_tools.workflow は、workflow の状態をJSONファイルとして保存します。

デフォルトでは ~/.strands/workflows ですが、AgentCore Runtime のコンテナ上では /tmp 配下に寄せておきます。

これは好みです！

agent/tech_selection_agent/agent.py

# JSONファイルの保存先を指定
DEFAULT_WORKFLOW_DIR = "/tmp/strands-workflows"
os.environ.setdefault("STRANDS_WORKFLOW_DIR", DEFAULT_WORKFLOW_DIR)

from strands_tools import workflow

workflow は import 時に STRANDS_WORKFLOW_DIR を読むため、from strands_tools import workflow より前に環境変数を設定しています。

3-2. 親Agentを作る

agent/tech_selection_agent/agent.py

def create_agent() -> Agent:
    region = get_region()
    model = BedrockModel(
        model_id=get_model_id(),
        boto_session=boto3.Session(region_name=region),
    )
    return Agent(
        model=model,
        tools=[workflow],
        system_prompt=SYSTEM_PROMPT,
        callback_handler=None,
    )

この親Agentは、技術選定レビュー本文を直接書く係ではありません。

役割は workflow ツールを呼び出すことです。

各レビュー担当のAgentは、workflow ツール内部でタスクごとに生成されます。

3-3. create と start

workflow ツールには、アクションという引数が存在します。

値	説明
`create`	タスク定義を保存する
`start`	タスク定義されたワークフローを実行する

まず create でタスク定義を保存し、次に start で実行します。

agent/tech_selection_agent/agent.py

def execute_tech_selection_workflow(prompt: str, workflow_id: str) -> dict[str, Any]:
    # 3-2 で作成した親エージェントを引数に格納する
    agent = create_agent()
    # 2-2 で定義したタスクを引数に格納する
    tasks = build_tasks(prompt)

    # タスク定義を保存する
    create_result = agent.tool.workflow(
        action="create",  # ここで`create`が指定されているので、「タスク定義が保存されている」と理解できる
        workflow_id=workflow_id,
        tasks=tasks,
    )
    ensure_tool_success(create_result, "create")

    # ここで`start`が指定されているので、「保存したタスクが実行される」と理解できる
    start_result = agent.tool.workflow(action="start", workflow_id=workflow_id)
    ensure_tool_success(start_result, "start")

create は状態ファイルを作るだけで、まだLLM実行は始まりません。

実際に各タスクが動くのは start です。start は同期実行なので、final_report まで終わるまで戻ってきません。

3-4. final_report を状態ファイルから取り出す

workflow start の戻り値は、「workflow が完了しました」というメッセージが中心です。

今回欲しいのは、最後の final_report タスクが生成したMarkdown本文です。そのため、workflow が保存したJSONを読みます。

agent/tech_selection_agent/agent.py

workflow_data = load_workflow_data(workflow_id, get_workflow_dir())
final_report = extract_task_text(workflow_data, "final_report")
if not final_report:
    final_report = extract_content_text(start_result)

状態ファイルは次の場所に保存されます。

/tmp/strands-workflows/<workflow_id>.json

この中の task_results.final_report.result から text block だけを抜き出して、最終レスポンスにしています。

親エージェントが最終レポートを読んでユーザにレスポンスを返す ということはしていません。最終レポートの内容をそのまま出力するようにしています。

4. AgentCore Runtime の entrypoint を作る

AgentCore Runtime から呼ばれる entrypoint は agent/tech_selection_agent/runtime_app.py に置きます。

agent/tech_selection_agent/runtime_app.py

from bedrock_agentcore.runtime import BedrockAgentCoreApp, PingStatus

from .agent import execute_tech_selection_workflow


app = BedrockAgentCoreApp()

入力は {"prompt": "..."} に寄せています。

agent/tech_selection_agent/runtime_app.py

def invoke_agent(payload: dict) -> dict:
    prompt = payload.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("payload.prompt must be a non-empty string")

    request_id = str(uuid.uuid4())
    workflow_id = f"tech-selection-{request_id}"

    result = execute_tech_selection_workflow(
        prompt=prompt.strip(),
        workflow_id=workflow_id,
    )

    return {
        "response": result["response"],
        "evidence": {
            "requestId": request_id,
            "workflow": {
                "workflowId": workflow_id,
                "status": result["workflow_status"],
                "taskCount": result["task_count"],
            },
        },
    }

workflow_id は状態ファイル名にもなるため、リクエストごとに uuid を付けて衝突しないようにしています。

AgentCore Runtime の entrypoint と health check は以下です。

@app.entrypoint
def invoke(payload: dict) -> dict:
    return invoke_agent(payload)


@app.ping
def ping() -> PingStatus:
    return PingStatus.HEALTHY

レスポンスには、最終レポート本文と、どの workflow が動いたかを確認するための evidence を含めています。

応答イメージは以下のようになります。

{
  "response": "# 技術選定レビュー\n\n## 結論\n条件付き採用...",
  "evidence": {
    "requestId": "5f3d...",
    "workflow": {
      "workflowId": "tech-selection-5f3d...",
      "status": "completed",
      "taskCount": 9
    }
  }
}

5. Runtime コンテナを作る

AgentCore Runtime へ載せるため、エージェントを Docker 化します。

agent/Dockerfile

FROM public.ecr.aws/docker/library/python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV PORT=8080
ENV STRANDS_WORKFLOW_DIR=/tmp/strands-workflows

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY tech_selection_agent ./tech_selection_agent

EXPOSE 8080

CMD ["python", "-m", "tech_selection_agent.runtime_app"]

ここでも STRANDS_WORKFLOW_DIR=/tmp/strands-workflows を指定しています。

agent.py 側でもデフォルト値を設定していますが、Runtime コンテナとしては Dockerfile と CDK の環境変数でも明示しておくと挙動を追いやすいのかなと思っています。（差分が生まれないように注意が必要かなとは思います。）

6. CDKでAgentCore Runtimeを作る

CDKスタックは infra/agentcore_tech_selection_stack.py に実装しました。

今回作るリソースは主に以下です。

ECR Repository
DockerImageAsset
ECRDeployment
AgentCore Runtime
Runtime 用 IAM Role