Serverless Operations, inc

>_cd /blog/id_m9jf_8e_bdu

title

Amazon Bedrock AgentCore Identity (2) : Agentから外部サービスへ2LO（M2M）認証でアクセスする

前回の記事では Agentを呼び出す際の認証を、 IAM Signv4 と Cognito を用いた OAuth と２種類の方法で試しました。この記事ではAgent が外部サービスにアクセスを行う（API呼び出しなど）の認証を行っていきます。

2LO と 3LO

Outbound で外部サービスのトークンを取得する方法には、大きく2つの OAuth 2.0 フローがあります。AgentCore Identity は、client credentials grant（2LO とも呼ばれる）と、authorization code grant（3LO とも呼ばれる）の両方をネイティブにサポートしています。

2LO（Two-Legged OAuth / M2M: Machine to Machine）

登場人物が2者(エージェントと、アクセス先のサービス)のフローです。ユーザーの同意画面は登場しません。2LO フローでは、エージェントはユーザーの操作を介さずに、リソースサーバーに対して自分自身として直接認証します。

「エージェント自身の権限」で動くのが本質です。たとえば、社内 API を叩く、自社のサービスアカウントとしてデータを取得する、といった特定のユーザーに紐づかないアクセスに向いています。裏側では OAuth の client credentials grant が使われます。

3LO（Three-Legged OAuth / ユーザー委譲）

登場人物が3者（ユーザー、エージェント、外部サービス）のフローです。ここに「ユーザーの同意」というステップが加わります。3LO フローは、外部サービスにあるユーザー固有のデータへアクセスするために、ユーザーの明示的な同意と認可を可能にします。

「エージェントが、ユーザーに代わって、ユーザーのデータにアクセスする」のが本質です。例えば「エージェントが私にかわって Google Drive を読む」などがそれにあたります。流れとしては、エージェントが同意用の URL を生成し、ユーザーがブラウザでその URL を開いて「許可」を押すと、初めてエージェントがそのユーザーのデータにアクセスできるようになります。裏側では OAuth の authorization code grant が使われます。

2LO（M2M）

3LO（ユーザー委譲）

登場人物

エージェント ↔ サービス

ユーザー ↔ エージェント ↔ サービス

ユーザー同意

不要

必要（初回のみ、以降はVault活用）

誰として動くか

エージェント自身

ユーザーの代理

OAuth grant

client credentials

authorization code

auth_flow の指定

M2M

USER_FEDERATION

向く用途

社内API・サービスアカウント

個人のDrive/メール/カレンダー

さっそくやってみる

ではこの記事ではまず2LO (M2M)を試してみます。

この記事では第一回の記事で作成されたCognito ユーザープールを再利用しますので、前回の手順を終わらせておいてください。

1. Cognito の環境準備

setup_m2m.sh を作成します。

#!/bin/bash
set -e

# 第一回で作ったユーザープールIDを指定
export POOL_ID=us-west-2_Yskjectap
export REGION=us-west-2

# ドメインは AWS 全体で一意である必要があるため、サフィックスをランダム化
export DOMAIN_PREFIX="agentcore-m2m-$(head /dev/urandom | tr -dc a-z0-9 | head -c 8)"

# 1. リソースサーバーと独自スコープを作成
#    identifier: API を表す識別子 / scope: read アクセスを表す独自スコープ
aws cognito-idp create-resource-server \
  --user-pool-id "$POOL_ID" \
  --identifier "my-api" \
  --name "MyProtectedAPI" \
  --scopes ScopeName=read,ScopeDescription="Read access" \
  --region "$REGION" > /dev/null

# 2. ユーザープールドメインを作成（トークンエンドポイント有効化に必須）
aws cognito-idp create-user-pool-domain \
  --user-pool-id "$POOL_ID" \
  --domain "$DOMAIN_PREFIX" \
  --region "$REGION" > /dev/null

# 3. M2M 用アプリクライアントを作成（client_credentials 対応・シークレット付き）
CLIENT_JSON=$(aws cognito-idp create-user-pool-client \
  --user-pool-id "$POOL_ID" \
  --client-name "M2MClient" \
  --generate-secret \
  --allowed-o-auth-flows "client_credentials" \
  --allowed-o-auth-scopes "my-api/read" \
  --allowed-o-auth-flows-user-pool-client \
  --region "$REGION")

