Auth¶

The auth surface in mixpanel_data 0.4.0 is organized around three independent axes — Account, Project, Workspace — with three first-class account types, a single resolver, fluent in-session switching via Workspace.use(), and a Cowork bridge for remote authentication.

Explore on DeepWiki

🤖 Configuration Reference → (updated for 0.4.0)

Ask questions about account types, session axes, OAuth, the Cowork bridge, or in-session switching.

Overview¶

import mixpanel_data as mp

# Construct a Workspace from active config
ws = mp.Workspace()

# Override per Workspace (env > param > target > bridge > [active] > default_project)
ws = mp.Workspace(account="team", project="3713224")
ws = mp.Workspace(target="ecom")
ws = mp.Workspace(session=mp.Session(account=..., project=..., workspace=...))

# In-session switching — fluent, O(1), no re-auth on project swap
ws.use(project="3018488").events()
ws.use(account="personal").events()    # rebuilds auth header; preserves underlying HTTP client
ws.use(target="ecom").events()         # applies all three axes atomically
ws.use(workspace=3448414).events()

# Functional namespaces (also re-exported as mp.accounts / mp.session / mp.targets)
summaries = mp.accounts.list()
mp.accounts.use("team")
active = mp.session.show()             # ActiveSession
mp.targets.add("ecom", account="team", project="3018488", workspace=3448414)

See Configuration for the full setup walkthrough.

Account Types¶

Account is a Pydantic discriminated union over three first-class variants. The type field selects the variant; each variant carries the credentials it needs.

from mixpanel_data import (
    Account,                      # discriminated union type
    ServiceAccount,               # type == "service_account"
    OAuthBrowserAccount,          # type == "oauth_browser"
    OAuthTokenAccount,            # type == "oauth_token"
    AccountType,                  # Literal["service_account" | "oauth_browser" | "oauth_token"]
    Region,                       # Literal["us" | "eu" | "in"]
)

account: Account = ServiceAccount(
    name="team",
    region="us",
    default_project="3018488",
    username="team-mp...",
    secret="...",
)

if isinstance(account, ServiceAccount):
    print(f"SA {account.name} → project {account.default_project}")

ServiceAccount¶

Long-lived HTTP Basic Auth credentials. Best for CI / scripts / unattended automation.

mixpanel_data.ServiceAccount ¶

Bases: _AccountBase

Basic-auth service account credentials.

Long-lived credentials provisioned via the Mixpanel UI ("Service Accounts" section). Encodes username:secret as base64 for the Authorization header per the Mixpanel REST API spec.

Example

sa = ServiceAccount(
    name="team", region="us",
    username="sa.user", secret=SecretStr("hunter2"),
)
header = sa.auth_header(token_resolver=None)
# "Basic c2EudXNlcjpodW50ZXIy"

type `class-attribute` `instance-attribute` ¶

type: Literal['service_account'] = 'service_account'

Discriminator value for this variant.

username `instance-attribute` ¶

username: Annotated[str, Field(min_length=1)]

Service account username (e.g. sa.demo).

secret `instance-attribute` ¶

secret: SecretStr

Service account secret. Redacted in repr/str via Pydantic.

auth_header ¶

auth_header(*, token_resolver: TokenResolver | None = None) -> str

Return the Authorization header value for HTTP requests.

