Function get_jwt_token_sync

agentops/validation.py:77–102

Synchronous wrapper for get_jwt_token - runs async function in event loop. 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 error

(api_key: Optional[str] = None)

def get_jwt_token_sync(api_key: Optional[str] = None) -> Optional[str]:
    """
    Synchronous wrapper for get_jwt_token - runs async function in event loop.

    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:
        import concurrent.futures

        # Always run in a separate thread to avoid event loop issues
        def run_in_thread():
            return asyncio.run(get_jwt_token(api_key))

        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(run_in_thread)
            return future.result()
    except Exception as e:
        logger.warning(f"Failed to get JWT token synchronously: {e}")
        return None


def get_trace_details(trace_id: str, jwt_token: str) -> Dict[str, Any]: