SDK de Python

Instalação

pip install mb-suite

• Python 3.9 ou posterior. Sem dependências: usa a biblioteca padrão (urllib), ou o cliente HTTP que você conectar (veja «Transportes»).
• Só no seu servidor: uma API key é um segredo. Nunca a coloque em um navegador nem em um app móvel.
• Uma API key (mbs_live_…) ou um token OAuth (mbo_…): veja Autenticação.

Seu primeiro pedido

from mb_suite import MBSuite, response_meta

mb = MBSuite()  # sem api_key, lê a variável de ambiente MB_API_KEY

me = mb.platform.me()  # GET /v1/me
print(me["actor"]["uid"], me["org"]["id"])
print(response_meta(me).request_id)  # req_…: inclua-o se escrever para nós

Cada operação tem um id estável e o método o segue tal qual: crm.deals.create é mb.crm.deals.create() e crm.custom_fields.list é mb.crm.custom_fields.list(). Um trecho que é palavra reservada do Python leva um sublinhado no fim: contacts.contacts.bulk.import é mb.contacts.contacts.bulk.import_(). A Referência mostra o id de cada uma.

1. 1Primeiro, os parâmetros de rota, como texto e na ordem da rota: mb.crm.deals.get("deal_123").
2. 2Depois, os parâmetros: um dict em snake_case, igual à API (a query de um GET, o corpo do resto).
3. 3No fim, as opções do pedido como argumentos nomeados: organization, idempotency_key, confirm_token, timeout, max_retries e extra_headers. Uma opção que não existe é um TypeError.

Listar, criar e editar

page = mb.crm.deals.list({"status": "open", "limit": 50})

created = mb.crm.deals.create({
    "title": "Plan anual — Acme",
    "value": 4800,
    "currency": "USD",
    "pipeline_id": "PIPELINE_ID",
    "stage_id": "STAGE_ID",
})
deal_id = created["deal_id"]

mb.crm.deals.move(deal_id, {"stage_id": "NEXT_STAGE_ID"})
mb.crm.deals.update(deal_id, {"updates": {"value": 5200}})
deal = mb.crm.deals.get(deal_id)

Os resultados são dict comuns, o JSON da API. Os parâmetros e os resultados são tipados a partir do contrato com TypedDict (em mb_suite.types) e cada método traz a descrição da sua operação: o seu editor a mostra, e o mypy ou o Pyright os verificam.

Paginação

Cada lista paginada tem um gêmeo …_all() que percorre todas as páginas e entrega os itens um por um, pedindo cada página conforme você avança. Uma página pode vir vazia com has_more em True: o percurso continua. Saia do laço quando tiver o que precisa. Veja Paginação.

# Segue next_cursor até has_more ser False; limit é o tamanho de cada página
for deal in mb.crm.deals.list_all({"status": "open", "limit": 200}):
    print(deal["id"], deal["title"])

Prévia e confirmação

As operações que pedem confirmação —as em massa, as irreversíveis, as que enviam ou mudam algo fora do MB Suite ou têm um custo à parte, e as que sempre pedem— têm .dry_run(): não escreve nada e devolve a prévia e confirm(), que executa a mesma operação com os mesmos parâmetros e o confirm_token da prévia (vale 10 minutos). A sua idempotency_key, passe-a para .dry_run() ou para confirm(): ela vai com a confirmação, nunca com a prévia, que não escreve nada. Veja Prévia e confirmação.

run = mb.crm.deals.bulk.delete.dry_run({"deal_ids": ["DEAL_A", "DEAL_B"]})

print(run.preview["summary"], run.preview["counts"])  # nada foi excluído ainda

result = run.confirm()  # dentro de 10 minutos
print(result["deleted"])

• Chamá-la sem token lança ConfirmationRequiredError.
• Se o confirm_token chegar por outro caminho (uma aprovação, um agente), passe-o como opção: mb.crm.deals.bulk.delete(params, confirm_token=token).

Novas tentativas e idempotência

• Cada mutação sai com uma Idempotency-Key (um UUID que o SDK gera, ou a sua) e a mantém nas novas tentativas: se uma resposta se perde, a nova tentativa devolve a guardada em vez de executar duas vezes.
• São repetidos sozinhos, até 3 vezes com espera exponencial (0,5 s, 1 s, 2 s): os erros de rede, os timeouts, um 429 (rate_limited, external_rate_limited, um cooldown_active curto), um 409 idempotency_in_progress e os 5xx (menos o 502 ai_generation_failed). Se a resposta traz Retry-After, espera exatamente isso.
• Não são repetidos: o resto dos 4xx, um 429 quota_exceeded (a cota diária volta às 00:00 UTC), um 502 ai_generation_failed (repetir chama a IA de novo e gasta créditos se ela responder: decida você) nem um Retry-After de mais de 60 segundos. O erro volta com retry_after para você reprogramar.