PARAMETER	DESCRIPTION
`token_resolver`	Ignored for service accounts (kept for signature parity with the other variants). TYPE: `TokenResolver \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	The `Basic <base64>` header value.

Source code in src/mixpanel_data/_internal/auth/account.py

def auth_header(
    self,
    *,
    token_resolver: TokenResolver | None = None,  # noqa: ARG002 — signature parity with OAuth variants
) -> str:
    """Return the ``Authorization`` header value for HTTP requests.

    Args:
        token_resolver: Ignored for service accounts (kept for signature
            parity with the other variants).

    Returns:
        The ``Basic <base64>`` header value.
    """
    raw = f"{self.username}:{self.secret.get_secret_value()}"
    encoded = base64.b64encode(raw.encode()).decode("ascii")
    return f"Basic {encoded}"

is_long_lived ¶

is_long_lived() -> bool

Return whether this account survives across restarts without refresh.

RETURNS	DESCRIPTION
`bool`	`True` — service account credentials never expire.

Source code in src/mixpanel_data/_internal/auth/account.py

def is_long_lived(self) -> bool:
    """Return whether this account survives across restarts without refresh.

    Returns:
        ``True`` — service account credentials never expire.
    """
    return True

OAuthBrowserAccount¶

PKCE browser flow; access/refresh tokens persisted at ~/.mp/accounts/{name}/tokens.json and auto-refreshed on expiry.

mixpanel_data.OAuthBrowserAccount ¶

Bases: _AccountBase

OAuth account authenticated via PKCE browser flow.

The Account itself carries no secret — tokens are persisted at ~/.mp/accounts/{name}/tokens.json and produced on demand by a :class:TokenResolver.

Example

a = OAuthBrowserAccount(name="me", region="us")
header = a.auth_header(token_resolver=resolver)
# "Bearer <access-token>"

type `class-attribute` `instance-attribute` ¶

type: Literal['oauth_browser'] = 'oauth_browser'

Discriminator value for this variant.

auth_header ¶

auth_header(*, token_resolver: TokenResolver) -> str

Return the Authorization header value for HTTP requests.

PARAMETER	DESCRIPTION
`token_resolver`	Resolver responsible for loading + refreshing the on-disk token. Required. TYPE: `TokenResolver`

RETURNS	DESCRIPTION
`str`	The `Bearer <token>` header value.

Source code in src/mixpanel_data/_internal/auth/account.py

def auth_header(self, *, token_resolver: TokenResolver) -> str:
    """Return the ``Authorization`` header value for HTTP requests.

    Args:
        token_resolver: Resolver responsible for loading + refreshing the
            on-disk token. Required.

    Returns:
        The ``Bearer <token>`` header value.
    """
    token = token_resolver.get_browser_token(self.name, self.region)
    return f"Bearer {token}"

is_long_lived ¶

is_long_lived() -> bool

Return whether this account survives across restarts without refresh.

RETURNS	DESCRIPTION
`bool`	`True` — refresh-token-driven re-issuance keeps the bearer valid.

Source code in src/mixpanel_data/_internal/auth/account.py

def is_long_lived(self) -> bool:
    """Return whether this account survives across restarts without refresh.

    Returns:
        ``True`` — refresh-token-driven re-issuance keeps the bearer valid.
    """
    return True

OAuthTokenAccount¶

Static bearer token (CI / agents) — supplied inline or via an env var (token_env).

mixpanel_data.OAuthTokenAccount ¶

Bases: _AccountBase

OAuth account using a static bearer token (CI, agents, ephemeral runs).

Exactly one of token (inline SecretStr) or token_env (env-var name) must be provided — never both, never neither. This is enforced at construction time by :meth:_validate_exactly_one_token_source.

Example

OAuthTokenAccount(name="ci", region="us", token=SecretStr("xyz"))
OAuthTokenAccount(name="agent", region="eu", token_env="MP_OAUTH_TOKEN")

type `class-attribute` `instance-attribute` ¶

type: Literal['oauth_token'] = 'oauth_token'

Discriminator value for this variant.

token `class-attribute` `instance-attribute` ¶

token: SecretStr | None = None

Inline static bearer token (mutually exclusive with token_env).

token_env `class-attribute` `instance-attribute` ¶

token_env: str | None = None

Env-var name to read the bearer from at resolution time.

auth_header ¶

auth_header(*, token_resolver: TokenResolver) -> str

Return the Authorization header value for HTTP requests.

PARAMETER	DESCRIPTION
`token_resolver`	Resolver responsible for materializing the token (from inline `SecretStr` or env var). Required. TYPE: `TokenResolver`

RETURNS	DESCRIPTION
`str`	The `Bearer <token>` header value.

Source code in src/mixpanel_data/_internal/auth/account.py

def auth_header(self, *, token_resolver: TokenResolver) -> str:
    """Return the ``Authorization`` header value for HTTP requests.

    Args:
        token_resolver: Resolver responsible for materializing the token
            (from inline ``SecretStr`` or env var). Required.

    Returns:
        The ``Bearer <token>`` header value.
    """
    token = token_resolver.get_static_token(self)
    return f"Bearer {token}"

is_long_lived ¶

is_long_lived() -> bool

Return whether this account survives across restarts without refresh.

RETURNS	DESCRIPTION
`bool`	`False` — the caller controls token rotation; no refresh path.

Source code in src/mixpanel_data/_internal/auth/account.py

def is_long_lived(self) -> bool:
    """Return whether this account survives across restarts without refresh.

    Returns:
        ``False`` — the caller controls token rotation; no refresh path.
    """
    return False

Session Axes¶

A Session is the immutable resolved state for a single Workspace at construction time — account, project, optional workspace, and the auth headers they generate.

mixpanel_data.Session ¶

Bases: BaseModel

Immutable in-memory tuple of (Account, Project, optional WorkspaceRef).

Holds the resolved auth/scope state for a single chain of API calls. Switching to a different account, project, or workspace produces a new Session via :meth:replace; the original is never mutated.

Workspace is optional: a session with workspace=None lazy-resolves on the first workspace-scoped API call (per FR-025).

account `instance-attribute` ¶

account: Account

Resolved account (one of the three discriminated variants).

project `instance-attribute` ¶

project: Project

Resolved Mixpanel project.

workspace `class-attribute` `instance-attribute` ¶

workspace: WorkspaceRef | None = None

Resolved workspace; None triggers lazy resolution on first use.

headers `class-attribute` `instance-attribute` ¶

headers: Mapping[str, str] = Field(default_factory=dict)

Custom HTTP headers attached at resolution time.

Populated from [settings].custom_header and/or bridge.headers. Never read from os.environ after Session construction (per FR-014).

Wrapped in :class:types.MappingProxyType after validation, so any in-place mutation (session.headers["X"] = "Y") raises :class:TypeError instead of silently sharing state across sessions. Consumers that need a mutable copy should use dict(session.headers).

project_id `property` ¶

project_id: str

Return the project's numeric string ID.

RETURNS	DESCRIPTION
`str`	`self.project.id`.

workspace_id `property` ¶

workspace_id: int | None

Return the workspace ID if set, else None.

RETURNS	DESCRIPTION
`int \| None`	`self.workspace.id` or `None`.

region `property` ¶

region: Region

Return the account's region.

RETURNS	DESCRIPTION
`Region`	`self.account.region`.

auth_header ¶

auth_header(*, token_resolver: TokenResolver | None) -> str

Return the Authorization header for HTTP requests.

PARAMETER	DESCRIPTION
`token_resolver`	Required for OAuth accounts; ignored for `ServiceAccount`. TYPE: `TokenResolver \| None`

RETURNS	DESCRIPTION
`str`	The header value (`Basic ...` or `Bearer ...`).

Source code in src/mixpanel_data/_internal/auth/session.py

def auth_header(self, *, token_resolver: TokenResolver | None) -> str:
    """Return the ``Authorization`` header for HTTP requests.

    Args:
        token_resolver: Required for OAuth accounts; ignored for
            ``ServiceAccount``.

    Returns:
        The header value (``Basic ...`` or ``Bearer ...``).
    """
    # Only OAuth variants need a resolver; type-narrowed by the discriminator.
    if self.account.type == "service_account":
        return self.account.auth_header(token_resolver=token_resolver)
    if token_resolver is None:
        raise TypeError(
            "TokenResolver is required to compute auth_header for OAuth accounts"
        )
    return self.account.auth_header(token_resolver=token_resolver)

replace ¶

replace(
    *,
    account: Account | None = None,
    project: Project | None = None,
    workspace: WorkspaceRef | None | _SentinelType = _SENTINEL,
    headers: Mapping[str, str] | _SentinelType = _SENTINEL,
) -> Session

Return a new Session with the supplied axes swapped in.

Workspace and headers use a sentinel because None (resp. {}) is a valid replacement value, semantically distinct from "do not touch this axis".

PARAMETER	DESCRIPTION
`account`	Replacement account; omitted preserves the current value. TYPE: `Account \| None` DEFAULT: `None`
`project`	Replacement project; omitted preserves the current value. TYPE: `Project \| None` DEFAULT: `None`
`workspace`	Replacement workspace; `None` clears the workspace (re-triggering lazy resolution); omitting the kwarg preserves the current value. TYPE: `WorkspaceRef \| None \| _SentinelType` DEFAULT: `_SENTINEL`
`headers`	Replacement headers map; `{}` clears all custom headers; omitting the kwarg preserves the current value. TYPE: `Mapping[str, str] \| _SentinelType` DEFAULT: `_SENTINEL`

RETURNS	DESCRIPTION
`Session`	A new :class:`Session` instance; the original is unchanged.

Source code in src/mixpanel_data/_internal/auth/session.py

def replace(
    self,
    *,
    account: Account | None = None,
    project: Project | None = None,
    workspace: WorkspaceRef | None | _SentinelType = _SENTINEL,
    headers: Mapping[str, str] | _SentinelType = _SENTINEL,
) -> Session:
    """Return a new Session with the supplied axes swapped in.

    Workspace and headers use a sentinel because ``None`` (resp. ``{}``)
    is a valid replacement value, semantically distinct from "do not
    touch this axis".

    Args:
        account: Replacement account; omitted preserves the current value.
        project: Replacement project; omitted preserves the current value.
        workspace: Replacement workspace; ``None`` clears the workspace
            (re-triggering lazy resolution); omitting the kwarg preserves
            the current value.
        headers: Replacement headers map; ``{}`` clears all custom headers;
            omitting the kwarg preserves the current value.

    Returns:
        A new :class:`Session` instance; the original is unchanged.
    """
    update: dict[str, Any] = {}
    if account is not None:
        update["account"] = account
    if project is not None:
        update["project"] = project
    if workspace is not _SENTINEL:
        update["workspace"] = workspace
    if headers is not _SENTINEL:
        update["headers"] = headers
    return self.model_copy(update=update)

mixpanel_data.Project ¶

Bases: BaseModel

Mixpanel project reference.

Project IDs come from the Mixpanel API as numeric strings. timezone and organization_id are populated when the resolver has access to a /me response; both are optional.

id `instance-attribute` ¶

id: Annotated[ProjectId, Field(min_length=1, pattern='^\\d+$')]

Numeric project ID (Mixpanel's wire format is a digit string).

name `class-attribute` `instance-attribute` ¶

name: str | None = None

Display name from /me, when known.

organization_id `class-attribute` `instance-attribute` ¶

organization_id: int | None = None

Owning organization ID from /me, when known.

timezone `class-attribute` `instance-attribute` ¶

timezone: str | None = None

Project timezone (e.g. "US/Pacific") from /me, when known.

mixpanel_data.WorkspaceRef ¶

Bases: BaseModel

Mixpanel workspace reference (cohort/dashboard scoping unit).

The data model is named WorkspaceRef to avoid colliding with the public Workspace facade class. Public re-export keeps the WorkspaceRef name.

The optional project_id lets a :class:Session cross-check that the workspace actually belongs to the bound project — every workspace lives inside exactly one project, and routing a workspace ID to the wrong project returns 400/404 deep inside the API call rather than at session construction. When populated (typically from a /me enumeration), the session-level model_validator raises :class:ValueError on mismatch instead of letting the bug surface as an HTTP error mid-query.

id `instance-attribute` ¶

id: Annotated[WorkspaceId, Field(gt=0)]

Positive integer workspace ID assigned by Mixpanel.

name `class-attribute` `instance-attribute` ¶

name: str | None = None

Display name from /me or /projects/{pid}/workspaces/public.

is_default `class-attribute` `instance-attribute` ¶

is_default: bool | None = None

Whether this is the project's default workspace, when known.

project_id `class-attribute` `instance-attribute` ¶

project_id: ProjectId | None = None

Owning project ID, when known.

Populated by /me enumeration paths so :class:Session can verify project ↔ workspace coupling. Left None when the workspace was constructed from a bare ID (e.g. MP_WORKSPACE_ID=N) — in that case the cross-check degrades to "trust the caller" rather than raising a spurious mismatch error.

mixpanel_data.auth_types.ActiveSession ¶

Bases: BaseModel

Persisted shape of the [active] block in ~/.mp/config.toml.

Only account and workspace live in [active]. Project lives on the account itself as Account.default_project — switching accounts implicitly switches projects (per FR-033). Any legacy [active].project field is rejected by extra="forbid" to surface the migration loudly.

Both fields are optional — environment variables or per-command flags can supply each one independently.

account `class-attribute` `instance-attribute` ¶

account: AccountName | None = None

Local config name of the active account (must reference [accounts.NAME]).

workspace `class-attribute` `instance-attribute` ¶

workspace: WorkspaceId | None = None

Active workspace ID (positive int) or None for lazy resolution.

Workspace.use() — In-Session Switching¶

Workspace.use() is the only in-session switching method. It returns self for fluent chaining and preserves the underlying httpx.Client and per-account /me cache across switches, so cross-project / cross-account iteration is O(1) per turn.

import mixpanel_data as mp

ws = mp.Workspace()                                # active session

# In-session switching (returns self for chaining)
ws.use(account="team")                              # implicitly clears workspace
ws.use(project="3018488")
ws.use(workspace=3448414)
ws.use(target="ecom")                               # apply all three at once

# Persist the new state
ws.use(project="3018488", persist=True)             # writes account.default_project; [active] only stores account + workspace

# Fluent chain
result = ws.use(project="3018488").segmentation(
    "Login", from_date="2026-04-01", to_date="2026-04-21"
)

Switching the active account clears the workspace (workspaces are project-scoped). The project re-resolves on account swap via env > explicit > new account's default_project. There is no silent cross-axis fallback: if an axis can't be resolved on the new account, use() raises ConfigError.

Swap one or more session axes in place; return self for chaining.

target= is mutually exclusive with account=/project=/ workspace=. The HTTP transport is preserved across all switches (per Research R5).

When account= is supplied, the project axis re-resolves through the FR-017 chain ending at the new account's default_project (env MP_PROJECT_ID > explicit project= > new account's default_project). If no source provides a project, the call raises :class:ConfigError per FR-033 — the prior session's project is NEVER carried forward across an account swap because cross-account project access is not guaranteed. The workspace axis is cleared on account swap (workspaces are project-scoped; the prior workspace doesn't apply to the new project) — explicit workspace= or MP_WORKSPACE_ID env override is honored.

PARAMETER	DESCRIPTION
`account`	Replacement account name. TYPE: `str \| None` DEFAULT: `None`
`project`	Replacement project ID. TYPE: `str \| None` DEFAULT: `None`
`workspace`	Replacement workspace ID. TYPE: `int \| None` DEFAULT: `None`
`target`	Apply this target's three axes atomically. TYPE: `str \| None` DEFAULT: `None`
`persist`	When `True`, also write the new state to `[active]`. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Workspace`	`self` for fluent chaining.

RAISES	DESCRIPTION
`ValueError`	Mutually exclusive args, or referenced name missing.
`OAuthError`	New auth header construction fails (atomic on success).
`ConfigError`	`account=` swap cannot resolve a project axis.

Source code in src/mixpanel_data/workspace.py

