0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@leomarokun(れおまろ)

AgentCore Gateway と Lambda で決定論的な処理をツール化する AI エージェントの「Workflow as Tools」を理解

0
Last updated at Posted at 2026-08-29

はじめに

この記事は是非以下の記事とセットで読んでください

https://qiita.com/leomarokun/items/10c29c5eab780dfab3cb

前回の記事では、「非決定的な調査は AgentCore harness に、決定論的な Slack 通知は Step Functions 側に」という形で処理を分離しました。これはワークフロー(Step Functions)が主導権を持ち、AI エージェントをその中の 1 ステップとして呼び出す構成です。

今回は逆方向のアプローチを試します。

決定論的な処理(レポートを Slack に投稿する)そのものを、AI エージェントのツールとして提供する「Workflow as Tools」　パターンです。

主導権は AI エージェント側にあり、必要に応じて決定論的な処理を「ツール」として呼び出します。

Workflow as Tools を理解する

前回のおさらいです。同じ入力に対して常に同じ結果が返る性質を 決定論的、
同じ入力でも結果が一通りに定まらない性質を 非決定的
と呼びます。
「調べてまとめる」は非決定的、「Slack に投稿する」は決定論的です。

この 2 つを分離する方法は 2 通りあります。

image.png

前回はワークフローが決定論的な処理の実行を保証していました。「調査ステートの後に必ず通知ステートを通る」という構造そのものが保証になっていたからです。

今回はその保証の作り方が変わります。
決定論的な処理(Slack への投稿)を Lambda に切り出し、AgentCore Gateway 経由で MCP ツールとして公開します。ツールの中身(投稿フォーマット、リトライ、エラーハンドリング)は決定論的なコードなので安定していますが、「そのツールを呼ぶかどうか」の判断はエージェントに残ります。

やってみた

0. 簡易構成図

image.png

1. 決定論的パート: Lambda を Gateway でツール化する

完成版は「3. デプロイする」で示す main.tf をコピーして実行すれば作れるようにしてますので、ここではポイントだけを整理します。

Slack に投稿するだけの Lambda を用意します。ソースコードは、「3. デプロイする」で示す lambda/post_to_slack.py を参照してください。

この Lambda を Bedrock AgentCore Gateway の Lambda target として登録すると、text を受け取る post_to_slack という 1 つの MCP ツールとして公開されます。

resource "aws_bedrockagentcore_gateway" "tools" {
  name     = "aws-news-tool"
  role_arn = aws_iam_role.gateway.arn

  authorizer_type = "AWS_IAM"
  protocol_type   = "MCP"
}

resource "aws_bedrockagentcore_gateway_target" "post_to_slack" {
  name               = "slack"
  gateway_identifier = aws_bedrockagentcore_gateway.tools.gateway_id

  credential_provider_configuration {
    gateway_iam_role {}
  }

  target_configuration {
    mcp {
      lambda {
        lambda_arn = aws_lambda_function.post_to_slack.arn

        tool_schema {
          inline_payload {
            name        = "post_to_slack"
            description = "まとめたテキストを Slack チャンネルに投稿する。調査や要約が完了して、内容を共有する準備ができたら呼び出すこと。"

            input_schema {
              type = "object"
              property {
                name        = "text"
                type        = "string"
                description = "Slack に投稿する本文"
                required    = true
              }
            }
          }
        }
      }
    }
  }
}

2. 非決定的パート: harness にツールとして追加する

こちらも完成版は「3. デプロイする」で示す main.tf をコピーして実行すれば作れるようにしてますので、ここではポイントだけを整理します。

Bedrock AgentCore Harness には、前回と同じ AWS Knowledge MCP に加えて、この Gateway を agentcore_gateway タイプのツールとして追加します。

tool {
  type = "agentcore_gateway"
  name = "SlackTools"

  config {
    agentcore_gateway {
      gateway_arn = aws_bedrockagentcore_gateway.tools.gateway_arn
      outbound_auth {
        aws_iam = true
      }
    }
  }
}

3. デプロイする

Slack の Incoming Webhook URL を用意して、以下をそのままコピーして実行すれば動きます。

ディレクトリ構成は main.tf と lambda/post_to_slack.py の 2 ファイルです。

main.tf（全文） 
// Terraform 設定ファイル
terraform {
  required_version = ">= 1.5"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = ">= 6.61"
    }
    archive = {
      source  = "hashicorp/archive"
      version = ">= 2.4"
    }
  }
}

