Amazon Bedrock AgentCore + Amazon API Gateway + AWS Lambdaでストリーム応答を実装してみた

はじめに

こんにちは、コンサルティング部の神野です。

2025年11月19日、アップデートによりAmazon API Gateway（以下API Gateway）でストリーム応答が可能になりました！！
API Gateway + AWS Lambda（以下Lambda）の組み合わせで、Bedrockを呼び出す際にタイムアウトが気になるケースがあったので嬉しいアップデートですね。

https://aws.amazon.com/blogs/compute/building-responsive-apis-with-amazon-api-gateway-response-streaming/

早速弊社のsuzuki.ryoさんがこちらの記事でAmazon Bedrockとの組み合わせを試されていますので、ぜひご覧ください。

https://dev.classmethod.jp/articles/api-gateway-response-streaming-bedrock/

今回は、このAPI Gatewayのストリーム応答を使って、Amazon Bedrock AgentCore Runtime（以下AgentCore）をラップしたストリーミングAPIを作ってみました！

AgentCoreのURL形式

AgentCore Runtimeを呼び出すURLは、以下のような形式になっています。

https://bedrock-agentcore.ap-northeast-1.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aap-northeast-1%3A123456789012%3Aagent-runtime%2Fabcd1234/invocations

フロントエンドから直接リクエストを送る際に、URLにアカウントIDが含まれていたり、ARNをエンコードしないといけないのが気になるな・・・と思っていました。また、カスタムドメインも使いたいケースもありますよね。

こういったケースでは、CloudFrontを被せたり、Lambda Function URLsを使う方法もありますが、今回はアップデートによって可能になったLambda + API Gatewayで実装してみました！

参考：CloudFrontでラップする方法

AWSが公式ブログでやり方を紹介しているので、AgentCoreをCloudFrontでラップしたい場合は下記をご参照ください。

https://aws.amazon.com/jp/blogs/machine-learning/set-up-custom-domain-names-for-amazon-bedrock-agentcore-runtime-agents/

API Gatewayを選んだ理由

CloudFrontやLambda Function URLsでもラップは可能ですが、API Gateway + Lambdaを採用すると以下のような利点があると思いました。

• 認証・認可の統合が簡単
  • API Key、Amazon Cognito、Lambda Authorizerなど、豊富な認証方式を設定できます
• レート制限や使用量プランが標準機能
  • API Keyごとの使用量制限やスロットリングを設定できます
• APIとしての管理がしやすい
  • ステージ管理なども可能です

CloudFrontでラップする場合でも下記は嬉しいポイントですがAPI Gateway + Lambdaでも同じことが言えます。

• AWS WAFと統合
• カスタムドメインの設定

APIを管理する観点で、便利な機能が揃っているので組み合わせると嬉しいポイントもあるかなと思い今回組み合わせてみたいと思った次第です！

それでは、早速実装していきましょう！

アーキテクチャ概要

今回構築するアーキテクチャは以下の通りです。

LambdaはAPI Gatewayからのみ実行できるようにし、AgentCore RuntimeはIAM認証でLambdaから実行できるようにします。
CloudFrontのみでラップする場合と比較して、IAM認証を使用するこのアーキテクチャでは、AgentCoreのURLを直接実行されるリスクを低減できると思います。

前提条件

本記事は以下の環境で実施しました。

• AWS CDK 2.221.0
• Python 3.13.6

リポジトリ

今回の実装の完全なコードは、以下のGitHubリポジトリをご参照ください。

https://github.com/yuu551/agentcore-api-gateway-streaming

今回AgentCore + LambdaはCDKで実装し、API Gatewayはコンソール上で手動で設定する作りとしています。

AgentCore エージェントの実装

まずは、AgentCore側のエージェント実装から見ていきます。今回はシンプルな実装にしました。

main.py

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel

# Bedrock Modelの設定
model = BedrockModel(
    model_id="anthropic.claude-3-5-haiku-20241022-v1:0",
    params={"max_tokens": 4096, "temperature": 0.7},
    region="us-west-2",
)
agent = Agent(model=model)

app = BedrockAgentCoreApp()

