Can you give a minimal latest example of voice agent with aws transcribe for stt and aws polly for tts

Thanks for reaching out, mithun. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

does the above reuse a common boto3 session for authentication and authorisation when running in aws cloud ?

Thanks — great question. You’re asking whether the minimal example reuses a common boto3 session for authentication/authorization when running in AWS (e.g., EC2, ECS, Lambda). I’ll clarify how credentials are handled. Short answer • The example does not explicitly create or manage a

boto3

session. • The LiveKit AWS plugins rely on the AWS SDK’s default credential resolution (which uses

boto3

under the hood). • When running in AWS without explicit environment credentials, the SDK will automatically use the instance/ECS/Lambda IAM role, so you effectively get the managed credentials provided by AWS. What that means in practice • If you set AWS environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION), those credentials will be used. • If you do not set environment credentials and the code runs on AWS infrastructure with an attached IAM role, the SDK will pick up the role credentials automatically (no extra code required). • The LiveKit plugins abstract away direct

boto3

session creation; they do not require you to pass a

boto3.Session

object. When you might want an explicit boto3 session • You need to use a specific set of credentials different from the environment or instance role. • You want to control retries, custom config, or a shared session object across multiple libraries. In those cases you would create and pass a

boto3.Session

yourself to the code that needs it — but the minimal example doesn’t expose that option. If you’d like, I can: • Show a short example of how to create a shared

boto3.Session

and pass it into a LiveKit AWS plugin (if the plugin supports it), or • Show best practices for running this agent on ECS/EC2/Lambda using IAM roles and no environment credentials. Which would help you more? Sources: AWS AI and LiveKit | LiveKit Documentation | Amazon Polly TTS plugin guide | LiveKit Documentation | Amazon Nova Sonic integration guide | LiveKit Documentation | Amazon Transcribe plugin guide | LiveKit Documentation | Amazon Bedrock LLM plugin guide | LiveKit Documentation

here is the issue I am running into - Issue: Voice agent process killed as "unresponsive" in EKS — root cause is per-plugin IMDS credential fetching blocking the event loop Environment: LiveKit Agents SDK (Python), livekit-plugins-aws (STT = Amazon Transcribe, TTS = Amazon Polly, LLM = Amazon Bedrock), running in EKS with an IAM instance role (no static credentials in env vars). Symptom: Agent process is killed with

"process is unresponsive, killing process"

within ~60s of starting in the cloud. Works fine locally with docker-compose (where AWS credentials are static env vars). Root cause (confirmed from pod logs): 1. Each AWS plugin (stt.STT, tts.TTS, bedrock_llm.LLM) creates its own independent aioboto3.Session when no session= argument is passed. Each session has its own IMDS credential cache. stt.STT uses the AWS CRT SDK (

aws_sdk_transcribe_streaming

) — a completely different credential stack from aioboto3. It does not share any credential cache with the TTS/LLM sessions. 1. In push-to-talk mode (no audio flowing), Transcribe streaming silently disconnects and reconnects every ~15 seconds. On each reconnect, the CRT SDK fetches fresh credentials from IMDS. This call blocks the asyncio event loop for ~300ms. 2. Two reconnects within the LiveKit IPC ping window (60s default) → cumulative event loop block exceeds the ping timeout → worker process is killed. Question: Is there a supported way to pass pre-fetched static credentials to stt.STT so it doesn't call IMDS on every Transcribe reconnect? The constructor accepts credentials: stt.Credentials(access_key_id, secret_access_key, session_token) — is this the correct approach, and are there any caveats (e.g. token expiry, refresh)? Also: tts.TTS and bedrock_llm.LLM both accept session: aioboto3.Session | None. Is it safe/correct to pass a single shared aioboto3.Session instance to both so they share the IMDS credential cache?

Thanks — this is a clear description of the problem and very helpful logs. I’ll respond to each question directly and offer practical recommendations you can try quickly. Acknowledgement • You’re seeing repeated IMDS calls from multiple AWS plugin stacks (aioboto3 sessions for TTS/LLM and the CRT SDK for Transcribe STT). Those IMDS calls block the asyncio event loop (~300ms) on reconnects, and multiple reconnects within the LiveKit ping window cause the worker to be killed. You want to avoid per-plugin IMDS fetches by reusing credentials or sessions. Direct answers 1) Can you pass pre-fetched static credentials to stt.STT to avoid IMDS calls? • Yes. The

