AgentCore GatewayのターゲットにStep Functionsを指定できるようになりました

こんにちは、ふくちです。

AgentCore GatewayのターゲットにStep Functionsが選べるようになったので、一通り触ってみました。

何ができるのか

既存のステートマシンを、そのままMCPツールとしてエージェントに渡せます。現状でも既存のAPI GatewayやLambdaでできたことがStep Functionsにも拡張したような形ですね。

エージェントが決めるのはinput(いつ・どんな入力で動かすか)だけで、それ以降の処理(どのステップをどの順で実行するかなど)はステートマシン側の定義次第です。

ツールにできるのはStep FunctionsのAPI10個です(今のところ)。これらを分類すると3種類くらいになりそうです。

種類	操作
ステートマシンを固定するもの	StartExecution, StartSyncExecution, ListExecutions, DescribeStateMachine
実行ARNを受け取るもの	DescribeExecution, GetExecutionHistory, StopExecution
タスクトークンを受け取るもの	SendTaskSuccess, SendTaskFailure, SendTaskHeartbeat

ここでいう固定(ドキュメントではpinと記載されているもの)は、ターゲット作成時、操作ごとに対象のステートマシンのARNを決めておくことです。設定では各操作の parameterValues.stateMachineArn にARNを書きます。
例えばStartExecutionを注文処理のステートマシンに固定すると、エージェントから見えるツールの引数は input だけになり、呼べば必ずそのステートマシンが起動します。エージェントが別のARNを無理やり渡しても cannot set parameter(s): [/stateMachineArn] で弾かれます。

一方、実行ARNやタスクトークンを受け取る6つは、対象をエージェントが引数で渡すので固定できません。どの実行まで触れるかはGatewayのサービスロールのIAMポリシー次第です。

タスクトークンは、ステートマシンが途中で止まって外からの返事を待つときに、Step Functionsが発行する文字列です。返事をする側はこのトークンを付けてSendTask系の操作を呼びます。具体的な使い方は後の返金審査のところで書きます。

試してみる

先にサンプルのリポジトリを共有しておきます。メインはClaudeが作ってくれて、私が確認しました。

動作確認用に、2つのステートマシンを用意しました。どちらも簡単なもので、外部のサービスを呼ぶのは返金審査のSQSへの送信だけです。

処理	タイプ	処理
注文処理	Standard	注文金額をチェックし在庫引当 → 出荷待ち → 出荷の順に進んで、追跡番号を返す
返金審査	Standard	返金申請を受けると、審査依頼をSQSに送って止まる。エージェントの判断が返ってきたら、承認なら返金し、承認でなければ失敗で終わる

Gatewayには次の3つのターゲットを登録しました。ツール名は <ターゲット名>___<操作名>（例: orders___StartExecution）になります。

ターゲット	操作	固定先
orders	StartExecution, ListExecutions, DescribeStateMachine, DescribeExecution, GetExecutionHistory, StopExecution	注文処理
refund	StartExecution, ListExecutions, DescribeStateMachine	返金審査
tasks	SendTaskSuccess, SendTaskFailure, SendTaskHeartbeat	なし

エージェントはStrands Agentsで作り、IAM認証でAgentCore Gatewayとつないでいます。

コンソールで確認する

AgentCore Gatewayのターゲット追加画面で、ターゲットタイプに「コネクター」、その下の一覧で「Step Functions」を選びます。
扱いとしてはナレッジベースと同じなんですね。

あとはステートマシンを1つ選んで、そのステートマシンに対して実行して良い操作を選ぶだけです。
最初は10個全部の操作が選択されており、Standardのステートマシンを選ぶとStartSyncExecutionはExpress専用という警告が出ます。

StartSyncExecutionを外せばOKですね。

ちなみに1ターゲットにつき設定できるステートマシンは1つだけです。StartSyncExecutionだけExpressに向けるような操作ごとの固定は、APIから作る必要があります。

エージェントから呼んでみる

まずは注文処理から試してみます。

「注文D-3001(8,000円)を出荷して、追跡番号を教えて」と頼むと、エージェントは orders___StartExecution で注文処理を起動し、orders___DescribeExecution で終わるまで状態を確かめてから、追跡番号を返してくれました。

↓

出荷待ちを600秒にした注文で「止めて」と頼むと、orders___StopExecution で止めてくれました。

