fastn-auth (Python)

Python SDK for Fastn connector authentication. Provides a simple async interface for initiating OAuth and credential-based connector authentication flows.

Installation

pip install fastn-auth

Quick Start

import asyncio
from fastn_auth import FastnAuth

async def main():
    # Create a client
    client = FastnAuth(
        space_id="your-space-id",
        api_key="your-api-key",  # or use auth_token instead
        base_url="https://live.fastn.ai/api",  # optional
    )

    # Initialize an authentication session
    session = await client.initialize(
        connector_id="google-sheets",
        org_id="org-id",      # optional
        tenant_id="tenant-id" # optional
    )

    # Redirect the user to complete OAuth
    print(f"Redirect user to: {session.redirect_url}")

    # Wait for the user to complete authentication
    result = await session.wait_for_completion()

    print("Authentication complete!", result.credentials)

asyncio.run(main())

API Reference

FastnAuth

The main client class for initializing authentication flows.

Constructor

FastnAuth(
    space_id: str,
    api_key: str | None = None,
    auth_token: str | None = None,
    base_url: str | None = None,
)

Parameter	Type	Required	Description
space_id	str	Yes	The space/workspace ID
api_key	str	One of api_key or auth_token	API key — sent as x-fastn-api-key header
auth_token	str	One of api_key or auth_token	Bearer token — sent as Authorization: Bearer {token} header
base_url	str	No	Base URL for the Fastn API (defaults to https://live.fastn.ai/api)

Methods

async initialize(...) -> AuthSession

Initialize a connector authentication flow.

async def initialize(
    self,
    connector_id: str,
    org_id: str | None = None,
    tenant_id: str | None = None,
    connection_id: str | None = None,
) -> AuthSession

Parameter	Type	Required	Description
connector_id	str	Yes	The connector ID (e.g., "google-sheets", "salesforce")
org_id	str	No	Organization ID (defaults to "community")
tenant_id	str	No	Tenant ID
connection_id	str	No	Connection instance ID

async get_credentials(options: GetCredentialsOptions) -> Credentials

Fetch credentials for an already-authenticated connector without starting a new OAuth flow.

from fastn_auth import FastnAuth, GetCredentialsOptions

credentials = await client.get_credentials(
    GetCredentialsOptions(
        connector_id="google-sheets",
        org_id="org-id",        # optional
        tenant_id="tenant-id",  # optional
        connection_id="conn-id" # optional
    )
)

Parameter	Type	Required	Description
connector_id	str	Yes	The connector ID
org_id	str	No	Organization ID (defaults to "community")
tenant_id	str	No	Tenant ID
connection_id	str	No	Connection instance ID

---

AuthSession

Represents an active authentication session returned by client.initialize().

Properties

Property	Type	Description
id	str	Unique identifier for this session
state_key	str	State key for the OAuth flow
redirect_url	str	URL to redirect the user to for OAuth authorization

Methods

async wait_for_completion(options: PollOptions | None = None) -> AuthResult

Poll for the authentication status until it reaches ACTIVE or FAILED, or until the timeout is exceeded.

@dataclass
class PollOptions:
    interval: float = 2.0   # Polling interval in seconds
    timeout: float = 300.0  # Maximum wait time in seconds (5 minutes)

Returns an AuthResult object:

@dataclass
class AuthResult:
    status: AuthStatus
    credentials: Credentials | None = None
    error_message: str | None = None

async get_status() -> StatusResponse

Get the current status of the authentication session.

@dataclass
class StatusResponse:
    status: AuthStatus
    error_message: str | None = None

async get_credentials() -> Credentials

Fetch the credentials for the authenticated connector after the OAuth flow completes.

@dataclass
class Credentials:
    access_token: str | None = None
    expires_in: int | None = None
    [key: str]: Any # connector-specific fields

---

Error Handling

The SDK provides custom exception classes for different failure scenarios:

from fastn_auth import (
    FastnAuthError,
    TimeoutError,
    AuthenticationError,
    NetworkError,
    InvalidResponseError,
)

try:
    result = await session.wait_for_completion()
except TimeoutError:
    print("Authentication timed out")
except AuthenticationError as e:
    print(f"Authentication failed: {e.message}")
except NetworkError as e:
    print(f"Network error: {e.message}, status: {e.status_code}")
except InvalidResponseError as e:
    print(f"Unexpected API response: {e.message}")

Error Class	Code	Description
FastnAuthError	—	Base class for all SDK errors
TimeoutError	TIMEOUT	wait_for_completion() exceeded the timeout
AuthenticationError	AUTH_FAILED	The authentication flow returned FAILED
NetworkError	NETWORK_ERROR	HTTP request failed; includes status_code
InvalidResponseError	INVALID_RESPONSE	API response is missing required fields

---

Status Lifecycle

Status	Description
INACTIVE	OAuth initiated, awaiting user authorization
ACTIVE	Connector successfully authenticated
FAILED	Authentication failed

---

Requirements

• Python 3.9 or higher
• aiohttp >= 3.8.0

License

MIT