Connect & Grants

This page covers every surface for getting a credential into the system and managing its lifecycle: the OAuth provider catalog, Connect sessions (user-consent flows), grant CRUD, agent-delegation revoke, managed-secret grant creation, and human-in-the-loop approvals.

# Provider catalog
catalog = await app.oauth_providers.list()

# Connect flows
results = await app.connect(providers=["google"])
session = await app.create_connect_session(allowed_providers=["github"])
results = await app.poll_connect_session(session.session_token)
session = await app.create_connect_session_for_error(caught_error)
session = await app.create_managed_secret_connect_session(...)

# End-user sign-in (IDP login)
auth         = await app.authenticate()                 # CLI convenience
auth_session = await app.create_auth_session()          # headless / remote user
result       = await app.poll_auth_session(auth_session.session_token)

# Grants
page   = await app.list_grants()
await app.revoke_grant(grant_id, reason="rotation")
result = await app.create_managed_secret_grant(...)

# Delegations
await app.revoke_delegation(grant_id, agent_id)  # operator
await agent.revoke_delegation(grant_id)          # agent self-revoke

# Approvals
final  = await app.await_approval(approval_id, timeout=600.0)
status = await app.get_approval_status(approval_id)

oauth_providers.list()

Fetch the OAuth provider catalog: the connectable providers configured for the platform and, per provider, the scopes you can request. Use it to drive provider pickers and scope selection instead of hardcoding scope strings.

async def list(*, force_refresh: bool = False) -> OAuthProviderCatalog

Parameter	Type	Default	Description
force_refresh	bool	False	Bypass the in-process cache and re-fetch.

Results are cached in-process for 5 minutes. The catalog lists only active (connectable) providers. Available on both App and Agent as app.oauth_providers / agent.oauth_providers.

Returns: OAuthProviderCatalog — providers keyed by id, plus get_default_scopes(provider) / get_required_scopes(provider) helpers.

Raises: BackendError (malformed catalog response), AlterSDKError, NetworkError. Requires the providers:read scope.

from alter_sdk import App

app = App(api_key="alter_rk_…")

catalog = await app.oauth_providers.list()
provider = catalog.providers["<provider_id>"]
print(provider.display_name, provider.default_scopes)

# Pre-fill a Connect flow with the provider's default scopes.
scopes = catalog.get_default_scopes("<provider_id>")

connect()

Headless OAuth flow for CLI tools, notebooks, and backend scripts. Mints a session, opens the browser, polls until the user completes consent.

async def connect(
    *,
    providers: list[str] | None = None,
    timeout: int = 300,
    poll_interval: float = 2.0,
    open_browser: bool = True,
    grant_policy: dict[str, Any] | None = None,
) -> list[ConnectResult]

Parameter	Type	Default	Description
providers	list[str] | None	None	Restrict to specific providers. None allows any provider configured on the app.
timeout	int	300	Max seconds to wait for completion.
poll_interval	float	2.0	Seconds between poll requests.
open_browser	bool	True	True opens the URL in the default browser; False prints the URL for the user to open manually (headless servers).
grant_policy	dict | None	None	TTL bounds: max_ttl_seconds (ceiling), default_ttl_seconds (pre-selected).

Returns: list of ConnectResult — one per provider the user authorized.

Raises: ConnectTimeoutError, ConnectDeniedError, ConnectConfigError, ConnectFlowError, AlterSDKError, NetworkError.

from alter_sdk import App

app = App(api_key="alter_rk_…")

results = await app.connect(providers=["google"], timeout=180)
for r in results:
    print(r.grant_id, r.provider_id, r.account_identifier)

create_connect_session()

Mint a Connect session and receive the connect_url. Use when the UI hosts its own redirect / popup / mobile flow rather than the headless connect() browser bridge.

async def create_connect_session(
    *,
    allowed_providers: list[str] | None = None,
    return_url: str | None = None,
    allowed_origin: str | None = None,
    metadata: dict[str, Any] | None = None,
    grant_policy: dict[str, Any] | None = None,
    required_scopes: dict[str, list[str]] | None = None,
    agent: str | None = None,
    user_token: str | None = None,
) -> ConnectSession

