Python SDK

pip install vendo-sdk

# With HTTP session helper (requests):
pip install "vendo-sdk[http]"

# With async support (aiohttp):
pip install "vendo-sdk[async]"

import vendo

vendo.token(slug)                        # str — fetch single token
vendo.tokens(slugs)                      # dict[str, str | None]
vendo.session(slug)                      # requests.Session (requires [http])
vendo.connect_url(slug, *, return_to=None, state=None)  # str (Vendo mode only)
vendo.is_vendo_mode()                    # bool

from vendo import Vendo, AsyncVendo

client = Vendo(api_key=None, base_url=None)
client.token(slug)
client.tokens(slugs)
client.session(slug)
client.connect_url(slug, *, return_to=None, state=None)
client.invalidate(slug)

# Scoping
client.for_request(headers)  # → Vendo (Vendo mode only)
client.for_user(jwt)         # → Vendo (Vendo mode only)

# Sub-APIs (instance attributes)
client.connections           # ConnectionsAPI
client.integrations          # IntegrationsAPI
client.billing               # BillingAPI (Vendo mode only)
client.events                # EventsAPI (Vendo mode only)
client.webhooks              # WebhooksAPI (works in both modes)
client.data                  # DataAPI — Composio proxy (Vendo mode only)

import vendo

vendo.sso.is_enabled()
vendo.sso.identity_from_headers(headers)        # VendoIdentity | None

vendo.deployment.tenant_id()                    # str | None
vendo.deployment.deployment_id()                # str | None
vendo.deployment.is_running_on_vendo()          # bool — checks VENDO_DEPLOYMENT_ID

vendo.reconciler.bootstrap(env_file=..., mapping=...)
vendo.reconciler.start(env_file=..., mapping=..., on_change="restart")

from vendo import AsyncVendo, aio

async with AsyncVendo() as client:
    tok = await client.token(slug)
    conns = await client.connections.list()

# Module-level async shortcuts:
await aio.token(slug)
await aio.tokens(slugs)
await aio.connections.list()
await aio.integrations.list()
await aio.billing.balance()

client.connections.list()          # list[Connection]
client.connections.get(slug)       # Connection | None

client.integrations.list()         # list[Integration]
client.integrations.get(slug)      # Integration | None
client.integrations.env_vars(slug) # list[str] — BYOK env-var names

client.billing.balance()           # Balance
client.billing.spend_caps()        # SpendCaps
client.billing.usage(period="month")  # dict

for event in client.events.subscribe():
    print(event.type, event.data)

# Constructor — pass secret + max_age_sec here.
hooks = WebhooksAPI(secret="whsec_...", max_age_sec=300)
event = hooks.verify(headers, body)  # body: str → WebhookEvent

# Or via the client (uses VENDO_WEBHOOK_SECRET from env):
client.webhooks.verify(headers, body)

client.data.execute("SLACK_SEND_MESSAGE", {"channel": "#general", "text": "hi"})

import vendo

# mapping is a callable returning {VAR_NAME: value}.
vendo.reconciler.bootstrap(env_file="/app/.env", mapping=lambda: {"X": "1"})

# on_change defaults to "restart"; accepts "reload-hook" or a callable.
reconciler = vendo.reconciler.start(
    env_file="/app/.env",
    mapping=lambda: {"X": "1"},
    on_change="restart",
)

from vendo.errors import (
    VendoError,
    NotConnected,
    NeedsReauth,
    AuthError,
    RateLimited,
    BalanceExhausted,
    SpendCapExceeded,
    UpstreamError,
    ValidationError,
    IdempotencyConflict,
    VendoOnlyFeature,
    IdentityNotPresent,
)

from vendo.testing import MockClient, MockAsyncClient, fake_connection