Metadata-Version: 2.5
Name: appnigma-sdk
Version: 2.1.0
Summary: The package a customer's product installs to reach its Appnigma microserver
Project-URL: Homepage, https://docs.appnigma.ai/sdk
Project-URL: Source, https://github.com/Appnigma/ap-sdks/tree/main/python
Project-URL: Issues, https://github.com/Appnigma/ap-sdks/issues
Project-URL: Changelog, https://github.com/Appnigma/ap-sdks/releases
Author-email: Appnigma <support@appnigma.ai>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.11
Requires-Dist: python-dateutil>=2.8.2
Requires-Dist: typing-extensions>=4.7.1
Requires-Dist: urllib3<3.0.0,>=2.7.0
Provides-Extra: dev
Requires-Dist: build~=1.3; extra == 'dev'
Requires-Dist: pytest~=9.0; extra == 'dev'
Requires-Dist: ruff~=0.17.0; extra == 'dev'
Requires-Dist: twine~=7.0; extra == 'dev'
Description-Content-Type: text/markdown

# appnigma-sdk

The package your product installs to reach its Appnigma microserver. It
connects end customers, reads and writes their contacts, companies and deals
in one shape whichever CRM they use, runs your own functions, and calls any
platform's API. One package for every customer. Nothing about your account is
inside it, and it holds no secret: the key is yours to supply.

- Python 3.11 or newer. Tested on 3.11, 3.12, 3.13, 3.14 and 3.15.
- The same calls, error codes and retry rules as the TypeScript package
  `@appnigma/sdk`, with names in Python style.
- Not tied to Django, Flask or any framework.

## Install

```sh
pip install appnigma-sdk
```

The package is installed as `appnigma-sdk` and imported as `appnigma`.

## The key

Every call carries your SDK key. Appnigma gives you your key when your
microserver is set up; ask Appnigma for a new one if you need to replace or
cancel it. It starts with `apn_live_` and is shown once. Keep it on your
server, in the `APPNIGMA_SDK_KEY` environment variable, never in a browser or
a repo. The client reads it from there:

```python
from appnigma import Appnigma

client = Appnigma()
```

If you keep the key somewhere else, pass it in; a passed key wins:

```python
client = Appnigma(key=vault.read("appnigma-sdk-key"))
```

A client started with no key at all, or with a key that does not start with
`apn_live_`, raises `AuthError` at once, before any call.

Every argument is passed by name. Every answer is a plain `dict` with keys in
Python style (`expires_at`, `input_schema`), except content that is not ours,
which comes back exactly as it was sent: a function's `output` and its
schemas, a platform's `headers` and `body`, a CRM record's `raw` and the names
in its `custom_fields`.

## Connecting an end customer

Before calls can work for an end customer, they connect their account on that
platform once. Any platform switched on for your company can be connected, and
the end customer approves your own app there. Your product asks for a link,
shows it to the end customer, and they approve on the platform.

1. Ask for a link. It works once and expires after 15 minutes. Opening it does
   not use it up: it is used when the end customer presses Connect, so a mail
   scanner or a link preview opening it first does no harm.
2. Send the end customer to the link's `url`. They log in to the platform and
   approve.
3. When they are done, they land on your `return_url` with `status=connected`,
   or with `status=failed` and a `reason`. `return_url` must be HTTPS and at
   most 2048 bytes when written as a JSON string, about 2046 plain characters.
4. Confirm with `connections.get`. The connection is written before the end
   customer lands, so the answer is already there.
5. To see everyone at once, with every connection of theirs, use
   `connections.list` below.

```python
link = client.connections.create_link(
    end_customer_id="lawfirm-123",
    platform="hubspot",
    return_url="https://app.example.com/integrations/hubspot/done",
)
print(f"send the end customer to {link['url']} before {link['expires_at']}")

connection = client.connections.get(end_customer_id="lawfirm-123", platform="hubspot")
if connection["status"] == "connected":
    print("ready to read and write")
print(connection["logic_folder_key"])  # "ec_3f9a1c7e2b8d4a60"
```

`status` is `pending`, `connected`, `needs_reconnect` or `disconnected`.
`pending` is an end customer who was sent a connect link and has not approved
yet: they exist from the moment the link is made. An end customer
who is already connected can open a new link to connect again, for example
after changing HubSpot accounts. An end customer the API has never seen is a
`NotFoundError`, not a status.

`logic_folder_key` is where this end customer's own functions go in your logic
repo: `src/logic/end-customers/<logic_folder_key>/<platform>/`. It is one key
per end customer, the same on every platform, and the Microserver page shows
it next to them with a copy button. It is `None` only for a connection made
before the key existed, until Appnigma has filled it in.

An id outside the rule below is refused before the call with `InputError`, the
same refusal the server would give.

## Listing your end customers: connections.list

`connections.list` answers your end customers one page at a time, in id order:
each with their `end_customer_id`, their `logic_folder_key`,
`has_custom_logic` (`True` when a function written for them is live) and every
connection of theirs, with its `platform`, `status` and `connected_at` (ISO
8601 in UTC, `None` while pending). Give `platform` or `status` to keep only
the end customers with at least one connection that matches; each entry still
lists all of their connections. `limit` is 1 to 100, 50 if left out; `next` is
`None` on the last page, otherwise pass it back as `cursor`, exactly as it
came. It is the list to build your own integrations screen from, and the way
to find who must connect again:

```python
page = client.connections.list(status="needs_reconnect")
for end_customer in page["end_customers"]:
    lost = [c["platform"] for c in end_customer["connections"] if c["status"] == "needs_reconnect"]
    print(end_customer["end_customer_id"], "must connect again on", lost)

# More pages, while there is one.
if page["next"]:
    client.connections.list(status="needs_reconnect", cursor=page["next"])
```

A `platform` that is not switched on for your company is an `AuthError` with
`PLATFORM_NOT_ENABLED`; a `status` outside the four, a `limit` outside 1 to
100, or a `cursor` this list did not give you is an `InputError` naming the
field. An empty list is not an error. Reads are retried.

### Disconnecting an end customer

When an end customer leaves a platform, for example they remove it from your
Integrations page, disconnect them. Their login on that platform is removed
and the connection is deleted, so `connections.get` then raises `NotFoundError`
with `END_CUSTOMER_NOT_FOUND`. Their other platforms are left alone, and a new
link connects them again from the start. An end customer who was sent a link and never finished is
disconnected the same way.

```python
removed = client.connections.disconnect(end_customer_id="lawfirm-123", platform="hubspot")
print(removed["status"])  # "disconnected"
```

The answer is their id, the platform, `status` `disconnected` and their
`logic_folder_key`, which comes from their id and so is the same if they
connect again.

Disconnecting an end customer who is not connected, or disconnecting twice,
raises `NotFoundError` with `END_CUSTOMER_NOT_FOUND`. The call is made once and
never retried: if its answer is lost and you call again, that
`END_CUSTOMER_NOT_FOUND` means the first call worked. When the login cannot be
removed just now, it raises `ServerError` with `INTERNAL_ERROR`, nothing is
changed, and you can call again. The id rules above apply here too.

## CRM data in one shape: crm

`crm.contacts`, `crm.companies` and `crm.deals` read and write an end
customer's CRM in one shape, whichever CRM they use. Each has `list`, `get`,
`create` and `update`, and each takes `end_customer_id` and `platform`.
`end_customer_id` is your own id for your customer, as it is: an email, a UUID,
`ACME Inc / EU`. 1 to 255 characters, with no control characters and not only
spaces, kept exactly as you send it, so ids that differ only in case or spaces
are different end customers.

```python
page = client.crm.contacts.list(end_customer_id="lawfirm-123", platform="attio", limit=50)
for contact in page["records"]:
    email = contact["emails"][0]["value"] if contact["emails"] else None
    print(contact["full_name"], email)

# The next page, while there is one.
if page["next"]:
    client.crm.contacts.list(end_customer_id="lawfirm-123", platform="attio", cursor=page["next"])

deal = client.crm.deals.get(end_customer_id="lawfirm-123", platform="hubspot", id="1234")["record"]
print(deal["name"], deal["amount"], deal["currency"], deal["status"])
```

Every field is always there: a value the CRM does not have is `None`, never
left out.

| Object | Fields |
| --- | --- |
| Every record | `id`, `platform`, `created_at`, `updated_at`, `custom_fields`, and `raw`, the record exactly as the CRM sent it |
| Contact | `first_name`, `last_name`, `full_name`, `emails`, `phones` (the primary one first), `company_id`, `owner_id` |
| Company | `name`, `domain`, `owner_id` |
| Deal | `name`, `amount`, `currency`, `status` (`open`, `won` or `lost`), `stage`, `pipeline`, `probability`, `close_date`, `closed_at`, `company_id`, `owner_id` |

`custom_fields` is keyed by the CRM's own field names, and `raw` is exactly
what the CRM sent. A platform with no unified data yet for that object is a
`NotAvailableError` with `DATA_TYPE_NOT_AVAILABLE`; reach it with `request`
instead. An id that does not exist is a `NotFoundError` with
`RECORD_NOT_FOUND`. Reads are retried.

### Writing: create and update

`create` makes a record and `update` changes one, from the fields you give in
the shapes they are read in. Both answer the record as `get` reads it.

```python
contact = client.crm.contacts.create(
    end_customer_id="lawfirm-123",
    platform="hubspot",
    record={
        "first_name": "Jane",
        "last_name": "Doe",
        "emails": [{"value": "jane@firm.com", "type": "work", "primary": True}],
    },
)["record"]

client.crm.contacts.update(
    end_customer_id="lawfirm-123", platform="hubspot", id=contact["id"], record={"last_name": "King", "owner_id": None}
)

client.crm.deals.create(
    end_customer_id="lawfirm-123",
    platform="pipedrive",
    record={"name": "Retainer", "amount": 1200, "currency": "USD"},
    match_key="inv_1042",
)
```

- **What you can write.** The fields in the table above, except `id`,
  `platform`, the times, `custom_fields`, `raw` and a deal's `closed_at`, which
  are the CRM's to set. A stage or a pipeline is written as its id.
- **`None` clears a field;** a field left out is left alone.
- **Matching.** A contact with an email, or a company with a domain, updates
  the record the CRM already has with it rather than making a second one. A
  deal has no such field, so pass `match_key`, a key of your own such as your
  invoice id: the same key updates the same deal.
- **Repeats.** A create that matches, and an update, are retried like reads,
  because a repeat changes the same record. A create with nothing to match on
  is made once, because a repeat could make a second record.
- **Refusals.** A field this CRM cannot be told is an `InputError` naming it,
  sent nowhere. A write the end customer's connection lacks the scope for is
  an `AuthError` with `MISSING_SCOPES`, and `details["missing"]` names the
  scopes: the end customer has to connect again. A value the CRM itself
  refuses is a `PlatformRefusedError` with its reason.

## Which platforms: platforms

`platforms.list` names each platform switched on for you, and how much of it
the shared shapes cover.

```python
for platform in client.platforms.list()["platforms"]:
    print(platform["label"], platform["support"])
```

`support` is `supported` when `crm` reads and writes it, with `coverage`
naming each object and its actions, or `experimental` when only `request`
reaches it so far.

## Anything else on the platform: request

`request` calls the end customer's platform's own API, for anything the shared
shapes do not cover. It works on every platform switched on for you, including
ones with no unified data yet.

```python
answer = client.request(
    end_customer_id="lawfirm-123",
    platform="hubspot",
    method="GET",
    path="/crm/v3/objects/notes?limit=10",
)
if answer["status"] == 200:
    print(answer["body"])
```

`path` is a path on the platform's own API, with its query: never a full
address, and it cannot step out of that API, so `..` and `//host` are refused.
`body` is sent as JSON; leave it out on a `GET`.

The answer is the platform's own: its `status`, its `headers` (content type,
paging and rate limits) and its `body`, parsed from JSON when the platform
answered JSON and as text otherwise. A refusal from the platform comes back
the same way, with its status, rather than as an error: check
`answer["status"]`. A `GET` is retried like a read; any other method is made
once, because only you know whether repeating it is harmless.

## Your own functions: catalog

