Step Functions SDK Direct 完全ガイド Vol5 (Lambda削減+JSONata+Bedrock)

1. この記事について

1-1. 本記事のゴール

本記事は Step Functions SDK Direct Integration 完全ガイド (SF 実践編 Vol5 / 完結弾) です。

Step Functions では Lambda を介さずに AWS SDK API を直接呼び出す「SDK Direct Integration」を活用することで、Lambda 実行コストの削減・アーキテクチャの簡素化・レイテンシ短縮を同時に実現できます。しかし実務では「Optimized と AWS-SDK のどちらを選ぶか」「PascalCase 変換ルールで詰まる」「Bedrock 呼出の IAM 設定が分からない」という壁に直面します。

本記事は以下 6 つの実務論点を Terraform + ASL + CLI の 3 点セットで再現可能に解説します:

1. Optimized / AWS-SDK 選択 — §2 完全比較表 + 選択フロー
2. 7 サービス SDK Direct 実装 — DDB / SNS / SQS / EB / S3 / APIGW / Bedrock を §3-§5 で網羅
3. JSONata + ASL 組み込み関数 — Lambda を JSON 変換レイヤとして使わないテクを §6 で完全解説
4. IAM 最小権限設計 — §7 サービス別権限表 + Condition キー活用
5. 典型エラー 10 ケース — §7 で根本原因 + 修正手順
6. Lambda 置換コスト試算 — §3-QG1 で経営層向け月次差分を 3 シナリオ実算

本記事が読者にもたらす変化を「改善前/後」で示します:

1-2. 読者像

本記事が想定する読者は以下の前提知識を持つ方です:

Lambda を ECS/SFN に置換してコスト削減を推進している FinOps・SRE・アーキテクトの方に特に有用です。既存記事 SDK Integration 入門 (ID:1105) を読了済であれば、本記事で実践編の完結まで一気に到達できます。

1-3. なぜ今 SDK Direct を完結弾として置くか

SF 実践編シリーズは Vol1 で I/O フィルタ、Vol2 で Callback、Vol3 で Distributed Map、Vol4 で Express Workflow を積み上げてきました。これらを習得した読者が最後に直面するのが「結局 Lambda を完全に取り除けるのか？」という問いです。

SDK Direct Integration はその問いへの回答です。Optimized Integration が 10+ サービスに限定されるのに対し、AWS-SDK Integration は 200+ サービスをカバーします。2024-2026 に Bedrock / API Gateway / JSONata サポートが追加され、AI ワークロードを含む実務ユースケースの大半で Lambda 不要アーキテクチャが実現可能になりました。

完結弾として本記事を 5 連作の締めとする理由はもう一つあります。Lambda 置換コスト試算・IAM 設計・典型エラー集は「実装後の本番運用」で最初に直面する論点であり、入門編ではなく実践編最終巻で扱うべき深さを持っています。

なお SDK Direct Integration は 2023-2024 にかけて対応サービスが大幅に拡充されました。特に Bedrock (2023/11 GA) と JSONata サポート (2024/11) は AI ワークロード組込みの障壁を取り除いたターニングポイントです。今こそ「Lambda 不要アーキテクチャ」の完全解説を実践編最終巻として提供するタイミングです。

1-4. ID:1105 との差別化

既存の SDK Integration 入門 (ID:1105) は「Lambda を削って DDB/SQS/SNS/S3 を直接操作する入門ハンズオン」として公開済です。本記事はその発展版・完結弾として上述の 6 点で差別化されています。

本記事を読み終えると、ID:1105 のハンズオン体験を土台に、Optimized/SDK 選択 → 7 サービス実装 → JSONata 変換 → IAM 設計 → エラー対処 → コスト説明 のサイクルを実務で自走できるようになります。

ID:1105 との重複を避けるため、本記事では DDB / SQS / SNS / S3 の基本的な「Lambda を削る」手順の再説明は省略します。実装例は本番規模の複合パターン (ConditionExpression / BatchWriteItem / FIFO + MessageDeduplicationId 等) を中心に構成しています。

1-5. 関連記事

本記事は以下の記事を前提・補完として参照します。SF 実践編 Vol1-Vol4 は途中巻から参入可ですが、ID:1033 (SF 入門) は必読です。

1-6. SF 実践編 5 連作シリーズ全体マップ

SF 実践編は 5 巻で 1 つの完結した学習経路を構成しています。Vol1 → Vol5 の順に読むと「I/O 設計 → 非同期パターン → 大量並列 → 高頻度処理 → Lambda 不要化」の流れで段階的に実力が積み上がります。本巻 Vol5 は「SDK Direct Integration」を中心に置き、前 4 巻で習得した技術を統合する完結弾です。

1-7. 執筆方針

本記事は 2026-04 時点の AWS 仕様 に基づき執筆しています。各実装例は以下の 3 点セットで構成し、手元環境で再現可能なレベルで記述します:

• Terraform HCL: IAM ロール定義・State Machine デプロイ
• ASL (Amazon States Language) JSON: Task State の Resource / Parameters / JSONata 式
• AWS CLI: aws stepfunctions start-execution / describe-execution による動作確認

サービス仕様の変更や新機能の追加により内容が陳腐化する場合があります。AWS 公式 Step Functions Developer Guide を併せて参照ください。

各コード例は ap-northeast-1 (東京リージョン) で動作確認しています。リージョン固有のエンドポイントを使うサービス (Bedrock モデル ID 等) については本文中に注記します。Terraform バージョンは 1.5 以上、AWS provider は 5.x 以上を前提とします。

3 点セットのコード例はすべてコピー&ペーストで動作するレベルを目標に記述しており、ハンズオンとして実際に手元で試しながら読み進めることができます。

2. Optimized vs AWS-SDK 統合 全体像

Step Functions の SDK 統合には 2 種類あります。Optimized Integration と AWS-SDK Integration は名前が似ていますが、対応サービス数・ASL 記法・コスト・制約が大きく異なります。本章ではこの 2 種類を完全比較し、どちらをいつ選ぶかの判断軸を提示します。

2-1. Optimized Integration とは

Optimized Integration は、AWS が Step Functions 向けに最適化した統合パターンです。対応サービス数は限られていますが、.sync (同期) と .waitForTaskToken (非同期コールバック) の 2 パターンを提供し、ポーリング不要・ステートマシン内完結 を実現します。

Optimized Integration 対応サービス一覧 (2026-04 時点):

.sync:2 を指定すると Step Functions がポーリングを肩代わりし、リソース URI は arn:aws:states:::lambda:invoke.waitForTaskToken のように記述します。

ASL 記法 (Optimized .sync の例):

{
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
  "Parameters": {
 "FunctionName": "my-function",
 "Payload": {
"taskToken.$": "$$.Task.Token",
"input.$": "$"
 }
  }
}

.sync:2 の追加費用はなく、Lambda 実行ごとの Step Functions 状態遷移料金のみ発生します。

2-2. AWS-SDK Integration とは

AWS-SDK Integration は arn:aws:states:::aws-sdk:<service>:<action> という URN 形式で AWS SDK が提供する 200+ サービスを直接呼び出します。Optimized Integration でカバーされないサービス (DynamoDB / EventBridge / S3 / API Gateway / Secrets Manager 等) をすべて対象にできます。

ASL 記法 (AWS-SDK の例):

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
  "Parameters": {
 "TableName": "my-table",
 "Item": {
"pk": { "S.$": "$.userId" },
"data": { "S.$": "$.payload" }
 }
  }
}

ポイントは PascalCase 変換ルール です。AWS CLI では --item (小文字) ですが、ASL Parameters では Item (PascalCase) と記述する必要があります。詳細は §2-4 で解説します。

AWS-SDK Integration を呼び出せるサービス例:

2-3. Optimized vs AWS-SDK 完全比較表

使い分けの原則:

• Optimized を選ぶ条件: 処理完了を待つ必要がある (ポーリングを SFN に任せたい) / 対応サービスの .sync が必要
• AWS-SDK を選ぶ条件: Optimized 非対応サービスを使う / 即時レスポンスで十分 / DDB/S3/EB 等の 200+ サービスを利用

2-4. PascalCase 変換ルール

AWS-SDK Integration を使う際の最大の落とし穴が PascalCase 変換 です。AWS CLI では小文字+ハイフン (--table-name) ですが、ASL Parameters では PascalCase (TableName) で記述する必要があります。

変換例 (DynamoDB PutItem):