stt.STT

constructor accepts

credentials: stt.Credentials(access_key_id, secret_access_key, session_token)

. Passing pre-fetched credentials is the correct mechanism to prevent the Transcribe CRT SDK from calling IMDS on every reconnect. • Caveats: ◦ Token expiry: If you provide temporary credentials (with a session token), they will expire. The CRT SDK will not automatically refresh them for you if you pass static/one-time credentials. You must refresh and re-initialize the STT client (or provide new credentials) before expiry to avoid authentication failures. ◦ Refresh strategy: Implement a background refresher that obtains fresh credentials (from IMDS or an STS AssumeRole flow) and updates the STT instance before expiry. Prefer fetching credentials on a separate thread or in a non-blocking way to avoid blocking the asyncio loop. 2) Is it safe/correct to share a single aioboto3.Session between tts.TTS and bedrock_llm.LLM? • Yes. Passing a single shared

aioboto3.Session

instance to

tts.TTS(session=...)

and

bedrock_llm.LLM(session=...)

is the right approach to have them share credential caching and reduce duplicated IMDS calls. • Benefits: ◦ Shared credential cache reduces the number of IMDS calls and centralizes credential refresh behavior for the aioboto3-based plugins. ◦ Less chance of concurrent blocking IMDS fetches from multiple independent sessions. • Caveats: ◦ The CRT Transcribe SDK uses a different credential stack (not aioboto3) and will not share that aioboto3 session or its cache. You still need to address the CRT SDK separately (see below). ◦ Ensure the shared aioboto3.Session is created once at startup, not per-request or per-plugin instance. Recommended options / mitigation strategies Option A — Provide long-lived credentials to STT (quickest) 1. On startup, fetch credentials from IMDS once on a background thread (not in asyncio event loop) and pass them into

stt.STT(credentials=...)

. 2. If credentials are temporary, schedule a refresh before expiry and re-create or update the STT instance safely (again, perform IMDS calls on a thread or an async-friendly non-blocking mechanism). Pros: Minimal code changes, eliminates per-reconnect IMDS blocking within the CRT SDK. Cons: You must manage refresh lifecycle and re-initialize STT before tokens expire. Option B — Shared refreshing process + inject refreshed credentials 1. Create a single credential-manager component that: ◦ Obtains credentials via IMDS (or STS) on a background thread. ◦ Maintains current credentials and their expiry. ◦ Exposes safe, non-blocking access for code to read current credentials. 1. On refresh, update both the aioboto3 session (if possible) and re-create or update the CRT Transcribe client with new credentials. Pros: Centralized, robust refresh handling. Cons: Additional implementation (but stable for production). Option C — Avoid CRT IMDS blocking by ensuring CRT uses pre-fetched credentials or a non-blocking credential provider • If the CRT SDK supports being supplied with a credentials object (it does via

stt.Credentials

), supply it with credentials from your credential-manager so it never calls IMDS directly. • Ensure your credential-manager performs IMDS calls off the asyncio event loop (e.g., use ThreadPoolExecutor or native blocking calls in a dedicated thread). Practical implementation notes and patterns • Perform all blocking IMDS or boto3 calls in a thread executor: ◦ Use

asyncio.get_running_loop().run_in_executor(...)

or a dedicated background thread so the event loop is not blocked by IMDS calls. • Create a single shared aioboto3.Session early: ◦ Make it a module-level singleton or create it in your agent entrypoint and pass it into TTS/LLM constructors:

tts.TTS(session=shared_session)

and

bedrock_llm.LLM(session=shared_session)

. • For the CRT STT client: ◦ Construct it with

stt.Credentials(...)

filled from your credential-manager. ◦ When tokens near expiry, refresh credentials in the credential-manager and re-create the CRT STT client before expiry. • Keep reconnect intervals in mind: ◦ The CRT reconnects every ~15s in push-to-talk; ensure your refresh or re-init schedule avoids frequent blocking during reconnects. Minimal example sketch • High-level flow you can implement quickly: a. Start a background thread that calls IMDS once and stores credentials + expiry. b. Create a shared

aioboto3.Session

and pass it to TTS and LLM plugins. c. Initialize

stt.STT(credentials=stt.Credentials(...))

using the fetched credentials. d. Periodically refresh credentials in the background thread; when refreshed, re-initialize the CRT STT instance before expiry on a background thread to avoid blocking the event loop. If you’d like, I can provide: • A concrete code example showing a small credential-manager using

