SDK de Python

Instalación

pip install mb-suite

• Python 3.9 o posterior. Sin dependencias: usa la librería estándar (urllib), o el cliente HTTP que conectes (ver «Transportes»).
• Solo en tu servidor: una API key es un secreto. Nunca la pongas en un navegador ni en una app móvil.
• Una API key (mbs_live_…) o un token OAuth (mbo_…): ver Autenticación.

Tu primer pedido

from mb_suite import MBSuite, response_meta

mb = MBSuite()  # sin api_key, lee la variable de entorno MB_API_KEY

me = mb.platform.me()  # GET /v1/me
print(me["actor"]["uid"], me["org"]["id"])
print(response_meta(me).request_id)  # req_…: inclúyelo si nos escribes

Cada operación tiene un id estable y el método lo sigue tal cual: crm.deals.create es mb.crm.deals.create() y crm.custom_fields.list es mb.crm.custom_fields.list(). Un tramo que es palabra reservada de Python lleva un guion bajo al final: contacts.contacts.bulk.import es mb.contacts.contacts.bulk.import_(). La Referencia muestra el id de cada una.

1. 1Primero, los parámetros de ruta, como texto y en el orden de la ruta: mb.crm.deals.get("deal_123").
2. 2Después, los parámetros: un dict en snake_case, igual que la API (la query de un GET, el cuerpo del resto).
3. 3Al final, las opciones del pedido como argumentos con nombre: organization, idempotency_key, confirm_token, timeout, max_retries y extra_headers. Una opción que no existe es un TypeError.

Listar, crear y 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)

Los resultados son dict comunes, el JSON de la API. Los parámetros y los resultados están tipados desde el contrato con TypedDict (en mb_suite.types) y cada método trae la descripción de su operación: tu editor la muestra, y mypy o Pyright los verifican.

Paginación

Cada lista paginada tiene un gemelo …_all() que recorre todas las páginas y entrega los elementos uno por uno, pidiendo cada página a medida que avanzas. Una página puede venir vacía con has_more en True: el recorrido sigue igual. Sal del bucle cuando tengas lo que necesitas. Ver Paginación.

# Sigue next_cursor hasta que has_more sea False; limit es el tamaño de cada página
for deal in mb.crm.deals.list_all({"status": "open", "limit": 200}):
    print(deal["id"], deal["title"])

Vista previa y confirmación

Las operaciones que piden confirmación —las masivas, las irreversibles, las que mandan o cambian algo fuera de MB Suite o tienen un costo aparte, y las que la piden siempre— tienen .dry_run(): no escribe nada y devuelve la vista previa y confirm(), que ejecuta la misma operación con los mismos parámetros y el confirm_token de la vista previa (vale 10 minutos). Tu idempotency_key, pásala a .dry_run() o a confirm(): viaja con la confirmación, nunca con la vista previa, que no escribe nada. Ver Vista previa y confirmación.

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

print(run.preview["summary"], run.preview["counts"])  # todavía no se borró nada

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

• Llamarla sin token lanza ConfirmationRequiredError.
• Si el confirm_token llega por otro camino (una aprobación, un agente), pásalo como opción: mb.crm.deals.bulk.delete(params, confirm_token=token).

Reintentos e idempotencia

• Cada mutación sale con una Idempotency-Key (un UUID que genera el SDK, o la tuya) y la conserva en los reintentos: si una respuesta se pierde, el reintento devuelve la guardada en vez de ejecutar dos veces.
• Se reintentan solos, hasta 3 veces con espera exponencial (0,5 s, 1 s, 2 s): los errores de red, los timeouts, un 429 (rate_limited, external_rate_limited, un cooldown_active corto), un 409 idempotency_in_progress y los 5xx (menos el 502 ai_generation_failed). Si la respuesta trae Retry-After, espera exactamente eso.
• No se reintentan: el resto de los 4xx, un 429 quota_exceeded (el cupo diario vuelve a las 00:00 UTC), un 502 ai_generation_failed (repetirlo vuelve a llamar a la IA y gasta créditos si responde: lo decides tú) ni un Retry-After de más de 60 segundos. El error vuelve con retry_after para que lo reprogrames.

Para que una operación lógica sea segura entre tus propios reintentos, dale tu propia clave (vale 24 horas). La misma clave con otro cuerpo lanza IdempotencyError. Ver Idempotencia.

# La misma clave en un reintento tuyo (otro proceso, otro deploy) devuelve la respuesta guardada
mb.crm.deals.create(new_deal, idempotency_key=f"import-{row_id}")

Errores

Todo error hereda de MBSuiteError y trae el formato único de la API: type, code (estable: decide con él), message, status, param, request_id, doc_url, details y retry_after. La clase depende del type; los nombres no tapan los de 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"Reintenta en {err.retry_after} s")
except MBSuiteError as err:
    print(err.code, err.request_id, err.doc_url)
    raise

InvalidRequestError suma issues (qué campo y por qué), ConfirmationRequiredError suma reason (expired, mismatch), QuotaExceededError es un RateLimitError con quota y MaintenanceError es un ServiceUnavailableError. Todos los códigos están en Errores.

Cuentas

Una API key de workspace trabaja en las Cuentas de su alcance con MB-Organization. Elige la Cuenta para todo el cliente, para un pedido o con un cliente atado a ella. mb.platform.orgs.list() devuelve las que alcanza la API key. Ver Workspaces y Cuentas.

# Para todos los pedidos del cliente
mb = MBSuite(organization="ACCOUNT_ID")

# Para un pedido (None = la organización de la API key)
mb.crm.deals.list({}, organization="OTHER_ACCOUNT_ID")

# Un cliente para esa Cuenta; el original no cambia
account = mb.with_options(organization="ACCOUNT_ID")

Lo que trae cada respuesta

response_meta(resultado) devuelve el request_id, la versión, la ráfaga y el cupo del día (para frenar antes de un 429, ver Límites) y si fue la respuesta guardada de un pedido anterior. También es el atributo last_response del resultado, que no es una clave: no aparece en json.dumps ni en dict(resultado).

meta = response_meta(page)

meta.request_id           # "req_…"
meta.rate_limit           # LimitState(limit, remaining, reset): la ráfaga de la API key (10 s)
meta.quota                # LimitState(limit, remaining, reset): las llamadas de hoy
meta.idempotent_replayed  # True si es la respuesta guardada de un pedido anterior

El SDK manda MB-Version: 2026-11-01 en cada pedido: la versión que describen sus tipos. Si una respuesta llega de una versión en retiro, avisa una vez con un APIVersionDeprecationWarning y su fecha de Sunset. Ver Versiones.

Tipos y pedidos a mano

from mb_suite.types import CrmDealsCreateParams, CrmDealsListItem

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

# Cualquier endpoint por método y ruta, con los mismos reintentos y errores
mb.request("POST", "/crm/deals/DEAL_ID/archive", body={"archived": True})

Por cada operación (crm.deals.list → CrmDealsList): …Params, …Result y, en las listas paginadas, …Item. Por defecto el cliente habla con https://api.mb-suite.com/v1; base_url lo apunta a otra dirección (la misma API responde en https://mb-suite.com/api/platform/v1).

Transportes

Por defecto el SDK usa la librería estándar (certificados del sistema, las variables HTTP(S)_PROXY, sin seguir redirecciones). Para usar otro cliente HTTP, o probar sin red, pasa cualquier objeto con un método send(request) -> HTTPResponse: manda un pedido; los reintentos, las claves y los errores siguen en el 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 recibir eventos, verifica la firma MB-Signature de cada entrega como muestra la guía de Webhooks (tiene el fragmento de Python).