omnidapter-sdk

Python client for the Omnidapter API.

Omnidapter lets you connect to your users' calendars (and other services) through a single unified API. You create a link token, send the user through the Connect UI to authorise their account, and then read their data using the connection that comes back.

Installation

pip install omnidapter-sdk

Requires Python 3.10+.

Getting started

The SDK is async-native — every operation method returns a coroutine that must be awaited. OmnidapterClient is an async context manager; use async with to guarantee the underlying HTTP session is closed:

import asyncio
from omnidapter_sdk import OmnidapterClient


async def main() -> None:
    async with OmnidapterClient(
        base_url="https://api.example.com",
        api_key="omni_live_...",
    ) as client:
        providers = await client.providers.list_providers()
        for provider in providers:
            print(provider.key, provider.display_name)


asyncio.run(main())

For long-lived clients (e.g. one per FastAPI application), construct directly and call await client.close() at shutdown.

Every operation returns its payload directly — a list for list operations, the resource for single-resource operations. There is no .data wrapper. The per-request correlation id is surfaced on errors as exc.request_id (see Error handling).

Breaking change — unwrapped responses. Methods now return the payload directly instead of an ApiResponse envelope: replace result.data with result (e.g. providers.data → providers, conn.data.status → conn.status). List endpoints return the bare list; the paging meta is now reached via .raw on the domain accessor (e.g. await client.crm.raw.list_contacts(...) → the full {data, meta} page) — or just use the iter_* helpers. The previous correlation id on meta.request_id is now on the raised exception's .request_id.

Breaking change — async-native client (0.5.0). Prior versions (≤ 0.4.x) returned plain values from synchronous methods. The client flips every method to async def; existing call sites must add await and wrap construction in async with (or call await client.close() explicitly).

Providers

Providers are the services Omnidapter can connect to (e.g. Google, Microsoft). You typically list them once to populate a picker in your UI.

providers = await client.providers.list_providers()
for provider in providers:
    print(provider.key, provider.display_name)

# Fetch a single provider by its key
google = await client.providers.get_provider(provider_key="google")
print(google.display_name)

Connections

A connection represents an authorised link between one of your end users and a provider. Once a user completes the Connect flow, their connection appears here.

# List all connections, optionally filtering by status or provider
connections = await client.connections.list_connections(status="active", provider="google")
for conn in connections:
    print(conn.id, conn.provider_key, conn.status)

# Fetch a single connection
conn = await client.connections.get_connection(connection_id="conn_...")
print(conn.status)

# Revoke a connection — this also invalidates any stored credentials
await client.connections.delete_connection(connection_id="conn_...")

Link tokens

A link token is a short-lived, single-use token that grants an end user access to the Connect UI. Generate one server-side, pass it to your frontend, and redirect the user to connect_url. Omnidapter will handle the OAuth flow and create a connection on success.

from omnidapter_sdk.models import CreateLinkTokenRequest

result = await client.link_tokens.create_link_token(
    create_link_token_request=CreateLinkTokenRequest(
        end_user_id="user_123",           # your internal user ID
        allowed_providers=["google", "microsoft"],  # restrict which providers are shown
    )
)

token = result            # the payload is returned directly
print(token.token)       # lt_... — pass this to your frontend
print(token.connect_url) # redirect the user here to start the Connect flow
print(token.expires_at)  # datetime — tokens are short-lived, generate them on demand

Calendar

Once a user has a connection, you can read their calendar data. All calendar operations require a connection_id.

from datetime import datetime, timezone

# List the calendars available on a connection
calendars = await client.calendar.list_calendars(connection_id="conn_...")
for cal in calendars:
    print(cal.id, cal.name)

# List events within a time range across a specific calendar
events = await client.calendar.list_events(
    connection_id="conn_...",
    calendar_id="cal_...",
    start=datetime(2026, 4, 1, tzinfo=timezone.utc),
    end=datetime(2026, 4, 30, tzinfo=timezone.utc),
)
for event in events:
    print(event.id, event.title, event.start)

Contacts

