El SDK de Python

Empiece aquí

Si nunca ha usado Conecto, esta sección contiene todo el modelo mental. Cinco minutos aquí hacen evidente el resto de la página.

Qué es Conecto

Un widget de chat en su sitio web. Los visitantes escriben; su equipo, un agente de IA o su propio código responde. Todo lo siguiente sirve para introducir su código su código en ese bucle.

Los cinco sustantivos

Casi todos los métodos de este SDK aceptan o devuelven uno de estos objetos.

Las tres cosas que puede crear

¿Bot o plugin? Si quiere escribir las palabras que lee el visitante, cree un bot. Si prefiere dejar hablar a la IA y darle solo acceso a sus datos, como pedidos, suscripciones o existencias, cree un plugin. La mayoría de los equipos termina eligiendo un plugin.

Palabras que utiliza esta página

Jerga definida una sola vez. Solo los dos últimos términos son específicos de Conecto.

¿Qué sección necesito?

Instalar y autenticar

pip install conecto

Crear una credencial en Ajustes → Desarrolladores. Obtiene un ID de cliente (ck_…) and a secreto (cs_…). The secret is shown once and stored hashed, so put it somewhere your code can read and your repository cannot.

export CONECTO_CLIENT_ID=ck_your_client_id
export CONECTO_SECRET=cs_your_secret

from conecto import Conecto

client = Conecto()                    # reads the environment
client = Conecto("ck_...", "cs_...")  # or pass them directly

print(client.ping()["workspace"]["name"])

Call ping() once at startup. It raises AuthenticationError immediately if the credential is wrong or revoked — much better than finding out mid-conversation. A credential can cover the workspace or be scoped to one widget; a scoped one simply cannot see other widgets' conversations.

Inicio rápido

Tres cosas que puede hacer en los primeros cinco minutos.

Leer qué está ocurriendo

for convo in client.conversations.list(status="open"):
    who = convo.visitor.email if convo.visitor else "anonymous"
    print(f"#{convo.id}  {who}  {convo.preview}")

Responder a alguien

convo = client.conversations.get(4821)
print(convo.messages[-1].body)          # what they said last
convo.reply("On it — give me one moment.")

Enviar algo que no sea texto

from conecto import blocks

convo.reply("Here's how to reset it:", blocks=[
    blocks.image("https://cdn.you.com/reset.gif", alt="Reset flow"),
    blocks.buttons([
        blocks.reply_button("That worked"),
        blocks.link_button("Full guide", "https://docs.you.com/reset"),
    ]),
])

Así se estructura toda la biblioteca: un cliente con recursos, objetos capaces de actuar sobre sí mismos y builders para cualquier contenido estructurado.

El cliente

Cree un cliente y consérvelo. Agrupa conexiones y puede compartirse de forma segura entre hilos.

client = Conecto(
    client_id="ck_...",
    secret="cs_...",
    base_url="https://conecto.chat/api/v1",  # override for staging
    timeout=30.0,        # seconds per request
    max_retries=2,       # timeouts, 429s and 5xx
    app="acme-billing",  # added to the User-Agent; shows up in our logs
)

Entorno

Recursos

Desviación de versiones

Cada respuesta incluye la versión de la API del servidor. Regístrela al iniciar. Si cambia y desaparece un campo que utiliza, será lo primero que querrá conocer.

print(client.api_version)   # what the server last reported
print(client.sdk_version)   # what this SDK was built against

limits = client.meta.schema()["limits"]
assert limits["blocks_per_message"] >= 4

client.meta.schema() devuelve toda la API como JSON, incluidos endpoints, eventos, tipos de bloque, acciones reservadas, códigos de error y todos los límites. El mismo código aplica los límites y sirve la respuesta, por lo que no puede apartarse de la realidad.

Conversaciones

A thread between a visitor and your workspace. Messages have a sender de visitor, agent, bot o system.

Listar y leer

page = client.conversations.list(status="open", limit=25)

page.items                     # this page only
for convo in page:             # every page, fetched lazily
    ...

convo = client.conversations.get(4821)
convo.status, convo.visitor.email, convo.messages[-1].body

# Poll cheaply: only what is newer than what you already have.
fresh = client.conversations.messages(4821, since_id=last_seen_id)

Responder