AWS CLI での呼び出し:

aws dynamodb put-item \
  --table-name my-table \
  --item '{"pk": {"S": "user#123"}, "score": {"N": "99"}}' \
  --condition-expression "attribute_not_exists(pk)"

ASL Parameters での記述:

{
  "TableName": "my-table",
  "Item": {
 "pk": { "S.$": "$.userId" },
 "score": { "N.$": "States.Format('{}', $.score)" }
  },
  "ConditionExpression": "attribute_not_exists(pk)"
}

よくある PascalCase 変換ミス:

PascalCase ミスは実行時に States.Runtime エラーではなく InvalidParameterValue として表面化するため、デバッグで時間を取られがちです。§7 の典型エラー集で詳細な対処法を解説します。

2-5. 選択フローチャート

ユースケースに応じた統合タイプの選択フローを示します:

処理完了を待つ必要があるか？
├─ YES → Optimized Integration の対応サービスか？
│├─ YES → Optimized (.sync:2 または .waitForTaskToken) を選択
│└─ NO  → Lambda で中継 (Optimized + Lambda + SDK) か
│AWS-SDK + 外部ポーリング Loop を検討
└─ NO  → AWS-SDK Integration を選択
 └─ 対象 Action が 200+ サービスの範囲内か確認
 ├─ YES → aws-sdk:<service>:<action> で直接呼出
 └─ NO  → AWS ドキュメントで API 名を確認

典型ユースケース別の推奨統合タイプ:

2-6. 本記事での扱い方針

本記事では以下の方針で 2 統合タイプを扱います:

• AWS-SDK Integration を主軸: DDB / SNS / SQS / EB / S3 / APIGW / Bedrock の 7 サービス実装例を §3-§5 で完全解説
• Optimized Integration は補足: .sync:2 / .waitForTaskToken の基本は Vol2 (Callback / ID:1449) で解説済のため、本記事では差分のみ言及
• JSONata は §6 で独立解説: ${ expr } 記法は AWS-SDK の Parameters 内で活用でき、Lambda 置換の核心技術

実務では両方を組み合わせます。ECS タスク完了を Optimized で待ち、完了後に DDB を AWS-SDK で更新するパターンが典型例です。Vol4 (Express Workflow / ID:1500) との組み合わせで高頻度処理も可能です。

3. DynamoDB 直接操作 + Lambda 置換コスト比較

DynamoDB は SDK Direct Integration の花形サービスだ。PutItem / UpdateItem / Query / BatchWriteItem / TransactWriteItems の 5 アクションを Lambda なしで直接呼び出せる。本章では各 Action の ASL 記法・PascalCase 変換ルール・冪等性設計・バッチ処理の注意点を示し、Lambda 置換コストの実算で経営判断を支援する。

3-1. PutItem / UpdateItem / Query

SDK Direct の Resource は arn:aws:states:::aws-sdk:dynamodb:<Action> 形式を使う。AWS API の camelCase パラメータを PascalCase に変換して Parameters ブロックに記述することが最大のポイントだ。

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:PutItem",
  "Parameters": {
 "TableName": "orders",
 "Item": {
"order_id": { "S.$": "$.orderId" },
"status":{ "S": "PENDING" },
"amount":{ "N.$": "States.Format('{}', $.amount)" }
 },
 "ConditionExpression": "attribute_not_exists(order_id)",
 "ReturnValues": "NONE"
  },
  "ResultPath": null,
  "Next": "Done"
}

UpdateItem では ExpressionAttributeNames と ExpressionAttributeValues も PascalCase で記述する。Query の場合は KeyConditionExpression と IndexName をセットで指定する。いずれも AWS API ドキュメントの camelCase 名を先頭大文字に変換すれば正しい形式になる。

3-2. PascalCase 変換ルール

AWS API が camelCase を使うのに対し、ASL の Parameters ブロックは PascalCase への変換が必須だ。変換を誤ると InvalidParameterType または MissingRequiredParameter エラーが発生する。

実行ログの ExecutionFailed イベントで cause フィールドを確認すると、どのパラメータが誤っているかを即特定できる。

3-3. ConditionExpression + ReturnValues の ASL 記述

ConditionExpression は冪等性保証に不可欠だ。attribute_not_exists(pk) を付与することで重複書き込みを防ぎ、ステートマシンの再実行時にも安全性を確保できる。

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:PutItem",
  "Parameters": {
 "TableName": "idempotency_keys",
 "Item": {
"execution_id": { "S.$": "$$.Execution.Id" },
"created_at":{ "S.$": "$$.Execution.StartTime" }
 },
 "ConditionExpression": "attribute_not_exists(execution_id)"
  },
  "Catch": [{
 "ErrorEquals": ["DynamoDB.ConditionalCheckFailedException"],
 "Next": "AlreadyProcessed",
 "ResultPath": "$.error"
  }],
  "ResultPath": null,
  "Next": "MainProcess"
}

ReturnValues は ALL_NEW / ALL_OLD / UPDATED_NEW / UPDATED_OLD / NONE から選択し、更新後の値を後続ステートで参照する場合は ResultPath に格納する。

3-4. TransactWriteItems — 複数テーブル排他性

TransactWriteItems は最大 100 項目を 1 トランザクションで処理する。注文確定と在庫減算など、複数テーブルへの整合性が必要な場面に使う。

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:TransactWriteItems",
  "Parameters": {
 "TransactItems": [
{
  "Put": {
 "TableName": "orders",
 "Item": {
"order_id": { "S.$": "$.orderId" },
"status":{ "S": "CONFIRMED" }
 },
 "ConditionExpression": "attribute_exists(order_id)"
  }
},
{
  "Update": {
 "TableName": "inventory",
 "Key": { "sku": { "S.$": "$.sku" } },
 "UpdateExpression": "SET stock = stock - :qty",
 "ExpressionAttributeValues": {
":qty": { "N.$": "States.Format('{}', $.quantity)" },
":min": { "N": "0" }
 },
 "ConditionExpression": "stock >= :qty AND stock > :min"
  }
}
 ]
  },
  "Retry": [{
 "ErrorEquals": ["DynamoDB.TransactionConflictException"],
 "IntervalSeconds": 1,
 "MaxAttempts": 3,
 "BackoffRate": 2
  }],
  "Catch": [{
 "ErrorEquals": ["DynamoDB.TransactionCanceledException"],
 "Next": "HandleConflict",
 "ResultPath": "$.error"
  }],
  "ResultPath": null,
  "Next": "Confirm"
}

TransactionCanceledException の CancellationReasons に条件チェック失敗の詳細が含まれるため、ResultPath でエラー情報を保持してハンドリングする。

3-5. BatchWriteItem — 25 件上限とリトライ戦略

BatchWriteItem は最大 25 件を 1 リクエストで処理するが、一部失敗時に UnprocessedItems が返却される。ステートマシン側でリトライループを構成する必要がある。

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:BatchWriteItem",
  "Parameters": {
 "RequestItems": {
"batch_events.$": "$.batchItems"
 }
  },
  "Retry": [{
 "ErrorEquals": [
"DynamoDB.ProvisionedThroughputExceededException",
"DynamoDB.RequestLimitExceeded"
 ],
 "IntervalSeconds": 2,
 "MaxAttempts": 5,
 "BackoffRate": 2,
 "JitterStrategy": "FULL"
  }],
  "ResultPath": "$.batchResult",
  "Next": "CheckUnprocessed"
}

JitterStrategy: FULL は 2024-11 GA の機能で、サンダーハード問題を抑制する。CheckUnprocessed ステートで $.batchResult.UnprocessedItems の空判定を行い、残件があれば再度 BatchWriteItem に戻すループを組む。

3-6. Lambda 置換コスト試算 3 シナリオ

3-7. 3 点セット — Terraform + ASL + CLI

Terraform HCL — DDB テーブル + ステートマシン + IAM ポリシー

terraform {
  required_version = ">= 1.9.0"
  required_providers {
 aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
}

resource "aws_dynamodb_table" "orders" {
  name= "orders"
  billing_mode = "PAY_PER_REQUEST"
  hash_key  = "order_id"
  attribute { name = "order_id"; type = "S" }
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_iam_role_policy" "sf_ddb" {
  role = aws_iam_role.sf_exec.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Action = [
  "dynamodb:PutItem", "dynamodb:UpdateItem",
  "dynamodb:Query", "dynamodb:BatchWriteItem",
  "dynamodb:TransactWriteItems"
]
Resource = [
  aws_dynamodb_table.orders.arn,
  "${aws_dynamodb_table.orders.arn}/index/*"
]
 }]
  })
}

