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

使用 OAuth2 与 Google 云端硬盘集成

本入门教程将引导您完成开始为 AI 代理使用亚马逊 Bedrock I AgentCore dentity 的基本步骤。您将学习如何设置开发环境、安装必要的 SDK、创建第一个代理身份以及允许代理安全访问外部资源。

在本教程结束时，您将拥有一个工作代理，该代理可以通过 Ident AgentCore ity OAuth2 凭据提供商从 Google 检索访问令牌，并使用访问令牌从 Google 云端硬盘读取文件。有关 OAuth2 流程的详细信息，请参阅使用身份管理凭证提供商。 AgentCore

先决条件

在开始之前，你需要：

安装 SDK

要开始使用，请安装该bedrock-agentcore软件包：

pip install bedrock-agentcore

获取 Google 客户端 ID 和客户端密钥

要允许您的代理访问 Google 云端硬盘，您需要为代理获取 Google 客户端 ID 和客户端密钥。前往 Google 开发者控制台并按照以下步骤操作：

第 1 步：设置 OAuth 2.0 凭证提供商

使用之前使用以下 CLI 命令获得的 Google 客户端 ID 和客户端密钥创建新的 OAuth 2.0 凭据提供商： AWS

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": "<your-google-client-id>",
        "clientSecret": "<your-google-client-secret>"
      }
    }' \
--output json)

OAUTH2_CALLBACK_URL=$(echo $OAUTH2_CREDENTIAL_PROVIDER_RESPONSE | jq -r '.callbackUrl')

echo "OAuth2 Callback URL: $OAUTH2_CALLBACK_URL"

第 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装饰器按以下顺序运行：

第 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=""))

接下来做什么？

本节中的示例侧重于您可以根据特定用例进行调整的实际实现模式。您可以将代码作为代理或模型上下文协议 (MCP) 工具的一部分嵌入。如果您想使用 AgentCore 运行时托管代理代码或 MCP 工具，请关注托管代理或使用 Amazon Bedrock AgentCore Runtime 的工具，将上述代码复制到 Runtime。 AgentCore