Python SDK

Installation

pip install authpipe

Requires Python 3.10+. Uses httpx for HTTP and pydantic for types.

Configuration

Sync client

from authpipe import Authpipe

client = Authpipe(
    api_key="sk_...",
    base_url="https://api.authpipe.dev",  # optional
    api_version="2026-04-07",             # optional
    cache=True,                           # optional (default True)
    cache_ttl_seconds=300,                # optional (default 300)
)

Async client

from authpipe import AsyncAuthpipe

client = AsyncAuthpipe(api_key="sk_...")

Both clients support context managers:

# Sync
with Authpipe(api_key="sk_...") as client:
    result = client.get_credential(provider="slack", tenant_id="org_acme")

# Async
async with AsyncAuthpipe(api_key="sk_...") as client:
    result = await client.get_credential(provider="slack", tenant_id="org_acme")

Option	Type	Default	Description
api_key	str	—	Secret or publishable API key (required)
base_url	str	https://api.authpipe.dev	API base URL
api_version	str | None	None	API version header
cache	bool	True	Enable credential caching
cache_ttl_seconds	int	300	Cache TTL in seconds

Credentials

get_credential

All parameters are keyword-only.

result = client.get_credential(
    provider="slack",
    tenant_id="org_acme",
    user_id="user_jane",           # optional
    installation_id="inst_123",    # optional
    credential_for="any",          # optional: "any", "installation", "user"
)

result.credential       # str — access token or API key
result.credential_type  # str — "oauth_token", "api_key", etc.
result.source           # str — "connection" or "installation"
result.expires_at       # datetime | None
result.scopes           # list[str] | None

Async:

result = await client.get_credential(provider="slack", tenant_id="org_acme")

store_credential

result = client.store_credential(
    provider="sendgrid",
    tenant_id="org_acme",
    credential_type="api_key",    # "api_key" or "webhook_secret"
    credential="SG.xxxx",
    user_id="user_jane",          # optional
    installation_id="inst_123",   # optional
)

result.connection_id  # str
result.status         # str

OAuth Sessions

create_auth_session

session = client.create_auth_session(
    provider="google",
    tenant_id="org_acme",
    redirect_url="https://yourapp.com/callback",
    user_id="user_jane",                              # optional
    scopes=["drive.readonly"],                        # optional
    installation_id="inst_123",                       # optional
    template_variables={"shop": "acme-store"},        # optional
)

session.session_id         # str
session.authorization_url  # str — redirect the user here

create_install_session

session = client.create_install_session(
    provider="slack",
    tenant_id="org_acme",
    redirect_url="https://yourapp.com/installed",
    user_id="user_admin",                             # optional
    permissions=["channels:read", "chat:write"],      # optional
)

Connections

get_connection

conn = client.get_connection(connection_id="conn_abc123")

delete_connection

client.delete_connection(connection_id="conn_abc123")

refresh_connection

result = client.refresh_connection(connection_id="conn_abc123")
result.refreshed   # bool
result.expires_at  # datetime | None

search_connections

from authpipe.types import SearchRequest, SortDirective

resp = client.search_connections(SearchRequest(
    filter={"tenant_id": {"eq": "org_acme"}, "status": {"eq": "active"}},
    sort=[SortDirective(field="connected_at", direction="desc")],
    cursor="cursor_token",
    limit=25,
))

resp.data        # list[Connection]
resp.has_more    # bool
resp.next_cursor # str | None
resp.limit       # int

Installations

get_installation

inst = client.get_installation(installation_id="inst_abc123")

delete_installation

client.delete_installation(installation_id="inst_abc123")

search_installations

resp = client.search_installations(SearchRequest(
    filter={"tenant_id": {"eq": "org_acme"}},
))

Provider Configs

create_provider_config

config = client.create_provider_config(
    provider_id="slack",
    client_id="your-client-id",
    client_secret="your-client-secret",
    oauth_redirect_url="https://api.authpipe.dev/oauth/callback",
    attachment="user",
    enabled=True,
    scopes=["channels:read"],
)

get_provider_config

config = client.get_provider_config(config_id="pcfg_abc123")

update_provider_config

config = client.update_provider_config(
    config_id="pcfg_abc123",
    client_secret="new-secret",
    scopes=["channels:read", "chat:write"],
    bot_scopes=["chat:write"],
    signing_key="signing-secret",
    enabled=True,
)

delete_provider_config

client.delete_provider_config(config_id="pcfg_abc123")

search_provider_configs

resp = client.search_provider_configs(SearchRequest(
    filter={"provider_id": {"eq": "slack"}},
))

Providers (Catalog)

get_provider

provider = client.get_provider(provider_id="slack")

search_providers

resp = client.search_providers(SearchRequest(
    query="email",
    filter={"category": {"eq": "communication"}},
    limit=10,
))

Custom Providers

create_custom_provider

provider = client.create_custom_provider(
    id="my-internal-api",
    name="My Internal API",
    auth_api_key={"key_field": "x-api-key"},
)

get_custom_provider

provider = client.get_custom_provider(custom_provider_id="my-internal-api")

update_custom_provider

provider = client.update_custom_provider(
    custom_provider_id="my-internal-api",
    name="Updated Name",
)

delete_custom_provider

client.delete_custom_provider(custom_provider_id="my-internal-api")

Events

search_events

resp = client.search_events(SearchRequest(
    filter={"event_type": {"eq": "connection.created"}},
    sort=[SortDirective(field="created_at", direction="desc")],
    limit=50,
))

API Keys

list_api_keys

keys = client.list_api_keys()  # list[APIKey]

create_api_key

result = client.create_api_key(name="Production", role="secret")
result.id   # str
result.key  # str (only returned once)

revoke_api_key

client.revoke_api_key(api_key_id="key_abc123")

Webhook Config

get_webhook_config

config = client.get_webhook_config()
config.webhook_url   # str
config.has_secret_1  # bool
config.has_secret_2  # bool

set_webhook_config

result = client.set_webhook_config(webhook_url="https://yourapp.com/webhooks")
result.webhook_url  # str
result.secret       # str (save this)

rotate_webhook_secret

result = client.rotate_webhook_secret()
result.secret  # new secret

delete_webhook_config

client.delete_webhook_config()

Workspaces

get_workspace

ws = client.get_workspace(workspace_id="ws_abc123")

update_workspace

ws = client.update_workspace(workspace_id="ws_abc123", name="New Name")

Error handling

The SDK raises AuthpipeAPIError for HTTP errors:

from authpipe.errors import AuthpipeAPIError

try:
    result = client.get_credential(provider="slack", tenant_id="org_acme")
except AuthpipeAPIError as e:
    print(e.status_code)  # HTTP status
    print(e.code)         # Error code
    print(e.message)      # Human-readable message

Automatic retry on 429 and 5xx with exponential backoff (up to 3 attempts).

Pydantic types

All response types are Pydantic BaseModel subclasses:

from authpipe.types import (
    GetCredentialResult,
    StoreCredentialResult,
    CreateAuthSessionResult,
    Connection,
    Installation,
    ProviderConfig,
    Provider,
    SearchRequest,
    SearchResponse,
    SortDirective,
)