@app.entrypoint
async def entrypoint(payload):
    message = payload.get("prompt", "")
    # 非同期ストリーミングでメッセージを処理
    stream_messages = agent.stream_async(message)
    async for message in stream_messages:
        if "event" in message:
            yield message["event"]

Strands AgentsでHaikuを呼び出して、stream_asyncメソッドで非同期ストリーミング処理を行います。
yieldを使ってイベントを逐次返すことで、ストリーム応答を返却しています。

Lambda関数の実装

次に、Lambda関数の実装です。
コード全体はGitHubリポジトリを参照していただくとして、ここではポイントを抜粋して説明します。

awslambda.streamifyResponseの使用

Lambda Response Streamingを使うためには、awslambda.streamifyResponseでハンドラをラップする必要があります。

import type { APIGatewayProxyEvent, Context } from 'aws-lambda';

export const handler = awslambda.streamifyResponse(
  async (
    event: APIGatewayProxyEvent,
    responseStream: NodeJS.WritableStream,
    context: Context
  ) => {
    // ストリーミング処理
  }
);

通常のLambda関数と違って、responseStreamという書き込み可能なストリームが渡されるのが特徴です。

Server-Sent Events (SSE) 形式のレスポンス設定

クライアントにストリーミングレスポンスを返すために、Server-Sent Events (SSE) 形式を使います。

const responseMetadata = {
  statusCode: 200,
  headers: {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'X-Accel-Buffering': 'no',
  },
};

const httpStream = awslambda.HttpResponseStream.from(
  responseStream as Writable,
  responseMetadata
);

Content-Type: text/event-streamを設定することで、SSE形式のレスポンスになって、ブラウザのEventSourceやFetch APIでストリーミングを受信できるようになります。

AgentCore Runtime SDKの使用

AgentCore Runtimeを呼び出すために、@aws-sdk/client-bedrock-agentcoreを使用します。

import {
  BedrockAgentCoreClient,
  InvokeAgentRuntimeCommand,
} from '@aws-sdk/client-bedrock-agentcore';

const agentCoreClient = new BedrockAgentCoreClient({
  region: process.env.AWS_REGION || 'us-west-2',
});

const invokeCommand = new InvokeAgentRuntimeCommand({
  agentRuntimeArn: agentCoreArn,
  runtimeSessionId: requestParams.sessionId,
  payload: new TextEncoder().encode(
    JSON.stringify({ prompt: requestParams.prompt })
  ),
  qualifier: 'DEFAULT',
});

const runtimeResponse = await agentCoreClient.send(invokeCommand);

レスポンスはruntimeResponse.responseとしてストリームで返却されます。

Stream パイプラインによるデータ転送

最後に、AgentCore Runtimeからのストリームを、そのままクライアントへ転送します。

import { promisify } from 'util';
import { pipeline as streamPipeline, Readable } from 'stream';

const asyncPipeline = promisify(streamPipeline);

// AgentCore Runtimeからのストリーム → クライアントへのストリーム
await asyncPipeline(runtimeResponse.response as Readable, httpStream);

完全なコードはGitHubリポジトリをご参照ください。
今回はそのままAgentCoreからのレスポンスをそのまま返却していますが、Lambda関数で加工したり任意のフォーマットに変換してフロントエンドに返却することも可能です。

CDKでのデプロイ

AgentCore + Lambda関数をAWS CDKでデプロイします。
今回のCDK実装では、AgentCore RuntimeとLambda関数を一緒にデプロイできるようにしています！

CDKスタックの実装

コードの全体は下記となります。

import * as cdk from "aws-cdk-lib";
import * as path from "path";
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as iam from "aws-cdk-lib/aws-iam";
import * as nodejs from "aws-cdk-lib/aws-lambda-nodejs";
import * as logs from "aws-cdk-lib/aws-logs";
import * as agentcore from "@aws-cdk/aws-bedrock-agentcore-alpha";
import { Construct } from "constructs";