resource "aws_sfn_state_machine" "orders_sm" {
  name  = "orders-put-sm"
  role_arn = aws_iam_role.sf_exec.arn
  definition = file("statemachine/orders.asl.json")
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

ASL JSON — PutItem 完全実装 (Retry / Catch 付き)

{
  "Comment": "DDB PutItem with SDK Direct Integration",
  "StartAt": "PutOrder",
  "States": {
 "PutOrder": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:PutItem",
"Parameters": {
  "TableName": "orders",
  "Item": {
 "order_id": { "S.$": "$.orderId" },
 "status":{ "S": "PENDING" }
  },
  "ConditionExpression": "attribute_not_exists(order_id)"
},
"Retry": [{
  "ErrorEquals": ["DynamoDB.ProvisionedThroughputExceededException"],
  "MaxAttempts": 3,
  "BackoffRate": 2,
  "JitterStrategy": "FULL"
}],
"Catch": [{
  "ErrorEquals": ["DynamoDB.ConditionalCheckFailedException"],
  "Next": "AlreadyExists",
  "ResultPath": "$.error"
}],
"ResultPath": null,
"End": true
 },
 "AlreadyExists": {
"Type": "Fail",
"Error": "OrderAlreadyExists",
"Cause": "Duplicate order_id"
 }
  }
}

CLI — 実行・確認・IAM 検証

# ステートマシン実行
aws stepfunctions start-execution \
  --state-machine-arn arn:aws:states:ap-northeast-1:123456789012:stateMachine:orders-put-sm \
  --input '{"orderId":"ord-001","amount":9800}'

# DDB 書込確認
aws dynamodb get-item \
  --table-name orders \
  --key '{"order_id":{"S":"ord-001"}}' \
  --query 'Item'

# IAM 権限シミュレーション
aws iam simulate-principal-policy \
  --policy-source-arn arn:aws:iam::123456789012:role/sf-sdk-direct-exec \
  --action-names dynamodb:PutItem \
  --resource-arns arn:aws:dynamodb:ap-northeast-1:123456789012:table/orders

# TODO: 実機実行: 2026-04-XX

4. SNS + SQS + EventBridge 直接操作

Step Functions SDK Direct Integration で最も使われる通知・キュー・イベントバスの 3 サービスを横断的に解説する。Lambda を介さず ASL から直接 SNS/SQS/EventBridge を操作することで、マイクロサービス間連携のレイテンシを削減し、コストを最適化できる。

4-1. SNS Publish / PublishBatch

SNS への直接 Publish は arn:aws:states:::aws-sdk:sns:publish で行う。SDK Direct では、AWS API の camelCase パラメータを PascalCase に変換して Parameters に渡す。

{
  "Comment": "SNS Publish — SDK Direct Integration",
  "StartAt": "PublishToSNS",
  "States": {
 "PublishToSNS": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sns:publish",
"Parameters": {
  "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:order-events",
  "Message.$": "States.JsonToString($.payload)",
  "Subject": "Order Notification",
  "MessageAttributes": {
 "event_type": {
"DataType": "String",
"StringValue.$": "$.eventType"
 },
 "priority": {
"DataType": "Number",
"StringValue.$": "States.Format('{}', $.priority)"
 }
  }
},
"Retry": [
  {
 "ErrorEquals": ["SNS.SnsException", "States.TaskFailed"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"Catch": [
  {
 "ErrorEquals": ["States.ALL"],
 "Next": "HandleSNSError",
 "ResultPath": "$.error"
  }
],
"ResultPath": "$.snsResult",
"Next": "Done"
 },
 "HandleSNSError": {
"Type": "Pass",
"End": true
 },
 "Done": {
"Type": "Succeed"
 }
  }
}

PublishBatch は 1 回の API コールで最大 10 件のメッセージを送信できる。SNS FIFO トピックでは MessageGroupId と MessageDeduplicationId が必須になる。

{
  "Comment": "SNS PublishBatch — FIFO トピック対応",
  "StartAt": "BatchPublishToSNS",
  "States": {
 "BatchPublishToSNS": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sns:publishBatch",
"Parameters": {
  "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:order-events.fifo",
  "PublishBatchRequestEntries.$": "$.messages"
},
"Comment": "messages = [{Id, Message, MessageGroupId, MessageDeduplicationId, MessageAttributes}]",
"ResultPath": "$.batchResult",
"End": true
 }
  }
}

FIFO トピックへの PublishBatch では各エントリに MessageGroupId と MessageDeduplicationId を含める。重複排除 ID には States.UUID() を使うと冪等性を確保できる。

4-2. SQS SendMessage / SendMessageBatch

SQS への直接送信は arn:aws:states:::aws-sdk:sqs:sendMessage で行う。Optimized Integration の .waitForTaskToken とは異なり、SDK Direct は送信後すぐに次の State へ遷移する（ポーリング不要）。

{
  "Comment": "SQS SendMessage — 標準キュー",
  "StartAt": "SendToSQS",
  "States": {
 "SendToSQS": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sqs:sendMessage",
"Parameters": {
  "QueueUrl": "https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-queue",
  "MessageBody.$": "States.JsonToString($.orderPayload)",
  "DelaySeconds": 0,
  "MessageAttributes": {
 "source": {
"DataType": "String",
"StringValue": "step-functions-vol5"
 }
  }
},
"Retry": [
  {
 "ErrorEquals": ["SQS.SqsException", "States.TaskFailed"],
 "IntervalSeconds": 1,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.sqsResult",
"End": true
 }
  }
}

FIFO キューの場合: MessageGroupId と MessageDeduplicationId を追加する。

{
  "Comment": "SQS SendMessage — FIFO キュー + 冪等性保証",
  "StartAt": "SendToFIFOQueue",
  "States": {
 "SendToFIFOQueue": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sqs:sendMessage",
"Parameters": {
  "QueueUrl": "https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-queue.fifo",
  "MessageBody.$": "States.JsonToString($.orderPayload)",
  "MessageGroupId.$": "$.customerId",
  "MessageDeduplicationId.$": "States.UUID()"
},
"ResultPath": "$.sqsResult",
"End": true
 }
  }
}

SendMessageBatch は 1 回の API で最大 10 件を送信する。高スループット処理でのコスト削減効果が大きい。

{
  "Comment": "SQS SendMessageBatch — 高スループット処理",
  "StartAt": "BatchSendToSQS",
  "States": {
 "BatchSendToSQS": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sqs:sendMessageBatch",
"Parameters": {
  "QueueUrl": "https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-queue",
  "Entries.$": "$.batchItems"
},
"Comment": "batchItems = [{Id, MessageBody, DelaySeconds?, MessageAttributes?}]",
"ResultPath": "$.batchResult",
"End": true
 }
  }
}

4-3. EventBridge PutEvents

EventBridge へのイベント送信は arn:aws:states:::aws-sdk:eventbridge:putEvents を使う。カスタムバスを指定することでデフォルトバスのノイズを排除できる。

{
  "Comment": "EventBridge PutEvents — カスタムバス指定",
  "StartAt": "PutOrderEvent",
  "States": {
 "PutOrderEvent": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:eventbridge:putEvents",
"Parameters": {
  "Entries": [
 {
"Source": "com.myapp.orders",
"DetailType": "OrderPlaced",
"Detail.$": "States.JsonToString($.orderDetail)",
"EventBusName": "arn:aws:events:ap-northeast-1:123456789012:event-bus/order-bus",
"Time.$": "$$.Execution.StartTime"
 }
  ]
},
"Retry": [
  {
 "ErrorEquals": ["EventBridge.EventBridgeException", "States.TaskFailed"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.ebResult",
"End": true
 }
  }
}

Detail は文字列型が必須のため States.JsonToString() でオブジェクトを文字列化する点に注意。複数イベントを一括送信する場合は Entries 配列に最大 10 件まで含められる。

4-4. 3サービス横断の使い分け

4-5. SNS→SQS Fanout パターン (Terraform + ASL)

SNS と SQS を組み合わせた Fanout パターンは、1 つの SNS トピックから複数の SQS キューにメッセージを届ける標準構成。Step Functions から SNS に Publish するだけで複数のコンシューマへのデリバリが実現できる。

Terraform — Fanout リソース定義:

terraform {
  required_version = ">= 1.9.0"
  required_providers {
 aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
}

resource "aws_sns_topic" "order_events" {
  name = "order-events"
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_sqs_queue" "order_processing" {
  name  = "order-processing-queue"
  visibility_timeout_seconds = 30
  message_retention_seconds  = 86400
  tags  = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_sqs_queue" "inventory_notification" {
  name = "inventory-notification-queue"
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_sns_topic_subscription" "order_to_processing" {
  topic_arn = aws_sns_topic.order_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.order_processing.arn
  filter_policy = jsonencode({
 event_type = ["ORDER_PLACED", "ORDER_UPDATED"]
  })
}

resource "aws_sns_topic_subscription" "order_to_inventory" {
  topic_arn = aws_sns_topic.order_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.inventory_notification.arn
  filter_policy = jsonencode({
 event_type = ["ORDER_PLACED"]
  })
}

resource "aws_sqs_queue_policy" "order_processing_policy" {
  queue_url = aws_sqs_queue.order_processing.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "sns.amazonaws.com" }
Action = "sqs:SendMessage"
Resource  = aws_sqs_queue.order_processing.arn
Condition = {
  ArnEquals = { "aws:SourceArn" = aws_sns_topic.order_events.arn }
}
 }]
  })
}