Your microserver can also hold functions written for you.
`catalog` lists them, with the platform each one is for, its id, and the JSON
Schema of its input and output. Leave `platform` out for every platform,
or name one to narrow the list. Name an `end_customer_id` to narrow it to your
own functions and that end customer's; left out, every end customer's
functions are listed too.

```python
for function in client.catalog(platform="hubspot", end_customer_id="lawfirm-123")["functions"]:
    print(function["id"], function["platform"], function["name"], function["description"])
```

A function's `owner` is `{"type": "customer"}` when every end customer can use
it, or `{"type": "end_customer", "logic_folder_key": "..."}` when it was
written for one of them, named by the logic folder key `connections.get`
answers for them. An end customer's functions are their versions of yours,
which keep your function's id, and their own. `input_schema` and
`output_schema` are JSON Schema, as written. Each function has one live
version: a change to it replaces it when the change goes live. Until a
function has gone live for you, the list is empty, not an error.

`id` is what `run` takes as `function_id`. It stays the same when the
function's name or version changes, and it is `None` for a function that went
live before functions had ids. `always_on` is `True` for a function that also
runs by itself, and `conditionals` says when; it is empty when the function is
not always on. A conditional with `object`, `operation` and `timing`, such as
`{"object": "deal", "operation": "write", "timing": "before"}`, runs it before
or after a `crm` read or write of that object; one with `action` and `timing`
runs it around one of the platform's ready-made actions, such as
`get-call-transcript`; and `{}` runs it after every `crm` call and action on
its platform.

Always-on logic runs around your `crm` calls and actions. Before a call it can
change what the call sends, and when it fails the call stops with
`LOGIC_FAILED`: a `LogicFailedError`, never tried again, with nothing written
or returned. After a call it runs in the background, and never slows or fails
the call. The docs explain it, with an example of each kind:
https://docs.appnigma.ai/sdk/always-on-logic.

## Your own functions: run

`run` calls one of them, by platform and either its name or its id, for one
end customer. The input is checked against the function's shape before
anything runs, so a wrong or missing field is an `InputError` listing every
problem. Formats and patterns, such as an email or a fixed layout for a code,
are checked by the function itself, so those refusals arrive from it with the
same code. The answer is `{"output": ..., "ran_as": ...}`: the output in the
shape the function declared, with its keys as the function wrote them, and
whose function ran. `ran_as["owner"]` is `"end_customer"` when the end
customer has their own version of it and it ran, `"customer"` when yours did;
`ran_as["version"]` is the version that ran.

```python
result = client.run(
    end_customer_id="lawfirm-123",
    platform="hubspot",
    function_name="invoice-to-deal",
    input={"invoiceId": "inv_1042", "amount": 1250},
)
print("deal", result["output"]["dealId"], "ran as", result["ran_as"]["owner"], result["ran_as"]["version"])

# Or by the id catalog gives it.
client.run(
    end_customer_id="lawfirm-123",
    platform="hubspot",
    function_id="lg_7f3a9c2d41b8",
    input={"invoiceId": "inv_1042", "amount": 1250},
)
```

`platform` is required: the same name can exist on two platforms and mean two
different things. So is one of `function_name` and `function_id`, never both;
both, or neither, is an `InputError` raised before the call. A name finds your
own function, and the end customer's version of it runs instead when they
have one. An id finds the end customer's own version first, then yours; an id
only another end customer has is a `NotFoundError`. What runs is always the
function's live version. A function that is always on can be run this way
too: it runs once, whatever its `conditionals` say, and sets off no other
function. A name or id that exists only on another platform is a
`NotFoundError` whose `details["platforms"]` says which. A function that was
live and was removed from your logic is retired: calling it is a
`RetiredError` with `FUNCTION_RETIRED`, whose `details` name the function and
when it was retired. A platform that is not switched on for you is an
`AuthError` with `PLATFORM_NOT_ENABLED`. When the function's platform refuses
what it sent, it is a `PlatformRefusedError` carrying the platform's reason.
When the function itself fails, it is a `LogicFailedError` whose `message` is
what the function threw, with no logic named; a write it made before it failed
stays made.

`run` is made once and never retried: it is your own function, and only you
know whether running it twice is harmless. If it fails with a network or
server error, decide in your code whether to call again.

