本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 使用 OAuth2 与 Google 云端硬盘集成 本入门教程将引导您完成开始为 AI 代理使用亚马逊 Bedrock I AgentCore dentity 的基本步骤。您将学习如何设置开发环境、安装必要的 SDK、创建第一个代理身份以及允许代理安全访问外部资源。 在本教程结束时，您将拥有一个工作代理，该代理可以通过 Ident AgentCore ity OAuth2 凭据提供商从 Google 检索访问令牌，并使用访问令牌从 Google 云端硬盘读取文件。有关 OAuth2 流程的详细信息，请参阅[使用身份管理凭证提供商。 AgentCore ](identity-outbound-credential-provider.md) **Topics** + [先决条件](#identity-getting-started-prerequisites) + [第 1 步：设置 OAuth 2.0 凭证提供商](#identity-getting-started-step1) + [第 2 步：导入身份和身份验证模块](#identity-getting-started-step2) + [第 3 步：获取 OAuth 2.0 访问令牌](#identity-getting-started-step3) + [第 4 步：使用 OAuth2 访问令牌调用外部资源](#identity-getting-started-step4) + [接下来做什么？](#identity-getting-started-whats-next) ## 先决条件 在开始之前，你需要： + 具有适当权限的 Amazon 账户（例如，`BedrockAgentCoreFullAccess`） + Python 3.10 或更高版本 + 最新的 Amazon CLI 并`jq`已安装 + Amazon 证书和区域已配置 (`aws configure`) + 对 Python 编程的基本理解 **重要** `BedrockAgentCoreFullAccess`托管策略授予广泛的权限`GetWorkloadAccessTokenForUserId`，包括允许调用者使用任何用户标识符字符串颁发工作负载访问令牌，而无需验证 IdP 令牌。这适用于开发和测试。对于生产部署，创建遵循最低权限原则的自定义 IAM 策略，并将权限限制为仅限所需的特定操作。如果您的应用程序使用 JWT-based 身份验证（建议在生产环境中使用），则可以明确拒绝，`GetWorkloadAccessTokenForUserId`以确保所有用户身份都通过经过验证的 JWT 路径。有关更多信息，请参阅[获取工作负载访问令牌](get-workload-access-token.md)。 ### 安装 SDK 要开始使用，请安装该`bedrock-agentcore`软件包： ``` pip install bedrock-agentcore ``` ### 获取 Google 客户端 ID 和客户端密钥 要允许您的代理访问 Google 云端硬盘，您需要为代理获取 Google 客户端 ID 和客户端密钥。前往 [ Google 开发者控制台](https://console.developers.google.com/project)并按照以下步骤操作： 1. 在 Google 开发者控制台中创建项目 1. 启用 Google 云端硬盘 API 1. 配置 OAuth 同意屏幕 1. 为代理创建一个新的 Web 应用程序，例如 “我的代理 1” 1. 将以下 OAuth 2.0 范围添加到您的代理应用程序中：`https://www.googleapis.com/auth/drive.metadata.readonly` 1. 为新的 Web 应用程序创建 OAuth 2.0 凭据，并保存生成的 Google 客户端 ID 和客户端密钥 ## 第 1 步：设置 OAuth 2.0 凭证提供商 使用之前使用以下 CLI 命令获得的 Google 客户端 ID 和客户端密钥创建新的 OAuth 2.0 凭据提供商： Amazon ``` OAUTH2_CREDENTIAL_PROVIDER_RESPONSE=$(aws bedrock-agentcore-control create-oauth2-credential-provider \ --region us-east-1 \ --name "google-provider" \ --credential-provider-vendor "GoogleOauth2" \ --oauth2-provider-config-input '{ "googleOauth2ProviderConfig": { "clientId": "", "clientSecret": "" } }' \ --output json) OAUTH2_CALLBACK_URL=$(echo $OAUTH2_CREDENTIAL_PROVIDER_RESPONSE | jq -r '.callbackUrl') echo "OAuth2 Callback URL: $OAUTH2_CALLBACK_URL" ``` **注意** `callbackUrl`从上面的[CreateOauth2CredentialProvider](https://docs.amazonaws.cn/bedrock-agentcore-control/latest/APIReference/API_CreateOauth2CredentialProvider.html)响应中获取，然后将该 URI 添加到您的 Google 应用程序的重定向 URI 列表中。回调 URL 应如下所示：https://bedrock-agentcore.us-east-1.amazonaws.com/identities/oauth2/callback/\*\*\*\*\*\*\*\*-\*\*\*\*\*\*-\*\*\*\*\*\*-\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ## 第 2 步：导入身份和身份验证模块 将此导入语句添加到您的 Python 文件中： ``` from bedrock_agentcore.services.identity import IdentityClient from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key ``` ## 第 3 步：获取 OAuth 2.0 访问令牌 在上一步中创建了 Google 凭据提供商后，将`@requires_access_token`装饰器添加到需要 Google 访问令牌的代理代码中。从您的主机输出中复制授权网址，然后将其粘贴到浏览器中，并使用 Google 云端硬盘完成同意流程。 以下代码示例旨在集成到您的代理代码中，以调用授权工作流程。这不是可以独立复制和运行的独立代码。 ``` import asyncio # 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, # The callback URL to redirect to after the OAuth 2.0 token retrieval is complete callback_url='oauth2_callback_url_for_session_binding', ) async def write_to_google_drive(*, access_token: str): # Prints the access token obtained from Google print(access_token) asyncio.run(write_to_google_drive(access_token="")) ``` 在幕后，`@requires_access_token`装饰器按以下顺序运行： ![流程图显示 @requires_access_token 装饰器进行的 API 调用顺序](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/images/identity_access_token_flow.png) 1. 软件开发工具包对`CreateWorkloadIdentity``GetWorkloadAccessToken`、和进行 API 调用`GetResourceOauth2Token`。 1. 在本地运行代理代码时，SDK 会自动生成代理身份 ID 和随机用户 ID 以进行本地测试，并将它们存储在名为的本地文件中`.bedrock_agentcore.yaml`。 1. 使用 Runtime AgentCore 运行代理代码时，SDK 不会生成代理身份 ID 或随机用户 ID。相反，它使用分配的代理身份 ID 以及代理调用者传入的用户 ID 或 JWT 令牌。 1. 代理访问令牌是一种加密（不透明）令牌，包含代理身份 ID 和用户 ID。 1. AgentCore 身份服务将 Google 访问令牌存储在代币库中的代理身份 ID 和用户 ID 下。这会在代理身份、用户身份和 Google 访问令牌之间建立绑定。 1. [必须先完成](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/oauth2-authorization-url-session-binding.html)会话绑定流程，然后才能通过 I AgentCore dentity 将 Google 访问令牌返回给调用者。 ## 第 4 步：使用 OAuth2 访问令牌调用外部资源 代理通过上述步骤获得 Google 访问令牌后，它就可以使用访问令牌访问 Google 云端硬盘。以下是完整示例，其中列出了用户有权访问的前 10 个文件的名称和 ID。 首先，安装适用于 Python 的 Google 客户端库： ``` pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib ``` 然后，复制以下代码： ``` import asyncio from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError SCOPES = ["https://www.googleapis.com/auth/drive.metadata.readonly"] def main(access_token): """Shows basic usage of the Drive v3 API. Prints the names and ids of the first 10 files the user has access to. """ creds = Credentials(token=access_token, scopes=SCOPES) try: service = build("drive", "v3", credentials=creds) # Call the Drive v3 API results = ( service.files() .list(pageSize=10, fields="nextPageToken, files(id, name)") .execute() ) items = results.get("files", []) if not items: print("No files found.") return print("Files:") for item in items: print(f"{item['name']} ({item['id']})") except HttpError as error: # TODO(developer) - Handle errors from drive API. print(f"An error occurred: {error}") if __name__ == "__main__": # This annotation helps agent developer to obtain access tokens from external applications @requires_access_token( provider_name="google-provider", # Google OAuth2 scopes scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"], # 3LO flow auth_flow="USER_FEDERATION", # prints authorization URL to console on_auth_url=lambda x: print("Copy and paste this authorization url to your browser", x), force_authentication=True, callback_url='oauth2_callback_url_for_session_binding', ) async def read_from_google_drive(*, access_token: str): print(access_token) # You can see the access_token # Make API calls... main(access_token) asyncio.run(read_from_google_drive(access_token="")) ``` **注意** 有关处理[会话绑定的本地回调服务器实现示例](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/oauth2-authorization-url-session-binding.html)，请参阅上[的 oauth2\_callback\_server.py GitHub ](https://github.com/awslabs/amazon-bedrock-agentcore-samples/blob/main/01-features/05-authenticate-and-authorize/02-outbound-auth/02-outbound-auth-3lo/oauth2_callback_server.py) ## 接下来做什么？ 本节中的示例侧重于您可以根据特定用例进行调整的实际实现模式。您可以将代码作为代理或模型上下文协议 (MCP) 工具的一部分嵌入。如果您想使用 AgentCore 运行时托管代理代码或 MCP 工具，请关注[托管代理或使用 Amazon Bedrock AgentCore Runtime 的工具，将上述代码复制](agents-tools-runtime.md)到 Runtime。 AgentCore