Function get_jwt_token

agentops/validation.py:25–74

Exchange API key for JWT token asynchronously. Args: api_key: Optional API key. If not provided, uses AGENTOPS_API_KEY env var. Returns: JWT bearer token, or None if failed Note: This function never throws exceptions - all errors are handled gracefully

(api_key: Optional[str] = None)

async def get_jwt_token(api_key: Optional[str] = None) -> str:
    """
    Exchange API key for JWT token asynchronously.

    Args:
        api_key: Optional API key. If not provided, uses AGENTOPS_API_KEY env var.

    Returns:
        JWT bearer token, or None if failed

    Note:
        This function never throws exceptions - all errors are handled gracefully
    """
    try:
        if api_key is None:
            from agentops import get_client

            client = get_client()
            if client and client.config.api_key:
                api_key = client.config.api_key
            else:
                api_key = os.getenv("AGENTOPS_API_KEY")
                if not api_key:
                    logger.warning("No API key provided and AGENTOPS_API_KEY environment variable not set")
                    return None

        # Use a separate aiohttp session for validation to avoid conflicts
        import aiohttp

        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.agentops.ai/public/v1/auth/access_token",
                json={"api_key": api_key},
                timeout=aiohttp.ClientTimeout(total=10),
            ) as response:
                if response.status >= 400:
                    logger.warning(f"Failed to get JWT token: HTTP {response.status} - backend may be unavailable")
                    return None

                response_data = await response.json()

                if "bearer" not in response_data:
                    logger.warning("Failed to get JWT token: No bearer token in response")
                    return None

                return response_data["bearer"]

    except Exception as e:
        logger.warning(f"Failed to get JWT token: {e} - continuing without authentication")
        return None


def get_jwt_token_sync(api_key: Optional[str] = None) -> Optional[str]: