使用 OAuth2 与谷歌云端硬盘集成

本入门教程将引导你完成开始为人工智能代理使用 Amazon Bedrock AgentCore Identity 的基本步骤。您将学习如何设置开发环境、安装必要的软件开发工具包、创建您的第一个代理身份，以及如何允许代理安全地访问外部资源。

在本教程结束时，您将拥有一个工作代理，可以使用 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 命令 AWS 使用之前获得的 Google 客户端 ID 和客户端密钥创建新的 OAuth 2.0 凭据提供程序：

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 的谷歌客户端库：

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 R AgentCore untime 的工具，将上面的代码复制到 Runtim e 中。 AgentCore