Repository: livekit/agents
File: livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/stt.py (Python)

Code before:

        return CustomAwsCrtCredentialResolver()

    async def _run(self) -> None:
        while True:
            client = TranscribeStreamingClient(
                region=self._opts.region,
                credential_resolver=self._credential_resolver(),
            )

            live_config = {
                "language_code": self._opts.language,
                "media_sample_rate_hz": self._opts.sample_rate,
                "media_encoding": self._opts.encoding,
                "vocabulary_name": self._opts.vocabulary_name,
                "session_id": self._opts.session_id,
                "vocab_filter_method": self._opts.vocab_filter_method,
                "vocab_filter_name": self._opts.vocab_filter_name,
                "show_speaker_label": self._opts.show_speaker_label,
                "enable_channel_identification": self._opts.enable_channel_identification,
                "number_of_channels": self._opts.number_of_channels,
                "enable_partial_results_stabilization": self._opts.enable_partial_results_stabilization,  # noqa: E501
                "partial_results_stability": self._opts.partial_results_stability,
                "language_model_name": self._opts.language_model_name,
            }
            filtered_config = {k: v for k, v in live_config.items() if v and is_given(v)}
            stream = await client.start_stream_transcription(**filtered_config)  # type: ignore

            async def input_generator(stream: StartStreamTranscriptionEventStream) -> None:
                async for frame in self._input_ch:
                    if isinstance(frame, rtc.AudioFrame):
                        await stream.input_stream.send_audio_event(audio_chunk=frame.data.tobytes())
                await stream.input_stream.end_stream()  # type: ignore

            async def handle_transcript_events(stream: StartStreamTranscriptionEventStream) -> None:
                async for event in stream.output_stream:
                    if isinstance(event, TranscriptEvent):
                        self._process_transcript_event(event)

            tasks = [
                asyncio.create_task(input_generator(stream)),
                asyncio.create_task(handle_transcript_events(stream)),
            ]
            try:
                await asyncio.gather(*tasks)
            except BadRequestException as e:
                if e.message and e.message.startswith("Your request timed out"):
                    # AWS times out after 15s of inactivity, this tends to happen
                    # at the end of the session, when the input is gone, we'll ignore it and
                    # just treat it as a silent retry
                    logger.info("restarting transcribe session")

Review comment (bug):
Thank you @pabloFuente for this fix and @longcw for reviewing, while I wait for a release, I tried to patch this in and I believe we're missing the following in the case credentials are provided to the `STT`
```
                config_kwargs["aws_credentials_identity_resolver"] = (
                    StaticCredentialsResolver()
                )
```

The context is in deployed AWS environments when `self._credentials == None` and it uses `EnvironmentCredentialsResolver` did not work for me and the agents were absolutely silent with no error or debug logs as if session never started, still haven't figured out why. Regardless, since the `STT` module accepts `credentials` I provided that as a workaround passing it like so it gets tokens based on iam access of my containers.

```
            session = botocore.session.get_session()
            credentials = session.get_credentials()
            if credentials is None:
                raise RuntimeError("AWS credentials not found")
            frozen = credentials.get_frozen_credentials()
            return aws.STT(
                ...
                credentials=AWSCredentials(
                    access_key_id=frozen.access_key,
                    secret_access_key=frozen.secret_key,
                    session_token=frozen.token,
                ),
            )
```
but it was still not working and after some digging it seems the AWS SDK requires aws_credentials_identity_resolver to be set to `StaticCredentialsResolver` for it to use those properties 


Code after:
            stt=self,
            conn_options=conn_options,
            opts=self._config,
            credentials=self._credentials,
        )


class SpeechStream(stt.SpeechStream):
    def __init__(
        self,
        stt: STT,
        opts: STTOptions,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        credentials: Credentials | None = None,
    ) -> None:
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
        self._opts = opts
        self._credentials = credentials

    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
            else:
                config_kwargs["aws_credentials_identity_resolver"] = (
                    EnvironmentCredentialsResolver()
                )

            client: TranscribeStreamingClient = TranscribeStreamingClient(
                config=Config(**config_kwargs)
            )

            live_config = {
                "language_code": self._opts.language,
                "media_sample_rate_hertz": self._opts.sample_rate,
                "media_encoding": self._opts.encoding,
                "vocabulary_name": self._opts.vocabulary_name,
                "session_id": self._opts.session_id,
                "vocab_filter_method": self._opts.vocab_filter_method,
                "vocab_filter_name": self._opts.vocab_filter_name,
                "show_speaker_label": self._opts.show_speaker_label,
                "enable_channel_identification": self._opts.enable_channel_identification,
                "number_of_channels": self._opts.number_of_channels,
                "enable_partial_results_stabilization": self._opts.enable_partial_results_stabilization,
                "partial_results_stability": self._opts.partial_results_stability,
                "language_model_name": self._opts.language_model_name,
            }
            filtered_config = {k: v for k, v in live_config.items() if v and is_given(v)}