resource "aws_sqs_queue_policy" "inventory_notification_policy" {
  queue_url = aws_sqs_queue.inventory_notification.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "sns.amazonaws.com" }
Action = "sqs:SendMessage"
Resource  = aws_sqs_queue.inventory_notification.arn
Condition = {
  ArnEquals = { "aws:SourceArn" = aws_sns_topic.order_events.arn }
}
 }]
  })
}

resource "aws_iam_role" "sf_fanout_role" {
  name = "sf-fanout-vol5-role"
  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "states.amazonaws.com" }
Action = "sts:AssumeRole"
 }]
  })
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_iam_role_policy" "sf_sns_fanout_policy" {
  name = "sf-sns-publish-policy"
  role = aws_iam_role.sf_fanout_role.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["sns:Publish", "sns:PublishBatch"]
Resource = aws_sns_topic.order_events.arn
Condition = {
  StringEquals = { "aws:SourceAccount" = "123456789012" }
}
 }]
  })
}

resource "aws_sfn_state_machine" "fanout_workflow" {
  name = "fanout-workflow-vol5"
  role_arn= aws_iam_role.sf_fanout_role.arn
  definition = file("${path.module}/fanout-workflow.asl.json")
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

ASL — Fanout パターン State Machine:

{
  "Comment": "SNS→SQS Fanout パターン — SDK Direct Integration",
  "StartAt": "PublishOrderEvent",
  "States": {
 "PublishOrderEvent": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sns:publish",
"Parameters": {
  "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:order-events",
  "Message.$": "States.JsonToString($.order)",
  "Subject": "Order Event",
  "MessageAttributes": {
 "event_type": {
"DataType": "String",
"StringValue.$": "$.order.eventType"
 }
  }
},
"Retry": [
  {
 "ErrorEquals": ["SNS.SnsException", "States.TaskFailed"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.publishResult",
"Next": "RecordPublishResult"
 },
 "RecordPublishResult": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
"Parameters": {
  "TableName": "order-events-log",
  "Item": {
 "MessageId": { "S.$": "$.publishResult.MessageId" },
 "OrderId": { "S.$": "$.order.orderId" },
 "EventType": { "S.$": "$.order.eventType" },
 "PublishedAt": { "S.$": "$$.Execution.StartTime" }
  }
},
"ResultPath": "$.ddbResult",
"End": true
 }
  }
}

4-6. EB Rule → SF 起動 (逆方向) との対比

Step Functions から EB へ PutEvents する方向（順方向）とは逆に、EventBridge Rule のターゲットとして Step Functions State Machine を起動する逆方向パターンも重要。

resource "aws_cloudwatch_event_rule" "order_placed" {
  name  = "order-placed-rule"
  event_bus_name = "order-bus"
  event_pattern = jsonencode({
 source= ["com.myapp.orders"]
 detail-type = ["OrderPlaced"]
  })
}

resource "aws_cloudwatch_event_target" "trigger_sf" {
  rule  = aws_cloudwatch_event_rule.order_placed.name
  event_bus_name = "order-bus"
  arn= aws_sfn_state_machine.downstream_sm.arn
  role_arn = aws_iam_role.eb_sf_trigger_role.arn
}

4-7. 3点セット — SendMessage / PutEvents 実行確認

CLI — 実行と確認:

# State Machine 実行 (Fanout ワークフロー)
aws stepfunctions start-execution \
  --state-machine-arn "arn:aws:states:ap-northeast-1:123456789012:stateMachine:fanout-workflow-vol5" \
  --input '{
 "order": {
"orderId": "ORD-20260424-001",
"customerId": "CUST-100",
"amount": 15000,
"eventType": "ORDER_PLACED"
 }
  }'
# 実機実行: 2026-04-XX

# 実行結果確認
aws stepfunctions describe-execution \
  --execution-arn "arn:aws:states:ap-northeast-1:123456789012:execution:fanout-workflow-vol5:EXECUTION_ID"
# 実機実行: 2026-04-XX

# SQS キューのメッセージ数確認
aws sqs get-queue-attributes \
  --queue-url "https://sqs.ap-northeast-1.amazonaws.com/123456789012/order-processing-queue" \
  --attribute-names ApproximateNumberOfMessages
# 実機実行: 2026-04-XX

# EventBridge イベント手動送信テスト
aws events put-events \
  --entries '[{
 "Source": "com.myapp.orders",
 "DetailType": "OrderPlaced",
 "Detail": "{\"orderId\": \"ORD-TEST-001\"}",
 "EventBusName": "order-bus"
  }]'
# 実機実行: 2026-04-XX

5. S3 + API Gateway + Bedrock 直接操作

S3 オブジェクト操作、API Gateway 経由の HTTP 統合、Bedrock による AI モデル呼出しを SDK Direct で実装する。特に S3 Object → Bedrock → DynamoDB の完全フローは Lambda を一切使わずに AI 推論パイプラインを構築できる実践的なパターン。

5-1. S3 PutObject / GetObject / CopyObject

S3 オブジェクト操作は arn:aws:states:::aws-sdk:s3:putObject、getObject、copyObject などで行う。テキストデータは States.JsonToString() で文字列化して渡す。

{
  "Comment": "S3 PutObject → GetObject → CopyObject シーケンス",
  "StartAt": "PutS3Object",
  "States": {
 "PutS3Object": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:s3:putObject",
"Parameters": {
  "Bucket": "my-workflow-bucket",
  "Key.$": "States.Format('outputs/{}/{}.json', $.executionId, $.timestamp)",
  "Body.$": "States.JsonToString($.resultPayload)",
  "ContentType": "application/json",
  "ServerSideEncryption": "AES256",
  "Metadata": {
 "workflow-version": "v5",
 "source": "step-functions-sdk-direct"
  }
},
"Retry": [
  {
 "ErrorEquals": ["S3.S3Exception", "States.TaskFailed"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.s3PutResult",
"Next": "GetS3Object"
 },
 "GetS3Object": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:s3:getObject",
"Parameters": {
  "Bucket": "my-workflow-bucket",
  "Key.$": "States.Format('outputs/{}/{}.json', $.executionId, $.timestamp)"
},
"ResultSelector": {
  "body.$": "$.Body",
  "contentType.$": "$.ContentType",
  "contentLength.$": "$.ContentLength"
},
"ResultPath": "$.s3GetResult",
"Next": "CopyS3Object"
 },
 "CopyS3Object": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:s3:copyObject",
"Parameters": {
  "Bucket": "my-archive-bucket",
  "Key.$": "States.Format('archive/{}.json', $.executionId)",
  "CopySource.$": "States.Format('my-workflow-bucket/outputs/{}/{}.json', $.executionId, $.timestamp)",
  "ServerSideEncryption": "AES256"
},
"ResultPath": "$.s3CopyResult",
"End": true
 }
  }
}

S3 の GetObject レスポンスの Body は文字列として返るため、States.StringToJson() でパースして後続 State に渡す。大容量オブジェクト (> 5 MB) の場合はマルチパートアップロード (createMultipartUpload → uploadPart → completeMultipartUpload) のシーケンスで実装できる。

5-2. API Gateway HTTP Invoke