convo.reply("I can refund order #1284 — confirm?",
            buttons=["Yes, refund it", "Talk to a human"])

Al pulsar un botón de respuesta rápida, el texto vuelve como un mensaje normal del visitante; por tanto, el controlador no necesita ningún caso especial.

Dirección del hilo

convo.typing()                  # "…is typing", expires after ~8s
convo.reply("Working on it…")

convo.reply("Escalating: refund over policy limit.", internal=True)
convo.handoff()                 # flag for a human; routing rules apply
convo.assign(user_id=12)        # ids come from client.members.list()

convo.close()
convo.reopen()

Writing to a closed thread raises ConversationClosed. Reopen it first, or catch it — see Errores.

Contenido enriquecido

Cualquier mensaje puede incluir hasta diez bloques: imágenes y GIF, vídeo, audio, archivos, contenido incrustado de terceros, carruseles de tarjetas, listas de etiqueta/valor, botones y separadores. Se muestran en el widget y en la bandeja del agente.

Los bloques son diccionarios sencillos durante la transmisión y podrían escribirse a mano. Los builders existen porque una clave mal escrita se convierte en un cliente que dice que la imagen nunca apareció. Aplican las mismas reglas que el servidor cuando el traceback todavía apunta a su código.

from conecto import blocks

convo.reply("Here's your order:", blocks=[
    blocks.list_block([
        blocks.row("Placed", "12 July 2026"),
        blocks.row("Status", "Shipped"),
        blocks.row("Carrier", "DHL", subtitle="Tracking 4Z8871",
                   url="https://track.dhl.com/4Z8871"),
    ], title="Order #1042"),
    blocks.divider(),
    blocks.cards([
        blocks.card("Trail Runner 2",
                    subtitle="Road · Neutral",
                    price="89.00 USD",
                    badge="Back in stock",
                    image="https://cdn.you.com/tr2.jpg",
                    url="https://shop.you.com/p/tr2",
                    buttons=[blocks.link_button("Buy", "https://shop.you.com/cart")]),
    ]),
])

Cada tipo de bloque

Dos tipos de botones

blocks.reply_button("Yes, refund it")                    # sends that text back
blocks.reply_button("Yes, refund it", "confirm_refund")  # nice label, stable value
blocks.link_button("Read the policy", "https://you.com/refunds")  # opens a tab

Dé un valor explícito a un botón de respuesta cuando la etiqueta deba leerse bien, pero el texto que interpreta su bot no deba cambiar cada vez que alguien reescribe el contenido.

Enviar un GIF

A .gif works. A muted, looping, autoplaying video is usually the better trade — a fraction of the bytes, and nobody can tell.

blocks.video("https://cdn.you.com/reset.mp4",
             autoplay=True, loop=True, muted=True)

Reglas que aplican los builders

• Every URL must be https. The widget runs on an https page, so anything else is blocked by the browser anyway.
• Diez bloques por mensaje, diez tarjetas, doce filas de lista, seis botones.
• Long strings are truncated, not rejected. Structural mistakes raise BlockError.
• A YouTube link in a video block raises, and tells you to use embed.

La validación se realiza localmente. reply(blocks=…) checks the list before sending, so a typo raises BlockError from your line rather than coming back as an HTTP 400 from inside the client.

Subida de archivos

¿No hay una URL pública para ese GIF, recibo o ficha técnica? Suba el archivo, hasta 10 MB, y use el resultado.

media = client.media.upload("celebrate.gif")     # a path
media = client.media.upload(open("a.pdf", "rb")) # a file object
media = client.media.upload(raw_bytes, filename="receipt.png")

convo.reply("All set!", blocks=[blocks.image(media)])

A Media can be passed straight to blocks.image, blocks.video o blocks.file.

Use media.url, not media.absolute_url. url es una ruta que el widget resuelve respecto a su propio origen. Así, el mismo mensaje funciona en desarrollo, en una vista previa y en producción, incluso después de cambiar de dominio.

Contactos y visitantes

Contactos

The CRM. upsert is keyed on email, so it is safe to point a nightly sync at.

client.contacts.upsert(
    "maya@acme.io",
    name="Maya Silva",
    company="Acme",
    custom_fields={"plan": "scale", "mrr": "480"},
)