Para que uma operação lógica seja segura entre as suas próprias novas tentativas, dê a ela a sua própria chave (vale 24 horas). A mesma chave com outro corpo lança IdempotencyError. Veja Idempotência.

# A mesma chave numa nova tentativa sua (outro processo, outro deploy) devolve a resposta guardada
mb.crm.deals.create(new_deal, idempotency_key=f"import-{row_id}")

Erros

Todo erro herda de MBSuiteError e traz o formato único da API: type, code (estável: decida com ele), message, status, param, request_id, doc_url, details e retry_after. A classe depende do type; os nomes não escondem os do Python (PermissionDeniedError, APIConnectionError, APITimeoutError):

from mb_suite import MBSuiteError, NotFoundError, RateLimitError

try:
    deal = mb.crm.deals.get(deal_id)
except NotFoundError:
    deal = None
except RateLimitError as err:
    print(f"Tente de novo em {err.retry_after} s")
except MBSuiteError as err:
    print(err.code, err.request_id, err.doc_url)
    raise

InvalidRequestError soma issues (qual campo e por quê), ConfirmationRequiredError soma reason (expired, mismatch), QuotaExceededError é um RateLimitError com quota e MaintenanceError é um ServiceUnavailableError. Todos os códigos estão em Erros.

Contas

Uma API key de workspace trabalha nas Contas do seu alcance com MB-Organization. Escolha a Conta para todo o cliente, para um pedido ou com um cliente preso a ela. mb.platform.orgs.list() devolve as que a API key alcança. Veja Workspaces e Contas.

# Para todos os pedidos do cliente
mb = MBSuite(organization="ACCOUNT_ID")

# Para um pedido (None = a organização da API key)
mb.crm.deals.list({}, organization="OTHER_ACCOUNT_ID")

# Um cliente para essa Conta; o original não muda
account = mb.with_options(organization="ACCOUNT_ID")

O que cada resposta traz

response_meta(resultado) devolve o request_id, a versão, a rajada e a cota do dia (para frear antes de um 429, veja Limites) e se foi a resposta guardada de um pedido anterior. Também é o atributo last_response do resultado, que não é uma chave: não aparece em json.dumps nem em dict(resultado).

meta = response_meta(page)

meta.request_id           # "req_…"
meta.rate_limit           # LimitState(limit, remaining, reset): a rajada da API key (10 s)
meta.quota                # LimitState(limit, remaining, reset): as chamadas de hoje
meta.idempotent_replayed  # True se é a resposta guardada de um pedido anterior

O SDK envia MB-Version: 2026-11-01 em cada pedido: a versão que os seus tipos descrevem. Se uma resposta vier de uma versão em retirada, avisa uma vez com um APIVersionDeprecationWarning e a data de Sunset. Veja Versões.

Tipos e pedidos manuais

from mb_suite.types import CrmDealsCreateParams, CrmDealsListItem

new_deal: CrmDealsCreateParams = {"title": "Acme", "value": 4800, "currency": "USD", "pipeline_id": "P", "stage_id": "S"}

# Qualquer endpoint por método e rota, com as mesmas novas tentativas e erros
mb.request("POST", "/crm/deals/DEAL_ID/archive", body={"archived": True})

Para cada operação (crm.deals.list → CrmDealsList): …Params, …Result e, nas listas paginadas, …Item. Por padrão o cliente fala com https://api.mb-suite.com/v1; base_url o aponta para outro endereço (a mesma API responde em https://mb-suite.com/api/platform/v1).

Transportes

Por padrão o SDK usa a biblioteca padrão (certificados do sistema, as variáveis HTTP(S)_PROXY, sem seguir redirecionamentos). Para usar outro cliente HTTP, ou testar sem rede, passe qualquer objeto com um método send(request) -> HTTPResponse: envia um pedido; as novas tentativas, as chaves e os erros continuam no cliente.

import requests
from mb_suite import HTTPRequest, HTTPResponse, MBSuite

class RequestsTransport:
    def __init__(self) -> None:
        self.session = requests.Session()

    def send(self, request: HTTPRequest) -> HTTPResponse:
        r = self.session.request(request.method, request.url, headers=request.headers, data=request.body,
                                 timeout=request.timeout, allow_redirects=False)
        return HTTPResponse(status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()}, body=r.content)

mb = MBSuite(transport=RequestsTransport())

Para receber eventos, verifique a assinatura MB-Signature de cada entrega como mostra o guia de Webhooks (tem o trecho de Python).