export M2M_CLIENT_ID=$(echo "$CLIENT_JSON" | jq -r '.UserPoolClient.ClientId')
export M2M_CLIENT_SECRET=$(echo "$CLIENT_JSON" | jq -r '.UserPoolClient.ClientSecret')

# トークンエンドポイント（ドメインベース。cognito-idp のパスとは別物）
export TOKEN_ENDPOINT="https://${DOMAIN_PREFIX}.auth.${REGION}.amazoncognito.com/oauth2/token"

echo "----------------------------------------"
echo "Domain prefix : $DOMAIN_PREFIX"
echo "Token endpoint: $TOKEN_ENDPOINT"
echo "M2M Client ID : $M2M_CLIENT_ID"
echo "M2M Secret    : $M2M_CLIENT_SECRET"
echo "Scope         : my-api/read"
echo "----------------------------------------"

実行します。

chmod +x setup_m2m.sh
source setup_m2m.sh

出力される以下の値をメモしておきます。

----------------------------------------
Domain prefix : agentcore-m2m-uy7ad0pl
Token endpoint: https://agentcore-m2m-uy7ad0pl.auth.us-west-2.amazoncognito.com/oauth2/token
M2M Client ID : 5od8etreqjlk8qpeq0bnpqbk3b
M2M Secret    : qiaoj34u6iethoo7frs1rfn0bqvmv3o8384aqln4a224qpbou27
Scope         : my-api/read
----------------------------------------

ではトークンの取得をテストします。

curl -s -X POST "$TOKEN_ENDPOINT" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -u "${M2M_CLIENT_ID}:${M2M_CLIENT_SECRET}" \
  -d "grant_type=client_credentials&scope=my-api/read" | jq

以下の様な値が出れば成功です。

{
  "access_token": "eyJraWQiOiJUUFJya1dNQXd5QUZPVDZBVnVCZC9oOGZJRVBDb0Rnc1R3ZXI5Qk1kaDZVPSIsImFsZyI6IlJTMjU2In0.eyJzdWIiOiI1b2Q4ZXRyZXFqbGs4cXBlcTBibnBxYmszYiIsInRva2VuX3VzZSI6ImFjY2VzcyIsInNjb3BlIjoibXktYXBpL3JlYWQiLCJhdXRoX3RpbWUiOjE3ODc0NjA3NjUsImlzcyI6Imh0dHBzOi8vY29nbml0by1pZHAudXMtd2VzdC0yLmFtYXpvbmF3cy5jb20vdXMtd2VzdC0yX1lza2plY3RhcCIsImV4cCI6MTc4NzQ2NDM2NSwiaWF0IjoxNzg3NDYwNzY1LCJ2ZXJzaW9uIjoyLCJqdGkiOiJlOGYxYWE5OS04YmQ5LTQ2N2QtODFlYS1jYTdlYjRjZjAwMmQiLCJjbGllbnRfaWQiOiI1b2Q4ZXRyZXFqbGs4cXBlcTBibnBxYmszYiJ9.zxyeJvB4FDOAwJtjW4C3wpNzOTB3mQkXe6glqqcOLsBbuhyYLKi-aXf7XF1gMvvT6cUQ-yExWjVsG_TNuq5Cb24C74ou2F5SHmGkKSm9tN1w3jen5xyqM2n2c9E5RbXwd8lpeWGUxUBdYdom-ViO-HOK_lPGwyxdTfMIlkQVEVHN6BCHsAqIpM9hwbofP4ixRxqNMyAu2th7PnUzVqZ0Anoww1txVTG9L_DgSsN6rsWRTHZGz9Z3r3SFu-oOM9L_KmBilIz69Y0kcDwvql8N2R3g0jfKzpB0IG3xFwwNDMafAP2oTb0b2xluiS72SfNmHqt2ORiccrUqG8vk8SI3gQ",
  "expires_in": 3600,
  "token_type": "Bearer"
}

2. テスト用APIの準備

protected_api.py を作成します。

"""
保護された自前 API（2LO デモ用）
Cognito が発行した M2M アクセストークンを検証し、正しければデータを返す。
"""
import json
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

import jwt  # PyJWT
from jwt import PyJWKClient

# ---- 設定：自分の環境に合わせる ----
REGION = "us-west-2"
POOL_ID = "us-west-2_Yskjectap"
EXPECTED_SCOPE = "my-api/read"
EXPECTED_CLIENT_ID = "5od8etreqjlk8qpeq0bnpqbk3b"  # M2M Client ID
# ------------------------------------