// AWS Provider 設定
provider "aws" {
  region = var.region
}

// 変数定義
variable "region" {
  type    = string
  default = "ap-northeast-1"
}

variable "slack_webhook_url" {
  type      = string
  sensitive = true
}

variable "model_id" {
  type    = string
  default = "global.amazon.nova-2-lite-v1:0"
}

// ローカル変数定義
locals {
  name         = "aws-news-tool"
  harness_name = "aws_news_tool_agent"
}

// 現在の AWS アカウント情報を取得
data "aws_caller_identity" "current" {}

########################################
# 1. Slack 通知 Lambda (決定論的パート)
########################################

// Lambda 関数のコードを zip ファイルにアーカイブ
data "archive_file" "post_to_slack" {
  type        = "zip"
  source_file = "${path.module}/lambda/post_to_slack.py"
  output_path = "${path.module}/lambda/post_to_slack.zip"
}

// Lambda 実行用の IAM ロールを作成
resource "aws_iam_role" "lambda" {
  name = "${local.name}-lambda-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Principal = { Service = "lambda.amazonaws.com" }
      Action    = "sts:AssumeRole"
    }]
  })
}

// Lambda 実行用の IAM ポリシーをロールにアタッチ
resource "aws_iam_role_policy_attachment" "lambda_logs" {
  role       = aws_iam_role.lambda.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

// Slack 通知用の Lambda 関数を作成
resource "aws_lambda_function" "post_to_slack" {
  function_name    = "${local.name}-post-to-slack"
  role             = aws_iam_role.lambda.arn
  handler          = "post_to_slack.handler"
  runtime          = "python3.13"
  timeout          = 10
  filename         = data.archive_file.post_to_slack.output_path
  source_code_hash = data.archive_file.post_to_slack.output_base64sha256

  environment {
    variables = {
      SLACK_WEBHOOK_URL = var.slack_webhook_url
    }
  }
}

########################################
# 2. AgentCore Gateway (Lambda を MCP ツール化)
########################################

// Gateway 実行用の IAM ロールを作成
resource "aws_iam_role" "gateway" {
  name = "${local.name}-gateway-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Principal = { Service = "bedrock-agentcore.amazonaws.com" }
      Action    = "sts:AssumeRole"
    }]
  })
}

// Gateway 実行用の IAM ポリシーをロールにアタッチ
resource "aws_iam_role_policy" "gateway_invoke_lambda" {
  role = aws_iam_role.gateway.name

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Sid      = "InvokePostToSlack"
      Effect   = "Allow"
      Action   = "lambda:InvokeFunction"
      Resource = aws_lambda_function.post_to_slack.arn
    }]
  })
}

// Bedrock AgentCore Gateway を作成
resource "aws_bedrockagentcore_gateway" "tools" {
  name     = local.name
  role_arn = aws_iam_role.gateway.arn

  authorizer_type = "AWS_IAM"
  protocol_type   = "MCP"
}

// Gateway に Slack 投稿用のターゲットを追加
resource "aws_bedrockagentcore_gateway_target" "post_to_slack" {
  name               = "slack"
  gateway_identifier = aws_bedrockagentcore_gateway.tools.gateway_id
  description        = "まとめたレポートを Slack に投稿する"

  credential_provider_configuration {
    gateway_iam_role {}
  }

  target_configuration {
    mcp {
      lambda {
        lambda_arn = aws_lambda_function.post_to_slack.arn

        tool_schema {
          inline_payload {
            name        = "post_to_slack"
            description = "まとめたテキストを Slack チャンネルに投稿する。調査や要約が完了して、内容を共有する準備ができたら呼び出すこと。"

            input_schema {
              type = "object"

              property {
                name        = "text"
                type        = "string"
                description = "Slack に投稿する本文"
                required    = true
              }
            }
          }
        }
      }
    }
  }
}

########################################
# 3. AgentCore harness (非決定的パート)
########################################

// harness 実行用の IAM ロールを作成
resource "aws_iam_role" "harness" {
  name = "${local.name}-harness-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Principal = { Service = "bedrock-agentcore.amazonaws.com" }
      Action    = "sts:AssumeRole"
      Condition = {
        StringEquals = { "aws:SourceAccount" = data.aws_caller_identity.current.account_id }
        ArnLike      = { "aws:SourceArn" = "arn:aws:bedrock-agentcore:${var.region}:${data.aws_caller_identity.current.account_id}:*" }
      }
    }]
  })
}