export class CdkStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // =====================
    // AgentCore Runtime
    // =====================

    // ローカルのコードをビルド
    const agentRuntimeArtifact = agentcore.AgentRuntimeArtifact.fromAsset(
      path.join(__dirname, "../agent"),
    );

    // ランダムなAgent名を生成
    const randomSuffix = Math.random().toString(36).substring(2, 8);
    const runtimeName = `agentcore_runtime_${randomSuffix}`;

    // AgentCore Runtime を作成
    const runtime = new agentcore.Runtime(this, "AgentCoreRuntime", {
      runtimeName: runtimeName,
      agentRuntimeArtifact: agentRuntimeArtifact,
      description: "Strands Agent deployed via CDK L2 Construct",
    });

    // Bedrock の呼び出し権限を追加
    runtime.addToRolePolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          "bedrock:InvokeModel",
          "bedrock:InvokeModelWithResponseStream",
        ],
        resources: [`arn:aws:bedrock:${this.region}::foundation-model/*`],
      }),
    );

    // =====================
    // Lambda 関数（AgentCore Proxy）
    // =====================

    // Lambda 関数用のロググループ
    const lambdaLogGroup = new logs.LogGroup(this, "LambdaLogGroup", {
      logGroupName: "/aws/lambda/agentcore-proxy",
      retention: logs.RetentionDays.ONE_WEEK,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Lambda関数の作成
    const proxyFunction = new nodejs.NodejsFunction(
      this,
      "AgentCoreProxyFunction",
      {
        functionName: "agentcore-proxy",
        entry: path.join(__dirname, "../lambda/agentcore-proxy/index.ts"),
        handler: "handler",
        runtime: lambda.Runtime.NODEJS_24_X,
        timeout: cdk.Duration.minutes(15),
        memorySize: 512,
        environment: {
          AGENT_ARN: runtime.agentRuntimeArn,
        },
        logGroup: lambdaLogGroup,
      },
    );

    // AgentCore Runtime呼び出し権限を付与
    proxyFunction.addToRolePolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ["bedrock-agentcore:InvokeAgentRuntime"],
        resources: [
          runtime.agentRuntimeArn,
          `${runtime.agentRuntimeArn}/runtime-endpoint/*`,
        ],
      }),
    );

    // =====================
    // Outputs
    // =====================

    new cdk.CfnOutput(this, "RuntimeName", {
      value: runtimeName,
      description: "Name of the AgentCore Runtime",
    });

    new cdk.CfnOutput(this, "RuntimeArn", {
      value: runtime.agentRuntimeArn,
      description: "ARN of the AgentCore Runtime",
    });

    new cdk.CfnOutput(this, "RuntimeId", {
      value: runtime.agentRuntimeId,
      description: "ID of the AgentCore Runtime",
    });

    new cdk.CfnOutput(this, "ProxyFunctionName", {
      value: proxyFunction.functionName,
      description: "AgentCore Proxy Lambda Function Name",
    });

    new cdk.CfnOutput(this, "ProxyFunctionArn", {
      value: proxyFunction.functionArn,
      description: "AgentCore Proxy Lambda Function ARN",
    });
  }
}

デプロイ

CDKでデプロイします。AgentCore RuntimeとLambda関数が一緒にデプロイされます！

npm install
cdk bootstrap  # 初回のみ
cdk deploy

デプロイが完了すると、Lambda関数名が出力されます。

 ✅  AgentCoreProxyStack

✨  Deployment time: 88.67s

Outputs:
AgentCoreProxyStack.ProxyFunctionArn = arn:aws:lambda:us-west-2:xxx:function:agentcore-proxy
AgentCoreProxyStack.ProxyFunctionName = agentcore-proxy
AgentCoreProxyStack.RuntimeArn = arn:aws:bedrock-agentcore:us-west-2:xxx:runtime/agentcore_runtime_xxx
AgentCoreProxyStack.RuntimeId = agentcore_runtime_xxx
AgentCoreProxyStack.RuntimeName = agentcore_runtime_xxx
Stack ARN:
arn:aws:cloudformation:us-west-2:xxx:stack/AgentCoreProxyStack/xxx

このLambda関数名（agentcore-proxy）をメモしておきます。
次のAPI Gateway設定で使用します。

API Gatewayの手動設定