def use(
    self,
    *,
    account: str | None = None,
    project: str | None = None,
    workspace: int | None = None,
    target: str | None = None,
    persist: bool = False,
) -> Workspace:
    """Swap one or more session axes in place; return ``self`` for chaining.

    ``target=`` is mutually exclusive with ``account=``/``project=``/
    ``workspace=``. The HTTP transport is preserved across all switches
    (per Research R5).

    When ``account=`` is supplied, the project axis re-resolves through
    the FR-017 chain ending at the new account's ``default_project``
    (env ``MP_PROJECT_ID`` > explicit ``project=`` > new account's
    ``default_project``). If no source provides a project, the call
    raises :class:`ConfigError` per FR-033 — the prior session's
    project is NEVER carried forward across an account swap because
    cross-account project access is not guaranteed. The workspace
    axis is cleared on account swap (workspaces are project-scoped;
    the prior workspace doesn't apply to the new project) — explicit
    ``workspace=`` or ``MP_WORKSPACE_ID`` env override is honored.

    Args:
        account: Replacement account name.
        project: Replacement project ID.
        workspace: Replacement workspace ID.
        target: Apply this target's three axes atomically.
        persist: When ``True``, also write the new state to ``[active]``.

    Returns:
        ``self`` for fluent chaining.

    Raises:
        ValueError: Mutually exclusive args, or referenced name missing.
        OAuthError: New auth header construction fails (atomic on success).
        ConfigError: ``account=`` swap cannot resolve a project axis.
    """
    if target is not None and (
        account is not None or project is not None or workspace is not None
    ):
        raise ValueError(
            "`target=` is mutually exclusive with `account=`/`project=`/`workspace=`."
        )

    cm = ConfigManager()
    client = self._require_api_client()
    new_account_obj: _AccountUnion | None = None
    new_project_obj: _Project | None = None
    new_workspace_obj: _WorkspaceRef | None = None
    if target is not None:
        # Route through the same resolver as Workspace() construction so
        # env > param > target > bridge > config ordering applies (FR-017).
        # Without this, mid-process env-var overrides would be honored at
        # construction but silently ignored on `ws.use(target=...)`.
        sess = _resolve_session(
            target=target,
            config=cm,
            bridge=_load_bridge(),
        )
        new_account_obj = sess.account
        new_project_obj = sess.project
        new_workspace_obj = sess.workspace
    elif account is not None:
        # Explicit account swap: the user told us which account to use,
        # so the env-vars-override-param rule (FR-017) on the account
        # axis doesn't apply here — load the requested account directly.
        # Project re-resolves through the FR-017 chain ending at the
        # NEW account's default_project (env > explicit > new account's
        # default); raises ConfigError if nothing resolves (per FR-033,
        # cross-account project access is not guaranteed).
        # Workspace is cleared (workspaces are project-scoped; the
        # prior workspace is meaningless under the new account/project)
        # — explicit `workspace=` overrides the clear, and env override
        # via MP_WORKSPACE_ID still applies for parity with FR-017.
        new_account_obj = cm.get_account(account)
        br = _load_bridge()
        project_id = _resolve_project_axis(
            explicit=project,
            target_project=None,
            bridge=br,
            account=new_account_obj,
        )
        if project_id is None:
            raise ConfigError(_format_no_project_error(new_account_obj))
        new_project_obj = _Project(id=project_id)
        # Account-swap intentionally clears workspace per FR-033 (workspaces
        # are project-scoped; the prior workspace doesn't apply to the new
        # project). Only an explicit ``workspace=`` kwarg or a validated
        # ``MP_WORKSPACE_ID`` env var can populate it. We bypass
        # ``resolve_workspace_axis`` because that consults ``[active].workspace``
        # — which is exactly the fallback we need to skip here.
        if workspace is not None:
            new_workspace_obj = _WorkspaceRef(id=workspace)
        else:
            env_ws = _env_workspace_id()
            new_workspace_obj = (
                _WorkspaceRef(id=env_ws) if env_ws is not None else None
            )
    else:
        new_project_obj = _Project(id=project) if project is not None else None
        new_workspace_obj = (
            _WorkspaceRef(id=workspace) if workspace is not None else None
        )
    client.use(
        account=new_account_obj,
        project=new_project_obj,
        workspace=new_workspace_obj,
    )
    self._session = client.session

    # Clear lazy services so subsequent reads of `project` / `account` /
    # `workspaces()` / `_me_svc` observe the new session rather than the
    # prior one.
    self._account_name = self._session.account.name
    self._initial_workspace_id = (
        self._session.workspace.id if self._session.workspace else None
    )
    self._discovery = None
    self._live_query = None
    self._me_service = None

    if persist:
        self._persist_active()
    return self

Snapshot mode (parallel iteration)¶

For parallel cross-project iteration, snapshot the resolved Session and construct a fresh Workspace per task:

from concurrent.futures import ThreadPoolExecutor
import mixpanel_data as mp

ws = mp.Workspace()
sessions = [
    ws.session.replace(project=mp.Project(id=p.id))
    for p in ws.projects()
]

def event_count(s: mp.Session) -> int:
    return len(mp.Workspace(session=s).events())

with ThreadPoolExecutor(max_workers=4) as pool:
    counts = list(pool.map(event_count, sessions))

Functional Namespaces¶

The auth surface exposes three module-level namespaces re-exported from mixpanel_data. These are the canonical Python API for managing accounts, the active session, and saved targets.

`mp.accounts`¶

Account lifecycle: register, switch, probe, OAuth flows, bridge export.

mixpanel_data.accounts ¶

Public mp.accounts namespace.

Thin wrapper around :class:~mixpanel_data._internal.config.ConfigManager exposing account CRUD, switching, and probing operations as the canonical Python API for mp account ... CLI commands and the plugin's auth_manager.py.

Reference: specs/042-auth-architecture-redesign/contracts/python-api.md §5.

list ¶

list() -> builtins.list[AccountSummary]

Return all configured accounts as AccountSummary records.

RETURNS	DESCRIPTION
`list[AccountSummary]`	Sorted-by-name list of summaries.

Source code in src/mixpanel_data/accounts.py

def list() -> builtins.list[AccountSummary]:  # noqa: A001 — public namespace shadow
    """Return all configured accounts as ``AccountSummary`` records.

    Returns:
        Sorted-by-name list of summaries.
    """
    return _config().list_accounts()

add ¶

add(
    name: str,
    *,
    type: AccountType,
    region: Region,
    default_project: str | None = None,
    username: str | None = None,
    secret: SecretStr | str | None = None,
    token: SecretStr | str | None = None,
    token_env: str | None = None,
) -> AccountSummary

Add a new account.

Per FR-004, default_project is required at add-time for service_account and oauth_token. For oauth_browser it is OPTIONAL — the value gets backfilled by login(name) after the PKCE flow completes via a /me lookup.

Per FR-045, the first account added auto-promotes to [active].account. Subsequent accounts do not.

PARAMETER	DESCRIPTION
`name`	Account name (must match `^[a-zA-Z0-9_-]{1,64}$`). TYPE: `str`
`type`	One of `service_account` / `oauth_browser` / `oauth_token`. TYPE: `AccountType`
`region`	One of `us` / `eu` / `in`. TYPE: `Region`
`default_project`	Numeric project ID. Required for SA / oauth_token; optional for oauth_browser (backfilled by `login()`). TYPE: `str \| None` DEFAULT: `None`
`username`	Required for `service_account`. TYPE: `str \| None` DEFAULT: `None`
`secret`	Required for `service_account`. TYPE: `SecretStr \| str \| None` DEFAULT: `None`
`token`	For `oauth_token` (mutually exclusive with `token_env`). TYPE: `SecretStr \| str \| None` DEFAULT: `None`
`token_env`	For `oauth_token` (mutually exclusive with `token`). TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AccountSummary`	class:`AccountSummary` for the new account.

RAISES	DESCRIPTION
`ConfigError`	Validation failure or duplicate name.

Source code in src/mixpanel_data/accounts.py

def add(
    name: str,
    *,
    type: AccountType,  # noqa: A002 — matches contracts/python-api.md
    region: Region,
    default_project: str | None = None,
    username: str | None = None,
    secret: SecretStr | str | None = None,
    token: SecretStr | str | None = None,
    token_env: str | None = None,
) -> AccountSummary:
    """Add a new account.

    Per FR-004, ``default_project`` is required at add-time for
    ``service_account`` and ``oauth_token``. For ``oauth_browser`` it is
    OPTIONAL — the value gets backfilled by ``login(name)`` after the PKCE
    flow completes via a ``/me`` lookup.

    Per FR-045, the first account added auto-promotes to
    ``[active].account``. Subsequent accounts do not.

    Args:
        name: Account name (must match ``^[a-zA-Z0-9_-]{1,64}$``).
        type: One of ``service_account`` / ``oauth_browser`` / ``oauth_token``.
        region: One of ``us`` / ``eu`` / ``in``.
        default_project: Numeric project ID. Required for SA / oauth_token;
            optional for oauth_browser (backfilled by ``login()``).
        username: Required for ``service_account``.
        secret: Required for ``service_account``.
        token: For ``oauth_token`` (mutually exclusive with ``token_env``).
        token_env: For ``oauth_token`` (mutually exclusive with ``token``).

    Returns:
        :class:`AccountSummary` for the new account.

    Raises:
        ConfigError: Validation failure or duplicate name.
    """
    cm = _config()
    # Compose the add-and-promote-as-active sequence in a single _mutate()
    # transaction so a fresh process never sees the new account without its
    # promoted [active].account when it was the first account added.
    with cm._mutate() as raw:
        is_first = not (raw.get("accounts") or {})
        cm._apply_add_account(
            raw,
            name,
            type=type,
            region=region,
            default_project=default_project,
            username=username,
            secret=secret,
            token=token,
            token_env=token_env,
        )
        if is_first:
            cm._apply_set_active(raw, account=name)
    return show(name)

update ¶

update(
    name: str,
    *,
    region: Region | None = None,
    default_project: str | None = None,
    username: str | None = None,
    secret: SecretStr | str | None = None,
    token: SecretStr | str | None = None,
    token_env: str | None = None,
) -> AccountSummary

Update fields on an existing account in place.

Type cannot be changed via this function (remove + re-add for that). Type-incompatible fields raise ConfigError.

PARAMETER	DESCRIPTION
`name`	Account to update. TYPE: `str`
`region`	New region. TYPE: `Region \| None` DEFAULT: `None`
`default_project`	New `default_project` (digit string). TYPE: `str \| None` DEFAULT: `None`
`username`	New username (service_account only). TYPE: `str \| None` DEFAULT: `None`
`secret`	New secret (service_account only). TYPE: `SecretStr \| str \| None` DEFAULT: `None`
`token`	New inline token (oauth_token only). TYPE: `SecretStr \| str \| None` DEFAULT: `None`
`token_env`	New env-var name (oauth_token only). TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Updated`	class:`AccountSummary`. TYPE: `AccountSummary`

RAISES	DESCRIPTION
`ConfigError`	Account not found, type-incompatible field, or validation failure.

Source code in src/mixpanel_data/accounts.py

def update(
    name: str,
    *,
    region: Region | None = None,
    default_project: str | None = None,
    username: str | None = None,
    secret: SecretStr | str | None = None,
    token: SecretStr | str | None = None,
    token_env: str | None = None,
) -> AccountSummary:
    """Update fields on an existing account in place.

    Type cannot be changed via this function (remove + re-add for that).
    Type-incompatible fields raise ``ConfigError``.

    Args:
        name: Account to update.
        region: New region.
        default_project: New ``default_project`` (digit string).
        username: New username (service_account only).
        secret: New secret (service_account only).
        token: New inline token (oauth_token only).
        token_env: New env-var name (oauth_token only).

    Returns:
        Updated :class:`AccountSummary`.

    Raises:
        ConfigError: Account not found, type-incompatible field, or
            validation failure.
    """
    _config().update_account(
        name,
        region=region,
        default_project=default_project,
        username=username,
        secret=secret,
        token=token,
        token_env=token_env,
    )
    return show(name)

remove ¶

remove(name: str, *, force: bool = False) -> builtins.list[str]

Remove an account.

PARAMETER	DESCRIPTION
`name`	Account name. TYPE: `str`
`force`	When `True`, remove even if referenced by targets. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`list[str]`	List of orphaned target names (empty unless `force=True` and
`list[str]`	the account had references).

RAISES	DESCRIPTION
`ConfigError`	Missing account.
`AccountInUseError`	Referenced and `force=False`.