Parameter	Type	Default	Description
allowed_providers	list[str] | None	None	Restrict to specific providers.
return_url	str | None	None	URL to redirect after completion.
allowed_origin	str | None	None	Allowed origin for postMessage communication in popup flows.
metadata	dict | None	None	Request metadata (ip_address, user_agent) recorded with the session.
grant_policy	dict | None	None	TTL bounds. Validated against the input keyset — a typo (maxTtlSeconds instead of max_ttl_seconds) raises AlterValueError.
required_scopes	dict[str, list[str]] | None	None	Per-provider scope ceiling. Each scope must be in the app’s configured allowlist; the backend rejects attempts to widen with scope_not_allowed.
agent	str | None	None	Managed-agent UUID or name. When set, the resulting grant is delegated to this agent.
user_token	str | None	None	Explicit user JWT for identity binding. Wins over user_token_getter.

Returns: ConnectSession.

session = await app.create_connect_session(
    allowed_providers=["github"],
    allowed_origin="https://app.example.com",
    grant_policy={"max_ttl_seconds": 86400, "default_ttl_seconds": 3600},
)
return {"connect_url": session.connect_url, "session_token": session.session_token}

poll_connect_session()

Poll a Connect session token to completion. Used when the application minted the session itself and needs to block until the user finishes consent.

async def poll_connect_session(
    session_token: str,
    *,
    timeout: int = 300,
    poll_interval: float = 2.0,
) -> list[ConnectResult]

Parameter	Type	Default	Description
session_token	str	—	Token from create_connect_session() or create_connect_session_for_error(). Positional.
timeout	int	300	Max seconds to wait.
poll_interval	float	2.0	Seconds between polls.

Returns: list of ConnectResult.

Raises: ConnectTimeoutError, ConnectDeniedError, ConnectConfigError, ConnectFlowError, AlterSDKError.

session = await app.create_connect_session(allowed_providers=["slack"])
# … hand connect_url to the user …
results = await app.poll_connect_session(session.session_token, timeout=600)

create_connect_session_for_error()

Mint a recovery Connect session directly from a caught error. The method threads provider_id from the error into allowed_providers=[...], and agent_id (when present) into agent=, so the recovery session re-binds the same delegation target.

async def create_connect_session_for_error(
    error: NoDelegatedGrantError | GrantNotFoundError | CredentialRevokedError,
    *,
    allowed_origin: str | None = None,
    return_url: str | None = None,
    metadata: dict[str, Any] | None = None,
    grant_policy: dict[str, Any] | None = None,
    required_scopes: dict[str, list[str]] | None = None,
    user_token: str | None = None,
) -> ConnectSession

Parameter	Type	Default	Description
error	typed exception	—	The caught exception. Must carry provider_id.
allowed_origin / return_url / metadata / grant_policy / required_scopes			Standard create_connect_session parameters. Not inferred from the error.
user_token	str | None	None	Explicit user JWT. Falls back to user_token_getter.

Raises: AlterValueError if error.provider_id is None (no recovery context), or if a NoDelegatedGrantError has no agent_id set.

from alter_sdk import NoDelegatedGrantError

try:
    resp = await agent.request("GET", url, provider="google")
except NoDelegatedGrantError as e:
    session = await agent.create_connect_session_for_error(
        e,
        allowed_origin="https://app.example.com",
    )
    redirect_user_to(session.connect_url)
    results = await agent.poll_connect_session(session.session_token)
    # Retry with results[0].grant_id

app_user_id is intentionally not threaded through — create_connect_session binds the session to a user via user_token, not app_user_id. Pass user_token= explicitly when user binding is required.

create_managed_secret_connect_session()

Mint a Connect session for the user→agent delegation flow on a managed secret. Symmetric to create_connect_session() but targets the managed_secret_agent_delegations primitive.

async def create_managed_secret_connect_session(
    *,
    template_slug: str,
    delegated_agent_id: str,
    user_token: str,
    requested_ttl_seconds: int | None = None,
    delegated_agent_name: str | None = None,
    allowed_origin: str | None = None,
    return_url: str | None = None,
) -> ManagedSecretConnectSession

Parameter	Type	Default	Description
template_slug	str	—	Canonical template slug (e.g. "stripe-api-key"). Kebab-case required.
delegated_agent_id	str	—	Agent UUID the user is being asked to authorize. Must belong to the calling app and be active.
user_token	str	—	IDP JWT identifying the consenting user. Required — managed-secret delegation has no system-principal fallback.
requested_ttl_seconds	int | None	None	Caller-suggested TTL. Effective TTL is the minimum of requested, template max, source expiry, and 90 days.
delegated_agent_name	str | None	None	Display name shown on consent screen. Defaults to the agent’s stored display_name.
allowed_origin	str | None	None	Popup-flow postMessage origin.
return_url	str | None	None	Mobile-redirect return URL.

Returns: ManagedSecretConnectSession. The user opens connect_url; on Approve the backend returns a delegation_id (in the postMessage payload or return_url query) which the agent then passes as grant_id to request().

