feat(bedrock): Add bearer token authentication support for AgentCore

[busla]

Description

This PR adds support for Cognito bearer token authentication to AWS Bedrock AgentCore, enabling dual authentication modes (AWS SigV4 IAM credentials OR bearer tokens).

Changes

• Modified sign_request() to check for bearerToken in optional_params
• When bearer token is present, uses Authorization: Bearer <token> header instead of AWS SigV4 signing
• Added verbose_logger.debug() statements for authentication method selection
• Maintains backward compatibility - defaults to SigV4 when no bearer token provided

Authentication Modes

1. AWS SigV4 (IAM) - Default

response = completion(
    model="bedrock/agentcore/arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-runtime",
    messages=[{"role": "user", "content": "Hello"}],
    # Uses AWS credentials from environment
)

2. Bearer Token (Cognito) - New

response = completion(
    model="bedrock/agentcore/arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-runtime",
    messages=[{"role": "user", "content": "Hello"}],
    bearerToken="<cognito-access-token>",
    runtimeSessionId="session-123",
    runtimeUserId="user-456"
)

Custom Callback Example

For LiteLLM Proxy deployments, you can use a custom callback to map HTTP headers to AgentCore parameters:

from litellm.integrations.custom_logger import CustomLogger

class InjectAgentCoreAuth(CustomLogger):
    async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type):
        from litellm.proxy.proxy_server import llm_model_list
        
        metadata = (data or {}).get("metadata") or {}
        headers = metadata.get("headers", {})
        
        # Extract bearer token from OpenID access token header
        bearer_token = next(
            (v for k, v in headers.items() if k.lower() == "x-openid-access-token"),
            None,
        )
        
        # Extract session and user IDs
        session_id = next(
            (v for k, v in headers.items() 
             if k.lower() in ("x-conversation-id", "x-amzn-bedrock-agentcore-runtime-session-id")),
            None,
        )
        user_id = next(
            (v for k, v in headers.items() 
             if k.lower() in ("x-user-id", "x-amzn-bedrock-agentcore-runtime-user-id")),
            None,
        )
        
        # Check if this is an AgentCore model
        model = data.get("model", "")
        litellm_model = next(
            (config.get("litellm_params", {}).get("model", "")
             for config in (llm_model_list or [])
             if config.get("model_name") == model),
            "",
        )
        
        if "bedrock/agentcore" in litellm_model:
            # Set parameters at top level - litellm moves them to optional_params internally
            if session_id:
                data["runtimeSessionId"] = session_id
            if user_id:
                data["runtimeUserId"] = user_id
            if bearer_token:
                data["bearerToken"] = bearer_token
        
        return data

# Register in litellm config
proxy_handler_instance = InjectAgentCoreAuth()

Testing

Tested with:

• ✅ AWS SigV4 authentication (IAM credentials)
• ✅ Bearer token authentication (Cognito)
• ✅ LibreChat integration with federated token support
• ✅ Session and user ID tracking
• ✅ Streaming responses with both auth modes

Breaking Changes

None - this is a backward-compatible addition. Existing AgentCore integrations continue to work with SigV4 authentication by default.

[vercel]

@busla is attempting to deploy a commit to the CLERKIEAI Team on Vercel.

A member of the Team first needs to authorize it.

[krrishdholakia]

@busla can we use the openai param - api_key instead? This would keep it consistent with other integrations

[krrishdholakia]

Thank you for this PR - btw!

[busla]

api_key

ahh, of course, good catch!