@dataclass
class KyvosConfig:
base_url: str = ""
auth_token: str = ""
username: str = ""
password: str = ""
timeout_seconds: int = 60
max_retries: int = 5
retry_backoff_factor: float = 2.0
mcp_server_url: str = ""
mcp_pat: str = ""
mcp_folder_name: str = ""
mcp_aispace_name: str = ""
client_mode: str = "rest"Methods:
from_env() -> KyvosConfig— Load from environment variables
class KyvosService:
def __init__(
self,
config: KyvosConfig | None = None,
client: httpx.Client | None = None,
base_url: str | None = None,
auth_token: str | None = None,
username: str | None = None,
password: str | None = None,
) -> NoneLifecycle:
initialize() -> None— Login and validate sessionensure_authenticated() -> None— Re-login if neededlogin() -> str— Explicit login, returns session IDlogout() -> None— End session
Connection:
create_or_update_connection_xml(...) -> tuple[str, bool]create_or_update_connection_json(...) -> tuple[str, bool]create_connection_json(...) -> str
Folders:
create_folder(name, folder_type) -> tuple[str, str]list_registered_datasets_in_folder(folder_name) -> dict[str, str]
Datasets:
create_dataset(table, connection_name, folder_id, folder_name) -> tuple[str, str]create_dataset_json(table, connection_name, folder_id, folder_name) -> tuple[str, str]refresh_dataset_columns(dataset_id) -> dictget_dataset_column_details(folder_name, dataset_name) -> list[dict]
DRD:
create_dataset_relationship_drd(drd_xml) -> Nonecreate_dataset_relationship_drd_json(drd_json) -> strresolve_dataset_relationship_id(folder_name, drd_name, ...) -> strget_drd_object_xml(drd_id) -> strparse_drd_structure(drd_xml) -> dict
Semantic Model:
create_semantic_model_xml(smodel_xml) -> Nonecreate_semantic_model_json(smodel_json) -> None
Dashboard:
create_dashboard(dashboard, model_id) -> dict[str, str]
Properties:
session_id -> str | None
Raised when a Kyvos API call fails after retries.
class KyvosAuthService:
def __init__(self, base_url, username, password, ...) -> NoneMethods:
login() -> str— Authenticate and return session tokenlogout() -> None— End sessionis_session_valid() -> bool— Check if session is still activeget_session_token() -> str | None— Get current token
Exceptions:
KyvosAuthError— Base auth errorKyvosLoginError— Login failedKyvosLogoutError— Logout failedKyvosSessionExpiredError— Session expiredKyvosConnectionError— Cannot connect
Abstract interface for Kyvos entity operations. Both KyvosService and future MCPKyvosClient implement this protocol.
Returns REST or MCP client based on config.client_mode.
Returns MCP query client if configured, else None.
Decorator with exponential backoff for Kyvos API calls.
Configurable retry decorator.
Override default retry parameters.
Pydantic models: ColumnSpec, TableSpec, DatasetSpec, RelationshipSpec, MeasureSpec, HierarchySpec, SemanticModelSpec, DashboardSpec, DashboardWidgetSpec.
Query-only client for mcp-kyvos-server v2.2.0.
Methods:
execute_query(mdx_query) -> dictget_aispace(aispace_name) -> dictget_semantic_model(model_name) -> dictsearch_docs(query) -> list[dict]