Python

The official Python SDK, maton-ai, wraps the Maton API with typed accessors and automatic retries.

Installation

pip install maton-ai

uv add maton-ai

poetry add maton-ai

Quickstart

import os
from maton_ai import Maton

maton = Maton(api_key=os.environ["MATON_API_KEY"])

gmail = maton.google_mail()
messages = gmail.message.list(q="is:unread", max_results=10)
gmail.message.send(to="alice@example.com", subject="hi", body="hello")

Client

maton = Maton(
    api_key=None,        # defaults to MATON_API_KEY env var
    connection=None,     # default connection id for every call
    timeout=30.0,        # per-request timeout, seconds
    max_retries=2,       # retry attempts on transient failures
    max_backoff=20.0,    # cap on a single backoff sleep, seconds
)

App Accessors

Every supported app is available as an accessor with resource-oriented methods:

slack = maton.slack(connection=slack_conn_id)
slack.message.send(channel="#general", text="deploy finished ✅")

gh = maton.github(connection=gh_conn_id)
gh.issue.create(repo="maton-ai/maton-py", title="bug: ...", body="...")

notion = maton.notion(connection=notion_conn_id)
notion.data_source.query(data_source_id="...", filter={"property": "Status", "status": {"equals": "Done"}})

hubspot = maton.hubspot(connection=hubspot_conn_id)
hubspot.contact.list(limit=25)

Selecting a connection

A connection can be set in three places. Precedence is per-call → accessor → constructor:

maton = Maton(connection=connection_id)              # 1. client default

gmail = maton.google_mail(connection=connection_id)  # 2. accessor
gmail.message.list(q="is:unread")

maton.google_mail.message.list(                      # 3. per-call (wins)
    q="is:unread", connection=connection_id
)

Connections

maton.connection.create(app="slack")                  # returns the PENDING connection + auth url
maton.connection.get(connection_id)
maton.connection.list(app="slack", status="ACTIVE")
maton.connection.delete(connection_id)

create returns the freshly created connection with a url field — open it in a browser to complete OAuth, then poll get until status is ACTIVE.

Passthrough

For any endpoint without a typed accessor, use maton.api to reach it through the gateway:

maton.api.get("slack", "/api/conversations.list", params={"limit": 10}, connection=connection_id)

maton.api.post(
    "google-mail",
    "/gmail/v1/users/me/messages/send",
    json={"raw": "..."},
    connection=connection_id,
)

Triggers register an event source (e.g. GitHub pull_request.opened) and fan matching events out to webhook destinations. Manage them through maton.trigger, with maton.trigger.destination and maton.trigger.event sub-resources:

trigger = maton.trigger.create(
    source="github",
    event_type="pull_request.opened",
    connection_id=gh_conn_id,
    parameters={"repository_full_name": "maton-ai/cli"},
    destinations=[{"url": "https://example.com/hook"}],
)
trigger_id = trigger["trigger"]["trigger_id"]

maton.trigger.list(source="github", status="ENABLED")
maton.trigger.get(trigger_id)
maton.trigger.update(trigger_id, status="DISABLED")
maton.trigger.delete(trigger_id)

# Destinations
dst = maton.trigger.destination.create(trigger_id, url="https://example.com/hook")
destination_id = dst["destination"]["destination_id"]
maton.trigger.destination.rotate_secret(trigger_id, destination_id)

# Events
events = maton.trigger.event.list(trigger_id, limit=20)
maton.trigger.event.replay(trigger_id, events["events"][0]["event_id"])

Watching events

watch polls for new events and yields each one as it arrives, checkpointing so restarts resume after the last handled event:

for event in maton.trigger.event.watch(trigger_id):
    handle(event)

Reliability

Every call goes through an httpx-based client with automatic retries. Retries fire on connection errors and on 429 / 500 / 502 / 503 / 504; other 4xx/5xx surface immediately. Backoff is truncated exponential with full jitter and honors a server-provided Retry-After header.

Errors

All failures raise a subclass of MatonError, each carrying status_code, request_id, and the parsed body:

Exception	When
`AccessDeniedError`	401/403 — bad/missing API key or insufficient scope
`ResourceNotFoundError`	404
`TooManyRequestsError`	429 — also exposes `.retry_after`
`ValidationError`	other 4xx (incl. 412 when an action needs a connection)
`InternalServerError`	5xx or unexpected upstream response
`APIConnectionError`	network/transport failure

from maton_ai import MatonError, TooManyRequestsError

try:
    maton.google_mail.message.list(q="is:unread", connection=connection_id)
except TooManyRequestsError as exc:
    print("slow down; retry after", exc.retry_after)
except MatonError as exc:
    print(exc.status_code, exc.request_id, exc.body)

Some apps wrap their vendor-specific error envelopes in a dedicated subclass (GitHubError, GoogleDriveError, LinearError, SlackError, StripeError) so you can catch them by app while still falling back to MatonError.

Python

On this page