Source code in src/mixpanel_data/accounts.py

def remove(name: str, *, force: bool = False) -> builtins.list[str]:
    """Remove an account.

    Args:
        name: Account name.
        force: When ``True``, remove even if referenced by targets.

    Returns:
        List of orphaned target names (empty unless ``force=True`` and
        the account had references).

    Raises:
        ConfigError: Missing account.
        AccountInUseError: Referenced and ``force=False``.
    """
    return _config().remove_account(name, force=force)

use ¶

use(name: str) -> None

Switch the active account, clearing any prior workspace pin.

The new account becomes [active].account and any prior [active].workspace is dropped — workspaces are project-scoped, so a leftover workspace ID from a different account would resolve to a foreign workspace (or a 404) on the next Workspace() construction. Project lives on the account itself as :attr:Account.default_project, so it travels with the new account automatically — no separate project axis to reset.

Both writes happen in a single _mutate() transaction so the next process never sees a half-swapped state.

PARAMETER	DESCRIPTION
`name`	Account to make active. TYPE: `str`

RAISES	DESCRIPTION
`ConfigError`	Account does not exist.

Source code in src/mixpanel_data/accounts.py

def use(name: str) -> None:
    """Switch the active account, clearing any prior workspace pin.

    The new account becomes ``[active].account`` and any prior
    ``[active].workspace`` is dropped — workspaces are project-scoped, so
    a leftover workspace ID from a different account would resolve to a
    foreign workspace (or a 404) on the next ``Workspace()`` construction.
    Project lives on the account itself as
    :attr:`Account.default_project`, so it travels with the new account
    automatically — no separate project axis to reset.

    Both writes happen in a single ``_mutate()`` transaction so the
    next process never sees a half-swapped state.

    Args:
        name: Account to make active.

    Raises:
        ConfigError: Account does not exist.
    """
    cm = _config()
    with cm._mutate() as raw:
        cm._apply_set_active(raw, account=name)
        cm._apply_clear_active(raw, workspace=True)

show ¶

show(name: str | None = None) -> AccountSummary

Return the named account summary, or the active one if no name given.

PARAMETER	DESCRIPTION
`name`	Account name; if `None`, the active account is shown. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AccountSummary`	class:`AccountSummary`.

RAISES	DESCRIPTION
`ConfigError`	Account not found OR no active account configured.

Source code in src/mixpanel_data/accounts.py

def show(name: str | None = None) -> AccountSummary:
    """Return the named account summary, or the active one if no name given.

    Args:
        name: Account name; if ``None``, the active account is shown.

    Returns:
        :class:`AccountSummary`.

    Raises:
        ConfigError: Account not found OR no active account configured.
    """
    cm = _config()
    if name is None:
        active = cm.get_active().account
        if not active:
            raise ConfigError("No active account configured.")
        name = active
    summaries = {s.name: s for s in cm.list_accounts()}
    if name not in summaries:
        raise ConfigError(f"Account '{name}' not found.")
    return summaries[name]

test ¶

test(name: str | None = None) -> AccountTestResult

Probe /me for the named account and return the structured result.

Resolves the named account (or active account when name is None), constructs a short-lived :class:MixpanelAPIClient against /me, and reports whether the credentials are accepted plus the authenticated user identity / accessible-project count from the response body.

Never raises — every failure mode (account not found, missing credentials, OAuth refresh failure, HTTP error) is captured in result.error so the CLI can render a structured failure message and downstream tooling can color accounts as needs_login / needs_token based on the error string.

PARAMETER	DESCRIPTION
`name`	Account to test; `None` means the active account. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AccountTestResult`	class:`AccountTestResult` — `ok=True` with `user` populated
`AccountTestResult`	on success, or `ok=False` with `error` describing the failure.

Source code in src/mixpanel_data/accounts.py

def test(name: str | None = None) -> AccountTestResult:
    """Probe ``/me`` for the named account and return the structured result.

    Resolves the named account (or active account when ``name`` is None),
    constructs a short-lived :class:`MixpanelAPIClient` against ``/me``,
    and reports whether the credentials are accepted plus the
    authenticated user identity / accessible-project count from the
    response body.

    Never raises — every failure mode (account not found, missing
    credentials, OAuth refresh failure, HTTP error) is captured in
    ``result.error`` so the CLI can render a structured failure message
    and downstream tooling can color accounts as
    ``needs_login`` / ``needs_token`` based on the error string.

    Args:
        name: Account to test; ``None`` means the active account.

    Returns:
        :class:`AccountTestResult` — ``ok=True`` with ``user`` populated
        on success, or ``ok=False`` with ``error`` describing the failure.
    """
    try:
        summary = show(name)
    except ConfigError as exc:
        return AccountTestResult(
            account_name=name or "(none)", ok=False, error=str(exc)
        )

    cm = _config()
    try:
        account = cm.get_account(summary.name)
    except ConfigError as exc:  # pragma: no cover — show() already validated
        return AccountTestResult(account_name=summary.name, ok=False, error=str(exc))

    # Lazy imports to keep import-time cheap (httpx + threading pull in lots).
    from mixpanel_data._internal.api_client import MixpanelAPIClient
    from mixpanel_data._internal.auth.session import Project, Session
    from mixpanel_data._internal.me import MeResponse

    # ``MixpanelAPIClient`` requires a project to construct a Session even
    # though ``/me`` itself is project-agnostic. Use the account's default
    # when present, falling back to ``"0"`` so probes still work for fresh
    # ``oauth_browser`` accounts that have not yet been login'd.
    placeholder_project = account.default_project or "0"
    probe_session = Session(
        account=account,
        project=Project(id=placeholder_project),
    )

    api_client = MixpanelAPIClient(session=probe_session)
    try:
        try:
            me_raw = api_client.me()
        except Exception as exc:  # noqa: BLE001 — capture every failure mode
            return AccountTestResult(
                account_name=summary.name,
                ok=False,
                error=f"/me probe failed: {exc}",
            )
        try:
            me_resp = MeResponse.model_validate(me_raw)
        except Exception as exc:  # noqa: BLE001 — malformed payload
            return AccountTestResult(
                account_name=summary.name,
                ok=False,
                error=f"/me response could not be parsed: {exc}",
            )
        user: MeUserInfo | None = None
        if me_resp.user_id is not None and me_resp.user_email is not None:
            user = MeUserInfo(id=me_resp.user_id, email=me_resp.user_email)
        project_count = len(me_resp.projects) if me_resp.projects else 0
        return AccountTestResult(
            account_name=summary.name,
            ok=True,
            user=user,
            accessible_project_count=project_count,
        )
    finally:
        api_client.close()

login ¶

login(name: str, *, open_browser: bool = True) -> OAuthLoginResult

Run the OAuth browser flow for an oauth_browser account.

Drives the full PKCE login dance:

Validate name resolves to an oauth_browser account.
Run :meth:OAuthFlow.login (PKCE + browser callback + token exchange).
Persist the resulting tokens atomically to ~/.mp/accounts/{name}/tokens.json.
Probe /me to capture the authenticated user identity and (when missing) backfill account.default_project with the first accessible project.

The browser is opened by default; pass open_browser=False to skip the call (useful for headless environments where the user copies the authorization URL manually).

PARAMETER	DESCRIPTION
`name`	Account name (must be `oauth_browser` type). TYPE: `str`
`open_browser`	Whether to launch the system browser. When False, the authorize URL is printed to stderr for manual copy (CLI flag: `mp account login NAME --no-browser`). TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`An`	class:`OAuthLoginResult` describing the persistence paths, TYPE: `OAuthLoginResult`
`OAuthLoginResult`	token expiry, and (best-effort) authenticated user identity.

RAISES	DESCRIPTION
`ConfigError`	`name` is not configured or is not `oauth_browser`.
`OAuthError`	Any leg of the PKCE flow fails (registration, browser, callback, token exchange).

Source code in src/mixpanel_data/accounts.py

def login(
    name: str,
    *,
    open_browser: bool = True,
) -> OAuthLoginResult:
    """Run the OAuth browser flow for an ``oauth_browser`` account.

    Drives the full PKCE login dance:

    1. Validate ``name`` resolves to an ``oauth_browser`` account.
    2. Run :meth:`OAuthFlow.login` (PKCE + browser callback + token exchange).
    3. Persist the resulting tokens atomically to
       ``~/.mp/accounts/{name}/tokens.json``.
    4. Probe ``/me`` to capture the authenticated user identity and
       (when missing) backfill ``account.default_project`` with the first
       accessible project.

    The browser is opened by default; pass ``open_browser=False`` to
    skip the call (useful for headless environments where the user copies
    the authorization URL manually).

    Args:
        name: Account name (must be ``oauth_browser`` type).
        open_browser: Whether to launch the system browser. When False,
            the authorize URL is printed to stderr for manual copy
            (CLI flag: ``mp account login NAME --no-browser``).

    Returns:
        An :class:`OAuthLoginResult` describing the persistence paths,
        token expiry, and (best-effort) authenticated user identity.

    Raises:
        ConfigError: ``name`` is not configured or is not ``oauth_browser``.
        OAuthError: Any leg of the PKCE flow fails (registration, browser,
            callback, token exchange).
    """
    cm = _config()
    account = cm.get_account(name)
    if not isinstance(account, OAuthBrowserAccount):
        raise ConfigError(
            f"`mp account login` is only valid for oauth_browser accounts; "
            f"'{name}' is type '{account.type}'."
        )

    # Lazy imports — pull in OAuthFlow / Workspace only when actually logging in.
    from mixpanel_data._internal.api_client import MixpanelAPIClient
    from mixpanel_data._internal.auth.flow import OAuthFlow
    from mixpanel_data._internal.auth.session import Project, Session
    from mixpanel_data._internal.me import MeResponse

    flow = OAuthFlow(region=account.region)
    # ``persist=False`` skips the v2 ``~/.mp/oauth/tokens_{region}.json``
    # write — v3 owns ``~/.mp/accounts/{name}/tokens.json`` exclusively.
    tokens = flow.login(persist=False, open_browser=open_browser)

    tokens_path = _persist_browser_tokens(name, tokens)

    # /me probe: validates the freshly minted bearer + backfills the
    # account's default_project on first login.
    user: MeUserInfo | None = None
    chosen_project = account.default_project
    placeholder_project = chosen_project or "0"
    probe_session = Session(
        account=account,
        project=Project(id=placeholder_project),
    )
    api_client = MixpanelAPIClient(session=probe_session)
    try:
        try:
            me_raw = api_client.me()
            me_resp = MeResponse.model_validate(me_raw)
        except Exception as exc:  # noqa: BLE001 — re-raise as OAuthError below
            raise OAuthError(
                f"Login succeeded but `/me` probe failed: {exc}",
                code="OAUTH_TOKEN_ERROR",
                details={"account_name": name, "region": account.region},
            ) from exc
        if me_resp.user_id is not None and me_resp.user_email is not None:
            user = MeUserInfo(id=me_resp.user_id, email=me_resp.user_email)
        if chosen_project is None and me_resp.projects:
            # ``me_resp.projects`` keys are str at runtime; cast to ProjectId
            # to satisfy the typed contract on ``Account.default_project``.
            chosen_project = ProjectId(next(iter(sorted(me_resp.projects))))
            cm.update_account(name, default_project=chosen_project)
    finally:
        api_client.close()

    return OAuthLoginResult(
        account_name=name,
        user=user,
        expires_at=tokens.expires_at,
        tokens_path=tokens_path,
        client_path=_client_info_path(account.region),
    )

