Connect Python and the CLI to AI Experiments on CloudFerro Cloud
Being able to open MLflow in a browser is a useful first check, but your Python program still needs its own connection settings and sign-in. Without them, it cannot send results to the instance.
The setup below pairs the instance address with a small authentication helper and checks the connection from a terminal using MLflow’s command-line interface (CLI). The helper obtains a bearer token that identifies your platform account, while instance membership and role permissions determine what your code may do.
Prerequisites
No. 1 A running instance and membership
Complete Your first AI Experiments instance on CloudFerro Cloud, or ask the owner of an existing instance for access and its tracking URL. Your sign-in email must be on the instance member list. AI Experiments members and roles on CloudFerro Cloud explains the project permission and instance roles needed for access. Use an experiment you own or one you have permission to edit.
No. 2 Platform credentials and a fresh two-factor code
Use the username and password for your CloudFerro portal account, together with the current six-digit TOTP code from your two-factor authentication app. The supplied helper uses the token endpoint https://identity.cloudferro.com/auth/realms/CloudFerro-Cloud/protocol/openid-connect/token and the public client mlflow-cli-authservice, as configured for these accounts.
No. 3 A Python environment and Bash terminal
Use Python 3.10 or newer with pip and support for virtual environments. The setup below installs MLflow 3.14.0, matching the interface used in this guide. A local computer is sufficient for the addition example. Python code for your first MLflow run on CloudFerro Cloud explains the program; save first_experiment.py and get_token.py in the working directory created below. A notebook should use this Python environment as its kernel and inherit the configured environment variables.
No. 4 MLflow client and tracking server concepts
Review how a Python client connects to a remote server in the MLflow tracking server documentation. For this procedure, the tracking server is your AI Experiments instance; use its Endpoint from My Portal when configuring the client.
Prepare the client environment
If you completed the Python example’s folder and package setup, the environment is already prepared. In your Bash terminal, return to it with:
cd "$HOME/ai-experiments-hello-world"
source .venv/bin/activate
You can then continue at Configure the instance connection below. A new terminal needs the connection settings and sign-in credentials again, but not another virtual environment.
If this is your first client setup, create the working directory:
mkdir -p "$HOME/ai-experiments-hello-world"
cd "$HOME/ai-experiments-hello-world"
Save both downloaded Python files in this directory. Keep their filenames unchanged. The virtual environment keeps the example’s Python packages separate from other projects. Create and activate it here, then install the client packages:
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "mlflow==3.14.0" requests
With .venv active, python uses its Python 3 interpreter.
Check that the two files are present and that the client imports successfully:
ls first_experiment.py get_token.py
python -c "import mlflow, requests; print('MLflow:', mlflow.__version__)"
The first command lists both filenames; the second prints MLflow: 3.14.0. The directory should contain:
ai-experiments-hello-world/
├── .venv/
├── first_experiment.py
└── get_token.py
Configure the instance connection
The instance Endpoint from My Portal supplies MLFLOW_TRACKING_URI. Replace the example URL below with that exact value; <instanceId> is a placeholder. Enter your username and password at the prompts, so the password is not part of a saved shell command. The password prompt hides what you type or paste; press Enter to submit it:
export MLFLOW_TRACKING_URI="https://mlflow-<instanceId>.mlops.cloudferro.com"
IFS= read -r -p "CloudFerro username/email: " MLFLOW_USERNAME
IFS= read -r -s -p "CloudFerro password: " MLFLOW_PASSWORD
printf '\n'
export MLFLOW_USERNAME MLFLOW_PASSWORD
unset MLFLOW_TOTP
Use the username or email and password entered on the CloudFerro sign-in page. The instance’s MLflow endpoint is separate from its S3 storage endpoint. Leaving MLFLOW_TOTP unset makes the helper request a current authenticator code each time.
Variable |
How the examples use it |
|---|---|
MLFLOW_TRACKING_URI |
The instance URL used by both Python and CLI examples. |
MLFLOW_USERNAME |
Platform username read by the authentication helper. |
MLFLOW_PASSWORD |
Platform password read by the authentication helper. |
MLFLOW_TOTP |
Optional environment variable containing a fresh two-factor code. When unset, the helper prompts for the code each time. The code itself is required for CloudFerro portal accounts. |
MLFLOW_TRACKING_TOKEN |
Token supplied to MLflow. The Python script assigns it after authentication; the CLI command below exports it in the shell. |
Keep credentials out of Python files and notebooks. MLFLOW_USERNAME and MLFLOW_PASSWORD are inputs to this helper; the token it obtains is what authenticates the MLflow client.
The token helper
Save get_token.py in your working directory:
"""Request an AI Experiments token using CloudFerro portal credentials."""
import os
import sys
import requests
TOKEN_URL = (
"https://identity.cloudferro.com/auth/realms/CloudFerro-Cloud/"
"protocol/openid-connect/token"
)
CLIENT_ID = "mlflow-cli-authservice"
def get_token() -> str:
missing = [name for name in ("MLFLOW_USERNAME", "MLFLOW_PASSWORD")
if not os.environ.get(name)]
if missing:
raise RuntimeError("Set these environment variables: " + ", ".join(missing))
totp = os.environ.get("MLFLOW_TOTP", "").strip()
if not totp:
print("TOTP code (6 digits): ", end="", file=sys.stderr, flush=True)
totp = input().strip()
response = requests.post(
TOKEN_URL,
data={
"client_id": CLIENT_ID,
"grant_type": "password",
"username": os.environ["MLFLOW_USERNAME"],
"password": os.environ["MLFLOW_PASSWORD"],
"totp": totp,
},
timeout=30,
)
try:
response.raise_for_status()
except requests.HTTPError:
try:
details = response.json()
except ValueError:
details = {}
fields = []
if isinstance(details, dict):
for name in ("error", "error_description"):
value = details.get(name)
if isinstance(value, str) and value:
fields.append(f"{name}={value}")
message = "; ".join(fields) or "No OAuth error details returned."
for secret in (os.environ["MLFLOW_USERNAME"],
os.environ["MLFLOW_PASSWORD"], totp):
if secret:
message = message.replace(secret, "[redacted]")
raise RuntimeError(
f"Token request failed (HTTP {response.status_code}): {message}"
) from None
token = response.json().get("access_token")
if not isinstance(token, str) or not token:
raise RuntimeError("The identity service returned no access token.")
return token
if __name__ == "__main__":
print(get_token())
The helper requests a token with mlflow-cli-authservice and waits up to 30 seconds for the identity service response. Its prompt goes to the terminal’s error stream, leaving standard output available for the token when Bash captures it.
CloudFerro portal accounts require TOTP two-factor authentication. A missing or expired code can produce 401 Invalid user credentials even when the password is correct. Codes rotate every 30 seconds. Enter a fresh code at the prompt, or set MLFLOW_TOTP immediately before the request. Leave that variable unset for the repeated interactive runs in this guide.
Check the connection from the CLI
Run this command in the configured terminal, from the directory containing get_token.py:
MLFLOW_TRACKING_TOKEN="$(python get_token.py)" &&
export MLFLOW_TRACKING_TOKEN &&
mlflow experiments search
The search runs only if token acquisition succeeds. It lists experiments visible to your account; the example’s first-experiment will appear after its first run. A successful search checks authentication and read access. The calculation will also check permission to log a run and upload an artifact.
If the helper reports error=invalid_grant; error_description=Invalid user credentials, re-enter the username and password with the prompts above and use a fresh code from the authenticator entry for that account. This message does not identify which login field was rejected. A successful browser session alone does not confirm the credentials stored in this terminal.
Resume in a new terminal
The virtual environment and Python files remain on disk when you close the terminal. In a new Bash terminal, return to the directory and activate the environment:
cd "$HOME/ai-experiments-hello-world"
source .venv/bin/activate
Set MLFLOW_TRACKING_URI again and repeat the username and password prompts above. You do not need to recreate .venv or reinstall the packages for each run. The calculation requests a new token when it starts.
Python and notebook use
The supplied calculation imports get_token and assigns the token to MLFLOW_TRACKING_TOKEN before calling MLflow. Other scripts or notebook cells can use the same setup:
import os
import mlflow
from get_token import get_token
mlflow.set_tracking_uri(os.environ["MLFLOW_TRACKING_URI"])
os.environ["MLFLOW_TRACKING_TOKEN"] = get_token()
Keep get_token.py in the notebook’s working directory or on its Python import path. A locally launched notebook can inherit settings from its launch shell. In a browser-hosted notebook, set them in the Python kernel: exporting variables in a separate JupyterLab terminal does not update an already running kernel. This setup connects the client; first_experiment.py supplies the logging operations.
The same program can run in an MLOps Studio notebook. Both Python files must be uploaded beside the notebook, and the selected kernel needs mlflow and requests. After configuring authentication in that kernel, an IPython cell can execute the existing example:
%run first_experiment.py --run-name notebook-first-total
The calculation executes in MLOps Studio and records its result in the configured AI Experiments instance. The notebook still asks for a fresh TOTP code.
Token expiry
Tokens are short-lived, about 15 minutes. The token response also contains a refresh_token tied to your SSO session, which lives much longer. Do not cache tokens on disk.
Long-running training jobs
Two limitations rule out naive token rotation in a running job:
The client reads MLFLOW_TRACKING_TOKEN once, at startup. Updating the variable later has no effect.
Framework integrations (TensorFlow/Keras autologging, etc.) call the tracking server internally, so you cannot catch their 401 responses yourself.
The solution is a small helper module that plugs into the client’s auth mechanism and refreshes the token before every request, using the refresh token grant. Download cloudferro_auth.py and save it in your project. The file contains the following code:
"""Reusable CloudFerro authentication for MLflow tracking with automatic token refresh.
MLflow reads MLFLOW_TRACKING_TOKEN once, when the tracking store is created, so
rotating that variable in a running process does not work. Instead, this module
registers a custom MLflow request auth provider (named "cloudferro") that MLflow
calls before *every* request. The provider returns a requests-compatible auth
object which keeps the CloudFerro SSO session alive via the refresh token grant.
This covers all MLflow traffic, including framework autologging (TensorFlow,
Keras, sklearn, ...), which calls MLflow internally and cannot be wrapped in
your own retry logic.
Refresh strategy: each refresh is scheduled at half of whichever deadline
comes first — the access token lifetime or the REFRESH TOKEN'S OWN expiry,
which encodes the SSO session idle deadline. Refreshing right before the
access token expires is too late whenever the session idle timeout is shorter
than the access token lifetime: refresh tokens die with the idle timeout
(`invalid_grant: Token is not active`). Each refresh resets the idle timer, so
staying ahead of it keeps the session alive indefinitely (up to the SSO
session max). If a refresh still fails, the session falls back to a full
re-login (password grant), prompting for a fresh TOTP code instead of killing
the run.
Usage:
import cloudferro_auth
cloudferro_auth.setup_mlflow("https://mlflow-<instanceId>.mlops.cloudferro.com")
mlflow.set_experiment("my-experiment")
...
Required environment variables:
MLFLOW_USERNAME - platform username (same as the web UI login)
MLFLOW_PASSWORD - platform password
Optional:
MLFLOW_TOTP - 6-digit 2FA code; prompted interactively when unset.
CLOUDFERRO_TOKEN_URL - defaults to the CloudFerro Cloud production realm
CLOUDFERRO_CLIENT_ID - defaults to "mlflow-cli-authservice"
"""
import base64
import json
import os
import sys
import threading
import time
import requests
from mlflow.tracking.request_auth.abstract_request_auth_provider import RequestAuthProvider
# Defaults point at the CloudFerro Cloud production realm. Override with
# CLOUDFERRO_TOKEN_URL / CLOUDFERRO_CLIENT_ID for other environments.
DEFAULT_TOKEN_URL = (
"https://identity.cloudferro.com/auth/realms/CloudFerro-Cloud/protocol/openid-connect/token"
)
DEFAULT_CLIENT_ID = "mlflow-cli-authservice"
PROVIDER_NAME = "cloudferro"
# Schedule each refresh at half of the relevant lifetime (access token or
# refresh token — see module docstring), never more often than the floor.
REFRESH_LIFETIME_FRACTION = 0.5
MIN_REFRESH_DELAY_SECONDS = 30
def _jwt_exp(token) -> float | None:
"""Read the exp claim from a JWT without verifying the signature."""
try:
payload = token.split(".")[1]
payload += "=" * (-len(payload) % 4)
return float(json.loads(base64.urlsafe_b64decode(payload))["exp"])
except Exception:
return None
class CloudFerroSession(requests.auth.AuthBase):
"""Holds the SSO session and injects a fresh Bearer token into every request."""
def __init__(self):
self.token_url = os.environ.get("CLOUDFERRO_TOKEN_URL", DEFAULT_TOKEN_URL)
self.client_id = os.environ.get("CLOUDFERRO_CLIENT_ID", DEFAULT_CLIENT_ID)
self.username = os.environ["MLFLOW_USERNAME"]
self.password = os.environ["MLFLOW_PASSWORD"]
self._lock = threading.Lock()
self._access_token = None
self._refresh_token = None
self._expires_at = 0.0
self._refresh_after = 0.0
self._refresh_exp = None
self._last_expires_in = 0
def _request(self, payload):
response = requests.post(self.token_url, data=payload, timeout=30)
if response.status_code != 200:
try:
error = response.json()
detail = f"{error.get('error')}: {error.get('error_description')}"
except ValueError:
detail = response.text
raise RuntimeError(
f"Token request failed (HTTP {response.status_code}): {detail}"
)
return response.json()
def _store(self, data):
self._access_token = data["access_token"]
# The server rotates the refresh token on every refresh — always keep the latest one.
self._refresh_token = data.get("refresh_token", self._refresh_token)
self._last_expires_in = data.get("expires_in", 300)
now = time.time()
self._expires_at = now + self._last_expires_in
# The refresh token's exp encodes the SSO session idle deadline.
# Refresh at half of whichever lifetime runs out first, so the
# session idle timer never trips before we use the token.
self._refresh_exp = _jwt_exp(self._refresh_token)
access_half = self._last_expires_in * REFRESH_LIFETIME_FRACTION
if self._refresh_exp is not None:
refresh_half = max(
(self._refresh_exp - now) * REFRESH_LIFETIME_FRACTION,
MIN_REFRESH_DELAY_SECONDS,
)
self._refresh_after = now + min(access_half, refresh_half)
else:
self._refresh_after = now + access_half
def _password_grant(self, reprompt: bool = False):
totp = os.environ.get("MLFLOW_TOTP", "").strip()
if sys.stdin.isatty():
prompt = (
"Fresh TOTP code for re-login (6 digits, Enter to reuse MLFLOW_TOTP or skip): "
if reprompt
else "TOTP code (6 digits, Enter to skip if your account has no 2FA): "
)
entered = input(prompt).strip()
totp = entered or totp
payload = {
"client_id": self.client_id,
"grant_type": "password",
"username": self.username,
"password": self.password,
}
if totp:
payload["totp"] = totp
self._store(self._request(payload))
refresh_ttl = (
int(self._refresh_exp - time.time()) if self._refresh_exp is not None else "unknown"
)
print(
f"[cloudferro_auth] logged in: access token valid {self._last_expires_in}s, "
f"refresh token valid ~{refresh_ttl}s",
file=sys.stderr,
)
def _refresh_grant(self):
self._store(
self._request(
{
"client_id": self.client_id,
"grant_type": "refresh_token",
"refresh_token": self._refresh_token,
}
)
)
print(
f"[cloudferro_auth] access token refreshed "
f"(next refresh in {int(self._refresh_after - time.time())}s)",
file=sys.stderr,
)
def token(self) -> str:
"""Return a valid access token, logging in, refreshing, or re-logging in as needed."""
with self._lock:
if self._access_token is None:
self._password_grant()
elif time.time() >= self._refresh_after:
try:
self._refresh_grant()
except RuntimeError as exc:
# The SSO session is gone (e.g. idle timeout). Do not kill
# the run — fall back to a full re-login with a fresh TOTP.
print(
f"[cloudferro_auth] refresh failed ({exc}); falling back to re-login",
file=sys.stderr,
)
self._password_grant(reprompt=True)
return self._access_token
def __call__(self, request):
request.headers["Authorization"] = f"Bearer {self.token()}"
return request
_session = None
def get_session() -> CloudFerroSession:
global _session
if _session is None:
_session = CloudFerroSession()
return _session
class CloudFerroRequestAuthProvider(RequestAuthProvider):
"""MLflow request auth provider. get_auth() is called before every MLflow request."""
def get_name(self):
return PROVIDER_NAME
def get_auth(self):
return get_session()
def _register_provider():
# No public registration API — the registry is a module-level singleton
# that MLflow consults via fetch_auth() on every request.
from mlflow.tracking.request_auth.registry import _request_auth_provider_registry
if not any(p.get_name() == PROVIDER_NAME for p in _request_auth_provider_registry):
_request_auth_provider_registry.register(CloudFerroRequestAuthProvider)
def setup_mlflow(tracking_uri: str) -> None:
"""Register the auth provider, perform the initial login, and set the tracking URI.
Fails fast on bad credentials or TOTP, before any training work starts.
"""
import mlflow
_register_provider()
# A static token would take precedence over the auth provider — make sure it is gone.
os.environ.pop("MLFLOW_TRACKING_TOKEN", None)
os.environ["MLFLOW_TRACKING_AUTH"] = PROVIDER_NAME
get_session().token()
mlflow.set_tracking_uri(tracking_uri)
Then activate it once at the top of your experiment:
import cloudferro_auth
cloudferro_auth.setup_mlflow("https://mlflow-<instanceId>.mlops.cloudferro.com")
setup_mlflow() logs you in once, prompting for a single TOTP code at the start of the process, and points the client at your instance. From then on every request, including autologging calls, automatically carries a fresh token. The module schedules each refresh at half of whichever deadline comes first: the access token’s lifetime or the refresh token’s own expiry, which encodes the SSO session idle timeout. This prevents the session idle timer from expiring. Each successful refresh resets that timer, keeping the session alive for the whole run. At login the module prints both token lifetimes to stderr, so you can see the limits upfront. If a refresh still fails, for example because the session expired while the job was paused, it does not kill your run. It falls back to a full re-login and asks for a fresh TOTP code.
The same setup covers training code that uses framework integrations: Keras/TensorFlow (mlflow.tensorflow.autolog()), scikit-learn, PyTorch, XGBoost and others. Those libraries call the tracking server through the same MLflow client, so every request they make goes through the provider and carries a fresh token too. Your training code needs no token handling. One setup_mlflow() call covers both your explicit mlflow.log_* calls and everything the integrations log internally.
If a long run returns 400 invalid_grant: Token is not active, the SSO session idle timeout elapsed before the token was refreshed. Use the current cloudferro_auth.py above, which refreshes halfway through the token lifetime and re-logs in on failure. If the error persists, ask your administrator to raise the session idle timeout.
What to do next
With the connection check complete, keep this terminal open and continue at Run the example. If first-total is already recorded, the article Run and compare your first MLflow runs on CloudFerro Cloud continues with Find the result in the MLflow UI.
For a hosted Python environment, the article Create and check a Kubeflow notebook on CloudFerro Cloud covers creating and opening a notebook in MLOps Studio.
The article Models, prompts, and AI Gateway on CloudFerro Cloud uses the same authentication for a configured gateway endpoint.
Token, permission, and connection errors are collected in Troubleshooting AI Experiments on CloudFerro Cloud.