API Gateway の REST API を Step Functions から直接呼び出す場合、HTTP タスク (arn:aws:states:::http:invoke) を使う。2023 年に追加された SDK Direct の拡張で、任意の HTTP エンドポイントへのリクエストを ASL 内で記述できる。

{
  "Comment": "API Gateway HTTP Invoke — OAuth2 連携パターン",
  "StartAt": "GetOAuthToken",
  "States": {
 "GetOAuthToken": {
"Type": "Task",
"Resource": "arn:aws:states:::http:invoke",
"Parameters": {
  "ApiEndpoint": "https://auth.example.com/oauth2/token",
  "Method": "POST",
  "Headers": {
 "Content-Type": "application/x-www-form-urlencoded"
  },
  "RequestBody": "grant_type=client_credentials&client_id=MY_CLIENT_ID&client_secret=MY_SECRET",
  "Authentication": {
 "ConnectionArn": "arn:aws:events:ap-northeast-1:123456789012:connection/oauth-connection/CONNECTION_ID"
  }
},
"ResultSelector": {
  "accessToken.$": "$.ResponseBody.access_token",
  "expiresIn.$": "$.ResponseBody.expires_in"
},
"ResultPath": "$.auth",
"Next": "CallOrderAPI"
 },
 "CallOrderAPI": {
"Type": "Task",
"Resource": "arn:aws:states:::http:invoke",
"Parameters": {
  "ApiEndpoint.$": "States.Format('https://api.example.com/orders/{}', $.orderId)",
  "Method": "POST",
  "Headers": {
 "Content-Type": "application/json",
 "Authorization.$": "States.Format('Bearer {}', $.auth.accessToken)"
  },
  "RequestBody.$": "$.orderPayload"
},
"Retry": [
  {
 "ErrorEquals": ["States.Http.StatusCode.429", "States.Http.StatusCode.503"],
 "IntervalSeconds": 3,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"Catch": [
  {
 "ErrorEquals": ["States.Http.StatusCode.400", "States.Http.StatusCode.422"],
 "Next": "HandleValidationError",
 "ResultPath": "$.httpError"
  }
],
"ResultSelector": {
  "statusCode.$": "$.StatusCode",
  "body.$": "$.ResponseBody"
},
"ResultPath": "$.apiResult",
"End": true
 },
 "HandleValidationError": {
"Type": "Pass",
"End": true
 }
  }
}

HTTP タスクは EventBridge Connections を使って認証情報を安全に管理する。ConnectionArn に OAuth2 クライアント認証情報を登録しておけば、ASL 内にシークレットを直書きせずに済む。

5-3. Bedrock InvokeModel

{
  "Comment": "Bedrock InvokeModel — Claude 3 Sonnet 呼出し",
  "StartAt": "InvokeBedrock",
  "States": {
 "InvokeBedrock": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:bedrock:invokeModel",
"Parameters": {
  "ModelId": "anthropic.claude-3-sonnet-20240229-v1:0",
  "ContentType": "application/json",
  "Accept": "application/json",
  "Body": {
 "anthropic_version": "bedrock-2023-05-31",
 "max_tokens": 1024,
 "messages": [
{
  "role": "user",
  "content.$": "States.Format('以下のデータを要約してください: {}', States.JsonToString($.inputData))"
}
 ]
  }
},
"ResultSelector": {
  "rawBody.$": "$.Body"
},
"ResultPath": "$.bedrockRaw",
"Next": "ParseBedrockResponse"
 },
 "ParseBedrockResponse": {
"Type": "Pass",
"Parameters": {
  "parsedResponse.$": "States.StringToJson($.bedrockRaw.rawBody)"
},
"ResultPath": "$.bedrockParsed",
"Next": "ExtractContent"
 },
 "ExtractContent": {
"Type": "Pass",
"Parameters": {
  "summary.$": "$.bedrockParsed.parsedResponse.content[0].text"
},
"ResultPath": "$.aiResult",
"End": true
 }
  }
}

Titan テキストモデルのリクエスト Body 構造:

{
  "Body": {
 "inputText.$": "States.Format('次のテキストを要約: {}', $.inputText)",
 "textGenerationConfig": {
"maxTokenCount": 512,
"temperature": 0.7,
"topP": 0.9
 }
  }
}

5-4. Bedrock InvokeModelWithResponseStream (ストリーミング応答)

invokeModelWithResponseStream は SDK Direct で呼び出せるが、Step Functions は現時点でストリーミングレスポンスの逐次処理をネイティブサポートしていない。全ストリームが結合されたレスポンスとして返却される。

{
  "Comment": "Bedrock InvokeModelWithResponseStream — SFではストリームが結合返却",
  "StartAt": "InvokeBedrockStream",
  "States": {
 "InvokeBedrockStream": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:bedrock:invokeModelWithResponseStream",
"Parameters": {
  "ModelId": "anthropic.claude-3-sonnet-20240229-v1:0",
  "ContentType": "application/json",
  "Accept": "application/json",
  "Body": {
 "anthropic_version": "bedrock-2023-05-31",
 "max_tokens": 2048,
 "messages": [
{
  "role": "user",
  "content.$": "$.prompt"
}
 ]
  }
},
"TimeoutSeconds": 60,
"Retry": [
  {
 "ErrorEquals": ["Bedrock.BedrockException", "States.Timeout"],
 "IntervalSeconds": 5,
 "MaxAttempts": 2,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.streamResult",
"End": true
 }
  }
}

ストリーミングを活用した UI 表示が必要な場合は Lambda + Bedrock の組み合わせか AppSync + Bedrock の構成が現実的。SF 内での用途は「長文生成の結果を一括取得して後続 State で処理する」パターンに限定することを推奨する。

5-5. Bedrock + DynamoDB 組み合わせ (AI 応答を直接 DDB に書込)

Bedrock の推論結果を DynamoDB に直接書き込む場合、Lambda を使わず SF 内で invokeModel → putItem のシーケンスで実装できる。

{
  "Comment": "Bedrock InvokeModel → DDB PutItem — Lambda不要フロー",
  "StartAt": "GenerateSummary",
  "States": {
 "GenerateSummary": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:bedrock:invokeModel",
"Parameters": {
  "ModelId": "anthropic.claude-3-sonnet-20240229-v1:0",
  "ContentType": "application/json",
  "Accept": "application/json",
  "Body": {
 "anthropic_version": "bedrock-2023-05-31",
 "max_tokens": 512,
 "messages": [
{
  "role": "user",
  "content.$": "States.Format('商品レビューを日本語で3文にまとめてください: {}', $.reviewText)"
}
 ]
  }
},
"ResultSelector": {
  "rawBody.$": "$.Body"
},
"ResultPath": "$.bedrockRaw",
"Next": "ParseResponse"
 },
 "ParseResponse": {
"Type": "Pass",
"Parameters": {
  "parsed.$": "States.StringToJson($.bedrockRaw.rawBody)"
},
"ResultPath": "$.parsed",
"Next": "StoreToDDB"
 },
 "StoreToDDB": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
"Parameters": {
  "TableName": "review-summaries",
  "Item": {
 "ReviewId": { "S.$": "$.reviewId" },
 "ProductId": { "S.$": "$.productId" },
 "Summary": { "S.$": "$.parsed.parsed.content[0].text" },
 "ModelId": { "S": "anthropic.claude-3-sonnet-20240229-v1:0" },
 "GeneratedAt": { "S.$": "$$.Execution.StartTime" },
 "InputTokens": { "N.$": "States.Format('{}', $.parsed.parsed.usage.input_tokens)" },
 "OutputTokens": { "N.$": "States.Format('{}', $.parsed.parsed.usage.output_tokens)" }
  },
  "ConditionExpression": "attribute_not_exists(ReviewId)"
},
"Retry": [
  {
 "ErrorEquals": ["DynamoDB.DynamoDbException"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.ddbResult",
"End": true
 }
  }
}

5-6. S3 Object → Bedrock → DDB 書込 の完全フロー (Lambda 不要)

S3 に保存されたテキストデータを読み込み → Bedrock で AI 推論 → DynamoDB に結果を書き込む、完全に Lambda 不要のパイプラインを SDK Direct で構築する。

{
  "Comment": "S3 → Bedrock → DDB 完全 Lambda 不要フロー",
  "StartAt": "ReadFromS3",
  "States": {
 "ReadFromS3": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:s3:getObject",
"Parameters": {
  "Bucket.$": "$.sourceBucket",
  "Key.$": "$.sourceKey"
},
"ResultSelector": {
  "content.$": "$.Body"
},
"ResultPath": "$.s3Content",
"Next": "InvokeBedrockAnalysis"
 },
 "InvokeBedrockAnalysis": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:bedrock:invokeModel",
"Parameters": {
  "ModelId": "anthropic.claude-3-sonnet-20240229-v1:0",
  "ContentType": "application/json",
  "Accept": "application/json",
  "Body": {
 "anthropic_version": "bedrock-2023-05-31",
 "max_tokens": 1024,
 "messages": [
{
  "role": "user",
  "content.$": "States.Format('次のドキュメントを分析し、要点を抽出してください。\n\nドキュメント: {}', $.s3Content.content)"
}
 ]
  }
},
"ResultSelector": {
  "rawBody.$": "$.Body"
},
"ResultPath": "$.bedrockRaw",
"Next": "ParseAnalysisResult"
 },
 "ParseAnalysisResult": {
"Type": "Pass",
"Parameters": {
  "analysis.$": "States.StringToJson($.bedrockRaw.rawBody)"
},
"ResultPath": "$.bedrockParsed",
"Next": "StoreAnalysisToDDB"
 },
 "StoreAnalysisToDDB": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
"Parameters": {
  "TableName": "document-analysis",
  "Item": {
 "DocumentKey": { "S.$": "$.sourceKey" },
 "Bucket": { "S.$": "$.sourceBucket" },
 "AnalysisText": { "S.$": "$.bedrockParsed.analysis.content[0].text" },
 "AnalyzedAt": { "S.$": "$$.Execution.StartTime" },
 "ExecutionId": { "S.$": "$$.Execution.Id" }
  }
},
"Retry": [
  {
 "ErrorEquals": ["DynamoDB.DynamoDbException", "States.TaskFailed"],
 "IntervalSeconds": 2,
 "MaxAttempts": 3,
 "BackoffRate": 2.0
  }
],
"ResultPath": "$.storeResult",
"Next": "NotifyCompletion"
 },
 "NotifyCompletion": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sns:publish",
"Parameters": {
  "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:analysis-complete",
  "Message.$": "States.Format('Document analysis complete for: {}. Execution: {}', $.sourceKey, $$.Execution.Id)",
  "Subject": "Document Analysis Complete"
},
"ResultPath": "$.notifyResult",
"End": true
 }
  }
}