session = await app.create_managed_secret_connect_session(
    template_slug="stripe-api-key",
    delegated_agent_id=agent.id,
    user_token=user_jwt,
    requested_ttl_seconds=7 * 24 * 3600,
)
return {"connect_url": session.connect_url}

authenticate()

Trigger IDP login for an end user via the browser. Opens the app’s configured IDP login page, polls for completion, and returns the user’s JWT.

async def authenticate(*, timeout: float = 300.0) -> AuthResult

Parameter	Type	Default	Description
timeout	float	300.0	Max seconds to wait for the user to complete login.

Returns: AuthResult with user_token (JWT) and user_info.

Raises: AlterSDKError (no IDP configured), ConnectTimeoutError (user didn’t complete in time).

authenticate() is on App only — agents do not initiate user logins.

result = await app.authenticate(timeout=180.0)
# `result.user_token` is a JWT — keep it in memory; never log or print it.
email = result.user_info.get("email")

create_auth_session()

The split, headless counterpart to authenticate(). Mints a sign-in session and returns the IDP auth_url without opening a browser or mutating the instance. Hand auth_url to a user on any channel (a chat message, an MCP client, a printed link), then poll with poll_auth_session(). Because it installs no user_token_getter, it is safe on a shared App that resolves a JWT per request.

async def create_auth_session() -> AuthSession

Returns: AuthSession with session_token, auth_url, expires_in, and expires_at. session_token is persistable: a worker can store it, poll in the background, and resume after a restart.

Raises: AlterSDKError (no IDP configured, client closed), BackendError (malformed response), NetworkError. Requires the idp_users:write scope. App-only — agents do not start user logins.

auth_session = await app.create_auth_session()
# Send `auth_session.auth_url` to the user; never log it (it carries the session token).
return {"sign_in_url": auth_session.auth_url, "session": auth_session.session_token}

poll_auth_session()

The polling half of the link-based flow. Blocks until the user finishes IDP login, then returns their JWT. Like create_auth_session(), it installs no user_token_getter — the caller decides what to do with the token.

async def poll_auth_session(
    session_token: str,
    *,
    timeout: float = 300.0,
    poll_interval: float = 2.0,
) -> AuthResult

Parameter	Type	Default	Description
session_token	str	—	Token from create_auth_session().
timeout	float	300.0	Max seconds to wait. The TS SDK’s equivalent takes milliseconds.
poll_interval	float	2.0	Seconds between polls.

Transient network blips and non-200 responses are retried until the deadline; only a terminal IDP error, an expired session, or the timeout ends the loop.

Returns: AuthResult with user_token (JWT) and user_info.

Raises: AlterValueError (blank session_token), ConnectTimeoutError (deadline), AlterSDKError (closed, terminal IDP error / expired session). A permanent backend failure surfaces as its typed subclass (GrantNotFoundError, PolicyViolationError, ReAuthRequiredError, BackendError). Requires the idp_users:read scope.

auth_session = await app.create_auth_session()
result = await app.poll_auth_session(auth_session.session_token, timeout=600.0)

---

list_grants()

Paginated list of grants visible to the calling principal. The unified endpoint returns a discriminated-union result — each item is either an OAuthGrantItem (grant_kind="oauth") or a ManagedSecretGrantItem (grant_kind="managed_secret").

App.list_grants

