fix aws credentials when using env vals (#4403)

Author: chenghao-mou

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/stt.py

@@ -47,6 +47,7 @@
         TranscriptResultStream,
     )
     from smithy_aws_core.identity import (
+        AWSCredentialsIdentity,
         ContainerCredentialsResolver,
         EnvironmentCredentialsResolver,
         IMDSCredentialsResolver,
@@ -195,9 +196,22 @@ async def _run(self) -> None:
         while True:
             config_kwargs: dict[str, Any] = {"region": self._opts.region}
             if self._credentials:
-                config_kwargs["aws_access_key_id"] = self._credentials.access_key_id
-                config_kwargs["aws_secret_access_key"] = self._credentials.secret_access_key
-                config_kwargs["aws_session_token"] = self._credentials.session_token
+                # Use a credentials resolver for explicit credentials
+                # for some reason, Config with direct values doesn't work
+                class StaticCredsResolver:
+                    def __init__(self, creds: Credentials):
+                        self._identity = AWSCredentialsIdentity(
+                            access_key_id=creds.access_key_id,
+                            secret_access_key=creds.secret_access_key,
+                            session_token=creds.session_token,
+                        )
+
+                    async def get_identity(self, **kwargs: Any) -> AWSCredentialsIdentity:
+                        return self._identity
+
+                config_kwargs["aws_credentials_identity_resolver"] = StaticCredsResolver(
+                    self._credentials
+                )
             else:
                 config_kwargs["aws_credentials_identity_resolver"] = ChainedIdentityResolver(
                     resolvers=(
@@ -229,6 +243,8 @@ async def _run(self) -> None:
             }
             filtered_config = {k: v for k, v in live_config.items() if v and is_given(v)}
 
+            tasks: list[asyncio.Task[Any]] = []
+
             try:
                 stream = await client.start_stream_transcription(
                     input=StartStreamTranscriptionInput(**filtered_config)
@@ -286,14 +302,15 @@ async def handle_transcript_events(
                 else:
                     raise e
             finally:
-                # Close input stream first
-                await utils.aio.gracefully_cancel(tasks[0])
-
-                # Wait for output stream to close cleanly
-                try:
-                    await asyncio.wait_for(tasks[1], timeout=3.0)
-                except (asyncio.TimeoutError, asyncio.CancelledError):
-                    await utils.aio.gracefully_cancel(tasks[1])
+                if tasks:
+                    # Close input stream first
+                    await utils.aio.gracefully_cancel(tasks[0])
+
+                    # Wait for output stream to close cleanly
+                    try:
+                        await asyncio.wait_for(tasks[1], timeout=3.0)
+                    except (asyncio.TimeoutError, asyncio.CancelledError):
+                        await utils.aio.gracefully_cancel(tasks[1])
 
                 # Ensure gather future is retrieved to avoid "exception never retrieved"
                 with contextlib.suppress(Exception):