ISSUER = f"https://cognito-idp.{REGION}.amazonaws.com/{POOL_ID}"
JWKS_URL = f"{ISSUER}/.well-known/jwks.json"

# 署名検証用の公開鍵（JWKS）を取得するクライアント
jwks_client = PyJWKClient(JWKS_URL)


def verify_token(token: str) -> dict:
    """トークンの署名・issuer・有効期限・スコープ・client_id を検証する"""
    signing_key = jwks_client.get_signing_key_from_jwt(token)
    claims = jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],
        issuer=ISSUER,
        # M2M アクセストークンには aud が無いので audience 検証は無効化
        options={"verify_aud": False},
    )
    # スコープ検証（独自スコープが含まれているか）
    scopes = claims.get("scope", "").split()
    if EXPECTED_SCOPE not in scopes:
        raise ValueError(f"required scope '{EXPECTED_SCOPE}' not present")
    # client_id 検証（想定した M2M クライアントか）
    if claims.get("client_id") != EXPECTED_CLIENT_ID:
        raise ValueError("client_id mismatch")
    return claims


class Handler(BaseHTTPRequestHandler):
    def _send(self, code: int, body: dict):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(payload)

    def do_GET(self):
        if self.path != "/data":
            self._send(404, {"error": "not found"})
            return

        auth = self.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            self._send(401, {"error": "missing bearer token"})
            return

        token = auth[len("Bearer "):]
        try:
            claims = verify_token(token)
        except Exception as e:
            self._send(403, {"error": f"token verification failed: {e}"})
            return

        # 検証成功：保護されたデータを返す
        self._send(200, {
            "message": "Hello from the protected API!",
            "authorized_client": claims.get("client_id"),
            "scope": claims.get("scope"),
            "server_time": int(time.time()),
        })

    def log_message(self, *args):
        pass  # アクセスログを抑制


if __name__ == "__main__":
    port = 8081
    print(f"Protected API listening on http://localhost:{port}/data")
    HTTPServer(("0.0.0.0", port), Handler).serve_forever()

実行に必要なライブラリをインストールします。

pip install "PyJWT[crypto]"

準備が出来たらAPIを起動します。

python protected_api.py

ではこのままの状態でもう一つターミナルを開いて、以下の環境変数を設定します。それぞれの値は先ほどメモしたものに置き換えてください。

export TOKEN_ENDPOINT="https://agentcore-m2m-uy7ad0pl.auth.us-west-2.amazoncognito.com/oauth2/token"
export M2M_CLIENT_ID="5od8etreqjlk8qpeq0bnpqbk3b"
export M2M_CLIENT_SECRET="qiaoj34u6iethoo7frs1rfn0bqvmv3o8384aqln4a224qpbou27"
export REGION="us-west-2"
export POOL_ID="us-west-2_Yskjectap"
export DOMAIN_PREFIX="agentcore-m2m-uy7ad0pl"

ではトークンを取得してみます。

# トークン取得
export TOKEN=$(curl -s -X POST "$TOKEN_ENDPOINT" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -u "${M2M_CLIENT_ID}:${M2M_CLIENT_SECRET}" \
  -d "grant_type=client_credentials&scope=my-api/read" | jq -r '.access_token')

echo "len=${#TOKEN}"   # 今度は数百の数字になるはず

# トークン有りで叩く → 200
curl -i http://localhost:8081/data -H "Authorization: Bearer ${TOKEN}"

以下の様な出力がされれば完成です。

len=843
HTTP/1.0 200 OK
Server: BaseHTTP/0.6 Python/3.12.3
Date: Sun, 23 Aug 2026 05:00:24 GMT
Content-Type: application/json

{"message": "Hello from the protected API!", "authorized_client": "5od8etreqjlk8qpeq0bnpqbk3b", "scope": "my-api/read", "server_time": 1787461224}

3. AgentCore Identity の設定

では２番目のターミナルで以下を実行します。

aws bedrock-agentcore-control create-oauth2-credential-provider \
  --name "cognito-m2m-provider" \
  --credential-provider-vendor "CustomOauth2" \
  --oauth2-provider-config-input "{
    \"customOauth2ProviderConfig\": {
      \"clientId\": \"${M2M_CLIENT_ID}\",
      \"clientSecret\": \"${M2M_CLIENT_SECRET}\",
      \"oauthDiscovery\": {
        \"authorizationServerMetadata\": {
          \"issuer\": \"https://cognito-idp.${REGION}.amazonaws.com/${POOL_ID}\",
          \"authorizationEndpoint\": \"https://${DOMAIN_PREFIX}.auth.${REGION}.amazoncognito.com/oauth2/authorize\",
          \"tokenEndpoint\": \"${TOKEN_ENDPOINT}\",
          \"responseTypes\": [\"token\"]
        }
      }
    }
  }" \
  --region "${REGION}"