このフローは S3 → Bedrock → DDB → SNS の 4 サービスを Lambda 不要で直接連携する。従来の Lambda 経由に比べて Lambda 実行コスト・コールドスタートレイテンシを排除できる。

5-7. 3点セット — Bedrock 呼出実装 (Terraform + ASL + CLI)

Terraform — Bedrock + S3 + DDB 実行ロール:

terraform {
  required_version = ">= 1.9.0"
  required_providers {
 aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
}

resource "aws_iam_role" "sf_bedrock_role" {
  name = "sf-bedrock-sdk-direct-role"
  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "states.amazonaws.com" }
Action = "sts:AssumeRole"
Condition = {
  StringEquals = { "aws:SourceAccount" = "123456789012" }
  ArnLike = {
 "aws:SourceArn" = "arn:aws:states:ap-northeast-1:123456789012:stateMachine:*"
  }
}
 }]
  })
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_iam_role_policy" "sf_bedrock_invoke" {
  name = "sf-bedrock-invoke-policy"
  role = aws_iam_role.sf_bedrock_role.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Action = [
  "bedrock:InvokeModel",
  "bedrock:InvokeModelWithResponseStream"
]
Resource = [
  "arn:aws:bedrock:ap-northeast-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0",
  "arn:aws:bedrock:ap-northeast-1::foundation-model/amazon.titan-text-lite-v1"
]
 }]
  })
}

resource "aws_iam_role_policy" "sf_s3_access" {
  name = "sf-s3-access-policy"
  role = aws_iam_role.sf_bedrock_role.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["s3:GetObject", "s3:PutObject", "s3:CopyObject"]
Resource = "arn:aws:s3:::my-workflow-bucket/*"
Condition = {
  StringEquals = { "aws:SourceAccount" = "123456789012" }
}
 }]
  })
}

resource "aws_iam_role_policy" "sf_ddb_write" {
  name = "sf-ddb-write-policy"
  role = aws_iam_role.sf_bedrock_role.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:GetItem"]
Resource = "arn:aws:dynamodb:ap-northeast-1:123456789012:table/document-analysis"
Condition = {
  StringEquals = { "aws:SourceAccount" = "123456789012" }
}
 }]
  })
}

resource "aws_iam_role_policy" "sf_sns_notify" {
  name = "sf-sns-notify-policy"
  role = aws_iam_role.sf_bedrock_role.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["sns:Publish"]
Resource = "arn:aws:sns:ap-northeast-1:123456789012:analysis-complete"
 }]
  })
}

