Besorgen Sie sich das OAuth 2.0-Zugriffstoken

AgentCore Identity ermöglicht es Entwicklern, OAuth-Token entweder für vom Benutzer delegierten Zugriff oder für die Maschine-zu-Maschine-Authentifizierung auf der Grundlage der konfigurierten OAuth 2.0-Anmeldeinformationsanbieter zu erhalten. Der Dienst orchestriert den Authentifizierungsprozess zwischen dem Benutzer oder der Anwendung und dem nachgeschalteten Autorisierungsserver und ruft das resultierende Token ab und speichert es. Sobald das Token im AgentCore Identity Vault verfügbar ist, können autorisierte Agenten es abrufen und verwenden, um Aufrufe an Ressourcenserver zu autorisieren. Mit dem folgenden Beispielcode wird beispielsweise ein Token abgerufen, um im Namen eines Endnutzers mit Google Drive zu interagieren. Weitere Informationen finden Sie unter Integration mit Google Drive mithilfe von OAuth2 für ein vollständiges Beispiel.

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

Der Vorgang ähnelt dem Abrufen eines Tokens für Maschine-zu-Maschine-Aufrufe, wie im folgenden Beispiel gezeigt:

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

Speicherung und Verwendung von Tokens für automatische Aktualisierungen

AgentCore speichert und verwendet automatisch Aktualisierungstoken, sofern sie von OAuth2-Anbietern verfügbar sind, wodurch die Häufigkeit von Aufforderungen zur erneuten Autorisierung von Benutzern reduziert wird. Wenn Benutzer ihre Zustimmung zunächst über einen standardmäßigen OAuth2-Autorisierungscode erteilen, speichert das System sowohl Zugriffstoken als auch Aktualisierungstoken (falls vorhanden) im sicheren Token-Tresor. Auf diese Weise können Agenten automatisch neue Zugriffstoken erhalten, wenn die ursprünglichen Token ablaufen, was die Benutzererfahrung verbessert, indem wiederholte Einwilligungsanfragen minimiert werden.

Aktualisierungstoken haben in der Regel eine längere Lebensdauer als Zugriffstoken und haben eine Standardgültigkeitsdauer von etwa 30 Tagen im Vergleich zur kürzeren Lebensdauer von Zugriffstoken (oft 1—2 Stunden). Wenn ein Zugriffstoken abläuft, verwendet es AgentCore automatisch das gespeicherte Aktualisierungstoken, um ein neues Zugriffstoken vom Anbieter anzufordern. Wenn ein gültiges Aktualisierungstoken gespeichert ist, wird der Benutzerverbundfluss AgentCore übersprungen und direkt ein neues Zugriffstoken zurückgegeben. Wenn das Aktualisierungstoken ebenfalls abgelaufen oder ungültig ist, fordert das System den Benutzer erneut auf, eine vollständige Neuautorisierung vorzunehmen.

Für diese Funktion ist keine interne AgentCore Konfiguration erforderlich. Sie funktioniert automatisch, wenn Aktualisierungstoken in der Token-Antwort des OAuth2-Anbieters enthalten sind. Sie müssen Ihren OAuth2-Anbieter jedoch so konfigurieren, dass er Aktualisierungstoken in den Autorisierungsablauf einbezieht. Die spezifische Konfiguration hängt von Ihrem Anbieter ab:

"customParameters": { "access_type": "offline" }

"scopes": ["openid", "profile", "offline_access"]

"scopes": ["api", "refresh_token"]

"scopes": ["read:jira-user", "offline_access"]

Wenn Ihr Anbieter Aktualisierungstoken unterstützt und ordnungsgemäß konfiguriert ist, AgentCore speichert und verwaltet er diese automatisch ohne zusätzliche Einrichtung. Um gespeicherte Aktualisierungstoken zu löschen und Benutzer zur erneuten Authentifizierung zu zwingen, legen Sie forceAuthentication=true beim Aufrufen fest. GetResourceOauth2Token Dadurch wird das Aktualisierungstoken gelöscht und ein vollständiger Verbundablauf erzwungen. Informationen zur Konfiguration von OAuth2-Anbietern finden Sie unter Einrichtung und Konfiguration von Anbietern.

Autorisierungs-URLs an Anwendungsaufrufer streamen

