langfuse-credentials.py

import os
import uuid
import time
import boto3
import logging
from typing import Optional, Tuple

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

def get_parameter(param_name: str) -> Optional[str]:
    """
    Retrieve a parameter from AWS Parameter Store.
    """
    try:
        ssm = boto3.client('ssm')
        response = ssm.get_parameter(
            Name=f"/helloworld/langfuse/{param_name}",
            WithDecryption=True
        )
        return response['Parameter']['Value']
    except Exception as exc:
        logger.error("Error retrieving parameter %s: %s", param_name, exc)
        return None


def get_credentials() -> Tuple[Optional[str], Optional[str], Optional[str]]:
    """
    Retrieve Langfuse credentials from environment variables or AWS Parameter Store.
    """
    secret_key = os.environ.get('LANGFUSE_SECRET_KEY') or get_parameter('secret_key')
    public_key = os.environ.get('LANGFUSE_PUBLIC_KEY') or get_parameter('public_key')
    host = os.environ.get('LANGFUSE_HOST') or get_parameter('host')
    return secret_key, public_key, host