run_in_executor

, creation of a shared

aioboto3.Session

, and safe re-initialization of

stt.STT

with refreshed credentials. • A snippet showing how to re-create the CRT STT instance without blocking the asyncio event loop. Which would you prefer: a compact code example for immediate testing, or a more production-ready credential-manager pattern?

• yea - A concrete code example showing a small credential-manager using

run_in_executor

, creation of a shared

aioboto3.Session

, and safe re-initialization of

stt.STT

with refreshed credentials. • A snippet showing how to re-create the CRT STT instance without blocking the asyncio event loop.

Thanks — this is a great, clear request. Below is a concrete, minimal but practical pattern you can drop into your LiveKit agent to: • Run IMDS credential fetches off the asyncio event loop using run_in_executor. • Maintain a single shared aioboto3.Session for TTS and LLM plugins. • Provide pre-fetched credentials to the CRT-based Transcribe STT client and safely re-create it before token expiry without blocking the event loop. Key ideas (brief) • Use a background credential-manager that fetches credentials on a thread and exposes them safely to the async code. • Create a single shared

aioboto3.Session

at startup and pass it into TTS/LLM. • Initialize the CRT STT with

stt.Credentials(...)

from the credential-manager. When credentials near expiry, re-create the CRT STT client on the background thread and swap it into the async runtime without blocking. Code (minimal, focused)

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Optional
import aioboto3

# Replace these imports with actual LiveKit plugin imports
# from livekit.plugins import aws_stt as stt  # CRT STT wrapper
# from livekit.plugins import aws_tts as tts
# from livekit.plugins import bedrock_llm as llm

# Mocked types for illustration; replace with actual plugin types:
class stt:
    class Credentials:
        def __init__(self, access_key_id, secret_access_key, session_token, expiration_ts=None):
            self.access_key_id = access_key_id
            self.secret_access_key = secret_access_key
            self.session_token = session_token
            self.expiration_ts = expiration_ts

    class STT:
        def __init__(self, credentials: Optional['stt.Credentials'] = None):
            self._creds = credentials
            # Initialize CRT client here using credentials
        def close(self):
            # teardown CRT client
            pass

# Simple credential container
class AWSCreds:
    def __init__(self, access_key, secret, token, expiration_ts):
        self.access_key = access_key
        self.secret = secret
        self.token = token
        self.expiration_ts = expiration_ts

    @property
    def expires_in(self):
        return max(0, self.expiration_ts - time.time())

class CredentialManager:
    """
    Fetch credentials off the asyncio event loop (in a thread).
    Exposes current creds and signals when refreshed.
    """
    def __init__(self, refresh_threshold_seconds: int = 60):
        self._creds: Optional[AWSCreds] = None
        self._lock = asyncio.Lock()
        self._executor = ThreadPoolExecutor(max_workers=1)
        self._refresh_threshold_seconds = refresh_threshold_seconds
        self._stop = False
        self._refresh_task: Optional[asyncio.Task] = None

    def _blocking_fetch_imds_credentials(self) -> AWSCreds:
        """
        Blocking call that uses botocore/requests to fetch IMDS credentials.
        This function runs in a background thread. Replace with your preferred IMDS/STS call.
        """
        # Example using boto3 (blocking) to get credentials from the default session:
        import boto3
        session = boto3.Session()
        creds = session.get_credentials().get_frozen_credentials()
        # If using instance profile, you can get expiration via botocore metadata or STS
        # Here we set a conservative expiration (e.g., now + 3600) if not present.
        expiration_ts = time.time() + 3600
        # If you use STS AssumeRoleWithWebIdentity / metadata with expiry, parse it here.
        return AWSCreds(creds.access_key, creds.secret_key, creds.token, expiration_ts)

    async def start(self):
        # initial fetch synchronously (but off loop)
        await self._refresh_once()
        # start background refresher
        self._refresh_task = asyncio.create_task(self._periodic_refresh())

    async def stop(self):
        self._stop = True
        if self._refresh_task:
            await self._refresh_task
        self._executor.shutdown(wait=False)

    async def _refresh_once(self):
        loop = asyncio.get_running_loop()
        new_creds = await loop.run_in_executor(self._executor, self._blocking_fetch_imds_credentials)
        async with self._lock:
            self._creds = new_creds

    async def _periodic_refresh(self):
        while not self._stop:
            async with self._lock:
                creds = self._creds
            if creds is None:
                # no creds yet, fetch now
                await self._refresh_once()
                continue

            # Sleep until we need to refresh; wake up earlier if within threshold
            sleep_for = max(5, creds.expires_in - self._refresh_threshold_seconds)
            await asyncio.sleep(sleep_for)

            # Refresh in background (blocking)
            await self._refresh_once()

    async def get_current_credentials(self) -> Optional[AWSCreds]:
        async with self._lock:
            return self._creds

