取得 OAuth 2.0 存取權杖

AgentCore Identity 可讓開發人員根據設定的 OAuth 2.0 憑證提供者，取得使用者委派存取或machine-to-machine身分驗證的 OAuth 權杖。服務會協調使用者或應用程式與下游授權伺服器之間的身分驗證程序，並擷取和存放產生的字符。在 AgentCore Identity 保存庫中提供字符之後，授權代理程式就可以擷取字符，並使用它來授權對資源伺服器的呼叫。例如，以下範例程式碼會擷取權杖，以代表最終使用者與 Google Drive 互動。如需詳細資訊，請參閱使用 OAuth2 與 Google Drive 整合以取得完整範例。

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

程序類似於取得machine-to-machine呼叫的字符，如下列範例所示：

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 會自動儲存和管理它們，而無需額外設定。若要清除存放的重新整理字符並強制使用者重新驗證，請在呼叫 GetResourceOauth2Token forceAuthentication=true時設定 。這會清除重新整理字符並強制完整的聯合流程。如需有關設定 OAuth2 提供者的資訊，請參閱提供者設定和組態。

將授權 URLs 串流至應用程式呼叫者

對於三邊 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 流程中的資源指標

資源指標提供標準化的方式來指定哪些資源伺服器應接受 OAuth2 存取權杖。AgentCore 使用 Cognito 做為身分驗證提供者，其支援 RFC 8707 相容的資源指標，可讓您在權杖請求期間指定預期的資源伺服器。若要使用資源指標，您必須先設定授權伺服器，以使用 Cognito 的 CreateResourceServer API 來辨識特定資源伺服器。設定完成後，當您在權杖請求中指定資源指標時，Cognito 會在產生的權杖的 aud 宣告中包含對應的資源伺服器識別符，讓資源伺服器驗證權杖是否適用於其特定用途。這提供幾項重要優點：資源伺服器可以驗證權杖是否專門用於它們 （最低權限原則）、透過明確識別每個權杖目標的資源伺服器來改善可稽核性，並降低應用程式環境中不同服務間權杖濫用的風險。

透過 Cognito 的 RFC 8707 實作，AgentCore 可讓用戶端直接在授權和字符請求中指定資源伺服器，並覆寫預設對象參數。在 Cognito 中，RFC 中參考的「資源指標」對應於 ResourceServer 的「識別符」值。資源指標對於模型內容通訊協定 (MCP) 實作特別重要，有助於緩解 MCP 授權規格中概述的特定安全風險。資源指標對應至 RFC 9728 資源參數，確保 MCP 伺服器互動的適當權杖範圍。請注意，目前的實作支援單一資源繫結，這表示您可以為每個字符請求指定一個資源伺服器。

當您的代理程式需要存取具有特定安全要求的資源伺服器，或您需要對權杖對象驗證進行精細控制時，請使用資源指標。資源指標對於字符應限於特定客戶資源的多租戶應用程式特別有用。