omnidapter-sdk 0.8.0

Omnidapter API client SDK

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="[e-mail]")
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.