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 (🧪 Beta).

Beta Feature!

Please do not use in production or without explicit guidance from Albert. You might otherwise have a bad experience. This feature currently falls outside of the Albert support contract, but we'd love your feedback!

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.
chat_flags	ChatFlagCollection	Access to chat flag 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 os.getenv("ALBERT_TOKEN"),
    auth_manager=auth_manager,
)

chat_sessions

chat_sessions: ChatSessionCollection

chat_messages

chat_messages: ChatMessageCollection

chat_folders

chat_folders: ChatFolderCollection

chat_flags

chat_flags: ChatFlagCollection

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()