# How to wire into your LiveKit agent
async def agent_entrypoint():
    # Create credential manager and start it
    cred_mgr = CredentialManager(refresh_threshold_seconds=60)
    await cred_mgr.start()

    # Create shared aioboto3.Session for TTS and LLM (this shares credential caching for aioboto3)
    shared_session = aioboto3.Session()

    # Initialize TTS and LLM with shared_session (example)
    # tts_client = tts.TTS(session=shared_session)
    # llm_client = llm.LLM(session=shared_session)

    # Initialize CRT STT with pre-fetched credentials
    creds = await cred_mgr.get_current_credentials()
    if not creds:
        raise RuntimeError("Failed to obtain AWS credentials")

    crt_stt = await create_crt_stt_from_awcreds(creds)

    # Example: run main loop (your actual logic will be different)
    try:
        while True:
            # Use crt_stt for streaming. If creds near expiry, swap to a refreshed instance.
            # Check expiry and re-create CRT STT if necessary (do re-create off-loop)
            await asyncio.sleep(5)

            cur = await cred_mgr.get_current_credentials()
            if cur and cur.expires_in < 70:  # if less than ~70s left, refresh CRT instance proactively
                # Create a new CRT STT on background thread to avoid blocking event loop
                new_stt = await recreate_stt_in_executor(cred_mgr, crt_stt)
                # swap
                old = crt_stt
                crt_stt = new_stt
                # tidy up old one
                old.close()
    finally:
        # cleanup
        crt_stt.close()
        await cred_mgr.stop()

# Helper to create CRT STT from AWSCreds (runs in executor)
async def create_crt_stt_from_awcreds(creds: AWSCreds) -> stt.STT:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, _blocking_create_crt_stt, creds)

def _blocking_create_crt_stt(creds: AWSCreds) -> stt.STT:
    # Convert AWSCreds -> stt.Credentials and construct STT (blocking init)
    stt_creds = stt.Credentials(creds.access_key, creds.secret, creds.token)
    # This constructor may perform blocking operations; run it in executor
    return stt.STT(credentials=stt_creds)

async def recreate_stt_in_executor(cred_mgr: CredentialManager, old_stt: stt.STT) -> stt.STT:
    """
    Fetch freshest creds from cred_mgr and create a new CRT STT in executor.
    Returns new STT instance; caller should close old one (non-blocking).
    """
    creds = await cred_mgr.get_current_credentials()
    if not creds:
        raise RuntimeError("No credentials available for STT recreation")
    new_stt = await create_crt_stt_from_awcreds(creds)
    return new_stt

# If running as a script/demo:
if __name__ == "__main__":
    asyncio.run(agent_entrypoint())

Notes and integration tips • The credential-manager uses boto3 in a background thread to fetch credentials. You can replace

_blocking_fetch_imds_credentials

with whatever mechanism you prefer (botocore IMDS client, STS AssumeRole, or HTTP call to IMDSv2). • Important: All blocking calls (IMDS fetch, CRT STT creation) run inside

run_in_executor

or a ThreadPoolExecutor so the asyncio event loop never blocks. • Use a single shared

aioboto3.Session()

for

tts.TTS(session=shared_session)

and

bedrock_llm.LLM(session=shared_session)

so those plugins reuse credential caching and reduce duplicate IMDS calls. • For CRT STT: ◦ Pass

stt.Credentials(...)

built from the pre-fetched creds. ◦ Recreate the CRT-based STT client before temporary credentials expire. Do the recreate on a background thread (as shown) and swap instances atomically from the async code. • Swap logic: ◦ Create new STT instance in executor, then swap the reference in the async code and call

old.close()

so teardown is non-blocking. ◦ Avoid recreating STT too often; refresh only when expiry is approaching. Why this prevents the "unresponsive" kills • The CRT SDK will no longer call IMDS directly during reconnects because it uses credentials