// harness にポリシーをアタッチ
resource "aws_iam_role_policy" "harness" {
  role = aws_iam_role.harness.name

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid      = "BedrockModelInvocation"
        Effect   = "Allow"
        Action   = ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"]
        Resource = ["arn:aws:bedrock:*::foundation-model/*", "arn:aws:bedrock:${var.region}:${data.aws_caller_identity.current.account_id}:*"]
      },
      {
        Sid      = "EcrPublicPull"
        Effect   = "Allow"
        Action   = ["ecr-public:GetAuthorizationToken", "sts:GetServiceBearerToken"]
        Resource = "*"
      },
      {
        Sid      = "XRayTracing"
        Effect   = "Allow"
        Action   = ["xray:PutTraceSegments", "xray:PutTelemetryRecords", "xray:GetSamplingRules", "xray:GetSamplingTargets"]
        Resource = "*"
      },
      {
        Sid      = "Metrics"
        Effect   = "Allow"
        Action   = "cloudwatch:PutMetricData"
        Resource = "*"
        Condition = {
          StringEquals = { "cloudwatch:namespace" = "bedrock-agentcore" }
        }
      },
      {
        Sid    = "Logs"
        Effect = "Allow"
        Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents", "logs:DescribeLogStreams", "logs:DescribeLogGroups"]
        Resource = [
          "arn:aws:logs:${var.region}:${data.aws_caller_identity.current.account_id}:log-group:/aws/bedrock-agentcore/runtimes/*",
          "arn:aws:logs:${var.region}:${data.aws_caller_identity.current.account_id}:log-group:*",
        ]
      },
      {
        Sid      = "AgentCoreMemory"
        Effect   = "Allow"
        Action   = ["bedrock-agentcore:CreateEvent", "bedrock-agentcore:GetEvent", "bedrock-agentcore:ListEvents", "bedrock-agentcore:DeleteEvent", "bedrock-agentcore:RetrieveMemoryRecords"]
        Resource = "arn:aws:bedrock-agentcore:${var.region}:${data.aws_caller_identity.current.account_id}:memory/${local.harness_name}-*"
      },
      {
        Sid      = "WorkloadIdentity"
        Effect   = "Allow"
        Action   = ["bedrock-agentcore:GetWorkloadAccessToken", "bedrock-agentcore:GetWorkloadAccessTokenForJWT"]
        Resource = [
          "arn:aws:bedrock-agentcore:${var.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
          "arn:aws:bedrock-agentcore:${var.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/*",
        ]
      },
      {
        Sid      = "AgentCoreGatewayAccess"
        Effect   = "Allow"
        Action   = "bedrock-agentcore:InvokeGateway"
        Resource = aws_bedrockagentcore_gateway.tools.gateway_arn
      },
    ]
  })
}

// Bedrock AgentCore Harness を作成
resource "aws_bedrockagentcore_harness" "news" {
  harness_name       = local.harness_name
  execution_role_arn = aws_iam_role.harness.arn

  model {
    bedrock_model_config {
      model_id = var.model_id
    }
  }

  system_prompt {
    text = "あなたは AWS の最新情報を調べて日本語で要約するアシスタントです。調査には必ず AWS Knowledge MCP のツールを使い、推測で書かないでください。まとめが完成したら post_to_slack ツールを使って必ず Slack に投稿してください。"
  }

  tool {
    type = "remote_mcp"
    name = "AWSKnowledgeMCPServer"

    config {
      remote_mcp {
        url = "https://knowledge-mcp.global.api.aws"
      }
    }
  }

  tool {
    type = "agentcore_gateway"
    name = "SlackTools"

    config {
      agentcore_gateway {
        gateway_arn = aws_bedrockagentcore_gateway.tools.gateway_arn
        outbound_auth {
          aws_iam = true
        }
      }
    }
  }

  allowed_tools   = ["*"]
  max_iterations  = 15
  timeout_seconds = 600
}

// 出力変数
output "harness_arn" {
  value = aws_bedrockagentcore_harness.news.arn
}

output "gateway_arn" {
  value = aws_bedrockagentcore_gateway.tools.gateway_arn
}
lambda/post_to_slack.py（全文） 
import json
import os
import urllib.request


