AsyncAlbert Client

albert.AsyncAlbert

AsyncAlbert(
    *,
    base_url: str | None = None,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | None = None,
    session: AsyncAlbertSession | None = None,
)

Async client for interacting with the Albert chat API.

Uses httpx.AsyncClient under the hood and must be closed when no longer needed — either by calling await client.aclose() or by using the client as an async context manager (async with AsyncAlbert(...) as client).

Parameters:

Name	Type	Description	Default
base_url	str	The base URL of the Albert API. Falls back to the ALBERT_BASE_URL environment variable or "https://app.albertinvent.com".	None
token	str	A static JWT token. Ignored when auth_manager is provided. Defaults to the ALBERT_TOKEN environment variable.	None
auth_manager	AlbertClientCredentials | AlbertSSOClient	An authentication manager for OAuth2-based token refresh. Ignored if token is provided.	None
session	AsyncAlbertSession	A fully configured async session. When provided, base_url, token, and auth_manager are all ignored.	None

Attributes:

Name	Type	Description
session	AsyncAlbertSession	The internal async session used for authenticated requests.
chat_sessions	ChatSessionCollection	Access to chat session API methods.
chat_messages	ChatMessageCollection	Access to chat message API methods.
chat_folders	ChatFolderCollection	Access to chat folder API methods.

Examples:

One-off usage with an async context manager:

async with AsyncAlbert(base_url=..., token=...) as client:
    sessions = [s async for s in client.chat_sessions.get_all()]

Long-lived client via FastAPI lifespan:

@asynccontextmanager
async def lifespan(app):
    app.state.albert = AsyncAlbert(base_url=..., token=...)
    yield
    await app.state.albert.aclose()

Methods:

Name	Description
from_token	Create an AsyncAlbert client using a static token for authentication.
from_client_credentials	Create an AsyncAlbert client using client credentials authentication.
aclose	Close the underlying HTTP client.

Source code in src/albert/client.py

def __init__(
    self,
    *,
    base_url: str | None = None,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | None = None,
    session: AsyncAlbertSession | None = None,
):
    if session is not None:
        self.session = session
        return

    if auth_manager and base_url and base_url != auth_manager.base_url:
        raise ValueError("`base_url` must match the URL used by the auth manager.")

    resolved_base_url = (
        base_url
        or (auth_manager.base_url if auth_manager else None)
        or default_albert_base_url()
    )

    self.session = AsyncAlbertSession(
        base_url=resolved_base_url,
        token=token or os.getenv("ALBERT_TOKEN"),
        auth_manager=auth_manager,
    )

session

session = AsyncAlbertSession(
    base_url=resolved_base_url,
    token=token or getenv("ALBERT_TOKEN"),
    auth_manager=auth_manager,
)

chat_sessions

chat_sessions: ChatSessionCollection

chat_messages

chat_messages: ChatMessageCollection

chat_folders

chat_folders: ChatFolderCollection

from_token

from_token(
    *, base_url: str | None = None, token: str
) -> AsyncAlbert

Create an AsyncAlbert client using a static token for authentication.

Parameters:

Name	Type	Description	Default
base_url	str | None	The base URL of the Albert API. Falls back to the ALBERT_BASE_URL environment variable or "https://app.albertinvent.com".	None
token	str	A static JWT token used for all requests.	required

Returns:

Type	Description
AsyncAlbert	A configured async client authenticated with the given token.

Source code in src/albert/client.py

@classmethod
def from_token(cls, *, base_url: str | None = None, token: str) -> AsyncAlbert:
    """
    Create an AsyncAlbert client using a static token for authentication.

    Parameters
    ----------
    base_url : str | None, optional
        The base URL of the Albert API. Falls back to the ``ALBERT_BASE_URL``
        environment variable or "https://app.albertinvent.com".
    token : str
        A static JWT token used for all requests.

    Returns
    -------
    AsyncAlbert
        A configured async client authenticated with the given token.
    """
    return cls(base_url=base_url, token=token)

from_client_credentials

from_client_credentials(
    *,
    base_url: str | None = None,
    client_id: str,
    client_secret: str,
) -> AsyncAlbert

Create an AsyncAlbert client using client credentials authentication.

Parameters:

Name	Type	Description	Default
base_url	str | None	The base URL of the Albert API. Falls back to the ALBERT_BASE_URL environment variable or "https://app.albertinvent.com".	None
client_id	str	The OAuth2 client ID.	required
client_secret	str	The OAuth2 client secret.	required

Returns:

Type	Description
AsyncAlbert	A configured async client that refreshes tokens automatically.

Source code in src/albert/client.py

@classmethod
def from_client_credentials(
    cls,
    *,
    base_url: str | None = None,
    client_id: str,
    client_secret: str,
) -> AsyncAlbert:
    """
    Create an AsyncAlbert client using client credentials authentication.

    Parameters
    ----------
    base_url : str | None, optional
        The base URL of the Albert API. Falls back to the ``ALBERT_BASE_URL``
        environment variable or "https://app.albertinvent.com".
    client_id : str
        The OAuth2 client ID.
    client_secret : str
        The OAuth2 client secret.

    Returns
    -------
    AsyncAlbert
        A configured async client that refreshes tokens automatically.
    """
    resolved_base_url = base_url or default_albert_base_url()
    creds = AlbertClientCredentials(
        id=client_id,
        secret=SecretStr(client_secret),
        base_url=resolved_base_url,
    )
    return cls(auth_manager=creds)

aclose

aclose() -> None

Close the underlying HTTP client.

Source code in src/albert/client.py

async def aclose(self) -> None:
    """Close the underlying HTTP client."""
    await self.session.aclose()