AWS bedrock-agentcore medium security documentation change

Service: bedrock-agentcore · 2025-10-01 · Security-related medium

File: bedrock-agentcore/latest/devguide/identity-authentication.md

Summary

Added documentation for streaming authorization URLs in OAuth flows with implementation patterns (streaming response, callback, polling)

Security assessment

The change addresses secure handling of authorization URLs in OAuth flows by providing patterns to avoid exposing URLs through insecure channels like console logs. This prevents potential authorization code interception or phishing risks.

Diff

diff --git a/bedrock-agentcore/latest/devguide/identity-authentication.md b/bedrock-agentcore/latest/devguide/identity-authentication.md
index 6913635d3..f752694c6 100644
--- a//bedrock-agentcore/latest/devguide/identity-authentication.md
+++ b//bedrock-agentcore/latest/devguide/identity-authentication.md
@@ -4,0 +5,2 @@
+Streaming authorization URLs to application callers
+
@@ -36,0 +39,94 @@ The process is similar to obtain a token for machine-to-machine calls, as shown
+## Streaming authorization URLs to application callers
+
+For three-legged OAuth (3LO) flows, your agent needs to provide the authorization URL to the calling application so users can complete the consent flow. While the examples above show printing the URL to the console, production applications require streaming the URL back to the caller through your application's response mechanism.
+
+**Common implementation patterns:**
+
+**Streaming response pattern:** For applications that support streaming responses, you can send the authorization URL as part of the response stream:
+    
+    
+    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,
+    )
+    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))
+    
+
+**Callback pattern:** For applications using callbacks or webhooks, store the authorization URL and notify the caller:
+    
+    
+    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,
+    )
+    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"
+        })
+    
+
+**Polling pattern:** For applications that prefer polling, store the authorization URL in a retrievable location:
+    
+    
+    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,
+    )
+    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
+    
+
+Choose the pattern that best fits your application architecture. Streaming responses provide the best user experience for real-time applications, while callback and polling patterns work well for asynchronous or batch processing scenarios.
+
@@ -45 +141 @@ Get workload access token
-Obtain API key
+Scope credential access