Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

构建您的第一个经过身份验证的代理

本入门教程将引导您使用 Amazon Bedrock Identity 从头开始构建经过 AgentCore 身份验证的完整代理，并将帮助您开始在代理应用程序中实现身份功能。您将学习如何设置开发环境、使用 Cognito 创建身份验证基础架构、将代理部署到 AgentCore Runtime 以及测试完整的身份验证工作流程。

在本教程结束时，您将拥有一个完全部署的代理，该代理可以通过 OAuth2 流程对用户进行身份验证，安全地获取访问令牌，并演示完整的身份管理生命周期。您的代理将使用适当的 IAM 权限在 AgentCore Runtime 上运行，从而创建一个测试实验室环境，您可以在其中演示和测试集成能力。

先决条件

在开始之前，请确保您满足以下条件：

本教程要求您拥有 OAuth 2.0 授权服务器。如果您没有，步骤 1 将使用 Amazon Cognito 用户池为您创建一个用户池。如果您的 OAuth 2.0 授权服务器配置了客户端 ID、客户端密钥和用户，则可以继续执行步骤 2。该授权服务器将充当资源凭证提供者，代表向代理授予出站 OAuth 2.0 访问令牌的权限。

安装 SDK 和依赖项

为本指南创建一个文件夹，创建 Python 虚拟环境，然后安装 AgentCore SDK 和 Amazon Python 开发工具包 (boto3)。

mkdir agentcore-identity-quickstart
cd agentcore-identity-quickstart
python3 -m venv .venv
source .venv/bin/activate
pip install bedrock-agentcore boto3 strands-agents pyjwt

第 1 步：创建 Cognito 用户池（可选）

本教程需要 OAuth 2.0 授权服务器。如果您没有可供测试的，或者您想将测试与授权服务器分开，则此脚本将使用您的 Amazon 证书设置一个 Amazon Cognito 实例供您用作授权服务器。该脚本将创建：

删除 Cognito 用户池 AgentCoreIdentityQuickStartPool 也会删除关联的 client_id 和用户。

您可以选择将此脚本另存为 create_cognito.sh 并从命令行执行，也可以将脚本粘贴到命令行中。

#!/bin/bash

REGION=$(aws configure get region)

# Create user pool
USER_POOL_ID=$(aws cognito-idp create-user-pool \
  --pool-name AgentCoreIdentityQuickStartPool \
  --query 'UserPool.Id' \
  --no-cli-pager \
  --output text)

# Create user pool domain
DOMAIN_NAME="agentcore-quickstart-$(LC_ALL=C tr -dc 'a-z0-9' < /dev/urandom | head -c 5)"
aws cognito-idp create-user-pool-domain \
  --domain $DOMAIN_NAME \
  --no-cli-pager \
  --user-pool-id $USER_POOL_ID > /dev/null

# Create user pool client with secret and hosted UI settings
CLIENT_RESPONSE=$(aws cognito-idp create-user-pool-client \
  --user-pool-id $USER_POOL_ID \
  --client-name AgentCoreQuickStart \
  --generate-secret \
  --allowed-o-auth-flows "code" \
  --allowed-o-auth-scopes "openid" "profile" "email" \
  --allowed-o-auth-flows-user-pool-client \
  --supported-identity-providers "COGNITO" \
  --query 'UserPoolClient.{ClientId:ClientId,ClientSecret:ClientSecret}' \
  --output json)

CLIENT_ID=$(echo $CLIENT_RESPONSE | jq -r '.ClientId')
CLIENT_SECRET=$(echo $CLIENT_RESPONSE | jq -r '.ClientSecret')

# Generate random username and password
USERNAME="AgentCoreTestUser$(printf "%04d" $((RANDOM % 10000)))"
PASSWORD="$(LC_ALL=C tr -dc 'A-Za-z0-9!@#$%^&*()_+-=[]{}|;:,.<>?' < /dev/urandom | head -c 16)$(LC_ALL=C tr -dc '0-9' < /dev/urandom | head -c 1)"

# Create user with permanent password
aws cognito-idp admin-create-user \
  --user-pool-id $USER_POOL_ID \
  --username $USERNAME \
  --output text > /dev/null

aws cognito-idp admin-set-user-password \
  --user-pool-id $USER_POOL_ID \
  --username $USERNAME \
  --password $PASSWORD \
  --output text > /dev/null \
  --permanent

# Get region

ISSUER_URL="https://cognito-idp.$REGION.amazonaws.com/$USER_POOL_ID/.well-known/openid-configuration"
HOSTED_UI_URL="https://$DOMAIN_NAME.auth.$REGION.amazoncognito.com"

# Output results
echo "User Pool ID: $USER_POOL_ID"
echo "Client ID: $CLIENT_ID"
echo "Client Secret: $CLIENT_SECRET"
echo "Issuer URL: $ISSUER_URL"
echo "Hosted UI URL: $HOSTED_UI_URL"
echo "Test User: $USERNAME"
echo "Test Password: $PASSWORD"