def handler(event, context):
    text = event.get("text", "")
    if not text:
        return {"status": "error", "message": "text is required"}

    webhook_url = os.environ["SLACK_WEBHOOK_URL"]
    req = urllib.request.Request(
        webhook_url,
        data=json.dumps({"text": text}).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        body = resp.read().decode("utf-8")
        return {"status": "ok", "slackResponse": body}

# ディレクトリを作って移動
mkdir -p aws-news-tool-tf/lambda && cd aws-news-tool-tf

# 上記の main.tf と lambda/post_to_slack.py をこのディレクトリに保存してから

# terraform.tfvars を作って Slack の Incoming Webhook URL を書き込む
cat > terraform.tfvars <<EOF
slack_webhook_url = "https://hooks.slack.com/services/XXXX/YYYY/ZZZZ"
EOF

# Terraform を初期化してデプロイ
terraform init
terraform apply -auto-approve

今回は Step Functions を使わないので、boto3 の invoke_harness で直接動作確認します。

pip install --upgrade "boto3>=1.43" "botocore>=1.43"

import boto3, uuid

client = boto3.client("bedrock-agentcore", region_name="ap-northeast-1")

response = client.invoke_harness(
    harnessArn="<terraform output -raw harness_arn の値>",
    runtimeSessionId=str(uuid.uuid4()).ljust(33, "0"),  # 33文字以上必須
    messages=[{
        "role": "user",
        "content": [{"text": "AWS Knowledge MCP で AWS Step Functions の概要を1文で調べて、post_to_slack ツールを使って必ず Slack に投稿してください。"}],
    }],
)

text = ""
for event in response["stream"]:
    if "contentBlockDelta" in event:
        text += event["contentBlockDelta"]["delta"].get("text", "")
print(text)

まず、シンプルに Slack へ投稿されたことを確認してみましょう。

スクリーンショット 2026-08-29 16.49.11.png

次に、AWS マネージメントコンソールの BedrockAgentCore オブザバビリティのトレースを見て、post_to_slack が呼ばれたことを確認してみましょう。

スクリーンショット 2026-08-29 16.55.16.png

4. ツールを呼ぶかどうかは、結局 AI エージェント次第

前回の記事の教訓は「通知するかどうかを AI エージェントに判断させると、判断ミスで通知が飛ばないことがある」でした。Workflow as Tools パターンでこの問題がどう変わるかを、同じ BedrockAgentCore Harness に対して指示だけを変えて試してみます。

system_prompt messages(ユーザー発話) 結果
元のまま(「post_to_slack ツールを使って必ず Slack に投稿してください」を含む) 「調べて Slack に投稿してください」 post_to_slack が呼ばれた
元のまま(同上) 「調べてください」(投稿の指示なし) post_to_slack は呼ばれなかった

system_prompt に「post_to_slack ツールを使って必ず Slack に投稿してください」と明記していても、それだけでは投稿は保証されませんでした。ユーザー発話(messages)側に投稿を促す文言がないと、post_to_slack は呼ばれなかったのです。

裏を返せば、「いつ呼ぶか」はシステムプロンプトの指示だけでは決まらず、そのときどきのユーザー発話にも左右されており、構造的に保証されているわけではありません。

これが Step Functions 主導パターンとの決定的な違いです。前回はワークフローの構造そのものが「通知は必ず実行される」を保証していました。今回は、ツールを呼ぶかどうかの最終判断が AI エージェントに残るぶん、状況に応じて「今回は投稿しない」という柔軟な振る舞いも自然にできる代わりに、「必ず実行してほしい」処理には向いていません。

後片付け

terraform destroy -auto-approve

どちらを選ぶか

ワークフロー主導(前回) Workflow as Tools(今回)
主導権 Step Functions AgentCore harness
決定論的処理の実行保証 構造で保証(必ず通る) プロンプト/ツール説明文に依存
向いている用途 通知・監査ログなど「必ず実行してほしい」処理 状況に応じて呼ぶかどうかを判断してほしい処理
実装の主な要素 Step Functions の Catch Gateway の description

「絶対に落としたくない処理」はワークフロー主導で外に出し、「エージェントの判断に任せてよい処理」だけを Workflow as Tools にする、という使い分けにするとよさそうです。

参考リンク

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?