async def list_grants(
    *,
    provider_id: str | None = None,
    status: str | None = None,
    account: str | None = None,
    end_user_token: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> UnifiedGrantListResult

Parameter	Type	Default	Description
provider_id	str | None	None	Filter to one provider. OAuth: the provider id (e.g. "google"). Managed secret: the per-secret slug (e.g. "stripe-production"), unique per app.
status	str | None	None	Filter by grant status (e.g. "active", "expired", "revoked").
account	str | None	None	Filter by account_identifier (multi-account disambiguation).
end_user_token	str | None	None	Scope the list to one end user by their JWT — their own grants plus group grants they are a live member of. Takes precedence over a configured user_token_getter; an invalid token is rejected with 401, never a silent fall-through to app scope.
limit	int	100	Page size, 1-1000.
offset	int	0	Page offset.

Scope. With no end-user scoping, App.list_grants returns every grant the app owns — both OAuth and managed-secret — across every principal kind (user, group, system, agent). Construct the App with a user_token_getter, or pass end_user_token, to narrow the list to one end user: their own grants plus any group grants they are a live member of. All filters are composable and AND-ed; each only narrows the result.

from alter_sdk import App, Provider

app = App(api_key="alter_rk_…")

page = await app.list_grants(provider_id=Provider.GOOGLE, limit=50)
for item in page.grants:
    if item.grant_kind == "oauth":
        print(item.grant_id, item.account_identifier, item.delegated_agent_ids)

Agent.list_grants

async def list_grants(
    *,
    provider_id: str | None = None,
    status: str | None = None,
    account: str | None = None,
    end_user_token: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> UnifiedGrantListResult

Parameter	Type	Default	Description
provider_id	str | None	None	Filter to one provider. OAuth: the provider id (e.g. "google"). Managed secret: the per-secret slug (e.g. "stripe-production"), unique per app.
status	str | None	None	Filter by grant status (e.g. "active", "expired", "revoked").
account	str | None	None	Filter by account_identifier.
end_user_token	str | None	None	Scope the agent’s delegated OAuth grants to one end user by their JWT. Agent-owned managed-secret grants have no end user, so they are unaffected.
limit	int	100	Page size, 1-1000.
offset	int	0	Page offset.

Scope. Returns only the grants the calling agent can reach: OAuth grants delegated to it, plus managed-secret grants it owns or that are delegated to it. The access_via field on each item marks ownership versus delegation (oauth_delegation / ms_delegation). Everything else is invisible — grants belonging to other agents, grants of users who did not delegate to this agent, and generic system grants. All filters are composable and AND-ed.

page = await agent.list_grants()
for item in page.grants:
    if item.grant_kind == "oauth":
        print("OAuth grant:", item.provider_id, item.account_identifier)
    else:
        print("Managed secret:", item.managed_secret_slug, item.label)

Raises (both): AlterValueError on bad pagination input, BackendError on backend / wire-shape failure, ReAuthRequiredError on JWT validation failures.

revoke_grant()

App only. Permanently revokes the specified grant. Tokens are deleted from the vault and the grant status flips to "revoked" — the user must re-authorize to restore access.

async def revoke_grant(
    grant_id: str,
    *,
    reason: str | None = None,
) -> RevokeGrantResult

Parameter	Type	Default	Description
grant_id	str	—	Grant to revoke. Positional.
reason	str | None	None	Optional reason recorded in the audit log.

Returns: RevokeGrantResult.

Raises: AlterSDKError (closed), GrantNotFoundError, NetworkError, TimeoutError, BackendError.

result = await app.revoke_grant(grant_id, reason="key_rotation")
print(result.success, result.revoked_at)

Note

When user_token_getter is configured, the caller’s JWT is auto-included as user_token, and the backend scopes the revoke to that user’s grants only — preventing cross-user revocation. When no getter is configured, the call runs as the operator and may target any grant the app principal owns.

revoke_delegation()

Revoke an agent’s delegation on an OAuth grant. The OAuth grant itself stays active; only the named agent’s access path is removed.

App.revoke_delegation

Operator path — both grant_id and agent_id are required.

async def revoke_delegation(grant_id: str, agent_id: str) -> None

Parameter	Type	Default	Description
grant_id	str	—	OAuth grant whose delegation should be revoked.
agent_id	str	—	Agent UUID whose delegation should be revoked.

Raises: AlterValueError (empty inputs), BackendError (use_self_revoke_path if the caller is an agent).

await app.revoke_delegation(grant_id="…", agent_id="…")

Agent.revoke_delegation

Self-revoke — the agent revokes its own delegation. Idempotent.

async def revoke_delegation(grant_id: str) -> None

Parameter	Type	Default	Description
grant_id	str	—	OAuth grant whose delegation should be revoked.

Raises: BackendError (agent_key_required), NetworkError, TimeoutError.

await agent.revoke_delegation(grant_id="…")

create_managed_secret_grant()

App only. Create a managed-secret grant bound to a user / group / system / agent principal. The discriminator on principal decides the binding shape.

async def create_managed_secret_grant(
    managed_secret_id: str,
    *,
    principal: Principal,
    grant_policy: dict[str, Any] | None = None,
) -> CreateGrantResult

Parameter	Type	Default	Description
managed_secret_id	str	—	Managed-secret template id. Positional.
principal	Principal	—	Discriminated principal model (see below).
grant_policy	dict | None	None	Optional grant policy (e.g. {"expires_at": "2026-12-31T00:00:00Z"}). Validated against the input keyset.

Returns: CreateGrantResult.

Principal is a discriminated union over four shapes:

Type	Discriminator (type)	Required fields
UserPrincipal	"user"	user_token, label
GroupPrincipal	"group"	external_group_id, idp_id, label
SystemPrincipal	"system"	(label optional)
AgentPrincipal	"agent"	Not accepted by this SDK method. Create agent-bound managed-secret grants via the Developer Portal flow instead.

from alter_sdk import App
from alter_sdk.models import UserPrincipal, GroupPrincipal, SystemPrincipal

app = App(api_key="alter_rk_…")

# User-bound
result = await app.create_managed_secret_grant(
    managed_secret_id="ms_stripe_prod",
    principal=UserPrincipal(user_token=user_jwt, label="Acme Stripe"),
)

# Group-bound (requires an identity provider that supports group grants)
result = await app.create_managed_secret_grant(
    managed_secret_id="ms_datadog",
    principal=GroupPrincipal(
        idp_id="11111111-2222-3333-4444-555555555555",
        external_group_id="oncall-team",
        label="On-call rotation",
    ),
)

# System (server-to-server)
result = await app.create_managed_secret_grant(
    managed_secret_id="ms_internal_notion",
    principal=SystemPrincipal(label="weekly_report_cron"),
)

# Agent-bound principals are not accepted by this SDK call.
# Create agent-bound managed-secret grants via the Developer Portal.

The dict shorthand also works:

result = await app.create_managed_secret_grant(
    managed_secret_id="ms_openai",
    principal={"type": "user", "user_token": user_jwt, "label": "Alice"},
)

---

await_approval()

When a grant has requires_approval configured, proxy_request() returns a PendingApproval instead of executing the call. await_approval() blocks until the approval reaches a terminal state, then returns the proxied result.

async def await_approval(
    approval_id: str,
    *,
    timeout: float = 300.0,
    poll_interval: float = 2.0,
) -> ApprovalResult

Parameter	Type	Default	Description
approval_id	str	—	The approval row id (from PendingApproval.approval_id). Positional.
timeout	float	300.0	Local wait timeout in seconds. The approval row may still be pending on the backend after this elapses — the SDK simply gives up waiting.
poll_interval	float	2.0	Seconds between polls.

Returns: ApprovalResult on the executed terminal (the proxied provider response).

Raises:

• ApprovalDeniedError — approver explicitly denied.
• ApprovalExpiredError — approval window elapsed before any decision.
• ApprovalExecutionFailedError — approved but backend’s proxy execution failed (e.g. grant revoked between approval and execution).
• ApprovalTimeoutError — local wait elapsed before any decision. Re-poll later if the row is still pending.

Transient gateway errors (502, 503, 504) and network failures during the wait are retried silently until the deadline elapses — the approval row is fine on the backend and the next poll typically succeeds.

from alter_sdk.exceptions import ApprovalDeniedError, ApprovalTimeoutError
from alter_sdk.models import PendingApproval

result = await app.proxy_request(
    "POST",
    "https://api.stripe.com/v1/refunds",
    grant_id=grant_id,
    json_body={"charge": "ch_abc"},
)

if isinstance(result, PendingApproval):
    try:
        final = await app.await_approval(
            str(result.approval_id),
            timeout=600.0,
            poll_interval=3.0,
        )
        print("Approved:", final.status_code, final.body_json())
    except ApprovalDeniedError as e:
        print("Denied:", e.message)
    except ApprovalTimeoutError:
        # row may still be pending — re-poll later
        ...

get_approval_status()

Single-shot poll of an approval row. Returns the current status snapshot without blocking.

async def get_approval_status(approval_id: str) -> ApprovalStatus

Parameter	Type	Default	Description
approval_id	str	—	The approval row id.

Returns: ApprovalStatus. Branch on .status or .is_terminal.

status = await app.get_approval_status(approval_id)
if status.is_terminal:
    print("Decided:", status.status, "at", status.decided_at)
else:
    print("Still", status.status)

Status values

ApprovalStatus.status is one of:

Value	Terminal	Meaning
"pending"	no	Approver has not decided.
"approved"	no	Decision made; backend is about to execute the proxied call.
"executing"	no	Backend is currently running the proxied call.
"denied"	yes	Approver declined.
"expired"	yes	Approval window elapsed.
"executed"	yes	Proxied call completed; has_result=True once the result is ready to fetch.
"failed"	yes	Execution failed after approval.

await_approval() handles every terminal automatically and raises the matching typed exception (or returns ApprovalResult for executed).

Reading the result

ApprovalResult.body_b64 is base64-encoded. Use the helpers:

• body_bytes() -> bytes
• body_text(encoding: str = "utf-8") -> str
• body_json() -> Any
• .status_code — HTTP status from the provider
• .headers — provider response headers (dict)
• .approval_id — the approval row id (None for synchronous, non-HITL proxy_request results)