API Gatewayの設定で、ストリーム応答を有効にしてLambda関数と統合します。

REST APIの作成

まず、API Gatewayコンソールを開いて、REST APIを作成します。

1. コンソール上でAPI Gatwayのサービスページを開きます
2. 「APIを作成」をクリック
   
3. 「REST API」を選択し、「構築」をクリック
   
4. 以下を設定します。
   • 新しいAPI: 新しいAPI
   • API名: AgentCoreProxyAPI
   • エンドポイントタイプ: リージョン
   • Security policy:任意のポリシー（今回はSecurityPolicy_TLS13_1_2_2021_06を選択しました）
5. 「APIを作成」をクリック
   

リソースとメソッドの作成

次に、/invokeエンドポイントを作成します。

1. 「リソース」→「リソースを作成」をクリック
   
2. リソース名: invoke
3. 「リソースを作成」をクリック
   

もし異なるドメインからアクセスする場合は、CORSを有効にしましょう。

Lambda統合の設定

次にLambda関数との統合設定をします。

1. /invokeリソースを選択
2. 「メソッドを作成」をクリック
   
3. 以下を設定します：
   • メソッドタイプ: POST
   • 統合タイプ: Lambda関数
   • Lambdaプロキシ統合: オン
   • レスポンス転送モード:ストリーム
     • 今回のアップデートで追加された項目です
   • Lambda関数: agentcore-proxy
   • Lambdaリージョン: 任意のリージョン（今回ならus-west-2）
   • 統合のタイムアウト：900000
     • 今回はストリーミングレスポンスのため900000（15分）まで設定可能でした。
       • 900001に設定したらエラーが出ました。
4. 「メソッドを作成」をクリック
   

これでストリーム応答の設定が完了です！

APIのデプロイ

設定が完了したら、APIをデプロイします。

1. 「APIをデプロイ」をクリック
   
2. ステージ: 新しいステージ
3. ステージ名: test
4. 「デプロイ」をクリック
   

デプロイが完了すると、呼び出しURLが表示されます。

https://abc123xyz.execute-api.us-west-2.amazonaws.com/test

この呼び出しURLをメモしておきましょう。

動作確認

それでは、実際に動作を確認してみます！

curlでのテスト

まずは、シンプルにcurlでテストしてみます。

curl --no-buffer -X POST https://abc123xyz.execute-api.us-west-2.amazonaws.com/test/invoke \
  -H "Content-Type: application/json" \
  -d '{"prompt":"こんにちは、あなたは何ができますか？"}'

実際のレスポンス

ストリーミングで、以下のようなSSE形式のレスポンスが返ってきます！

data: {"event": {"messageStart": {"role": "assistant"}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "こ"}, "contentBlockIndex": 0}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "んにちは"}, "contentBlockIndex": 0}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "！"}, "contentBlockIndex": 0}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "私は"}, "contentBlockIndex": 0}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "、"}, "contentBlockIndex": 0}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "天"}, "contentBlockIndex": 0}}}

...

下記は実際に実行した動画です。（URLのところは変数に入れてマスクしていますが、API GatewayのURLを設定しています）

ストリーム応答でしっかり返ってきました！割と簡単に手軽に設定できましたね！

おわりに

今回は、API Gatewayのストリーム応答とLambda関数を組み合わせて、AgentCore Runtimeをラップしたストリーム応答処理を実装してみました！

AgentCoreとAPI Gateway + Lambdaを組み合わせることで、URLの抽象化だけでなく、認証・認可やレート制限、WAFとの統合などのAPI管理機能が簡単に使えるのが便利になるポイントと思います。CloudFrontやLambda Function URLsも選択肢としてありますが、APIとして管理するという観点では、API GatewayとLambdaをAgentCoreに組み合わせる選択肢もあるのかなと思いました！

本記事が少しでも参考になりましたら幸いです。最後までご覧いただきありがとうございました！

補足

ストリーム応答の制限事項

API Gatewayのスリリーム応答には、制限事項がありますので、
使用される際には公式ドキュメントをご確認ください。

https://docs.aws.amazon.com/apigateway/latest/developerguide/response-transfer-mode.html