resource "aws_sfn_state_machine" "s3_bedrock_ddb_workflow" {
  name = "s3-bedrock-ddb-workflow-vol5"
  role_arn= aws_iam_role.sf_bedrock_role.arn
  definition = file("${path.module}/s3-bedrock-ddb.asl.json")
  logging_configuration {
 log_destination  = "${aws_cloudwatch_log_group.sf_logs.arn}:*"
 include_execution_data = true
 level= "ALL"
  }
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

resource "aws_cloudwatch_log_group" "sf_logs" {
  name  = "/aws/states/s3-bedrock-ddb-workflow-vol5"
  retention_in_days = 14
  tags  = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

CLI — S3→Bedrock→DDB フロー実行と確認:

# テスト用ドキュメントを S3 にアップロード
aws s3 cp ./sample-document.txt s3://my-workflow-bucket/inputs/doc-001.txt
# 実機実行: 2026-04-XX

# State Machine 実行
aws stepfunctions start-execution \
  --state-machine-arn "arn:aws:states:ap-northeast-1:123456789012:stateMachine:s3-bedrock-ddb-workflow-vol5" \
  --input '{
 "sourceBucket": "my-workflow-bucket",
 "sourceKey": "inputs/doc-001.txt"
  }'
# 実機実行: 2026-04-XX

# 実行状態確認
aws stepfunctions describe-execution \
  --execution-arn "arn:aws:states:ap-northeast-1:123456789012:execution:s3-bedrock-ddb-workflow-vol5:EXECUTION_ID"
# 実機実行: 2026-04-XX

# DDB に結果が書き込まれたか確認
aws dynamodb get-item \
  --table-name document-analysis \
  --key '{"DocumentKey": {"S": "inputs/doc-001.txt"}}'
# 実機実行: 2026-04-XX

# IAM ポリシー検証 (Bedrock)
aws iam simulate-principal-policy \
  --policy-source-arn "arn:aws:iam::123456789012:role/sf-bedrock-sdk-direct-role" \
  --action-names "bedrock:InvokeModel" \
  --resource-arns "arn:aws:bedrock:ap-northeast-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0"
# 実機実行: 2026-04-XX

6. JSONata + ASL 組み込み関数

Step Functions は 2024 年に JSONata 式エンジンをネイティブサポートした。これまで Lambda に委ねていた JSON 変換・条件分岐・日付処理を State Machine 内で直接記述でき、AWS-SDK Integration と組み合わせることで 完全 Lambda レスの変換パイプライン が実現する。

6-1. JSONata の基礎 (${ expr } 記法)

JSONata は JSONPath とは別の式言語だ。State Machine レベルまたは State レベルで "QueryLanguage": "JSONata" を指定して有効化する。

AWS-SDK Integration の Parameters は PascalCase 必須だ。JSONata 式はその値部分に .$ で埋め込む。JSONPath モードの "key.$": "$.path" と構造は同じで、値部分を ${ expr } に変えるだけだ。

{
  "QueryLanguage": "JSONata",
  "Parameters": {
 "TableName": "orders",
 "Item": {
"OrderId":{ "S.$": "${ $.order_id }" },
"CreatedAt": { "S.$": "${ $now() }" },
"Amount": { "N.$": "${ $string($.amount * 1.1) }" }
 }
  }
}

6-2. ASL 組み込み関数一覧

States.* 名前空間の組み込み関数は JSONata 式内でも利用できる。

JSONata モードでは ${ States.UUID() } のように式内で直接呼び出す。JSONPath モードでは "key.$": "States.UUID()" と .$ サフィックスを使う。

6-3. JSONata 実用例

{
  "Type": "Pass",
  "QueryLanguage": "JSONata",
  "Parameters": {
 "priority.$": "${ $.score >= 80 ? 'HIGH' : $.score >= 50 ? 'MEDIUM' : 'LOW' }",
 "label.$": "${ $.status = 'active' ? '有効' : '無効' }",
 "summary.$":  "${ States.Format('User {} scored {}', $.user_id, $.score) }"
  }
}

{
  "Type": "Pass",
  "QueryLanguage": "JSONata",
  "Parameters": {
 "active_ids.$":  "${ $.items[status = 'active'].id }",
 "total_price.$": "${ $sum($.items.price) }",
 "partitioned.$": "${ States.ArrayPartition($.items, 25) }"
  }
}

{
  "Type": "Task",
  "Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
  "QueryLanguage": "JSONata",
  "Parameters": {
 "TableName": "events",
 "Item": {
"EventId":{ "S.$": "${ States.UUID() }" },
"CreatedAt": { "S.$": "${ $now() }" },
"ExpiresAt": { "N.$": "${ $string($toMillis($now()) div 1000 + 86400) }" },
"Merged": { "S.$": "${ States.JsonToString(States.JsonMerge($.meta, $.extra, false)) }" }
 }
  },
  "End": true
}

6-4. Vol1 ResultSelector との関係

Vol1 (5 大 I/O フィルタ / ID:1439) で解説した ResultSelector は、JSONata モードでは Output フィールドに統合できる。

JSONata モードでは Output フィールド 1 つで 5 フィルタの役割を担えるため設定量が削減される。段階移行では QueryLanguage を State 別に指定して影響範囲を限定するのが安全だ。

6-5. PassState + JSONata での前処理パターン

SDK Direct Integration の呼出前に PassState で入力を整形するパターンは、Lambda 起動コストを完全に排除できる。

{
  "PrepareItem": {
 "Type": "Pass",
 "QueryLanguage": "JSONata",
 "Parameters": {
"TableName": "orders",
"Item": {
  "OrderId":{ "S.$": "${ $.order_id }" },
  "Status": { "S.$": "${ $uppercase($.status) }" },
  "Amount": { "N.$": "${ $string($.amount) }" },
  "UpdatedAt": { "S.$": "${ $now() }" }
}
 },
 "Next": "PutOrder"
  }
}

PassState の Parameters で変換を完結させることで Task State 側をシンプルに保てる。PassState は Lambda 起動コスト ($0.20/100 万リクエスト) と比べて無視できる水準のコストだ。

6-6. JSONata 未サポート機能と Lambda 回帰の判断基準

JSONata が有効でも Lambda に戻すべきケースが存在する。

Lambda 回帰の判断基準: 1 フィールドあたり 50 文字超の式・暗号化処理・外部リソース参照が必要な場合は Lambda を検討する。JSONata 未経験のチームには States.Format / States.UUID から段階導入を推奨する。

6-7. 3 点セット (JSONata 活用の State Machine)

Terraform (terraform >= 1.9 / hashicorp/aws ~> 5.0)

resource "aws_sfn_state_machine" "jsonata_demo" {
  name  = "jsonata-sdk-direct-demo"
  role_arn = aws_iam_role.sfn_exec.arn
  type  = "EXPRESS"

  definition = jsonencode({
 Comment = "JSONata + SDK Direct Integration デモ"
 QueryLanguage = "JSONata"
 StartAt = "WriteRecord"
 States = {
WriteRecord = {
  Type = "Task"
  Resource= "arn:aws:states:::aws-sdk:dynamodb:putItem"
  QueryLanguage = "JSONata"
  Parameters = {
 TableName = "demo-orders"
 Item = {
OrderId= { "S.$" = "${ States.UUID() }" }
CreatedAt = { "S.$" = "${ $now() }" }
Priority  = { "S.$" = "${ $.score >= 80 ? 'HIGH' : 'LOW' }" }
 }
  }
  Retry = [{ ErrorEquals = ["States.TaskFailed"], MaxAttempts = 3, BackoffRate = 2.0 }]
  End = true
}
 }
  })

  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

ASL JSON

{
  "Comment": "JSONata 条件分岐/UUID/タイムスタンプを SDK Direct DDB 書込みと組み合わせる",
  "QueryLanguage": "JSONata",
  "StartAt": "WriteRecord",
  "States": {
 "WriteRecord": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
"Parameters": {
  "TableName": "records",
  "Item": {
 "Id": { "S.$": "${ States.UUID() }" },
 "Priority": { "S.$": "${ $.score >= 80 ? 'HIGH' : $.score >= 50 ? 'MEDIUM' : 'LOW' }" },
 "ActiveIds": { "S.$": "${ States.JsonToString($.items[status = 'active'].id) }" },
 "CreatedAt": { "S.$": "${ $now() }" }
  }
},
"Retry": [{ "ErrorEquals": ["States.TaskFailed"], "MaxAttempts": 3 }],
"End": true
 }
  }
}

CLI

# JSONata 式の動作確認実行
aws stepfunctions start-execution \
  --state-machine-arn "arn:aws:states:ap-northeast-1:ACCOUNT_ID:stateMachine:jsonata-sdk-direct-demo" \
  --input '{"score": 85, "items": [{"id": "a", "status": "active"}, {"id": "b", "status": "inactive"}]}'
# 実機実行: 2026-04-XX (AI生成禁止・実行できない場合はTODO置換)

# PassState 変換結果の確認
aws stepfunctions get-execution-history \
  --execution-arn "arn:aws:states:ap-northeast-1:ACCOUNT_ID:execution:jsonata-sdk-direct-demo:EXECUTION_ID" \
  --query 'events[?type==`PassStateExited`].stateExitedEventDetails' \
  --output json

# DDB 書込み確認
aws dynamodb get-item \
  --table-name records \
  --key '{"Id": {"S": "UUID-FROM-EXECUTION"}}' \
  --region ap-northeast-1
# 実機実行: 2026-04-XX

7. IAM 最小権限 + 典型エラー集

SDK Direct Integration を本番運用する上で最も詰まりやすいのが IAM 権限設計 と 実行時エラーの解読 です。本章では 7 サービス分の最小権限表と典型エラー 10 ケースを体系化し、AccessDenied や ActionNotSupported に出会っても即座に対処できる知識を提供します。

7-1. SF 実行ロールの基本構造

Step Functions が SDK Direct Integration で他の AWS サービスを呼び出すには、状態機械に紐付いた IAM 実行ロール に必要な権限を付与します。信頼ポリシーには states.amazonaws.com の引き受けを許可し、Confused Deputy 問題 を防ぐために aws:SourceArn / aws:SourceAccount Condition を付加します。

resource "aws_iam_role" "sf_execution" {
  name = "sf-sdk-direct-vol5-execution-role"
  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "states.amazonaws.com" }
Action = "sts:AssumeRole"
Condition = {
  ArnLike= { "aws:SourceArn"  = "arn:aws:states:${var.region}:${var.account_id}:stateMachine:*" }
  StringEquals = { "aws:SourceAccount" = var.account_id }
}
 }]
  })
  tags = { Module = "sf-sdk-direct-vol5", Env = "prod-sample" }
}

実行ロールに付与する許可ポリシーはサービスごとに最小限の Action に絞ります。次節の権限表をもとに aws_iam_role_policy を定義してください。

7-2. サービス別 IAM 最小権限表

7 サービスで必要な IAM Action・Resource スコープ・推奨 Condition を一覧化します。* ワイルドカードは本番禁止です。

7-3. SourceArn / SourceAccount Condition で cross-account 防御

aws:SourceArn と aws:SourceAccount を Condition に設定することで、Confused Deputy 問題 を防ぎます。Terraform の aws_iam_role_policy に以下の Condition ブロックを必ず付加してください。

Condition = {
  ArnLike= { "aws:SourceArn"  = "arn:aws:states:${var.region}:${var.account_id}:stateMachine:${var.sm_name}" }
  StringEquals = { "aws:SourceAccount" = var.account_id }
}

ArnLike でワイルドカード (stateMachine:sf-sdk-*) を使うと同一プロジェクトの複数 SM に対応できます。Bedrock など一部サービスは Condition キーが非対応のため aws:SourceAccount のみで対応します。

7-4. 典型エラー 10 ケース集

7-5. エラーの Catch 戦略