contact = client.contacts.find("maya@acme.io")     # or None
for c in client.contacts.list(query="acme"):
    print(c.email, c.custom_fields.get("plan"))

custom_fields es combinado, no se sustituyen. Dos tareas pueden administrar sus propias claves sin sobrescribirse. Hasta 30 claves por llamada.

Visitantes

A browser session on a widget, keyed by (widget_id, session). The session id lives in the visitor's browser; read it client-side and send it to your backend.

visitor = client.visitors.get(widget_id=7, session=session_id)
visitor.email, visitor.verified

Mensajes proactivos

Enviar a una sesión sin esperar una petición, por ejemplo novedades de envío, avisos de prueba o recuperación del carrito.

client.visitors.message(
    widget_id=7, session=session_id,
    body="Still thinking it over? Your cart is saved:",
    blocks=[blocks.cards([
        blocks.card("Trail Runner 2", price="89.00 USD",
                    image="https://cdn.you.com/tr2.jpg",
                    url="https://shop.you.com/cart"),
    ])],
)

Reutiliza la conversación activa de la sesión o abre una. Un widget abierto la muestra en la siguiente consulta; uno cerrado la muestra al volver a abrirse, con el tratamiento de no leído.

Tickets y artículos

Tickets

ticket = client.tickets.create(
    email="maya@acme.io",
    message="Card was declined on renewal.",
    priority="high",          # low · normal · high · urgent
)

client.tickets.update(ticket.id, status="pending", assignee_user_id=12)
ticket.reply("We've fixed the card on file — try again?")   # emails the requester
ticket.reply("Billing confirmed the retry.", internal=True)  # private note

Una respuesta pública reabre un ticket resuelto, porque responder a algo marcado como terminado casi siempre significa que empieza otra ronda.

Artículos del centro de ayuda

Útil en ambas direcciones: un bot puede buscar ahí antes de abrir un ticket y usted puede enviar documentación desde cualquier herramienta de autoría.

hits = client.articles.list(query="refund", published=True)
if hits:
    convo.reply(f"This might help: “{hits[0].title}”")

client.articles.create("Shipping times", "<p>2-4 business days.</p>",
                       published=True)

El HTML se limpia en el servidor con una lista de etiquetas seguras, mediante la misma canalización que usa el editor del panel. Los scripts y estilos se eliminan antes de guardar nada.

Crear un bot

A bot is a webhook and a reply. Bot handles the five things that are easy to get subtly wrong — signature verification, delivery deduplication, event routing, filtering out your own messages, and answering fast — so your code is the part that is yours.

from conecto import Conecto, Bot, blocks

client = Conecto()
bot = Bot(client, secret=WEBHOOK_SECRET)

@bot.on_message
def handle(ctx):
    text = ctx.text.lower()

    if "human" in text:
        ctx.reply("Of course — connecting you now.")
        ctx.handoff()
        return

    if "order" in text:
        if not ctx.verified:
            ctx.reply("What's the email on the order?", ask_email=True)
            return
        order = lookup_order(ctx.verified_email)     # your system
        ctx.reply("Here it is:", blocks=[
            blocks.list_block([
                blocks.row("Status", order["status"]),
                blocks.row("Carrier", order["carrier"]),
            ], title=order["number"]),
        ])
        return

    ctx.reply("I'm not sure about that one — let me get someone who is.")
    ctx.note(f"Bot could not classify: {ctx.text!r}")
    ctx.handoff()

Publicación

# Flask
app.add_url_rule("/conecto", view_func=bot.flask_view(), methods=["POST"])

# FastAPI (handlers run in a worker thread, so they never stall the loop)
app.post("/conecto")(bot.fastapi_route())

# Django (urls.py)
path("conecto/", csrf_exempt(bot.django_view()))

# No framework at all
from wsgiref.simple_server import make_server
make_server("", 8000, bot.wsgi_app()).serve_forever()

Después, suscriba el endpoint:

hook = client.webhooks.create(
    "https://bots.you.com/conecto",
    ["message.created", "conversation.created", "conversation.rated"],
)
print(hook.secret)   # store this — it is how you verify deliveries

Controladores

Qué recibe un controlador

¿Ejecuta más de un worker? Deduplication is in-process by default, so two workers can each handle the same delivery once. Pass your own store — anything with add(id) -> bool, backed by Redis or your database.