logout ¶

logout(name: str) -> None

Remove the on-disk OAuth tokens for an oauth_browser account.

PARAMETER	DESCRIPTION
`name`	Account name. TYPE: `str`

RAISES	DESCRIPTION
`ConfigError`	Account not found.

Source code in src/mixpanel_data/accounts.py

def logout(name: str) -> None:
    """Remove the on-disk OAuth tokens for an ``oauth_browser`` account.

    Args:
        name: Account name.

    Raises:
        ConfigError: Account not found.
    """
    summary = show(name)  # raises if missing
    tokens_path = account_dir(summary.name) / "tokens.json"
    if tokens_path.exists():
        tokens_path.unlink()

token ¶

token(name: str | None = None) -> str | None

Return the current bearer token for an OAuth account.

PARAMETER	DESCRIPTION
`name`	Account name; `None` means the active account. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str \| None`	For `service_account`: `None` (no bearer).
`str \| None`	For `oauth_browser`: the on-disk access token (raises `OAuthError`
`str \| None`	via the resolver if unavailable).
`str \| None`	For `oauth_token`: the inline / env-resolved token.

RAISES	DESCRIPTION
`ConfigError`	Account not found.
`OAuthError`	OAuth token cannot be resolved (missing tokens, missing env var, etc.).

Source code in src/mixpanel_data/accounts.py

def token(name: str | None = None) -> str | None:
    """Return the current bearer token for an OAuth account.

    Args:
        name: Account name; ``None`` means the active account.

    Returns:
        For ``service_account``: ``None`` (no bearer).
        For ``oauth_browser``: the on-disk access token (raises ``OAuthError``
        via the resolver if unavailable).
        For ``oauth_token``: the inline / env-resolved token.

    Raises:
        ConfigError: Account not found.
        OAuthError: OAuth token cannot be resolved (missing tokens, missing
            env var, etc.).
    """
    cm = _config()
    summary = show(name)
    account = cm.get_account(summary.name)
    resolver = OnDiskTokenResolver()
    if isinstance(account, ServiceAccount):
        return None
    if isinstance(account, OAuthBrowserAccount):
        return resolver.get_browser_token(account.name, account.region)
    if isinstance(account, OAuthTokenAccount):
        return resolver.get_static_token(account)
    raise ConfigError(  # pragma: no cover — Literal exhaustiveness
        f"Unknown account type for {summary.name!r}"
    )

export_bridge ¶

export_bridge(
    *,
    to: Path,
    account: str | None = None,
    project: str | None = None,
    workspace: int | None = None,
) -> Path

Export the named (or active) account as a v2 bridge file.

Resolves the account, attaches any [settings].custom_header as bridge.headers (B5 — header attaches in memory at resolution time for the consumer), and writes a 0o600 file at to via :func:bridge.export_bridge.

PARAMETER	DESCRIPTION
`to`	Destination path for the bridge file. TYPE: `Path`
`account`	Account to export; `None` means the active account. TYPE: `str \| None` DEFAULT: `None`
`project`	Optional pinned project ID. `None` omits the field. TYPE: `str \| None` DEFAULT: `None`
`workspace`	Optional pinned workspace ID. `None` omits the field. TYPE: `int \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Path`	The path that was written (same as `to`).

RAISES	DESCRIPTION
`ConfigError`	Account not found, no active account, or `BridgeFile` validation failure.
`OAuthError`	`account.type == "oauth_browser"` but no on-disk tokens are available.

Source code in src/mixpanel_data/accounts.py

def export_bridge(
    *,
    to: Path,
    account: str | None = None,
    project: str | None = None,
    workspace: int | None = None,
) -> Path:
    """Export the named (or active) account as a v2 bridge file.

    Resolves the account, attaches any ``[settings].custom_header`` as
    ``bridge.headers`` (B5 — header attaches in memory at resolution time
    for the consumer), and writes a 0o600 file at ``to`` via
    :func:`bridge.export_bridge`.

    Args:
        to: Destination path for the bridge file.
        account: Account to export; ``None`` means the active account.
        project: Optional pinned project ID. ``None`` omits the field.
        workspace: Optional pinned workspace ID. ``None`` omits the field.

    Returns:
        The path that was written (same as ``to``).

    Raises:
        ConfigError: Account not found, no active account, or
            ``BridgeFile`` validation failure.
        OAuthError: ``account.type == "oauth_browser"`` but no on-disk
            tokens are available.
    """
    from mixpanel_data._internal.auth.bridge import (
        export_bridge as _bridge_export,
    )

    cm = _config()
    name = account or cm.get_active().account
    if name is None:
        raise ConfigError("No account specified and no active account configured.")
    acct = cm.get_account(name)
    header = cm.get_custom_header()
    headers = {header[0]: header[1]} if header is not None else None
    return _bridge_export(
        acct,
        to=to,
        project=project,
        workspace=workspace,
        headers=headers,
        token_resolver=OnDiskTokenResolver(),
    )

remove_bridge ¶

remove_bridge(*, at: Path | None = None) -> bool

Remove the v2 bridge file at at (or the default path).

PARAMETER	DESCRIPTION
`at`	Bridge file path; `None` means `MP_AUTH_FILE` then the default search paths. TYPE: `Path \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	`True` if a file was deleted; `False` if none was found.

Source code in src/mixpanel_data/accounts.py

def remove_bridge(*, at: Path | None = None) -> bool:
    """Remove the v2 bridge file at ``at`` (or the default path).

    Args:
        at: Bridge file path; ``None`` means ``MP_AUTH_FILE`` then the
            default search paths.

    Returns:
        ``True`` if a file was deleted; ``False`` if none was found.
    """
    from mixpanel_data._internal.auth.bridge import (
        remove_bridge as _bridge_remove,
    )

    return _bridge_remove(at=at)

`mp.session`¶

Read and write the persisted [active] block.

mixpanel_data.session ¶

Public mp.session namespace.

Thin wrapper around :class:~mixpanel_data._internal.config.ConfigManager exposing the persisted [active] session and per-axis updates.

Note: this module shadows the :class:Session value type. Public callers access via import mixpanel_data; mp.session.show() (module) or import mixpanel_data; mp.Session(...) (the type).

Reference: specs/042-auth-architecture-redesign/contracts/python-api.md §7.

show ¶

show() -> ActiveSession

Return the persisted [active] block.

RETURNS	DESCRIPTION
`ActiveSession`	`ActiveSession` with `account` and `workspace` (each may be
`ActiveSession`	None). Project lives on the active account as
`ActiveSession`	`account.default_project` — to read it, fetch the account.

Source code in src/mixpanel_data/session.py

def show() -> ActiveSession:
    """Return the persisted ``[active]`` block.

    Returns:
        ``ActiveSession`` with ``account`` and ``workspace`` (each may be
        None). Project lives on the active account as
        ``account.default_project`` — to read it, fetch the account.
    """
    return _config().get_active()

use ¶

use(
    *,
    account: str | None = None,
    project: str | None = None,
    workspace: int | None = None,
    target: str | None = None,
) -> None

Update one or more axes in the persisted config.

account= and workspace= are written to [active]. project= is written to the active account's default_project (project lives on the account, not in [active]). target= is mutually exclusive with the per-axis kwargs and applies all three axes atomically (writing project to the target account's default_project).

All updates land in a single apply_session transaction so the on-disk state never reflects a partial swap (e.g., new account but stale project).

PARAMETER	DESCRIPTION
`account`	New active account name. TYPE: `str \| None` DEFAULT: `None`
`project`	New project ID (digit string) for the active account. TYPE: `str \| None` DEFAULT: `None`
`workspace`	New active workspace ID. TYPE: `int \| None` DEFAULT: `None`
`target`	Apply this target's three axes atomically. TYPE: `str \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	`target=` combined with any axis kwarg.
`ConfigError`	Referenced account or target not found, or `project=` supplied with no active account configured.

Source code in src/mixpanel_data/session.py

def use(
    *,
    account: str | None = None,
    project: str | None = None,
    workspace: int | None = None,
    target: str | None = None,
) -> None:
    """Update one or more axes in the persisted config.

    ``account=`` and ``workspace=`` are written to ``[active]``.
    ``project=`` is written to the **active account's** ``default_project``
    (project lives on the account, not in ``[active]``). ``target=`` is
    mutually exclusive with the per-axis kwargs and applies all three
    axes atomically (writing project to the target account's
    ``default_project``).

    All updates land in a single ``apply_session`` transaction so the
    on-disk state never reflects a partial swap (e.g., new account but
    stale project).

    Args:
        account: New active account name.
        project: New project ID (digit string) for the active account.
        workspace: New active workspace ID.
        target: Apply this target's three axes atomically.

    Raises:
        ValueError: ``target=`` combined with any axis kwarg.
        ConfigError: Referenced account or target not found, or
            ``project=`` supplied with no active account configured.
    """
    if target is not None and (
        account is not None or project is not None or workspace is not None
    ):
        raise ValueError(
            "`target=` is mutually exclusive with `account=`/`project=`/`workspace=`."
        )
    cm = _config()
    if target is not None:
        cm.apply_target(target)
        return
    cm.apply_session(account=account, project=project, workspace=workspace)

`mp.targets`¶

Manage saved (account, project, optional workspace) cursor positions.

mixpanel_data.targets ¶

Public mp.targets namespace.

Thin wrapper around :class:~mixpanel_data._internal.config.ConfigManager exposing target CRUD and activation. Targets are saved (account, project, workspace?) triples used as named cursor positions: mp.targets.use("ecom") writes all three axes to [active] in a single config save.

Reference: specs/042-auth-architecture-redesign/contracts/python-api.md §6.

list ¶

list() -> builtins.list[Target]

Return all configured targets sorted by name.

RETURNS	DESCRIPTION
`list[Target]`	Sorted list of :class:`Target` records.