Bei dreibeinigen OAuth-Flows (3LO) muss Ihr Agent der aufrufenden Anwendung die Autorisierungs-URL zur Verfügung stellen, damit die Benutzer den Zustimmungsprozess abschließen können. In den obigen Beispielen wird zwar gezeigt, dass die URL auf der Konsole gedruckt wird, bei Produktionsanwendungen muss die URL jedoch über den Antwortmechanismus Ihrer Anwendung zurück zum Anrufer gestreamt werden.

Allgemeine Implementierungsmuster

Streaming-Antwortmuster — Für Anwendungen, die Streaming-Antworten unterstützen, können Sie die Autorisierungs-URL als Teil des Antwortstreams senden:

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

Rückrufmuster — Für Anwendungen, die Callbacks oder Webhooks verwenden, speichern Sie die Autorisierungs-URL und benachrichtigen Sie den Anrufer:

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"
    })

Abfragemuster — Für Anwendungen, die Abfragen bevorzugen, speichern Sie die Autorisierungs-URL an einem abrufbaren Ort:

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

Wählen Sie das Muster, das am besten zu Ihrer Anwendungsarchitektur passt. Streaming-Antworten bieten die beste Benutzererfahrung für Echtzeitanwendungen, während Rückruf- und Abfragemuster sich gut für asynchrone oder Batch-Verarbeitungsszenarien eignen.

Ressourcenindikatoren in OAuth2-Flows AgentCore

Ressourcenindikatoren bieten eine standardisierte Methode, um anzugeben, welcher Ressourcenserver ein OAuth2-Zugriffstoken akzeptieren soll. AgentCore verwendet Cognito als Authentifizierungsanbieter, der RFC 8707-konforme Ressourcenindikatoren unterstützt, mit denen Sie den beabsichtigten Ressourcenserver bei Token-Anfragen angeben können. Um Ressourcenindikatoren verwenden zu können, müssen Sie zunächst den Autorisierungsserver so konfigurieren, dass er mithilfe der Cognito-API bestimmte Ressourcenserver erkennt. CreateResourceServer Wenn Sie nach der Konfiguration einen Ressourcenindikator in Ihrer Token-Anfrage angeben, nimmt Cognito die entsprechende Ressourcenserver-ID in den Aud-Claim des resultierenden Tokens auf, sodass der Ressourcenserver überprüfen kann, ob das Token für seine spezifische Verwendung bestimmt ist. Dies bietet mehrere wichtige Vorteile: Ressourcenserver können überprüfen, ob Token speziell für sie bestimmt sind (Prinzip der geringsten Rechte), verbesserte Überprüfbarkeit, da eindeutig identifiziert wird, auf welchen Ressourcenserver jedes Token abzielt, und ein geringeres Risiko eines Tokenmissbrauchs zwischen verschiedenen Diensten in Ihrer Anwendungsumgebung.

Durch die RFC 8707-Implementierung von Cognito AgentCore können Clients einen Ressourcenserver direkt in Autorisierungs- und Token-Anfragen angeben und so den standardmäßigen Zielgruppenparameter außer Kraft setzen. In Cognito entspricht der „Ressourcenindikator“, auf den sich der RFC bezieht, dem „Identifier“ ResourceServer-Wert. Ressourcenindikatoren sind besonders wichtig für Implementierungen des Model Context Protocol (MCP), wo sie dazu beitragen, spezifische Sicherheitsrisiken zu minimieren, die in der MCP-Autorisierungsspezifikation beschrieben sind. Der Ressourcenindikator entspricht dem Ressourcenparameter RFC 9728 und gewährleistet so den korrekten Token-Scoping für MCP-Serverinteraktionen. Beachten Sie, dass die aktuelle Implementierung die Bindung einzelner Ressourcen unterstützt, was bedeutet, dass Sie einen Ressourcenserver pro Token-Anfrage angeben können.

Verwenden Sie Ressourcenindikatoren, wenn Ihre Agenten auf Ressourcenserver mit bestimmten Sicherheitsanforderungen zugreifen müssen oder wenn Sie eine genaue Kontrolle über die Überprüfung der Token-Zielgruppe benötigen. Ressourcenindikatoren sind besonders nützlich für Mehrmandantenanwendungen, bei denen Tokens auf bestimmte Kundenressourcen beschränkt werden sollten.