| 195 | |
| 196 | @dataclass(frozen=True) |
| 197 | class _BedrockProviderDefinition: |
| 198 | configured_region: str | None |
| 199 | region_source: Literal["explicit", "environment"] | None |
| 200 | configured_base_url: httpx.URL | None |
| 201 | api_key: str | None = field(default=None, repr=False) |
| 202 | token_provider: BedrockTokenProvider | None = field(default=None, repr=False, compare=False) |
| 203 | use_environment_bearer: bool = False |
| 204 | profile: str | None = None |
| 205 | access_key_id: str | None = field(default=None, repr=False) |
| 206 | secret_access_key: str | None = field(default=None, repr=False) |
| 207 | session_token: str | None = field(default=None, repr=False) |
| 208 | credential_provider: AwsCredentialsProvider | None = field(default=None, repr=False, compare=False) |
| 209 | name: str = field(default="bedrock", init=False) |
| 210 | |
| 211 | def _aws_source(self) -> Literal["static", "profile", "provider", "default"]: |
| 212 | if self.access_key_id is not None: |
| 213 | return "static" |
| 214 | if self.profile is not None: |
| 215 | return "profile" |
| 216 | if self.credential_provider is not None: |
| 217 | return "provider" |
| 218 | return "default" |
| 219 | |
| 220 | def _resolve_aws_auth(self) -> tuple[BedrockAwsAuthConfig, BedrockAwsAuth | None]: |
| 221 | if self.configured_region is not None: |
| 222 | return ( |
| 223 | BedrockAwsAuthConfig( |
| 224 | region=self.configured_region, |
| 225 | source=self._aws_source(), |
| 226 | region_source=self.region_source or "explicit", |
| 227 | profile=self.profile, |
| 228 | access_key_id=self.access_key_id, |
| 229 | secret_access_key=self.secret_access_key, |
| 230 | session_token=self.session_token, |
| 231 | credentials_provider=self.credential_provider, |
| 232 | ), |
| 233 | None, |
| 234 | ) |
| 235 | |
| 236 | auth = BedrockAwsAuth.resolve( |
| 237 | region=None, |
| 238 | profile=self.profile, |
| 239 | access_key_id=self.access_key_id, |
| 240 | secret_access_key=self.secret_access_key, |
| 241 | session_token=self.session_token, |
| 242 | credentials_provider=self.credential_provider, |
| 243 | ) |
| 244 | return auth.config, auth |
| 245 | |
| 246 | def configure(self) -> _ProviderRuntime: |
| 247 | def environment_token() -> str: |
| 248 | token = os.environ.get("AWS_BEARER_TOKEN_BEDROCK") |
| 249 | if not token: |
| 250 | raise OpenAIError( |
| 251 | "Could not find credentials for Bedrock. Pass a bearer credential or AWS credentials to " |
| 252 | "`bedrock(...)`, set `AWS_BEARER_TOKEN_BEDROCK`, or configure the default AWS credential chain." |
| 253 | ) |
| 254 | return token |