获取 OAuth 2.0 访问令牌

AgentCore Identity 使开发人员能够根据配置的 OAuth 2.0 凭据提供者获取 OAuth 令牌，用于用户委托的访问或机器对机器的身份验证。该服务将协调用户或应用程序与下游授权服务器之间的身份验证过程，并将检索和存储生成的令牌。一旦该令牌在 AgentCore 身份保管库中可用，授权代理就可以检索该令牌并使用它来授权对资源服务器的调用。例如，以下示例代码将检索代币以代表最终用户与 Google 云端硬盘互动。如需了解更多信息，请参阅使用 OAuth2 与 Google 云端硬盘集成，查看完整示例。

# Injects Google Access Token
@requires_access_token(
    # Uses the same credential provider name created above
    provider_name= "google-provider",
    # Requires Google OAuth2 scope to access Google Drive
    scopes= ["https://www.googleapis.com/auth/drive.metadata.readonly"],
    # Sets to OAuth 2.0 Authorization Code flow
    auth_flow= "USER_FEDERATION",
    # Prints authorization URL to console
    on_auth_url= lambda x: print("\nPlease copy and paste this URL in your browser:\n" + x),
    # If false, caches obtained access token
    force_authentication= False,
    callback_url='insert_oauth2_callback_url_for_session_binding',
)
async def write_to_google_drive(*, access_token: str):
    # Use the token to call Google Drive
    pass

# To invoke:
# asyncio.run(write_to_google_drive())

该过程类似于为机器间调用获取令牌，如以下示例所示：

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key

@requires_access_token(
    provider_name= "my-api-key-provider", # replace with your own credential provider name
    scopes= [],
    auth_flow= 'M2M',
)
async def need_token_2LO_async(*, access_token: str):
    # Use the access token
    pass

# To invoke:
# asyncio.run(need_token_2LO_async())

自动刷新令牌存储和使用情况

AgentCore 当 OAuth2 提供商提供刷新令牌时，会自动存储和使用刷新令牌，从而减少用户重新授权提示的频率。当用户最初通过标准的 OAuth2 授权码流程授予同意时，系统会将访问令牌和刷新令牌（如果提供）存储在安全令牌库中。这使代理能够在原始令牌到期时自动获取新的访问令牌，从而通过最大限度地减少重复的同意请求来改善用户体验。

刷新令牌的使用寿命通常比访问令牌长，默认有效期约为30天，而访问令牌的使用寿命较短（通常为1-2小时）。当访问令牌过期时， AgentCore 会自动使用存储的刷新令牌向提供商请求新的访问令牌。如果存储了有效的刷新令牌，则 AgentCore 跳过用户联合流程并直接返回新的访问令牌。如果刷新令牌也已过期或无效，则系统会回退为提示用户进行完全重新授权。

此功能无需在内部进行配置 AgentCore ——当 OAuth2 提供商的令牌响应中存在刷新令牌时，它会自动运行。但是，您必须将 OAuth2 提供商配置为在授权流程中包含刷新令牌。具体配置取决于您的提供商：

"customParameters": { "access_type": "offline" }

"scopes": ["openid", "profile", "offline_access"]

"scopes": ["api", "refresh_token"]

"scopes": ["read:jira-user", "offline_access"]

如果您的提供商支持刷新令牌并且配置正确， AgentCore 则无需额外设置即可自动存储和管理它们。要清除存储的刷新令牌并强制用户重新进行身份验证，请在调用forceAuthentication=true GetResourceOauth2Token时进行设置。这将清除刷新令牌并强制执行完整的联盟流程。有关配置 OAuth2 提供程序的信息，请参阅提供程序设置和配置。

向应用程序调用者直播授权网址

对于三足的 OAuth (3LO) 流程，您的代理需要向调用应用程序提供授权 URL，以便用户可以完成同意流程。虽然上面的示例显示了将 URL 打印到控制台，但生产应用程序需要通过应用程序的响应机制将 URL 流回调用方。

常见的实现模式

流式响应模式-对于支持流式响应的应用程序，您可以将授权网址作为响应流的一部分发送：

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Stream URL back to caller instead of printing
    on_auth_url=lambda url: stream_to_caller({
        "type": "authorization_required",
        "authorization_url": url,
        "message": "Please visit this URL to authorize access"
    }),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_streaming_auth(*, access_token: str):
    # Agent logic continues after user completes authorization
    return {"status": "success", "token_received": True}

def stream_to_caller(data):
    # Implementation depends on your streaming mechanism
    # Examples: WebSocket, Server-Sent Events, HTTP chunked response
    response_stream.send(json.dumps(data))

回调模式-对于使用回调或 webhook 的应用程序，请存储授权 URL 并通知调用者：

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL and trigger callback
    on_auth_url=lambda url: handle_auth_callback(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_callback_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def handle_auth_callback(authorization_url):
    # Store the URL associated with the request
    auth_store.save(request_id, {
        "authorization_url": authorization_url,
        "status": "pending_authorization"
    })

    # Notify the calling application
    callback_service.notify(callback_url, {
        "request_id": request_id,
        "authorization_url": authorization_url,
        "action_required": "user_authorization"
    })

轮询模式-对于喜欢轮询的应用程序，请将授权 URL 存储在可检索的位置：

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL for polling retrieval
    on_auth_url=lambda url: store_auth_url_for_polling(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_polling_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def store_auth_url_for_polling(authorization_url):
    # Store in database, cache, or session store
    session_store.set(f"auth_url:{session_id}", {
        "authorization_url": authorization_url,
        "created_at": datetime.utcnow(),
        "status": "pending"
    }, ttl=300)  # 5 minute expiration

选择最适合您的应用程序架构的模式。流式响应为实时应用程序提供了最佳的用户体验，而回调和轮询模式则适用于异步或批处理场景。

AgentCore OAuth2 流程中的资源指标

资源指示器提供了一种标准化的方法来指定哪个资源服务器应接受 OAuth2 访问令牌。 AgentCore 使用 Cognito 作为其身份验证提供程序，它支持符合 RFC 8707 标准的资源指标，允许您在令牌请求期间指定目标资源服务器。要使用资源指示器，必须先将授权服务器配置为使用 Cognito 的 CreateResourceServer API 识别特定的资源服务器。配置完成后，当您在令牌请求中指定资源指示符时，Cognito 会在生成的令牌的 aud 声明中包含相应的资源服务器标识符，从而使资源服务器能够验证令牌是否用于其特定用途。这提供了几个重要的好处：资源服务器可以验证令牌是否专门用于它们（最低权限原则），通过清楚地识别每个令牌针对哪个资源服务器来提高可审计性，以及降低应用程序环境中不同服务之间滥用令牌的风险。

通过 Cognito 的 RFC 8707 实现， AgentCore 使客户端能够直接在授权和令牌请求中指定资源服务器，从而覆盖默认的受众参数。在 Cognito 中，RFC 中提及的 “资源指标” 对应于的 “标识符” 值。ResourceServer资源指标对于模型上下文协议 (MCP) 实施尤其重要，它们有助于缓解 MCP 授权规范中概述的特定安全风险。资源指标对应于 RFC 9728 资源参数，可确保 MCP 服务器交互的令牌作用域正确。请注意，当前实现支持单资源绑定，这意味着您可以为每个令牌请求指定一个资源服务器。

当您的代理需要访问具有特定安全要求的资源服务器时，或者需要对令牌受众验证进行精细控制时，请使用资源指示器。资源指标对于多租户应用程序特别有用，在这些应用程序中，令牌应仅限于特定的客户资源。