class RedisSeen:
    def add(self, delivery_id):
        # True if new, False if we have handled it before
        return bool(redis.set(f"conecto:{delivery_id}", 1, nx=True, ex=86400))

bot = Bot(client, secret=WEBHOOK_SECRET, seen=RedisSeen())

Crear un plugin

Un bot responde a webhooks. Un plugin es la otra dirección: usted declara lo que pueden hacer sus sistemas y el agente de IA decide cuándo usarlo mientras habla con alguien. Nunca escribe la lógica de la conversación, solo la consulta.

Así es como una tienda, sistema de facturación o CRM que no conocemos se convierte en una integración de primera clase.

from conecto import Conecto, Plugin, parameter

plugin = Plugin("acme-store", base_url="https://api.acme.com/conecto")

@plugin.action(
    "catalog.search_products",
    description="Search the Acme catalog by keyword.",
    parameters=[parameter("query", required=True,
                          description="Search words from the visitor.")],
)
def search(req):
    hits = catalog.search(req.get("query"), limit=req.get("limit", 4))
    if not hits:
        return req.not_found()
    return req.ok(products=[
        {"title": p.name, "url": p.url, "image": p.image,
         "price_from": f"{p.price:.2f}", "currency": "USD",
         "available": p.stock > 0}
        for p in hits
    ])

@plugin.action(
    "orders.get_status",
    risk="verified_read",
    description="Status of one of the visitor's own orders.",
    parameters=[parameter("order_number", required=True)],
)
def order_status(req):
    order = orders.find(req.get("order_number"), email=req.verified_email)
    return req.ok(**order) if order else req.not_found()

Servir y registrar

# One route serves every action.
app.add_url_rule("/conecto/<path:action>", view_func=plugin.flask_view(),
                 methods=["POST"])

# Tell Conecto it exists. Idempotent — run it on every release.
integration = plugin.deploy(Conecto())
print(integration.signing_secret)   # set this in your environment

deploy() crea o actualiza la integración para que coincida con su código, elimina las acciones retiradas y la instala en sus widgets. No se expone nada a la IA hasta instalarla.

Los cinco tipos de respuesta

return req.ok(status="Shipped", carrier="DHL")   # success
return req.ok(result, blocks=[...])              # …plus rich content in the reply
return req.not_found()                           # nothing matched — a normal outcome
return req.verify_required()                     # "I need a proven identity"
return req.error("That subscription was already cancelled.")

Write error messages for the visitor, not for your logs: the AI may relay them. Returning a plain dict is also fine — it is treated as the result.

Niveles de riesgo

req.verified_email es la única identidad de confianza. An email in req.arguments is whatever the model parsed out of a chat. For verified_read y write the platform will not call you at all until we have proven who the visitor is, and the proven address is the one we hand you.

Acciones reservadas

Algunos nombres tienen un significado especial para la plataforma. Declare uno y su servicio se conectará a la función que ya existe para él.

Probarlo antes que un visitante

client.integrations.run("acme-store", "catalog.search_products",
                        {"query": "trail shoes"})
# {'status': 'ok', 'result': {...}, 'blocks': []}

Esto ejecuta la real call path — the signature, your credential, the checks that stop your URL pointing anywhere private, the timeout, the parsing — so what you see is exactly what the AI will get. Pass conversation_id= to run in the context of a live chat, which is the only way a verified action can succeed.

Escribir una buena descripción

El description is the highest-leverage string in your integration: it is what the AI reads to decide whether this action answers the question in front of it. Write it as an instruction to a capable colleague who cannot see your code.

Débil: «Consulta de pedidos.»
Fuerte: «Consulta el estado y la fecha de entrega de uno de los pedidos del visitante mediante su número. Úsalo siempre que pregunte dónde está algo o cuándo llegará.»

Qué garantiza la plataforma

• 8 segundos tiempo de espera, 128 KB límite de respuesta.
• Calls are signed with your signing_secret — same scheme as webhooks.
• Su URL se resuelve una vez y se fija a una IP pública; las redirecciones se rechazan.
• Todo lo que devuelve llega al modelo como datos, nunca como instrucciones. Las claves que parecen secretos se eliminan.
• Las integraciones personalizadas requieren un plan de pago Agente de IA. 20 por espacio de trabajo, 40 acciones cada una.

