Credentials

Everything on this page is imported from charter.auth, not charter. Charter never acquires or stores credentials. A CredentialProvider is the seam between the host application, which owns them, and the runtime, which needs a bearer token for the length of one request. A missing or rejected token raises CredentialError; the runtime never re-authenticates on its own.

`Credentials`

A bearer token, and optionally when it stops being valid.

str

required

The bearer token, sent as Authorization: Bearer ....

Optional[datetime]

When the token lapses. None means the caller has not said, and the token is treated as valid — the API is then the judge.

`Credentials.is_expired`

def is_expired(self, leeway_seconds: int = 0) -> bool: ...

Whether the token is past expires_at minus leeway_seconds. Credentials with no expires_at never report expired. A naive datetime is read as UTC, so a provider handing back a naive timestamp is not treated as far-future or far-past.

from datetime import datetime, timedelta, timezone

from charter.auth import Credentials

soon = Credentials(
    token="at-1", expires_at=datetime.now(timezone.utc) + timedelta(seconds=30)
)
assert soon.is_expired() is False
assert soon.is_expired(leeway_seconds=60) is True
assert Credentials(token="at-2").is_expired(leeway_seconds=3600) is False

`CredentialProvider`

class CredentialProvider(Protocol):
    async def get_credentials(self, provider: str) -> Credentials: ...

The protocol every provider satisfies. Runtime-checkable, so isinstance(obj, CredentialProvider) is a structural check on the method.

str

required

The identifier the tool was declared with — "google", "slack". It names the API, never the end user, so one implementation can serve several APIs. For who a call acts as, see SubjectProvider.

Anything with that one async method works: a secrets manager, a broker, a class of your own. A provider that caches can also define invalidate(credentials). It is not part of the protocol, and is called only if present: when the API refuses a token — a 401, or an envelope’s credential error — the runtime passes it the credentials it sent, so the next call can fetch fresh ones. The refused call still raises. A provider that holds nothing has nothing to drop and need not define it.

`StaticTokenProvider`

class StaticTokenProvider:
    def __init__(self, token: str, *, expires_at: Optional[datetime] = None) -> None: ...

Hands back one token, unchanged, for every provider. The right choice for a script, a test, or any place you already hold a valid access token. An empty token raises CredentialError at construction. Its repr masks the token.

from charter import oauth_tool_factory
from charter.auth import StaticTokenProvider

gmail = oauth_tool_factory(
    pack="mygmail",
    base_url="https://gmail.googleapis.com/",
    provider="google",
    credential_provider=StaticTokenProvider("test-token"),
)

`EnvTokenProvider`

class EnvTokenProvider:
    def __init__(self, var_name: str) -> None: ...

Reads the token from an environment variable on every call — re-read per call on purpose, so a sidecar that refreshes the variable is picked up without restarting the process.

str

required

The variable to read. An empty name raises CredentialError at construction; an unset or empty variable raises CredentialError at call time, naming the variable.

`CallbackProvider`

class CallbackProvider:
    def __init__(
        self, fn: Callable[[str], Union[Credentials, Awaitable[Credentials]]]
    ) -> None: ...

Delegates to a function of your own, sync or async — the hook for a credential store you already run. A non-callable raises CredentialError.

import asyncio

from charter.auth import CallbackProvider, Credentials

async def fetch(provider: str) -> Credentials:
    return Credentials(token=f"token-for-{provider}")

credentials = CallbackProvider(fetch)

async def main():
    assert (await credentials.get_credentials("google")).token == "token-for-google"

asyncio.run(main())

provider names the API, not the person: one of these serves one identity. For many, see SubjectProvider; to refresh a grant rather than read a stored token, see OAuth2Client.

`SubjectProvider`

class SubjectProvider:
    def __init__(
        self,
        factory: Callable[[str], Union[CredentialProvider, Awaitable[CredentialProvider]]],
        *,
        max_subjects: int = 1000,
    ) -> None: ...

One credential provider per end user, resolved per call. Give it a factory that builds a provider for a subject — typically reading that user’s refresh token from your database and returning an OAuth2Client.