Retry 上限到達後に Catch が発動します。States.ALL をフォールバックとして全エラーを補足し、ResultPath でエラー情報を後段ステートへ渡します。SDK Direct 特有のエラー (Throttling / AccessDenied 等) を Retry に、最終フォールバックを States.ALL Catch に分離するパターンが推奨です。

{
  "Retry": [{"ErrorEquals": ["DynamoDB.ProvisionedThroughputExceededException"],
 "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2.0}],
  "Catch": [{"ErrorEquals": ["States.ALL"], "Next": "HandleError", "ResultPath": "$.error"}]
}

7-6. デバッグ手順

7-7. 3 点セット (IAM Terraform + エラーハンドリング ASL + CLI)

Terraform: 7 サービス一括 IAM ポリシー

resource "aws_iam_role_policy" "sf_sdk_all_services" {
  name= "sf-sdk-direct-vol5-all-services"
  role= aws_iam_role.sf_execution.id
  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{ Effect = "Allow"
  Action= ["dynamodb:PutItem","dynamodb:GetItem","dynamodb:UpdateItem",
  "dynamodb:DeleteItem","dynamodb:Query","dynamodb:BatchWriteItem","dynamodb:TransactWriteItems"]
  Resource = "arn:aws:dynamodb:${var.region}:${var.account_id}:table/${var.table_name}" },
{ Effect = "Allow"; Action = ["sns:Publish","sns:PublishBatch"]
  Resource = "arn:aws:sns:${var.region}:${var.account_id}:${var.topic_name}" },
{ Effect = "Allow"; Action = ["sqs:SendMessage","sqs:SendMessageBatch"]
  Resource = "arn:aws:sqs:${var.region}:${var.account_id}:${var.queue_name}" },
{ Effect = "Allow"; Action = ["events:PutEvents"]
  Resource = "arn:aws:events:${var.region}:${var.account_id}:event-bus/${var.event_bus_name}" },
{ Effect = "Allow"; Action = ["s3:GetObject","s3:PutObject","s3:CopyObject","s3:DeleteObject","s3:ListObjectsV2"]
  Resource = "arn:aws:s3:::${var.bucket_name}/*" },
{ Effect = "Allow"; Action = ["execute-api:Invoke"]
  Resource = "arn:aws:execute-api:${var.region}:${var.account_id}:${var.api_id}/*" },
{ Effect = "Allow"; Action = ["bedrock:InvokeModel"]
  Resource = "arn:aws:bedrock:${var.region}::foundation-model/${var.bedrock_model_id}" }
 ]
  })
}

ASL: Retry + Catch 付きエラーハンドリングフロー

{
  "Comment": "SDK Direct Integration — Retry/Catch サンプル (vol5)",
  "StartAt": "WriteToDDB",
  "States": {
 "WriteToDDB": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
"Parameters": {
  "TableName": "orders",
  "Item": {
 "id": { "S.$": "$.order_id" },
 "status": { "S": "PROCESSING" },
 "created_at": { "S.$": "$$.Execution.StartTime" }
  }
},
"TimeoutSeconds": 10,
"Retry": [{"ErrorEquals": ["DynamoDB.ProvisionedThroughputExceededException"],
  "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2.0}],
"Catch": [{"ErrorEquals": ["States.ALL"], "Next": "NotifyError", "ResultPath": "$.ddb_error"}],
"Next": "Succeed"
 },
 "NotifyError": {
"Type": "Task",
"Resource": "arn:aws:states:::aws-sdk:sns:publish",
"Parameters": {
  "TopicArn.$": "$.alert_topic_arn",
  "Message.$": "States.Format('DDB write error: {}', $.ddb_error.Cause)"
},
"End": true
 },
 "Succeed": { "Type": "Succeed" }
  }
}

CLI: 実行確認コマンド

# 実行開始
aws stepfunctions start-execution--state-machine-arn arn:aws:states:ap-northeast-1:ACCOUNT:stateMachine:sf-sdk-direct-vol5--input '{"order_id":"ORD-001","alert_topic_arn":"arn:aws:sns:ap-northeast-1:ACCOUNT:alerts"}'
# TODO: 実機実行: 2026-04-XX

# 失敗実行の履歴確認
aws stepfunctions get-execution-history--execution-arn EXEC_ARN--query 'events[?type==`TaskFailed`]'
# TODO: 実機実行: 2026-04-XX

8. まとめ + 5連作完結宣言

Vol1 から始まった SF 実践編シリーズが、Vol5 の本記事で完結します。I/O フィルタ (Vol1) → Callback (Vol2) → Distributed Map (Vol3) → Express Workflow (Vol4) → SDK Direct Integration (Vol5) の 5 巻で、Lambda に頼らない Step Functions 実践スキルが完全に体系化されました。

5 巻を通じて読者が身に付けたのは、単なる「Lambda を減らす」テクニックではなく、ワークフローのオーケストレーション設計力 です。I/O フィルタで変換を理解し、Callback で非同期を制御し、Distributed Map で並列処理を拡張し、Express Workflow で高頻度処理を最適化し、SDK Direct で外部サービスとのインテグレーションを完成させる — これが SF 実践編の全体像です。

8-1. 6 実務論点振り返りマトリクス

本記事で扱った 6 つの実務論点と各章での成果物を整理します。

これら 6 論点は相互に連携しています。§2 で選んだ AWS-SDK Integration が §7 の IAM 最小権限表の対象になり、§6 の JSONata 変換が §3–§5 の各サービス呼び出しの前段として組み合わさります。Lambda を挟まず 7 サービスを直接操作する本番稼働の状態機械を設計・デプロイ・運用するサイクルを自走できるようになることが本記事の最終ゴールです。

8-2. チートシート (サービス別 ASL 記法 × IAM × JSONata)

共通注意点: ASL Resource の <action> は camelCase (先頭小文字)、Parameters の key は PascalCase (先頭大文字)。この 2 ルールを徹底することで §7-4 のPascalCase 誤り系エラー (ケース 4–6) を未然に防げます。

本記事を読了後、以下の項目を自力で実行できれば学習達成です:

• ASL Resource URI を 7 サービス分すべて記述できる
• JSONata ${ expr } と States.Format / States.JsonMerge を Lambda なしで活用できる
• 7 サービス分の IAM 最小権限ポリシーを Terraform で記述し aws:SourceArn Condition を付加できる
• 典型エラー 10 ケースのいずれかに直面しても 1 分以内に同定し対処できる
• Lambda 置換コスト試算を 3 シナリオで経営層向けに説明できる

8-3. Vol1–Vol5 5連作完結サマリ

SF 実践編シリーズ全 5 巻の要点と本記事との接続ポイントをまとめます。Vol1 → Vol5 の順に読むと「I/O 設計 → 非同期パターン → 大量並列 → 高頻度処理 → Lambda 不要化」の流れで段階的に実力が積み上がります。

5 連作を完読した読者の次のステップとして AWS CDK の StepFunctions Constructs (State Machine を TypeScript/Python で定義) や EventBridge Pipes との連携 (SF をパイプのターゲットとして組み込むイベント駆動アーキテクチャ) への発展をお勧めします。また X-Ray + CloudWatch Insights で SM 実行のボトルネックを可視化することで、本番運用の完成度をさらに高めることができます。

8-4. 関連記事表

本記事の前提・補完として参照した記事一覧です。§7 の Catch 戦略の詳細は Retry/Catch/Timeout 完全ガイド (ID:1057) で、Optimized 統合の基礎は Express vs Standard 比較 (ID:1101) で扱っています。

本番デプロイ前チェックリスト: SDK Direct Integration を本番環境へ投入する前に、以下の項目を確認してください。

8-5. 5連作完結メッセージ

8-6. 次のステップへ

SF 実践編 Vol1–Vol5 を通じて学んだ技術を土台に、次の発展テーマへ進みましょう。Vol1 では JSONata の前提となる I/O フィルタを復習でき、Vol3 では本記事 §3 で触れた DDB + Distributed Map 複合パターンを深掘りできます。AWS 公式ドキュメントでは SDK Integration で利用可能な全 API の最新仕様を確認できます。

→ Vol1 5大I/Oフィルタを復習 (JSONataの前提)

→ Vol3 Distributed Map 本番運用を復習 (DDB+DMap)

→ AWS 公式 SDK Integration doc (最新 API)

SF 実践編 Vol1–Vol5 で学んだ知識を実プロジェクトへ適用し、Lambda に頼らない堅牢なワークフロー設計を実践してください。5 連作の技術体系が、あなたの AWS 設計力の土台となります。