Identidad

Cualquier operación que afecte a los datos de una persona espera hasta que sepamos quién es. Esto sucede de dos formas: mediante un código enviado por correo o gracias a la validación de su sitio para un usuario que ya ha iniciado sesión.

# your backend, right after your own auth check
client.visitors.identify(
    widget_id=7,
    session=conecto_session,          # read from the visitor's browser
    email=user.email,
    name=user.name,
    verified=True,                    # the vouch
    verify_hours=24,                  # max 72, default 12
    data={"plan": user.plan, "customer_since": "2024"},
)

# when they log out
client.visitors.unverify(widget_id=7, session=conecto_session)

Mientras dura la validación, Conecto considera demostrado el correo: el widget conoce el nombre, los chats se vinculan al contacto correcto y los flujos que de otro modo pedirían un código por correo, como consultas de pedidos, reembolsos, datos de cuenta o acciones verificadas de su plugin, continúan sin él.

Valide solo las sesiones que su backend haya autenticado realmente. Su validación se considera tan sólida como nuestro propio código por correo porque realmente autenticó a la persona. Tiene duración limitada y ámbito de sesión, por lo que un navegador nuevo necesita una nueva validación. El navegador nunca puede marcarse a sí mismo como verificado, ya que esa llamada está autenticada y debe hacerse en su servidor.

Webhooks y eventos

Bot se ocupa de todo esto. Use los helpers de bajo nivel cuando quiera verificar en el borde y procesar en una cola.

from conecto import webhooks

event = webhooks.parse(request.get_data(), request.headers, secret=SECRET)

event.id                  # the delivery id — dedupe on this
event.event               # "message.created"
event.conversation_id
event.text
event.is_visitor_message  # excludes your own bot's messages
event.visitor.verified

Verificar con el cuerpo sin procesar. Re-serializing JSON changes the bytes and the signature stops matching — this is the most common first-integration bug there is. Use request.get_data() in Flask, request.body in Django, await request.body() in FastAPI. Passing a parsed dict raises with this exact advice.

Eventos

Semántica de entrega

Las entregas son al menos una vez y sin orden garantizado: if we cannot confirm you got one we send it again, and they will not always arrive in the order things happened. They also time out after 2.5 seconds — so answer 200 immediately and do the work afterwards.

• Every payload carries an id (also in X-Conecto-Delivery). Store it and skip ids you have already handled.
• Pass that same id as idempotency_key when you reply, and a redelivery can never make your bot speak twice.
• X-Conecto-Timestamp permite rechazar entregas antiguas; el SDK lo comprueba con una tolerancia predeterminada de cinco minutos.

signature = webhooks.sign(SECRET, raw_body)      # what we send
webhooks.verify(raw_body, header_value, SECRET)  # True / False, never raises

Llamadas de integración que realizamos a your plugin use the identical scheme with the integration's signing_secret — one verification routine covers both directions.

Errores y reintentos

Every failure raises a subclass of ConectoError, so you can catch exactly what you expect.

from conecto import ConversationClosed, RateLimited, NotFoundError

try:
    convo.reply("Refunded — you'll see it in 3-5 days.")
except ConversationClosed:
    convo.reopen()
    convo.reply("Refunded — you'll see it in 3-5 days.")
except NotFoundError:
    log.warning("conversation vanished")

Every exception carries code, message, status and the raw response.

Reintentos

Timeouts, rate limits (429) and server errors (5xx) are retried for you. Each retry waits a little longer than the last, with a random amount added so that a thousand clients recovering at once do not all come back in the same instant — and if we send a Retry-After, that wins.

Las escrituras también se reintentan, lo que normalmente sería peligroso. Aquí es seguro porque cada escritura incluye una clave de idempotencia y el reintento reutiliza la misma. El servidor reconoce el segundo intento como la misma llamada y no lo repite.

client = Conecto(max_retries=0)   # opt out entirely

Límites importantes

client.meta.schema()["limits"] siempre contiene los números en directo.

Paginación

List endpoints return a Page you can use three ways.

page = client.contacts.list(query="acme")

page.items        # just this page — a list, so len() and [0] work
page.has_more     # is there another?

