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

# 获取 OAuth 2.0 访问令牌
<a name="identity-authentication"></a>

AgentCore 身份使开发人员能够根据配置的 OAuth 2.0 凭据提供商获取 OAuth 令牌，用于用户委托访问权限或机器间身份验证。该服务将协调用户或应用程序与下游授权服务器之间的身份验证过程，并将检索和存储生成的令牌。一旦该令牌在 I AgentCore dentity Vault 中可用，授权代理就可以检索该令牌并使用它来授权对资源服务器的调用。例如，以下示例代码将检索一个令牌，用于代表最终用户与 Google 云端硬盘进行交互。有关更多信息，请参阅使用 OAuth2 [ 与 Google 云端硬盘](identity-getting-started-google.md)集成，查看完整示例。

```
# 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())
```

**Topics**
+ [自动刷新令牌存储和使用](#automatic-refresh-token-storage)
+ [向应用程序调用者流式传输授权 URL](#authorization-url-streaming)
+ [AgentCore OAuth2 流程中的资源指标](#resource-indicators-cognito)

## 自动刷新令牌存储和使用
<a name="automatic-refresh-token-storage"></a>

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

**重要**  
不能保证返回的访问令牌 AgentCore 是有效的。联邦提供商端的客户可以撤销令牌， AgentCore 但无法检测。如果令牌无效，则使用`forceAuthentication: true`强制执行新的身份验证流程并获取有效的访问令牌。

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

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


| Provider | 需要配置 | 
| --- | --- | 
|  **Google**  | 通话`customParameters`时包括`access_type=offline`在内 `GetResourceOauth2Token` <pre>"customParameters": { "access_type": "offline" }</pre> | 
|  **微软 **  | 调用时包含`offline_access`在`scopes`参数中 `GetResourceOauth2Token` <pre>"scopes": ["openid", "profile", "offline_access"]</pre> | 
|  **Salesforce**  | 调用时包含`refresh_token`在`scopes`参数中 `GetResourceOauth2Token` <pre>"scopes": ["api", "refresh_token"]</pre> | 
|  **Atlassian **  | 调用时包含`offline_access`在`scopes`参数中 `GetResourceOauth2Token` <pre>"scopes": ["read:jira-user", "offline_access"]</pre> | 
|  **GitHub**  | 无需额外 AgentCore 配置。在您的 GitHub 应用程序设置中启用 User-to-server 令牌到期功能。启用此功能后，刷新令牌会自动存储。 | 
|  **Slack**  | 无需额外 AgentCore 配置。在 Slack 应用程序设置中启用 “代币轮换” 功能。启用此功能后，刷新令牌会自动返回。 | 
|  **LinkedIn**  | 无需额外 AgentCore 配置。在 LinkedIn 应用程序配置中启用刷新令牌设置。 | 
|  **其他提供商 **  | 一些提供商需要在其提供商设置中进行配置，而不是 API 参数。有关刷新令牌要求，请查阅提供商的文档。 | 

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

## 向应用程序调用者流式传输授权 URL
<a name="authorization-url-streaming"></a>

对于三段式 OAuth (3LO) 流程，您的代理需要向调用应用程序提供授权 URL，以便用户完成同意流程。虽然上面的示例显示了将 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 流程中的资源指标
<a name="resource-indicators-cognito"></a>

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

通过 Cognito 的 [ RFC 8707 ](https://www.rfc-editor.org/rfc/rfc8707.html) 实现， AgentCore 使客户端能够直接在授权和令牌请求中指定资源服务器，覆盖默认受众参数。在 Cognito 中，RFC 中提及的 “资源指标” 对应[ResourceServer的 ](https://docs.amazonaws.cn/cognito-user-identity-pools/latest/APIReference/API_CreateResourceServer.html) “标识符” 值。资源指标对于模型上下文协议 (MCP) 的实施尤其重要，它们有助于缓解MCP授权规范中概述的特定安全风险。资源指示器对应于 RFC 9728 资源参数，确保了 MCP 服务器交互的正确令牌范围。请注意，当前实现支持单资源绑定，这意味着您可以为每个令牌请求指定一个资源服务器。

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