続いて返金審査をやってみます。
返金審査では、ワークフローが途中で止まってエージェントの判断を待ちます。この待ち合わせでタスクトークンを使います。

設定するのは、ステートマシンの定義(ASL)の中の、審査依頼をSQSに送るTask状態です。今回は次のように書きました。

"RequestAgentReview": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sqs:sendMessage.waitForTaskToken",
  "Arguments": {
    "QueueUrl": "<SQSキューのURL>",
    "MessageBody": {
      "taskToken": "{% $states.context.Task.Token %}",
      "executionArn": "{% $states.context.Execution.Id %}",
      "request": "{% $states.input %}"
    }
  },
  "HeartbeatSeconds": 120,
  "TimeoutSeconds": 3600,
  "Output": "{% $merge([$states.input, {'review': $states.result}]) %}",
  "Next": "IsApproved"
}

ポイントは2つです。

• Resourceには任意の待ち状態を発生させたい操作に.waitForTaskTokenを付ける
  • ここではSQSにメッセージを送る操作(sendMessage)の末尾につけている
  • こうするとStep FunctionsはSQSにメッセージを送ったあと、返事が来るまで実行停止する
• メッセージ本文となるMessageBodyに$states.context.Task.Tokenでタスクトークンを入れる
  • 返事する側(ここではエージェント)はこのトークンでどの実行のどのタスクへの返事かを伝える

サンプルでは、商品が割れていたので返金処理を依頼する、みたいなケースで試してみました。

1. エージェントが refund___StartExecution で返金審査を起動する
2. ワークフローがタスクトークンと申請内容をSQSに送って止まる
   
3. エージェントがSQSを読む自作のツールでメッセージを受け取り、申請内容を見て判断する
4. エージェントが判断結果をoutput(例: {"approved": true, "comment": "..."})に入れて tasks___SendTaskSuccess を呼ぶ。これによってワークフローが再開し、承認されていれば返金する
   
   ※承認されていなければ拒否です
   
5. 申請そのものを受け付けられないときは、tasks___SendTaskFailure を呼ぶこともできる
   

エージェントの審査に時間がかかるときは、tasks___SendTaskHeartbeat で作業中だと伝えることもできます。上の定義ではHeartbeatSecondsを120にしているので、120秒のあいだ返事もハートビートもなければタイムアウトする形ですね。

ちなみにトークンは1回しか使えず、使い終わったものをもう一度送るとTaskTimedOutになりました。
また、SendTask系はIAMで対象の実行を絞れないので、権限さえあれば、トークンを持っている人は誰でもそのタスクを完了できます。トークンは秘密情報として扱うべきですね。

なお、AgentCore GatewayのコネクターでStep Functionsを繋ぐだけではタスクトークンをエージェントに届けてくれません。今回のSQSのように、受け渡しの経路は自分で用意する必要があります。

ハマりどころ

ツールの引数に日本語をそのまま入れると、Step Functionsに届いた時点で?????になりました。同じ内容をboto3から直接SendTaskFailureで送ると正しく入るので、多分コネクタ側の問題っぽいです。

inputとoutputはJSON文字列なので、json.dumps(..., ensure_ascii=True) で \uXXXX にしてから渡せば大丈夫でした。ただSendTaskFailureのerrorとcauseはただの文字列なので逃げ道がなく、英語で書くしかなさそうです。

また、SendTask系はparameterValuesなしで追加するとドキュメントに書かれていますが、名前だけを並べるとConnector configurations must not be empty.で弾かれます。空の parameterValues を付ければ通ります。コンソールで作ったターゲットにも空のものが付いていました。
省略してはいけないっていう制約らしいです。

agc.create_gateway_target(
    gatewayIdentifier=GATEWAY_ID,
    name="tasks",
    targetConfiguration={"mcp": {"connector": {
        "source": {"connectorId": "states", "version": "1.0.0"},
        "configurations": [
            {"name": "SendTaskSuccess", "parameterValues": {}},
            {"name": "SendTaskFailure", "parameterValues": {}},
            {"name": "SendTaskHeartbeat", "parameterValues": {}},
        ],
    }}},
    credentialProviderConfigurations=[{"credentialProviderType": "GATEWAY_IAM_ROLE"}],
)

まとめ

選択肢が増えるのは良いことですね〜