for contact in page:   # every page, fetched lazily as you consume
    ...

page.all()        # everything, collected into one list

Iterating is lazy: next(iter(page)) costs exactly one request no matter how many contacts exist. all() is fine for contacts and tickets; think twice on a busy workspace's conversations, where iterating and stopping early is usually what you meant.

Probando

The client takes a transport, which is the seam for tests. Anything with get, post, patch, delete y request works — no network, no credentials, no mocking library.

class FakeTransport:
    base_url = "https://conecto.test/api/v1"

    def __init__(self):
        self.calls = []
        self.responses = {}

    def request(self, method, path, **kw):
        self.calls.append((method, path, kw.get("json_body")))
        return self.responses.get(f"{method} {path}", {})

    def get(self, p, **kw): return self.request("GET", p, **kw)
    def post(self, p, json_body=None, **kw): return self.request("POST", p, json_body=json_body, **kw)
    def patch(self, p, json_body=None, **kw): return self.request("PATCH", p, json_body=json_body, **kw)
    def delete(self, p, **kw): return self.request("DELETE", p, **kw)


def test_bot_answers_refunds():
    transport = FakeTransport()
    client = Conecto(transport=transport)
    client.conversations.reply(1, "Refunded.")
    assert transport.calls[0][1] == "/conversations/1/messages/"

Para bots y plugins, omita la verificación de firma en las pruebas en lugar de firmar fixtures:

bot = Bot(client, verify=False)
plugin = Plugin("acme", base_url="https://x.test", verify_signatures=False)

Referencia de métodos

Conversaciones

Contactos

Visitantes

Integraciones

Tickets, artículos, widgets, contenido multimedia, webhooks y metadatos

Modelos

Every returned object keeps the payload it was built from. Attribute access is the nice path; obj.raw is always there for a field this SDK version does not know about yet, so a release is never what stands between you and a new API field.

convo.status          # typed
convo.raw["status"]   # the same thing
convo["status"]       # shorthand for the raw payload
convo.to_dict()       # exactly what the API sent

Recetas

Desviar un ticket mediante su centro de ayuda

@bot.on_message
def deflect(ctx):
    hits = ctx.client.articles.list(query=ctx.text, published=True)
    if hits:
        ctx.reply(f"This might help: “{hits[0].title}”. Did that solve it?",
                  buttons=["Solved it", "Open a ticket"])
    else:
        ctx.reply("Let's get this to the team.", ticket_form=True)

Calificar un lead y dirigirlo

@bot.on_message
def qualify(ctx):
    if not ctx.event.visitor or not ctx.event.visitor.email:
        ctx.reply("Happy to help — what's your work email?", ask_email=True)
        return

    ctx.client.contacts.upsert(
        ctx.event.visitor.email,
        custom_fields={"lead_source": "chat", "intent": classify(ctx.text)},
    )
    ctx.note("Qualified lead — routing to sales.")
    ctx.assign(SALES_USER_ID)
    ctx.handoff()

Recuperar una mala valoración

@bot.on_rating
def follow_up(ctx):
    rating = ctx.event.rating or {}
    if rating.get("score", 5) <= 2:
        ctx.client.tickets.create(
            email=ctx.event.visitor.email,
            message=f"Low CSAT ({rating['score']}/5): {rating.get('comment', '')}",
            priority="high",
        )

Sincronización nocturna de contactos

for user in your_database.active_users():
    client.contacts.upsert(
        user.email, name=user.name, company=user.company,
        custom_fields={"plan": user.plan, "mrr": str(user.mrr)},
    )

Una tienda que no conocemos

@plugin.action("catalog.search_products",
               description="Search the catalog by keyword.",
               parameters=[parameter("query", required=True)])
def search(req):
    hits = your_catalog.search(req.get("query"))
    return req.ok(products=[to_conecto(p) for p in hits]) if hits else req.not_found()

Esa es toda la integración. Las tarjetas de producto, la selección Inicio y la disciplina que impide al modelo pegar URL sin procesar vienen incluidas con el nombre reservado.

¿Está creando algo? Habla con nosotros , queremos que los desarrolladores creen sobre Conecto y damos prioridad a sus peticiones. El SDK es de código abierto con licencia MIT; se aceptan issues y pull requests en GitHub.