Source code in src/mixpanel_data/targets.py

def list() -> builtins.list[Target]:  # noqa: A001 — public namespace shadow
    """Return all configured targets sorted by name.

    Returns:
        Sorted list of :class:`Target` records.
    """
    return _config().list_targets()

add ¶

add(
    name: str, *, account: str, project: str, workspace: int | None = None
) -> Target

Add a new target block.

PARAMETER	DESCRIPTION
`name`	Target name (block key). TYPE: `str`
`account`	Referenced account name (must exist). TYPE: `str`
`project`	Project ID (digit string). TYPE: `str`
`workspace`	Optional workspace ID. TYPE: `int \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Target`	The constructed :class:`Target`.

RAISES	DESCRIPTION
`ConfigError`	Duplicate name, missing account, or validation failure.

Source code in src/mixpanel_data/targets.py

def add(
    name: str,
    *,
    account: str,
    project: str,
    workspace: int | None = None,
) -> Target:
    """Add a new target block.

    Args:
        name: Target name (block key).
        account: Referenced account name (must exist).
        project: Project ID (digit string).
        workspace: Optional workspace ID.

    Returns:
        The constructed :class:`Target`.

    Raises:
        ConfigError: Duplicate name, missing account, or validation failure.
    """
    return _config().add_target(
        name, account=account, project=project, workspace=workspace
    )

remove ¶

remove(name: str) -> None

Remove a target block.

PARAMETER	DESCRIPTION
`name`	Target to remove. TYPE: `str`

RAISES	DESCRIPTION
`ConfigError`	Target does not exist.

Source code in src/mixpanel_data/targets.py

def remove(name: str) -> None:
    """Remove a target block.

    Args:
        name: Target to remove.

    Raises:
        ConfigError: Target does not exist.
    """
    _config().remove_target(name)

use ¶

use(name: str) -> None

Apply the target — write all three axes to [active] atomically.

PARAMETER	DESCRIPTION
`name`	Target to apply. TYPE: `str`

RAISES	DESCRIPTION
`ConfigError`	Target does not exist OR its referenced account is gone.

Source code in src/mixpanel_data/targets.py

def use(name: str) -> None:
    """Apply the target — write all three axes to ``[active]`` atomically.

    Args:
        name: Target to apply.

    Raises:
        ConfigError: Target does not exist OR its referenced account is gone.
    """
    _config().apply_target(name)

show ¶

show(name: str) -> Target

Return the named :class:Target.

PARAMETER	DESCRIPTION
`name`	Target name. TYPE: `str`

RETURNS	DESCRIPTION
`Target`	The Target record.

RAISES	DESCRIPTION
`ConfigError`	Target does not exist.

Source code in src/mixpanel_data/targets.py

def show(name: str) -> Target:
    """Return the named :class:`Target`.

    Args:
        name: Target name.

    Returns:
        The Target record.

    Raises:
        ConfigError: Target does not exist.
    """
    return _config().get_target(name)

Result Types¶

Read-only structured results returned from the namespaces above.

AccountSummary¶

mixpanel_data.AccountSummary ¶

Bases: BaseModel

Read-only summary of a configured account for mp account list.

Fields are derived from the persisted [accounts.NAME] block plus runtime context (is_active, referenced_by_targets). Status reflects the most recent mp account test outcome — "untested" is the default for accounts that have never been tested in this session.

Example

summary = AccountSummary(
    name="team", type="service_account", region="us",
    status="ok", is_active=True,
)

name `instance-attribute` ¶

name: str

Local config name (matches the TOML block key).

type `instance-attribute` ¶

type: AccountType

Discriminator value of the underlying Account variant.

region `instance-attribute` ¶

region: Region

Mixpanel region — us, eu, or in.

status `class-attribute` `instance-attribute` ¶

status: Literal['ok', 'needs_login', 'needs_token', 'untested'] = 'untested'

Result of the most recent mp account test (or "untested").

is_active `class-attribute` `instance-attribute` ¶

is_active: bool = False

True if [active].account == name.

referenced_by_targets `class-attribute` `instance-attribute` ¶

referenced_by_targets: list[str] = Field(default_factory=list)

Names of targets that reference this account.

AccountTestResult¶

mixpanel_data.AccountTestResult ¶

Bases: BaseModel

Outcome of mp account test NAME — captures the /me probe.

Never raises — error context is captured in error so the CLI can print structured failure messages and mp account list can color accounts as needs_login / needs_token based on the error code.

The ok/error fields are paired by an invariant: ok=True iff error is None. Constructing the model with both ok=True and a non-empty error (or ok=False and error=None) raises :class:pydantic.ValidationError to prevent ambiguous result states that would force callers to guess the right field to read.

account_name `instance-attribute` ¶

account_name: str

Account that was tested.

ok `instance-attribute` ¶

ok: bool

True if the /me request succeeded with valid credentials.

user `class-attribute` `instance-attribute` ¶

user: MeUserInfo | None = None

Authenticated principal identity, when ok is True.

accessible_project_count `class-attribute` `instance-attribute` ¶

accessible_project_count: int | None = None

Number of projects the account can read from /me.

error `class-attribute` `instance-attribute` ¶

error: str | None = None

Human-readable failure reason when ok is False.

OAuthLoginResult¶

mixpanel_data.OAuthLoginResult ¶

Bases: BaseModel

Outcome of mp.accounts.login(name) — captures the PKCE flow result.

Returned after a successful OAuth browser flow. user is populated from the immediate /me probe issued after the token exchange so callers can confirm "you are now logged in as alice@example.com" without needing a follow-up call.

account_name `instance-attribute` ¶

account_name: str

Account that was authenticated.

user `class-attribute` `instance-attribute` ¶

user: MeUserInfo | None = None

Authenticated principal identity from the post-login /me probe.

expires_at `class-attribute` `instance-attribute` ¶

expires_at: datetime | None = None

Access-token expiry (UTC) from the token endpoint response.

tokens_path `instance-attribute` ¶

tokens_path: Path

Where the tokens were persisted (~/.mp/accounts/{name}/tokens.json).

client_path `instance-attribute` ¶

client_path: Path

Where the DCR client info was persisted (~/.mp/accounts/{name}/client.json).

Target¶

mixpanel_data.Target ¶

Bases: BaseModel

A saved (account, project, workspace?) triple persisted in [targets.NAME].

Targets are named cursor positions: mp target use prod writes all three axes to [active] in a single config save. Workspace is optional — when omitted, the target resolves to the project's default workspace at use time (per FR-025 lazy resolution).

name `instance-attribute` ¶

name: TargetName

Local target name (matches the TOML block key).

account `instance-attribute` ¶

account: AccountName

Local config name of the referenced account (must exist).

project `instance-attribute` ¶

project: Annotated[ProjectId, Field(min_length=1, pattern='^\\d+$')]