## Your own guide

One page, written for your company: install, the key, connecting an end
customer to each platform switched on for you, reading and writing records,
`run`, every
one of your functions with its fields and an example call, and every error
with what to do about it. `guide` answers it as markdown, built from what is
live right now.

```python
from pathlib import Path

Path("appnigma-guide.md").write_text(client.guide()["guide"])
```

## The custom logic portal: portal

The portal is a page, inside your product, where one end customer sees their
own logic, edits it, puts it live and keeps the keys it reads in Settings. They
only ever see and change their own. It is switched on per company; until it is,
the call is an `AuthError` with `PORTAL_NOT_ENABLED`.

Your server asks for the link, never your page, because the call needs your
key. A link works once and expires after 15 minutes, so ask for a new one each
time you show the portal.

```python
link = client.portal.create_link(end_customer_id="lawfirm-123")
print(f"show it in an iframe, or open it in a new tab: {link['url']}")
```

Show the link's `url` in an iframe on one of your sites, or open it in a new
tab. Only the sites on your company's list can show it in an iframe; write to
support@appnigma.ai to add yours. A session lasts an hour. When the link was
already used, or the session ends, the portal tells your page with a message,
so your page can ask your server for a new link:

```js
const portalOrigin = new URL(portalUrl).origin; // the url your server sent
window.addEventListener("message", (event) => {
  if (event.origin !== portalOrigin) return;
  if (event.data?.type === "appnigma.portal.link-expired" || event.data?.type === "appnigma.portal.session-ended") {
    showPortal(); // your code: ask your server for a new link and set the iframe's src
  }
});
```

## Telling failures apart

Every failure is an exception you can catch by class. All of them inherit from
`AppnigmaError`, so one `except AppnigmaError` catches anything from
Appnigma. Every error carries `code`, `message`, `status` and `request_id`;
`str(error)` is the message. Quote the request id when you write to
support@appnigma.ai, and never include your key or a token.

| Class | When | Extra field |
| --- | --- | --- |
| `InputError` | 400. A field is wrong or missing. Also raised locally for a bad setting, a bad end customer id, or a `run` given both `function_name` and `function_id` or neither, before any call. | `problems`, a list of `{"field": ..., "problem": ...}` |
| `AuthError` | 401 or 403. The key is missing, unknown or cancelled, the microserver is not enabled, the platform or the custom logic portal is not switched on for you (`PLATFORM_NOT_ENABLED`, `PORTAL_NOT_ENABLED`), or the end customer's connection lacks a scope a write needs (`MISSING_SCOPES`, with `details["missing"]`). Also raised at start with no key. | |
| `NotFoundError` | 404. No connection for that end customer on that platform, no such function, or no such record (`RECORD_NOT_FOUND`). When the function exists on another platform, `details["platforms"]` names it. | |
| `ReconnectError` | 409 with `CONNECTION_NEEDS_RECONNECT`. The end customer has to approve again. | |
| `NotAvailableError` | Any other 409. The platform has no unified data for that yet (`DATA_TYPE_NOT_AVAILABLE`); use `request`. | |
| `RetiredError` | 410 with `FUNCTION_RETIRED`. The function was live and was removed from your logic; it will not answer again. | |
| `PlatformRefusedError` | 422 with `PLATFORM_REFUSED`. The end customer's platform refused the call; `message` is its reason. Never retried: fix what you sent. | `platform_status`, the status the platform answered, or `None` |
| `LogicFailedError` | 422 with `LOGIC_FAILED`. One of your functions, or always-on logic around the call, failed. Always-on logic before a `crm.*` call or an action stops it, so nothing was written or returned, and `message` names the logic and its id; a function called with `run` may have written before it failed, and `message` is what it threw. Never retried: the same call fails the same way until the logic changes. | `logic_ids`, the ids of the always-on logic the server named; empty on `run` |
| `RateLimitError` | 429, after the retries. Too many calls, or the platform asked us to slow down. | `retry_after_seconds`, or `None` when the server named no wait |
| `ServerError` | Any 5xx, after the retries. Also any successful answer whose body is not JSON. | |
| `NetworkError` | The API could not be reached, or an attempt ran out of time, after the retries. `status` and `request_id` are `None`. | `__cause__`, the underlying exception |

