WorkOS Python Library

from workos import WorkOSClient

client = WorkOSClient(api_key="sk_1234", client_id="client_1234")

# List organizations
page = client.organizations.list_organizations()
for org in page.auto_paging_iter():
    print(org.name)

# Create an organization
org = client.organizations.create_organization(name="Acme Corp")
print(org.id)

from workos import AsyncWorkOSClient

async_client = AsyncWorkOSClient(api_key="sk_1234", client_id="client_1234")

page = await async_client.organizations.list_organizations()
async for org in page.auto_paging_iter():
    print(org.name)

# Iterate through all pages automatically
for user in client.user_management.list_users().auto_paging_iter():
    print(user.email)

# Or work with a single page
page = client.user_management.list_users(limit=10)
print(page.data)        # List of items on this page
print(page.has_more())  # Whether more pages exist
print(page.after)       # Cursor for the next page

from workos import NotFoundError, RateLimitExceededError

try:
    client.organizations.get_organization("org_nonexistent")
except NotFoundError as e:
    print(f"Not found: {e.message}")
    print(f"Request ID: {e.request_id}")
except RateLimitExceededError as e:
    print(f"Retry after: {e.retry_after} seconds")

import httpx2
from workos import AsyncWorkOSClient, WorkOSClient

client = WorkOSClient(
    api_key="sk_...",
    http_client=httpx2.Client(proxy="http://proxy.internal:3128", verify="/etc/ssl/corp.pem"),
)

async_client = AsyncWorkOSClient(
    api_key="sk_...",
    http_client=httpx2.AsyncClient(limits=httpx2.Limits(max_connections=20)),
)

import asyncio

import aiohttp
import yarl

from workos import (
    AsyncWorkOSClient,
    HTTPResponse,
    TransportConnectError,
    TransportError,
    TransportTimeout,
)


class AiohttpBackend:
    def __init__(self, session: aiohttp.ClientSession) -> None:
        self._session = session

    async def request(self, method, url, *, headers, content, timeout) -> HTTPResponse:
        try:
            async with self._session.request(
                method,
                yarl.URL(url, encoded=True),  # keep the SDK's percent-encoding intact
                headers=headers,
                data=content,
                timeout=aiohttp.ClientTimeout(total=timeout),
                allow_redirects=True,
            ) as resp:
                body = await resp.read()
                return HTTPResponse(resp.status, resp.headers, body, method, str(resp.url))
        except (asyncio.TimeoutError, aiohttp.ServerTimeoutError) as exc:  # before connection errors
            raise TransportTimeout(str(exc)) from exc
        except aiohttp.ClientConnectionError as exc:
            raise TransportConnectError(str(exc)) from exc
        except aiohttp.ClientError as exc:
            raise TransportError(str(exc)) from exc

    async def close(self) -> None:
        await self._session.close()


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        client = AsyncWorkOSClient(api_key="sk_...", http_client=AiohttpBackend(session))
        page = await client.organizations.list_organizations()

result = client.organizations.list_organizations(
    request_options={
        "timeout": 10,
        "max_retries": 5,
        "extra_headers": {"X-Custom": "value"},
        "idempotency_key": "my-key",
        "base_url": "https://staging.workos.com/",
    }
)