Numeric project ID (Mixpanel's wire format).

workspace `class-attribute` `instance-attribute` ¶

workspace: Annotated[WorkspaceId, Field(gt=0)] | None = None

Optional workspace ID (must be a positive integer when set); None defers to lazy resolution. Mirrors WorkspaceRef.id's PositiveInt constraint so bad values fail at construction rather than corrupting downstream config.

Credential Resolution Chain¶

When constructing a Workspace, each axis is resolved independently in this priority order:

Environment variables — the resolver reads MP_USERNAME + MP_SECRET + MP_PROJECT_ID + MP_REGION (service-account quad), MP_OAUTH_TOKEN + MP_PROJECT_ID + MP_REGION (OAuth-token triple), MP_PROJECT_ID (project axis), and MP_WORKSPACE_ID (workspace axis). MP_ACCOUNT is not consumed by the Python resolver — it only feeds the CLI's --account / -a flag via Typer's envvar= default.
Constructor / CLI param — Workspace(account="..."), mp -a NAME ....
Saved target — Workspace(target="ecom"), mp -t ecom ....
Bridge file — MP_AUTH_FILE or ~/.claude/mixpanel/auth.json.
Persisted active session — the [active] block in ~/.mp/config.toml.
Account default — account.default_project for the project axis.

See Configuration → Credential Resolution Chain for examples.

Cowork Bridge (v2)¶

The Cowork bridge is a v2 JSON file that lets a remote VM authenticate against Mixpanel using your host machine's account and tokens. It embeds the full Account, optional OAuth tokens, and optional pinned project/workspace/headers.

from pathlib import Path
import mixpanel_data as mp

# On the host
mp.accounts.export_bridge(to=Path("~/.claude/mixpanel/auth.json").expanduser())
mp.accounts.remove_bridge()

# CLI equivalents
mp account export-bridge --to ~/.claude/mixpanel/auth.json
mp account remove-bridge
mp session --bridge          # show bridge-resolved state

Default search order: MP_AUTH_FILE → ~/.claude/mixpanel/auth.json → ./mixpanel_auth.json.

mixpanel_data.auth_types.BridgeFile ¶

Bases: BaseModel

Cowork credential bridge file — v2 schema.

Embeds a full :class:~mixpanel_data._internal.auth.account.Account record (with secrets inline) plus optional project / workspace pinning and a custom-headers map.

Example

{
  "version": 2,
  "account": {"type": "oauth_browser", "name": "personal", "region": "us"},
  "tokens": {"access_token": "...", "refresh_token": "...",
             "expires_at": "2026-04-22T12:00:00Z",
             "token_type": "Bearer", "scope": "read"},
  "project": "3713224",
  "workspace": 3448413,
  "headers": {"X-Mixpanel-Cluster": "internal-1"}
}

version `class-attribute` `instance-attribute` ¶

version: Literal[2] = 2

Bridge schema version — always 2.

account `instance-attribute` ¶

account: Account

Full Account discriminated-union record (with secrets inline by design).

tokens `class-attribute` `instance-attribute` ¶

tokens: OAuthTokens | None = None

OAuth tokens — required iff account.type == "oauth_browser".

project `class-attribute` `instance-attribute` ¶

project: Annotated[str | None, Field(default=None, pattern='^\\d+$')] = None

Optional pinned project ID (numeric string).

workspace `class-attribute` `instance-attribute` ¶

workspace: PositiveInt | None = None

Optional pinned workspace ID.

headers `class-attribute` `instance-attribute` ¶

headers: dict[str, str] = Field(default_factory=dict)

Custom HTTP headers attached to outbound requests at resolution time.

mixpanel_data.auth_types.load_bridge ¶

load_bridge(path: Path | None = None) -> BridgeFile | None

Load and validate a v2 bridge file from disk.

Resolves the path in this order:

Argument path (if not None).
$MP_AUTH_FILE env var (if set).
Default search paths (~/.claude/mixpanel/auth.json, then <cwd>/mixpanel_auth.json) — first existing file wins.

PARAMETER	DESCRIPTION
`path`	Optional explicit bridge path. TYPE: `Path \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`BridgeFile \| None`	The parsed :class:`BridgeFile`, or `None` if no candidate
`BridgeFile \| None`	path exists.

RAISES	DESCRIPTION
`ConfigError`	If a candidate file exists but is malformed or fails schema validation.

Source code in src/mixpanel_data/_internal/auth/bridge.py

def load_bridge(path: Path | None = None) -> BridgeFile | None:
    """Load and validate a v2 bridge file from disk.

    Resolves the path in this order:

    1. Argument ``path`` (if not None).
    2. ``$MP_AUTH_FILE`` env var (if set).
    3. Default search paths (``~/.claude/mixpanel/auth.json``, then
       ``<cwd>/mixpanel_auth.json``) — first existing file wins.

    Args:
        path: Optional explicit bridge path.

    Returns:
        The parsed :class:`BridgeFile`, or ``None`` if no candidate
        path exists.

    Raises:
        ConfigError: If a candidate file exists but is malformed or fails
            schema validation.
    """
    candidates: list[Path] = []
    if path is not None:
        candidates.append(path)
    elif "MP_AUTH_FILE" in os.environ and os.environ["MP_AUTH_FILE"]:
        candidates.append(Path(os.environ["MP_AUTH_FILE"]))
    else:
        candidates.extend(default_bridge_search_paths())

    for candidate in candidates:
        if not candidate.exists():
            continue
        try:
            payload = json.loads(candidate.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            raise ConfigError(
                f"Could not read bridge file at {candidate}: {exc}",
                details={"path": str(candidate)},
            ) from exc
        try:
            return _bridge_adapter.validate_python(payload)
        except ValidationError as exc:
            raise ConfigError(
                f"Invalid bridge file at {candidate}: "
                f"{exc.errors(include_url=False)[0]['msg']}",
                details={"path": str(candidate)},
            ) from exc
    return None

OAuth Token Plumbing¶

Low-level types for OAuth token handling. Most users never touch these directly — mp.accounts.login(name) drives the full flow and OnDiskTokenResolver materializes refreshed tokens automatically.

OAuthTokens¶

mixpanel_data.auth_types.OAuthTokens ¶

Bases: BaseModel

Immutable OAuth 2.0 token set with expiry tracking.

Stores access and optional refresh tokens along with metadata from the token response. The is_expired method includes a 30-second safety buffer to avoid using tokens that are about to expire.

ATTRIBUTE	DESCRIPTION
`access_token`	The OAuth access token (redacted in output). TYPE: `SecretStr`
`refresh_token`	The OAuth refresh token, if provided (redacted in output). TYPE: `SecretStr \| None`
`expires_at`	UTC datetime when the access token expires. TYPE: `datetime`
`scope`	Space-separated list of granted scopes. TYPE: `str`
`token_type`	Token type, typically `"Bearer"`. TYPE: `str`

access_token `instance-attribute` ¶

access_token: SecretStr

The OAuth access token (redacted in output).

refresh_token `class-attribute` `instance-attribute` ¶

refresh_token: SecretStr | None = None

The OAuth refresh token, if provided (redacted in output).

expires_at `instance-attribute` ¶

expires_at: datetime

UTC datetime when the access token expires.

Must be timezone-aware. Naive datetimes are rejected at validation time so a downstream consumer can never accidentally compare against an aware datetime.now(timezone.utc) and silently fall through the expiry check (Fix 25).

scope `instance-attribute` ¶

scope: str

Space-separated list of granted scopes.

token_type `instance-attribute` ¶

token_type: str

Token type, typically 'Bearer'.

is_expired ¶

is_expired() -> bool

Check whether the access token is expired or about to expire.

Uses a 30-second safety buffer to avoid sending tokens that are about to expire during in-flight requests.

RETURNS	DESCRIPTION
`bool`	True if the token is expired or will expire within 30 seconds.

Example

tokens = OAuthTokens.from_token_response(
    {"access_token": "x", "expires_in": 10,
     "scope": "read", "token_type": "Bearer"}
)
assert tokens.is_expired()  # 10s < 30s buffer

Source code in src/mixpanel_data/_internal/auth/token.py

def is_expired(self) -> bool:
    """Check whether the access token is expired or about to expire.

    Uses a 30-second safety buffer to avoid sending tokens that are
    about to expire during in-flight requests.

    Returns:
        True if the token is expired or will expire within 30 seconds.

    Example:
        ```python
        tokens = OAuthTokens.from_token_response(
            {"access_token": "x", "expires_in": 10,
             "scope": "read", "token_type": "Bearer"}
        )
        assert tokens.is_expired()  # 10s < 30s buffer
        ```
    """
    return datetime.now(timezone.utc) + timedelta(seconds=30) >= self.expires_at

from_token_response `classmethod` ¶

from_token_response(data: dict[str, object]) -> OAuthTokens

Create an OAuthTokens instance from a raw token endpoint response.

Computes expires_at by adding the expires_in value (in seconds) to the current UTC time.

PARAMETER	DESCRIPTION
`data`	Raw JSON response from the token endpoint. Must contain `access_token`, `expires_in`, `scope`, and `token_type`. May contain `refresh_token`. TYPE: `dict[str, object]`

RETURNS	DESCRIPTION
`OAuthTokens`	A new frozen OAuthTokens instance.

RAISES	DESCRIPTION
`KeyError`	If required keys are missing from `data`.
`ValueError`	If `expires_in` cannot be converted to an int.

Example

response = {
    "access_token": "eyJ...",
    "refresh_token": "dGhp...",
    "expires_in": 3600,
    "scope": "read:project",
    "token_type": "Bearer",
}
tokens = OAuthTokens.from_token_response(response)

Source code in src/mixpanel_data/_internal/auth/token.py

@classmethod
def from_token_response(cls, data: dict[str, object]) -> OAuthTokens:
    """Create an OAuthTokens instance from a raw token endpoint response.

    Computes ``expires_at`` by adding the ``expires_in`` value (in seconds)
    to the current UTC time.

    Args:
        data: Raw JSON response from the token endpoint. Must contain
            ``access_token``, ``expires_in``, ``scope``, and ``token_type``.
            May contain ``refresh_token``.

    Returns:
        A new frozen OAuthTokens instance.

    Raises:
        KeyError: If required keys are missing from ``data``.
        ValueError: If ``expires_in`` cannot be converted to an int.

    Example:
        ```python
        response = {
            "access_token": "eyJ...",
            "refresh_token": "dGhp...",
            "expires_in": 3600,
            "scope": "read:project",
            "token_type": "Bearer",
        }
        tokens = OAuthTokens.from_token_response(response)
        ```
    """
    expires_in_raw = data["expires_in"]
    expires_in = int(str(expires_in_raw))
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)

    raw_refresh = data.get("refresh_token")
    refresh_token: SecretStr | None = None
    if raw_refresh is not None:
        refresh_token = SecretStr(str(raw_refresh))

    return cls(
        access_token=SecretStr(str(data["access_token"])),
        refresh_token=refresh_token,
        expires_at=expires_at,
        scope=str(data.get("scope", "")),
        token_type=str(data["token_type"]),
    )

OAuthClientInfo¶

mixpanel_data.auth_types.OAuthClientInfo ¶

Bases: BaseModel

Immutable OAuth client registration metadata.

Stores client information from Dynamic Client Registration (RFC 7591) for reuse across sessions without re-registering.

ATTRIBUTE	DESCRIPTION
`client_id`	The OAuth client identifier. TYPE: `str`
`region`	Mixpanel data residency region (`us`, `eu`, or `in`). TYPE: `str`
`redirect_uri`	The redirect URI registered with the authorization server. TYPE: `str`
`scope`	Space-separated list of requested scopes. TYPE: `str`
`created_at`	UTC datetime when the client was registered. TYPE: `datetime`

client_id `instance-attribute` ¶

client_id: str

The OAuth client identifier.

region `instance-attribute` ¶

region: str

Mixpanel data residency region (us, eu, or in).

redirect_uri `instance-attribute` ¶

redirect_uri: str

The redirect URI registered with the authorization server.

scope `instance-attribute` ¶

scope: str

Space-separated list of requested scopes.

created_at `instance-attribute` ¶

created_at: datetime

UTC datetime when the client was registered.

TokenResolver Protocol¶

mixpanel_data.auth_types.TokenResolver ¶

Bases: Protocol

Protocol for producing bearer tokens for OAuth accounts.

Implementations decide how to fetch (and refresh) tokens for the two OAuth account variants. Concrete implementations live in :mod:mixpanel_data._internal.auth.token_resolver.

get_browser_token ¶

get_browser_token(name: str, region: Region) -> str

Return a fresh access token for an :class:OAuthBrowserAccount.

PARAMETER	DESCRIPTION
`name`	Account name (used to locate persisted tokens on disk). TYPE: `str`
`region`	Mixpanel region (used by some implementations). TYPE: `Region`

RETURNS	DESCRIPTION
`str`	The current access token (no `Bearer` prefix).

Source code in src/mixpanel_data/_internal/auth/account.py

def get_browser_token(self, name: str, region: Region) -> str:
    """Return a fresh access token for an :class:`OAuthBrowserAccount`.

    Args:
        name: Account name (used to locate persisted tokens on disk).
        region: Mixpanel region (used by some implementations).

    Returns:
        The current access token (no ``Bearer`` prefix).
    """
    ...

get_static_token ¶

get_static_token(account: OAuthTokenAccount) -> str

Return the static bearer for an :class:OAuthTokenAccount.

PARAMETER	DESCRIPTION
`account`	The account whose `token` or `token_env` to resolve. TYPE: `OAuthTokenAccount`

RETURNS	DESCRIPTION
`str`	The bearer token (no `Bearer` prefix).

Source code in src/mixpanel_data/_internal/auth/account.py

def get_static_token(self, account: OAuthTokenAccount) -> str:
    """Return the static bearer for an :class:`OAuthTokenAccount`.

    Args:
        account: The account whose ``token`` or ``token_env`` to resolve.

    Returns:
        The bearer token (no ``Bearer`` prefix).
    """
    ...

OnDiskTokenResolver¶

mixpanel_data.auth_types.OnDiskTokenResolver ¶

Bases: TokenResolver

Default resolver: tokens live on disk per account.

Reads OAuth browser tokens from ~/.mp/accounts/{name}/tokens.json written by :class:OAuthFlow. Reads static tokens from either the inline token field on the account or the environment variable named in token_env.

The resolver is intentionally I/O-light: the only side effects are reading files that already exist and (for expired browser tokens) refreshing via :meth:_refresh_and_persist, which delegates to :class:OAuthFlow.refresh_tokens and rewrites ~/.mp/accounts/{name}/tokens.json atomically via atomic_write_bytes. All failures surface as :class:OAuthError so callers can give actionable error messages.

get_browser_token ¶

get_browser_token(name: str, region: Region) -> str

Return a fresh access token for an :class:OAuthBrowserAccount.

Reads ~/.mp/accounts/{name}/tokens.json, checks the recorded expires_at (with a 30s safety buffer), and returns the token if not expired. If expired, refreshes via :meth:_refresh_and_persist; raises :class:OAuthError(code="OAUTH_REFRESH_ERROR") if no refresh token is recorded.

PARAMETER	DESCRIPTION
`name`	Account name (used to locate the tokens file). TYPE: `str`
`region`	Mixpanel region (kept for parity with the protocol; used by some refresh paths). TYPE: `Region`

RETURNS	DESCRIPTION
`str`	The current access token (no `Bearer` prefix).

RAISES	DESCRIPTION
`OAuthError`	If the tokens file is missing, malformed, expired without a refresh token, or refresh fails.

Source code in src/mixpanel_data/_internal/auth/token_resolver.py

def get_browser_token(self, name: str, region: Region) -> str:
    """Return a fresh access token for an :class:`OAuthBrowserAccount`.

    Reads ``~/.mp/accounts/{name}/tokens.json``, checks the recorded
    ``expires_at`` (with a 30s safety buffer), and returns the token
    if not expired. If expired, refreshes via
    :meth:`_refresh_and_persist`; raises
    :class:`OAuthError(code="OAUTH_REFRESH_ERROR")` if no refresh
    token is recorded.

    Args:
        name: Account name (used to locate the tokens file).
        region: Mixpanel region (kept for parity with the protocol;
            used by some refresh paths).

    Returns:
        The current access token (no ``Bearer`` prefix).

    Raises:
        OAuthError: If the tokens file is missing, malformed, expired
            without a refresh token, or refresh fails.
    """
    path = _account_tokens_path(name)
    if not path.exists():
        raise OAuthError(
            (
                f"No OAuth tokens found for account '{name}'. "
                f"Run `mp account login {name}` to authenticate."
            ),
            code="OAUTH_TOKEN_ERROR",
            details={"account_name": name, "path": str(path)},
        )
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise OAuthError(
            f"Could not read OAuth tokens for account '{name}' from {path}: {exc}",
            code="OAUTH_TOKEN_ERROR",
            details={"account_name": name, "path": str(path)},
        ) from exc

    # Single source of truth for parsing — `OAuthTokens` enforces the
    # tz-aware expiry invariant and the secret-wrapping in one place.
    # Any drift between how tokens are written vs read is structurally
    # impossible because both paths now go through the same model.
    try:
        tokens = OAuthTokens.model_validate_json(raw)
    except ValidationError as exc:
        raise OAuthError(
            (
                f"OAuth tokens for account '{name}' at {path} are malformed "
                f"or missing required fields. Re-run `mp account login {name}`."
            ),
            code="OAUTH_TOKEN_ERROR",
            details={
                "account_name": name,
                "path": str(path),
                "validation_error": str(exc),
            },
        ) from exc

    if tokens.is_expired():
        if tokens.refresh_token is None:
            raise OAuthError(
                (
                    f"OAuth access token for account '{name}' has "
                    f"expired and no refresh token is available. "
                    f"Re-run `mp account login {name}`."
                ),
                code="OAUTH_TOKEN_ERROR",
                details={
                    "account_name": name,
                    "region": region,
                    "path": str(path),
                },
            )
        return self._refresh_and_persist(
            name=name,
            region=region,
            path=path,
            tokens=tokens,
        )

    return tokens.access_token.get_secret_value()

get_static_token ¶

get_static_token(account: OAuthTokenAccount) -> str

Return the static bearer for an :class:OAuthTokenAccount.

Resolves the bearer from the inline token field if present; otherwise reads the environment variable named in token_env.

PARAMETER	DESCRIPTION
`account`	The account whose `token` or `token_env` to resolve. TYPE: `OAuthTokenAccount`

RETURNS	DESCRIPTION
`str`	The bearer token (no `Bearer` prefix).

RAISES	DESCRIPTION
`OAuthError`	If `token_env` is set but the env var is unset or empty.

Source code in src/mixpanel_data/_internal/auth/token_resolver.py

def get_static_token(self, account: OAuthTokenAccount) -> str:
    """Return the static bearer for an :class:`OAuthTokenAccount`.

    Resolves the bearer from the inline ``token`` field if present;
    otherwise reads the environment variable named in ``token_env``.

    Args:
        account: The account whose ``token`` or ``token_env`` to resolve.

    Returns:
        The bearer token (no ``Bearer`` prefix).

    Raises:
        OAuthError: If ``token_env`` is set but the env var is unset
            or empty.
    """
    if account.token is not None:
        return account.token.get_secret_value()
    env_name = account.token_env
    # The ``OAuthTokenAccount`` validator enforces ``token XOR token_env``,
    # so this branch is reachable only when ``token_env`` is set. We raise
    # explicitly (rather than ``assert env_name is not None``) so the
    # invariant survives ``python -O``, where assertions are stripped.
    if env_name is None:  # pragma: no cover — model invariant
        raise OAuthError(
            f"OAuth account '{account.name}' has neither `token` nor `token_env`.",
            code="OAUTH_TOKEN_ERROR",
            details={"account_name": account.name},
        )
    value = os.environ.get(env_name)
    if not value:
        raise OAuthError(
            (
                f"OAuth account '{account.name}' references env var "
                f"`{env_name}`, but it is not set or is empty."
            ),
            code="OAUTH_TOKEN_ERROR",
            details={"account_name": account.name, "env_var": env_name},
        )
    return value

Auth¶

Overview¶

Account Types¶

ServiceAccount¶

mixpanel_data.ServiceAccount ¶

type class-attribute instance-attribute ¶

username instance-attribute ¶

secret instance-attribute ¶

auth_header ¶

is_long_lived ¶

OAuthBrowserAccount¶

mixpanel_data.OAuthBrowserAccount ¶

type class-attribute instance-attribute ¶

auth_header ¶

is_long_lived ¶

OAuthTokenAccount¶

mixpanel_data.OAuthTokenAccount ¶

type class-attribute instance-attribute ¶

token class-attribute instance-attribute ¶

token_env class-attribute instance-attribute ¶

auth_header ¶

is_long_lived ¶

Session Axes¶

mixpanel_data.Session ¶

account instance-attribute ¶

project instance-attribute ¶

workspace class-attribute instance-attribute ¶

headers class-attribute instance-attribute ¶

project_id property ¶

workspace_id property ¶

region property ¶

auth_header ¶

replace ¶

mixpanel_data.Project ¶

id instance-attribute ¶

name class-attribute instance-attribute ¶

organization_id class-attribute instance-attribute ¶

timezone class-attribute instance-attribute ¶

mixpanel_data.WorkspaceRef ¶

id instance-attribute ¶

name class-attribute instance-attribute ¶

is_default class-attribute instance-attribute ¶

project_id class-attribute instance-attribute ¶

mixpanel_data.auth_types.ActiveSession ¶

account class-attribute instance-attribute ¶

workspace class-attribute instance-attribute ¶

Workspace.use() — In-Session Switching¶

Snapshot mode (parallel iteration)¶

Functional Namespaces¶

mp.accounts¶

mixpanel_data.accounts ¶

list ¶

add ¶

update ¶

remove ¶

use ¶

show ¶

test ¶

login ¶

logout ¶

token ¶

export_bridge ¶

remove_bridge ¶

mp.session¶

mixpanel_data.session ¶

show ¶

use ¶

mp.targets¶

mixpanel_data.targets ¶

list ¶

add ¶

remove ¶

use ¶

show ¶

Result Types¶

AccountSummary¶

mixpanel_data.AccountSummary ¶

name instance-attribute ¶

type instance-attribute ¶

region instance-attribute ¶

type `class-attribute` `instance-attribute` ¶

username `instance-attribute` ¶

secret `instance-attribute` ¶

type `class-attribute` `instance-attribute` ¶

type `class-attribute` `instance-attribute` ¶

token `class-attribute` `instance-attribute` ¶

token_env `class-attribute` `instance-attribute` ¶

account `instance-attribute` ¶

project `instance-attribute` ¶

workspace `class-attribute` `instance-attribute` ¶

headers `class-attribute` `instance-attribute` ¶

project_id `property` ¶

workspace_id `property` ¶

region `property` ¶

id `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

organization_id `class-attribute` `instance-attribute` ¶

timezone `class-attribute` `instance-attribute` ¶

id `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

is_default `class-attribute` `instance-attribute` ¶

project_id `class-attribute` `instance-attribute` ¶

account `class-attribute` `instance-attribute` ¶

workspace `class-attribute` `instance-attribute` ¶

`mp.accounts`¶

`mp.session`¶

`mp.targets`¶

name `instance-attribute` ¶

type `instance-attribute` ¶

region `instance-attribute` ¶

status `class-attribute` `instance-attribute` ¶

is_active `class-attribute` `instance-attribute` ¶

referenced_by_targets `class-attribute` `instance-attribute` ¶

account_name `instance-attribute` ¶

ok `instance-attribute` ¶

user `class-attribute` `instance-attribute` ¶

accessible_project_count `class-attribute` `instance-attribute` ¶

error `class-attribute` `instance-attribute` ¶

account_name `instance-attribute` ¶

user `class-attribute` `instance-attribute` ¶

expires_at `class-attribute` `instance-attribute` ¶

tokens_path `instance-attribute` ¶

client_path `instance-attribute` ¶

name `instance-attribute` ¶

account `instance-attribute` ¶

project `instance-attribute` ¶

workspace `class-attribute` `instance-attribute` ¶

version `class-attribute` `instance-attribute` ¶

account `instance-attribute` ¶

tokens `class-attribute` `instance-attribute` ¶

project `class-attribute` `instance-attribute` ¶

workspace `class-attribute` `instance-attribute` ¶

headers `class-attribute` `instance-attribute` ¶

access_token `instance-attribute` ¶

refresh_token `class-attribute` `instance-attribute` ¶

expires_at `instance-attribute` ¶

scope `instance-attribute` ¶

token_type `instance-attribute` ¶

from_token_response `classmethod` ¶

client_id `instance-attribute` ¶

region `instance-attribute` ¶

redirect_uri `instance-attribute` ¶

scope `instance-attribute` ¶

created_at `instance-attribute` ¶