もどってきたARNの環境変数にセットします。

export CRED_PROVIDER_NAME="cognito-m2m-provider"
grep -q 'CRED_PROVIDER_NAME' env.sh || echo 'export CRED_PROVIDER_NAME="cognito-m2m-provider"' >> env.sh

これまでは手で curl して取っていた M2M トークン取得していましたが、AgentCore Identity に丸ごと任せるコードに書き換えます。エージェントは「トークンをくれ」と言うだけで、取得・キャッシュ・注入は SDK と AgentCore がやってくれます。

agentcore_m2m.py を作成します。

"""
2LO Outbound デモ用エージェント。
AgentCore Identity から M2M トークンを取得し、保護 API を叩く。
"""
import urllib.request
import json

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.identity.auth import requires_access_token

app = BedrockAgentCoreApp()

# 保護 API のエンドポイント（Step 2 のサーバー）
PROTECTED_API_URL = "http://localhost:8081/data"


@requires_access_token(
    provider_name="cognito-m2m-provider",  # Step 3 で登録した名前
    scopes=["my-api/read"],                # 独自スコープ
    auth_flow="M2M",                       # 2LO（client credentials）
)
async def call_protected_api(*, access_token: str) -> str:
    """AgentCore から注入された M2M トークンで保護 API を叩く"""
    req = urllib.request.Request(
        PROTECTED_API_URL,
        headers={"Authorization": f"Bearer {access_token}"},
    )
    with urllib.request.urlopen(req) as resp:
        return resp.read().decode()


@app.entrypoint
async def invoke(payload):
    """エントリポイント：保護 API を叩いて結果を返す"""
    result = await call_protected_api(access_token="")
    return {"api_response": json.loads(result)}


if __name__ == "__main__":
    app.run()

次にこのAgentを実行する run_local.py を実行します。

import asyncio
from agentcore_m2m import call_protected_api

async def main():
    # access_token="" で呼ぶ → AgentCore が本物のトークンを注入
    result = await call_protected_api(access_token="")
    print("=== 保護 API からの応答 ===")
    print(result)

if __name__ == "__main__":
    asyncio.run(main())
Created a workload identity
Created an user id
=== 保護 API からの応答 ===
{"message": "Hello from the protected API!", "authorized_client": "5od8etreqjlk8qpeq0bnpqbk3b", "scope": "my-api/read", "server_time": 1787461820}

ログの Created a workload identity / Created an user id が成功しています。SDK がローカルテスト用にワークロード ID とユーザー ID を自動生成し、AgentCore Identity 経由でトークンを取得しました。そして保護 API から Hello from the protected API! が 200 で返ってきています。このコードにトークンも Client Secret も一切書いていない状態でcall_protected_api(access_token="") と空で呼んだにもかかわらず、本物のトークンを取得し置き換えています。手で curl していた資格情報がコードに含まれていなくても動的に取得して実行されています。

少しややこしいので以下で整理してみます。

curl 手動版

AgentCore 版

トークン取得

自分で /oauth2/token に POST

@requires_access_token が裏で取得

Client Secret

コマンド/コードに登場

Secrets Manager に隔離（MANAGED）

トークンの扱い

変数に入れて手で運ぶ

引数に自動注入、コードに現れない

キャッシュ/更新

都度取り直し（60分で失効）

AgentCore が自動管理

LLM への露出

—

トークンがコンテキストに乗らない

コードの流れは以下になっています。

  1. @requires_access_token が「cognito-m2m-provider でトークンをくれ」と AgentCore Identity に依頼
  2. AgentCore Identity が、そのプロバイダーに紐づく Secret をSecrets Manager から取り出す
  3. その Client Secret を使って Cognito の /oauth2/token に client credentials で問い合わせ
  4. 得たトークンを、関数の access_token 引数に注入
  5. 認証付きAPIにアクセス

Written by
編集部

亀田 治伸

Kameda Harunobu

  • Facebook->
  • X->
  • GitHub->

Share

Facebook->X->
Back
to list
<-
SERVERLESS
OPERATIONS