echo ""
echo "# Copy and paste these exports to set environment variables for later use:"
echo "export USER_POOL_ID='$USER_POOL_ID'"
echo "export CLIENT_ID='$CLIENT_ID'"
echo "export CLIENT_SECRET='$CLIENT_SECRET'"
echo "export ISSUER_URL='$ISSUER_URL'"
echo "export HOSTED_UI_URL='$HOSTED_UI_URL'"
echo "export COGNITO_USERNAME='$USERNAME'"
echo "export COGNITO_PASSWORD='$PASSWORD'"

第 2 步：创建凭证提供商

凭证提供商是您的代理访问外部服务的方式。创建凭证提供商，并使用授权服务器的 OAuth 2.0 客户端对其进行配置。

如果您使用自己的授权服务器，请设置环境变量 ISSUER_URLCLIENT_ID，并CLIENT_SECRET使用授权服务器中的相应值。如果您使用前面的脚本使用 Cognito 为您创建授权服务器，请将输出中的 EXPORT 语句复制到终端以设置环境变量。

您的代理代码将使用该凭证提供商来获取访问令牌，以代表您的用户行事。

第 2.5 步：将回调 URL 添加到 OAuth 2.0 授权服务器

为防止未经授权的重定向，请将从 OAuth 2.0 授权服务器检索到CreateOauth2CredentialProvider的回调 URL GetOauth2CredentialProvider 添加到您的 OAuth 2.0 授权服务器中。

如果您使用前面的脚本使用 Cognito 创建授权服务器，请将输出中的 EXPORT 语句复制到您的终端以设置环境变量，并使用 OAuth2 凭据提供商回调 URL 更新 Cognito 用户池客户端。

#!/bin/bash
aws cognito-idp update-user-pool-client \
    --user-pool-id $USER_POOL_ID \
    --client-id $CLIENT_ID \
    --client-name AgentCoreQuickStart \
    --allowed-o-auth-flows "code" \
    --allowed-o-auth-scopes "openid" "profile" "email" \
    --allowed-o-auth-flows-user-pool-client \
    --supported-identity-providers "COGNITO" \
    --callback-urls "$OAUTH2_CALLBACK_URL"

第 3 步：创建启动 OAuth 2.0 流程的示例代理

在此步骤中，我们将创建一个代理，该代理启动 OAuth 2.0 授权流程，以获取代币以用户的名义行事。为简单起见，代理不会代表用户实际拨打外部服务，但会向我们证明其已获得同意代表我们的测试用户行事。

代理代码

创建一个名为agentcoreidentityquickstart.py的文件并保存此代码。

"""
AgentCore Identity Outbound Token Agent

This agent demonstrates the USER_FEDERATION OAuth 2.0 flow.

It handles the OAuth 2.0 user consent flow and inspects the resulting OAuth 2.0 access token.
"""

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.identity import requires_access_token
import asyncio
import jwt
import logging

app = BedrockAgentCoreApp()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def decode_jwt(token):
    try:
        decoded = jwt.decode(token, options={"verify_signature": False})
        return decoded
    except Exception as e:
        return {"error": f"Error decoding JWT: {str(e)}"}

class StreamingQueue:
    def __init__(self):
        self.finished = False
        self.queue = asyncio.Queue()

    async def put(self, item):
        await self.queue.put(item)

    async def finish(self):
        self.finished = True
        await self.queue.put(None)

    async def stream(self):
        while True:
            item = await self.queue.get()
            if item is None and self.finished:
                break
            yield item

queue = StreamingQueue()

async def handle_auth_url(url):
    await queue.put(f"Authorization URL, please copy to your preferred browser: {url}")

@requires_access_token(
    provider_name="AgentCoreIdentityQuickStartProvider",
    scopes=["openid"],
    auth_flow="USER_FEDERATION",
    on_auth_url=handle_auth_url, # streams authorization URL to client
    force_authentication=True,
    callback_url='insert_oauth2_callback_url_for_session_binding',
)
async def introspect_with_decorator(*, access_token: str):
    """Introspect token using decorator"""
    logger.info("Inside introspect_with_decorator - decorator succeeded")
    await queue.put({
        "message": "Successfully received an access token to act on behalf of your user!",
        "token_claims": decode_jwt(access_token),
        "token_length": len(access_token),
        "token_preview": f"{access_token[:50]}...{access_token[-10:]}"
    })
    await queue.finish()

@app.entrypoint
async def agent_invocation(payload, context):
    """Handler that uses only the decorator approach"""
    logger.info("Agent invocation started")

    # Start the agent task and immediately begin streaming
    task = asyncio.create_task(introspect_with_decorator())

    # Stream items as they come in
    async for item in queue.stream():
        yield item

    # Wait for task completion
    await task

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

步骤 4：将代理部署到 AgentCore 运行时

我们将在 AgentCore Runtime 上托管这个代理。我们可以使用 AgentCore CLI 轻松完成此操作。