Callable[[str], CredentialProvider | Awaitable[CredentialProvider]]

required

Builds the provider for one subject. Sync or async. Returning None raises CredentialError naming the subject. A non-callable raises CredentialError at construction.

int

default:"1000"

How many built providers to keep, evicted least-recently-used. Below 1 raises ValueError. Each provider carries its own cached token and refresh lock, so eviction removes both together.

Methods

Credentials

Resolves the current subject, then delegates. Concurrent cold starts for one subject are single-flighted, so twelve simultaneous calls read your token store once.

str

The subject for this call. Never falls back to a default or to whoever went last: an unset subject raises CredentialError naming what to do about it.

None

Drop a subject’s provider — after a revocation, or a sign-out.

bool

Disconnect one subject. Builds their provider if it is not held (your factory reads their stored grant, as for a tool call) and calls its revoke() — OAuth2Client.revoke. The subject is named rather than read from current_subject, since a disconnect is a request about one user, not a tool call made as one. Returns True when the server confirmed the revocation and False when the grant was already gone. An OAuth2Client that reads your store through a GrantLoader is kept afterwards: it refuses the revoked grant while your store still holds it, so a call made in the moment before you delete the row is not sent with it, and it takes up a new grant as soon as one is stored. Any other provider is forgotten, and so is one whose revocation raised, so the next lookup builds it from your store again. A provider with no revoke(), such as StaticTokenProvider, raises CredentialError pointing at revoke_token. Deleting the stored grant is yours, afterwards.

None

Pass a refused token on to the current subject’s provider. The provider, and the grant it holds, are kept.

int

How many providers are currently held.

A subject’s first call reads your store once however many calls arrive together: the rest wait on that one. If the call doing the read is cancelled (its client hung up, a timeout), one of the waiters reads instead, rather than all of them ending as though they had been cancelled too. Two limits worth knowing. One of these belongs to one event loop: resolution is a read, a reorder and sometimes an insert, and that sequence is not atomic across threads. And a subject evicted mid-refresh, against a server that rotates, is rebuilt from whatever your store holds — which needs max_subjects distinct other users inside one refresh, so raise the cap rather than tuning around it.

`current_subject`

current_subject: ContextVar[str] = ContextVar("charter_current_subject")

Who the next tool call acts for. Read by your provider, never by the runtime: the executor still calls get_credentials(name) and knows nothing else. A ContextVar is per-task in asyncio, so concurrent requests sharing one set of tools cannot see each other’s subject. Set it directly with current_subject.set(...) when you own the reset, or use the context manager below.

`use_subject`

@contextmanager
def use_subject(subject: str) -> Iterator[str]: ...

Set current_subject for the duration of a block. Resets on the way out, including on an exception, so a failed request cannot leave its identity behind for the next one. An empty subject raises CredentialError.

import asyncio

from charter.auth import CallbackProvider, Credentials, SubjectProvider, use_subject

async def for_user(subject: str) -> CallbackProvider:
    return CallbackProvider(lambda provider: Credentials(token=f"{subject}:{provider}"))

credentials = SubjectProvider(for_user)

async def main():
    with use_subject("u1"):
        assert (await credentials.get_credentials("google")).token == "u1:google"

asyncio.run(main())

Authorization servers — declaring the server an OAuth2Client refreshes against
The OAuth flow — obtaining the grant in the first place
OAuth2Client — the provider that renews a token

Core

Schema

Authentication

Runtime

Operating

`Credentials`

`Credentials.is_expired`

`CredentialProvider`

`StaticTokenProvider`

`EnvTokenProvider`

`CallbackProvider`

`SubjectProvider`

Methods

`current_subject`

`use_subject`

​Credentials

​Credentials.is_expired

​CredentialProvider

​StaticTokenProvider

​EnvTokenProvider

​CallbackProvider

​SubjectProvider

​Methods

​current_subject

​use_subject

​Related

`Credentials`

`Credentials.is_expired`

`CredentialProvider`

`StaticTokenProvider`

`EnvTokenProvider`

`CallbackProvider`

`SubjectProvider`

Methods

`current_subject`

`use_subject`

Related