Plank help · updated 2026-10-15

Working with amoCRM

A ready amoCRM client for workspace scripts — batch reads by id, batch updates, one shared rate limit, safe retries — plus how to connect an account and how to run bulk work in minutes instead of an hour.

Agents: fetch the raw markdown of this page at /en/help/amocrm.md

Working with amoCRM

Every script that talks to amoCRM goes through ONE client file, scripts/amocrm/lib/amocrm.py. Do not write a urllib call of your own in each script, and never put time.sleep between calls: the client already keeps the connection open, holds one rate limit for every process in the workspace, and retries what is safe to retry.

Why this matters: one request per record, a pause after each, and a new TLS handshake every time turned a 650-row reconciliation into ~50 minutes and ran into the 20-minute command timeout. Read by id in batches, write in batches, and the same job is one to two minutes.

1. Install the client

mkdir -p scripts/amocrm/lib
test -e scripts/amocrm/lib/amocrm.py && cp scripts/amocrm/lib/amocrm.py scripts/amocrm/lib/amocrm.py.bak-$(date +%F)
curl -sSfL -o scripts/amocrm/lib/amocrm.py https://plank.md/help/assets/amocrm.py

If the workspace already has its own amocrm.py, back it up first (the second line) and check that the scripts still import: the client keeps the usual names — get(path, params), post(path, payload), patch(path, payload), load_json, save_json, access_token, ROOT, CREDENTIALS_PATH, TOKEN_PATH — and the error text amoCRM API failed (<status>): <body>. Run python3 -m py_compile over every script that imports it before you use them.

From a script:

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "amocrm" / "lib"))  # adjust the level
import amocrm

2. Connect the account

The client takes the first of these that is configured:

A long-lived token in the environment (recommended). In amoCRM: Settings → Integrations → create a private integration → «Keys and access» → generate a long-lived token. The person saves it under Credentials (see Credentials); the account address is not a secret and goes in scripts/amocrm/credentials.json as {"base_url": "https://<subdomain>.amocrm.ru"}. Run scripts as plank-secret run --env AMOCRM_TOKEN="<item label>" -- python3 scripts/amocrm/<script>.py. No token file is written anywhere. AMOCRM_BASE_URL in the environment overrides base_url.
A long-lived token in the file: credentials.json with base_url and long_lived_token.
OAuth: credentials.json with base_url, client_id, client_secret, redirect_uri, plus scripts/amocrm/token.json from the code exchange. The client refreshes the token before it expires and on a 401, one process at a time, and writes the new pair back — amoCRM rotates the refresh token on every refresh, so the file must stay writable and must never be copied elsewhere.

A token file never goes into a chat, a report or a shared folder. Check the connection without printing anything secret:

python3 -c "import sys; sys.path.insert(0, 'scripts/amocrm/lib'); import amocrm; a = amocrm.get('/api/v4/account'); print(a['name'], a['subdomain'])"

A rejected long-lived token raises «the long-lived token was rejected — it expired or was revoked»: the person issues a new one; do not try other credentials.

3. The client

Call	What it does
`get(path, params=None)`	One GET. `params` is a dict (a list value repeats the key: `{"filter[id][]": [1, 2]}`) or a list of pairs. 204 → `{}`.
`post(path, payload)` / `patch(path, payload)`	One write. Payload is a dict or a list of entities.
`get_all(path, params=None, key=None)`	Every page of a list (250 per page, follows `_links.next`); returns the `_embedded[key]` items, `key` defaulting to the last path segment.
`get_by_ids(path, ids, params=None)`	Many entities by id with `filter[id][]`, 150 ids per request. Missing ids are simply absent. Works on leads, contacts, companies, customers.
`patch_many(path, entities)`	Batch update, 50 entities per request; each entity carries its `id`. A failure says how many were written before it.
`parallel_map(fn, items, workers=4, errors="raise", progress_every=0)`	Per-item work on threads, results in input order. `errors="return"` puts the exception in the item's place instead of stopping; `progress_every=N` prints `done/total` to stderr.
`chunks(items, size)`	Split a list.

Built in, so scripts must not redo it:

Rate limit. amoCRM allows 7 requests per second per account and answers 429 above that (repeated excess can block access). The client spaces every request of every process in the workspace to 5 per second through a lock file. Parallel workers share that limit, so workers=4 is plenty.
Retries. 429 is retried after its Retry-After for reads and writes (nothing was processed). 502/503/504 and a dropped connection are retried for GET only.
A write whose answer was lost is never repeated. It raises «the write may or may not have happened; read the entity back before repeating it». Read it, then decide.

Useful endpoints: /api/v4/leads, /api/v4/contacts, /api/v4/companies, /api/v4/customers (only when «Customers» is switched on in the account), /api/v4/leads/pipelines, /api/v4/users, /api/v4/<entity>/custom_fields, /api/v4/events (what changed and when), /api/v4/<entity>/notes, /api/v4/tasks. Linked records come with with=contacts (or catalog_elements, leads) instead of one request per record. Custom field values are read and written as custom_fields_values: [{"field_id": …, "values": [{"value": …}]}]; look the ids up once from /custom_fields.

4. Bulk work

Snapshot, then reconcile. Comparing N rows with amoCRM is one get_by_ids (or one get_all with a filter), matched locally in a dict — never N single get calls in a loop.
Preview, then write — with the person's approval. A bulk change shows what will change where (a table or a CSV of id, field, old → new) and waits for an explicit yes. Being fast does not change this.
Write in batches. patch_many for many entities of one type. Verify with one get_by_ids over the ids you changed, not one read per record.
Per-item chains run in parallel. When each row needs its own sequence of calls, parallel_map(..., workers=4, errors="return") so one failure lands in the report instead of stopping the rest.
Long jobs go to the background. Anything over two or three minutes: nohup python3 -u job.py > job.log 2>&1 &, progress_every=50, results row by row in a CSV, and tell the person how far it got every minute or so — see Running commands.

# Example: set a field on many leads after the person approved the preview.
FIELD_ID = 123456
rows = [{"lead_id": 101, "value": "Да"}, ...]                    # from the approved preview
before = {lead["id"]: lead for lead in amocrm.get_by_ids("/api/v4/leads", [r["lead_id"] for r in rows])}
changes = [{"id": r["lead_id"], "custom_fields_values": [{"field_id": FIELD_ID, "values": [{"value": r["value"]}]}]}
           for r in rows if r["lead_id"] in before]
amocrm.patch_many("/api/v4/leads", changes)
after = amocrm.get_by_ids("/api/v4/leads", [c["id"] for c in changes])   # one read-back for all of them