在您的终端上安装 AgentCore CLI 并创建 Python Strands 代理项目。显式选项创建了基于代码的代理而不是工具：

npm install -g @aws/agentcore
agentcore create --name IdentityQuickstart --language Python --framework Strands \
  --model-provider Bedrock --memory none

将代理脚本复制到项目的代理目录中，替换默认代理：

cp agentcoreidentityquickstart.py IdentityQuickstart/app/IdentityQuickstart/main.py

将 JWT 依赖项添加到生成的项目中：

cd IdentityQuickstart/app/IdentityQuickstart
uv add pyjwt
cd ../..

然后部署你的项目：

agentcore deploy

CLI 合成 Amazon CDK 堆栈并将您的代理部署到运行时。 AgentCore 这大约需要 2-3 分钟。

更新代理的 IAM 政策，以便能够访问代币库和客户端密钥

AgentCore CLI 会在部署期间创建代理的执行角色，但该角色不会自动包含令牌库访问权限。您需要附加额外的策略以允许代理在运行时检索 OAuth 2.0 令牌。

此脚本从 Amazon CLI 检索您的账户和区域，从 CloudFormation 堆栈中查找代理的执行角色，并附加相应的策略。您可以复制并粘贴此脚本，也可以将其保存到文件中并执行。

#!/bin/bash

# Get account and region from AWS CLI
AWS_ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
REGION=$(aws configure get region)

# Get execution role from CloudFormation stack outputs
EXECUTION_ROLE=$(aws cloudformation describe-stack-resources \
  --stack-name AgentCore-IdentityQuickstart-prod \
  --query "StackResources[?ResourceType=='AWS::IAM::Role'].PhysicalResourceId" \
  --output text | head -1)

echo "Parsed values:"
echo "Execution Role: $EXECUTION_ROLE"
echo "Account: $AWS_ACCOUNT"
echo "Region: $REGION"

# Create the policy document with proper variable substitution
cat > agentcore-identity-policy.json << EOF
{
"Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AccessTokenVault",
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:GetResourceOauth2Token",
        "secretsmanager:GetSecretValue"
      ],
      "Resource": ["arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:workload-identity-directory/default/workload-identity/*",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:token-vault/default/oauth2credentialprovider/AgentCoreIdentityQuickStartProvider",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:workload-identity-directory/default",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:token-vault/default",
        "arn:aws:secretsmanager:$REGION:$AWS_ACCOUNT:secret:bedrock-agentcore-identity!default/oauth2/AgentCoreIdentityQuickStartProvider*"
      ]
    }
  ]
}
EOF

# Create the policy
POLICY_ARN=$(aws iam create-policy \
    --policy-name AgentCoreIdentityQuickStartPolicy$(LC_ALL=C tr -dc '0-9' < /dev/urandom | head -c 4) \
    --policy-document file://agentcore-identity-policy.json \
    --query 'Policy.Arn' \
    --output text)

# Extract role name from ARN and attach policy
ROLE_NAME=$(echo $EXECUTION_ROLE | awk -F'/' '{print $NF}')
aws iam attach-role-policy \
    --role-name $ROLE_NAME \
    --policy-arn $POLICY_ARN

echo "Policy created and attached: $POLICY_ARN"

# Cleanup
rm agentcore-identity-policy.json

第 5 步：调用代理

现在一切都设置好了，你可以调用代理了。在本演示中，我们将使用agentcore invoke命令和我们的 IAM 证书。使用 IAM 身份验证时，我们需要传递--user-id和--session-id参数。

agentcore invoke "TestPayload" --runtime IdentityQuickstart --user-id "SampleUserID" --session-id "ALongThirtyThreeCharacterMinimumSessionIdYouCanChangeThisAsYouNeed"

然后，代理将返回您的agentcore invoke命令的 URL。将该 URL 复制并粘贴到您的首选浏览器中，然后您将被重定向到授权服务器的登录页面。--user-id参数是您向 Id AgentCore entity 出示的用户 ID。--session-id参数是会话 ID，其长度必须至少为 33 个字符。

当浏览器出现提示时，在授权服务器上输入用户的用户名和密码，或使用您配置的首选身份验证方法。如果您使用步骤 1 中的脚本创建 Cognito 实例，则可以从终端历史记录中检索该实例。

您的浏览器应重定向到您配置的 OAuth2 回调 URL，该网址处理会话绑定流程。确保您的 OAuth2 回调服务器提供明确的成功和错误响应以指示授权状态。

调试

如果您遇到任何错误或意外行为，Amazon CloudWatch 日志中会捕获代理的输出。运行agentcore deploy后会提供日志尾随命令。

清理

完成后，运行，agentcore remove all然后agentcore deploy从项目目录中删除已部署的 AgentCore 运行时资源。然后删除 Amazon Cognito 用户池，分离并删除您创建的 IAM 政策，并删除凭证提供商。

安全最佳实践

处理身份信息时：