The contacts service is a shared Contact + Company surface. It is backed either by a CRM provider (HubSpot, Pipedrive, Salesforce, Zoho — which also expose companies) or by a directory provider (Google Contacts, Microsoft, CardDAV, Apple — contacts only). Both are reached the same way, via client.contacts (or conn.contacts on a connection-scoped client):

conn = client.connection("conn_...")

# List and read contacts
contacts = await conn.contacts.list_contacts(limit=50)
contact = await conn.contacts.get_contact(contact_id="contact_...")

# Search, create, update, delete
matches = await conn.contacts.search_contacts(query="ada@example.com")
created = await conn.contacts.create_contact(...)
updated = await conn.contacts.update_contact(contact_id="contact_...", ...)
await conn.contacts.delete_contact(contact_id="contact_...")

Company operations (list_companies, get_company, create_company, …) are available on CRM-backed connections; on a directory provider they raise an OmnidapterApiError with code == "unsupported_capability".

The older client.crm.list_contacts / client.crm.list_companies surface still works but is deprecated in favour of client.contacts.

Pagination

A plain list call returns one page of items as a bare list. To see the paging state, use .raw on the domain accessor — it returns the full {data, meta} page, where has_more is the signal to keep going (a full page does not mean the end of the data):

# Just the items for one page:
contacts = await client.crm.list_contacts(connection_id="conn_...", limit=50)
for contact in contacts:
    print(contact.id)

# The page envelope, when you need the paging cursor / has_more:
page = await client.crm.raw.list_contacts(connection_id="conn_...", limit=50)
print(page.meta.pagination.has_more)  # True → there's another page
print(page.meta.pagination.cursor)    # pass as `cursor=` to fetch the next page

Prefer the hand-written iter_* helpers, which walk every page for you and yield each item:

conn = client.connection("conn_...")

async for contact in conn.iter_contacts():
    print(contact.id)
async for booking in conn.iter_bookings():
    ...
async for event in conn.iter_events("cal_...", start=..., end=...):
    ...

# Connections page from the top-level client:
async for connection in client.iter_connections():
    print(connection.id)

Helpers: iter_contacts, iter_companies, iter_deals, iter_activities, iter_bookings, iter_events, iter_connections. Each accepts page_size and forwards list filters as keyword arguments. See docs/api/pagination.md for the full reference.

Error handling

The SDK raises a typed exception for any non-2xx response. Catch OmnidapterApiError — an unambiguous alias for the base of every HTTP error (the generated ApiException / NotFoundException / … names are generic and can clash with other libraries you import). Each exception carries:

• .status — the HTTP status code.
• .code — the stable error.code (e.g. scope_insufficient); branch on this, not on the message or status.
• .details — the error.details object (required_scopes, retry_after, …) when present.
• .request_id — the server's correlation id; quote it when reporting an issue.
• .body — the raw JSON error body, if you need it.

from omnidapter_sdk import OmnidapterApiError
from omnidapter_sdk.exceptions import NotFoundException

try:
    contact = await client.crm.get_contact(connection_id, contact_id)   # the Contact
except NotFoundException as e:
    log.warning("not found, request_id=%s", e.request_id)
except OmnidapterApiError as e:
    if e.code == "scope_insufficient":
        prompt_for_upgrade(e.details["required_scopes"])
    else:
        log.error("api error %s (%s), request_id=%s", e.status, e.code, e.request_id)

OmnidapterApiError covers HTTP-response errors; OmnidapterSDKError is the base of everything the SDK can raise (also the client-side misuse types like ApiTypeError/ApiKeyError, which — note — subclass Python's built-in TypeError/KeyError, so a broad except KeyError: in your own code could catch an ApiKeyError; prefer catching OmnidapterApiError/OmnidapterSDKError).

Notes

• The SDK is generated from the server's OpenAPI spec via scripts/generate_sdks.sh. Run that script after pulling changes to regenerate the client code.
• The transport is aiohttp (via openapi-generator's library=asyncio template). An aiohttp.ClientSession is opened lazily on the first request and closed when the OmnidapterClient is closed (via async with or await client.close()). Skipping the close emits a ResourceWarning from aiohttp at process exit.