TURN Server for Amazon Bedrock AgentCore WebRTC

# Metered TURN for an AgentCore WebRTC agent (example)
import asyncio, json, os, urllib.parse, urllib.request

import boto3
from aiortc import RTCConfiguration, RTCIceServer

METERED_DOMAIN = os.environ["METERED_DOMAIN"]  # e.g. yourapp.metered.live
SECRET_ID = os.environ["METERED_SECRET_ID"]    # e.g. metered/turn-credential
_secrets = boto3.client("secretsmanager", region_name=os.environ.get("AWS_REGION"))


def _fetch_ice_servers() -> list[dict]:
    secret = json.loads(_secrets.get_secret_value(SecretId=SECRET_ID)["SecretString"])
    query = urllib.parse.urlencode({"apiKey": secret["apiKey"]})
    url = f"https://{METERED_DOMAIN}/api/v1/turn/credentials?{query}"
    with urllib.request.urlopen(url, timeout=5) as resp:  # raises on HTTP errors
        return json.load(resp)  # [{"urls", "username", "credential"}, ...]


async def get_ice_servers() -> list[dict]:
    # boto3 and urllib block, so keep them off the event loop
    return await asyncio.to_thread(_fetch_ice_servers)


def rtc_configuration(servers: list[dict]) -> RTCConfiguration:
    return RTCConfiguration(iceServers=[
        RTCIceServer(urls=s["urls"], username=s.get("username"), credential=s.get("credential"))
        for s in servers
    ])

import metered_turn  # replaces the sample's managed-TURN helper module


@asynccontextmanager
async def lifespan(app: FastAPI):
    # The sample's signaling-channel setup line goes: Metered needs no channel
    yield
    for pc in peer_connections.values():
        await pc.close()


@app.post("/invocations")
async def invocations(request: dict, background_tasks: BackgroundTasks):
    action = request.get("action")
    if action == "ice_config":
        return await _handle_ice_config()  # now async, so await it
    # ...offer, ice_candidate and disconnect stay as they are


async def _handle_ice_config():
    # The browser gets the ICE server list; it never needs your Metered API key
    return {"iceServers": await metered_turn.get_ice_servers()}


async def _handle_offer(data, background_tasks):
    servers = await metered_turn.get_ice_servers()
    pc = RTCPeerConnection(metered_turn.rtc_configuration(servers))
    # ...the rest of the sample's offer handling stays the same