```python
from appnigma import (
    AuthError,
    InputError,
    LogicFailedError,
    NetworkError,
    NotFoundError,
    PlatformRefusedError,
    RateLimitError,
    ReconnectError,
    ServerError,
)

try:
    client.crm.contacts.create(end_customer_id="lawfirm-123", platform="hubspot", record={"first_name": "Jane"})
except InputError as error:
    for problem in error.problems:
        print(problem["field"], problem["problem"])
except AuthError as error:
    print("check the key:", error.code)
except NotFoundError:
    print("this end customer is not connected yet")
except ReconnectError:
    print("ask the end customer to connect again")
except PlatformRefusedError as error:
    print("the CRM refused it:", error.message)
except LogicFailedError as error:
    print("your logic failed:", error.message, error.logic_ids)
except RateLimitError as error:
    print(f"wait {error.retry_after_seconds} seconds")
except ServerError as error:
    print("our side failed, quote this id to support:", error.request_id)
except NetworkError as error:
    print("could not reach the API:", error.__cause__)
```

Or catch the base class:

```python
from appnigma import AppnigmaError

try:
    client.connections.get(end_customer_id="nobody-here", platform="hubspot")
except AppnigmaError as error:
    print(type(error).__name__, error.code, error.status, error.request_id)
```

`code` is one of the strings in `ERROR_CODES`, or whatever a newer server
sent. A code this version of the package does not know is kept as is, never
replaced.

## Retries and time limits

A call is tried again only when trying again is safe: the API could not be
reached, it answered 5xx, or it answered 429. A 400, 401, 403, 404, 409, 410
or 422 is never retried, because the same call would be refused the same way.
By default a call is made up to three times.

`connections.create_link`, `connections.disconnect`, `portal.create_link`, `run`, `request` with a
method other than `GET`, and a `crm` create with nothing to match on are the
exceptions: each is made once, never retried. Every call to
`create_link` makes a new link, and a repeat after a lost answer would leave a
working link nobody saw; `run` is your own function, and whether a repeat is
harmless is yours to know; a create with nothing to match on could make a
second record. If one of them fails, decide in your code whether to call
again.

The wait between tries is what the `Retry-After` header says when the server
sent one, otherwise half a second and then a second, with a little random
spread so many clients do not retry in the same instant. A `Retry-After`
longer than `max_retry_wait_ms` is not waited for: the call fails at once with
a `RateLimitError` that carries the wait, so your code decides what to do with
half a minute rather than sitting through it.

Every attempt has a time limit, 30 seconds by default. An attempt that runs out
of time counts as a network failure and is retried like one. Each call can set
its own limit:

```python
page = client.crm.contacts.list(end_customer_id="lawfirm-123", platform="hubspot", limit=1, timeout_ms=10_000)
print("one record:", len(page["records"]))
```

Times are in milliseconds, as in the TypeScript and Ruby packages, so the
numbers are the same in all three.

## Settings

All passed to `Appnigma()`, and all optional. `key` can be passed first
without its name; the rest are passed by name.

| Setting | Default | Meaning |
| --- | --- | --- |
| `key` | the `APPNIGMA_SDK_KEY` environment variable | Your SDK key. |
| `timeout_ms` | `30_000` | How long one attempt may take. A whole number of milliseconds, at most 2147483647. |
| `max_retries` | `2` | How many times a call is tried again. `0` means one attempt. |
| `max_retry_wait_ms` | `30_000` | The longest `Retry-After` the client will sit through, in milliseconds, at most 2147483647. |

Per call, by name: `timeout_ms`.

## What is exported

From `appnigma`: `Appnigma`, `AppnigmaError` and its ten subclasses, the
constants `ERROR_CODES`, `OBJECTS`, `CONNECTION_STATUSES` and
`PLATFORM_REQUEST_METHODS`, `__version__`, and `API_DESCRIPTION_VERSION`, the
version of the published description of the API that this version of the
package was generated from. Names that start with an underscore are not part
of the package's interface and can change in any release.

## License

MIT. See `LICENSE`.
