Ando Platform — Python SDK

Client Python sincrono per la Ando Platform API v1 (https://api.ando-ai.com/platform/v1). Unica dipendenza: httpx>=0.24.

Endpoint predefinito: https://api.ando-ai.com/platform/v1.

Requisiti di accesso

Questo pacchetto contiene solo il client SDK: non include né rende autonomamente disponibile il servizio Ando Platform. Per usarlo servono un account Ando attivo (o un accesso di valutazione), i relativi diritti di licenza e una API key emessa da Ando. Per richiedere l'attivazione, contatta Ando su ando-ai.com.

Installazione

Requisito: Python 3.9+.

pip install ando-ai

Uso

from ando_ai import AndoPlatform, ApiError, JobFailed

with AndoPlatform("ando_sk_live_...") as ando:
    accepted = ando.upload_document("contratto.pdf")
    ando.wait_for_job(accepted["job_id"])

    result = ando.answer("Qual è la durata del contratto?")
    print(result["answer"], result["citations"])

    ando.delete_document(accepted["document_id"])

Il client è un context manager; in alternativa chiama ando.close() a fine lavoro.

Collega gli account degli utenti della tua app

Registra la tua app e il suo URL HTTPS di ritorno nella Console. Crea una AppKey app_user_server con sources:manage e query:read, vincolata a quella app. Tieni l'AppKey solo sul tuo server. Il tuo backend, dopo aver autenticato l'utente con il login della tua app, passa il suo ID stabile (app_user_id) a ogni chiamata. Ando lo registra internamente senza creare un membro Workspace. La stessa organizzazione può usare anche Workspace: i grant restano separati.

with AndoPlatform("ando_sk_live_...") as ando:
    session = ando.start_end_user_connection(
        "outlook", "https://app.example.com/connected",
        app_user_id=str(current_user.id),
    )
    # Il browser dell'utente apre session["authorization_url"].
    status = ando.get_end_user_connection_session(
        session["session_id"], app_user_id=str(current_user.id),
    )
    # Nel callback della tua app, verifica la sessione login del browser.
    # ando_session e ando_proof arrivano dall'URL di ritorno; non registrarli.
    confirmed = ando.confirm_end_user_connection(
        ando_session, ando_proof, app_user_id=str(current_user.id),
    )
    connections = ando.list_end_user_connections(
        app_user_id=str(current_user.id),
    )
    result = ando.test_end_user_connection(
        connections["connections"][0]["connection_id"],
        app_user_id=str(current_user.id),
    )
    connection_id = connections["connections"][0]["connection_id"]
    cursor = None
    while True:
        page = ando.sync_end_user_connection(
            connection_id, app_user_id=str(current_user.id),
            page_token=cursor,
        )
        cursor = page["next_page_token"]
        if cursor is None:
            break
    hits = ando.query_end_user(
        "Quali documenti parlano del progetto?",
        app_user_id=str(current_user.id),
    )

Provider personali: gmail, google_drive, outlook, onedrive. Teams (teams) usa un account Microsoft aziendale e chiede separatamente il solo permesso di lettura delle chat. Non richiede i permessi di scrittura, canali o trascrizioni del connettore Workspace. Il callback conferma il grant solo se il backend della tua app associa la sessione browser allo stesso ID utente che ha avviato il collegamento; la prova scade dopo dieci minuti. preview_end_user_connection restituisce al massimo cinque metadati live. sync_end_user_connection indicizza una pagina per chiamata nello store TRAMA; il cursore restituito permette di continuare senza caricare tutta la fonte in una richiesta. query_end_user cerca solo i chunk dell'utente dell'app con provenienza e credenziale verificate. La route generale query() continua a servire il percorso legacy con membri Ando.

La sincronizzazione automatica usa gli stessi account e il medesimo flusso di ingestion. Dopo la conferma puoi leggere get_end_user_sync_status e, per email o Teams, scegliere la finestra con get_end_user_sync_settings e set_end_user_sync_settings (7d, 30d, 90d oppure custom con date YYYY-MM-DD, massimo 90 giorni). Per Drive, usa browse_end_user_items e set_end_user_drive_selection per scegliere cartelle o file; per Teams usa browse_end_user_items e set_end_user_chat_selection per scegliere le chat. get_end_user_selection mostra la selezione attuale. control_end_user_drive_sync permette pause, resume e force. La pagina manuale sync_end_user_connection accetta da 1 a 100 elementi; browse_end_user_items accetta da 1 a 100 elementi per pagina. Queste chiamate richiedono sources:manage; query_end_user richiede query:read. Tutte ricevono l'identità dello stesso utente in ogni chiamata.

Se la tua app preferisce che Ando verifichi direttamente i token OIDC degli utenti, configura issuer, audience e JWKS nella sezione avanzata e crea una chiave end_user_only. Le stesse chiamate SDK accettano allora end_user_token=... al posto di app_user_id=...; i due parametri non possono essere usati insieme. Il token viene inviato solo nella chiamata che lo specifica e non viene salvato sul client SDK.

Cosa puoi costruire — i tre livelli

Lo stesso stack, esposto a tre profondità crescenti. Non sono alternative: sono quanto lavoro fa il server al posto tuo.

1. query() — recuperi i chunk, generi tu

Per chi ha già il proprio LLM/prompt e vuole solo il retrieval (con citazioni e score).

hits = ando.query("penali per ritardata consegna", k=8)
for hit in hits["results"]:
    print(hit["score"], hit["section"], hit["text"][:120])
    # hit["document_id"] / hit["chunk_id"] = la citazione da mostrare

2. answer() — risposta sintetizzata e ancorata

Il server recupera e sintetizza, citando i chunk usati. La risposta rispecchia la lingua della domanda (non si forza nulla). Se il retrieval non trova nulla: answer è None e no_results è True.

result = ando.answer("Qual è la durata del contratto?", verify=True)
print(result["answer"])
print(result["citations"])            # [{document_id, chunk_id}, ...]
print(result["verification"])         # {verified, score, notes} | None

3. agentic_answer() — l'agente completo, anche sui dati strutturati (PREVIEW)

Questa è la differenza tra «ti do dei chunk» e «ho interrogato il tuo gestionale». Gira la stessa macchina della chat prodotto: escalation TRAMA T1→T2→T3 e, sulle sources strutturate pubblicate del project, text-to-SQL — quindi una domanda la cui risposta sta in una tabella viene risposta dalla tabella, non approssimata dai documenti.

Read-only per costruzione: nessuna scrittura, nessuna azione, nessuna persistenza. steps[] mostra cosa ha fatto l'agente (tier di retrieval, SQL eseguito con credenziali redatte, specialist scelto).

result = ando.agentic_answer(
    "Quali clienti hanno esposizione più critica e per quale importo?",
    k=25,
    with_diagnostics=True,       # tier, timing, plan cache
)
print(result["answer"])
print(result["sources_used"])    # es. ["database", "fs"]
for step in result["steps"]:
    print(step["type"], step)    # tipi sconosciuti = additivi, non ramificarci sopra

Una chiamata agentica è multi-step: il timeout di default è AGENTIC_TIMEOUT (120s), non i 30s del client. Si sovrascrive per chiamata con timeout=.

PREVIEW: POST /agentic/answer non è ancora GA — request e response possono cambiare. Vedi docs/PLATFORM_API_CONTRACT.md.

Conversazioni — memoria tra turni agentici

Un turno agentico è stateless di default. Apri un thread con create_conversation() e passa il suo id a agentic_answer(): i follow-up si risolvono contro la storia del thread («e il secondo?» funziona) e ogni scambio viene persistito per il turno successivo. Scope query:read.

conv = ando.create_conversation(title="Analisi esposizione")
ando.agentic_answer("Chi sono i primi 3 clienti per esposizione?",
                    conversation_id=conv["conversation_id"])
ando.agentic_answer("E il secondo?",                       # follow-up risolto
                    conversation_id=conv["conversation_id"])

ando.list_conversations()                        # thread della chiave, cursor-paginati
ando.get_conversation_turns(conv["conversation_id"])   # scambi, dal più vecchio
ando.delete_conversation(conv["conversation_id"])      # irreversibile

Restringere il campo: filters, sources, partition

Valgono su query() e answer(). Ogni filtro può solo restringere: la resource policy della chiave è sempre intersecata sopra e vince.

# scorciatoie da keyword (le due dimensioni più usate)
ando.answer("fatture scadute", sources=["fs"], partition="cliente-42")

# oggetto completo, con condizioni sui metadata di ingest
ando.query(
    "resi 2025",
    filters={
        "sources": ["fs", "drive"],
        "partition": "cliente-42",
        "metadata": {
            "op": "and",
            "conditions": [
                {"key": "doc_type", "op": "eq", "value": "invoice"},
                {"key": "year", "op": "gte", "value": 2025},
            ],
        },
    },
)

partition e metadata sono esattamente quelli impostati all'ingest — il cerchio si chiude sullo stesso client:

ando.upload_document(
    "fattura-2025-11.pdf",
    metadata={"doc_type": "invoice", "year": 2025},   # oggetto PIATTO, max 32 chiavi
    partition="cliente-42",                            # A-Za-z0-9._:- , max 128 char
)

Passare la stessa dimensione due volte (keyword e dentro filters) è un ValueError lato client, prima di qualunque HTTP: sceglierne una in silenzio cambierebbe quali documenti vengono cercati. I nomi di source sconosciuti non vengono rifiutati dal SDK: i valori enum sono additivi per contratto, decide il server.

Generazione documenti — generate()

Dai tuoi dati a un file docx/pptx/xlsx/pdf/md. Il contesto arriva da una query (retrieval TRAMA, resource policy rispettata) e/o da document_ids espliciti; almeno uno dei due è obbligatorio (altrimenti ValueError prima di qualunque HTTP). È sempre asincrono: 202 con un generation_id da pollare, poi si scarica il file. Il contenuto rispecchia la lingua di instructions/query. Scope unico per tutto: generate:write.

accepted = ando.generate(
    "docx",
    "Prepara un report sul fatturato 2025 con sintesi e dettaglio.",
    query="fatturato 2025",
)
gen = ando.wait_for_generation(accepted["generation_id"])   # -> completed | GenerationFailed
artifact = ando.download_generation(gen["generation_id"])   # DownloadedFile
artifact.save("report.docx")                                # bytes su disco

wait_for_generation condivide identica policy di retry/backoff di wait_for_job (transient rate_limited/5xx/rete ritentati entro la deadline); solleva GenerationFailed (con error_code) su failed, GenerationTimeout allo scadere.

CAD — analisi, DFM, conversione

Upload STEP/DXF asincroni (202 con job_id + result_id). Analisi e DFM tornano un report JSON, la conversione un file STEP. Scope cad:read per analyze/dfm-check/similar/jobs/results, cad:write per convert. Un job CAD è un job: wait_for_cad_job riusa JobFailed/JobTimeout.

job = ando.cad_analyze("bracket.step")                      # {result_id, job_id, kind: "analysis"}
ando.wait_for_cad_job(job["job_id"])
report = ando.get_cad_result(job["result_id"]).json()       # report JSON

conv = ando.cad_convert("profilo.dxf", "extrude", params={"height_mm": 10})
ando.wait_for_cad_job(conv["job_id"])
ando.get_cad_result(conv["result_id"]).save("solido.step")  # STEP su disco

cad_find_similar() è la ricerca per somiglianza query-by-upload (live): carichi uno STEP/IGES/STL, un disegno PDF o un'immagine; viene embeddato al volo e confrontato con il corpus CAD ingerito del project. Il file di query non viene mai salvato. La chiamata è sincrona — uno STEP grande può richiedere ~10-30s di tessellazione, dimensiona il timeout di conseguenza.

simili = ando.cad_find_similar("staffa.step", top_k=5)
for hit in simili["results"]:
    print(hit["filename"], hit["similarity"], hit["description"])

Azioni — il write-loop

La metà write simmetrica della Push API: chiedi all'app del project di fare qualcosa (aprire un ticket, spostare un ordine) e ne verifichi l'esito. declare (una tantum, sources:manage) → propose → execute (actions:write) → l'app fa complete. Le letture usano actions:read.

actions = ando.actions()

# dichiarazione una tantum (chiave sources:manage)
actions.declare(
    "acme-desk",
    "create_ticket",
    params_schema={
        "type": "object",
        "properties": {"subject": {"type": "string"}},
        "required": ["subject"],
    },
    risk="medium",
)

# per richiesta (chiave actions:write) — propose + execute in una chiamata
dispatched = actions.run(
    "acme-desk",
    "create_ticket",
    {"subject": "Refund request"},
    idempotency_key="refund-4711",   # un run ritentato replay-a la proposta
)
print(dispatched["id"], dispatched["status"])   # ... "dispatched"

Le azioni high/critical richiedono confirm_risk=True su run/execute. propose_action accetta un idempotency_key opzionale (header Idempotency-Key): una propose ritentata con la stessa chiave replay-a il primo 201 invece di duplicare. Ogni metodo esiste anche come funzione di modulo (propose_action(client, ...)) e come delega sul client (ando.propose_action(...)).

Capability con binding HTTP — dichiara, poi invoke

Dichiara la capability con un binding HTTP e Ando può chiamare direttamente il tuo sistema: invoke_connector_action è il fratello sincrono di propose + execute — un round trip, esito nella risposta.

# dichiarazione CON binding (chiave sources:manage)
actions.declare(
    "acme-desk",
    "create_ticket",
    params_schema={
        "type": "object",
        "properties": {"subject": {"type": "string"}},
        "required": ["subject"],
    },
    risk="medium",
    binding={
        "method": "POST",
        "path": "/tickets",                        # relativo, sul TUO sistema
        "body_template": {"title": "{{subject}}"}, # placeholder {{param}}
        "response": {"id_field": "id"},            # dove vive l'id creato
        "timeout_seconds": 15,
    },
)

# invoke sincrono (chiave actions:write)
esito = ando.invoke_connector_action(
    "acme-desk", "create_ticket", {"subject": "Refund request"}
)
print(esito["ok"], esito["status_code"], esito["response"])

I params sono validati contro il params_schema dichiarato e vale lo stesso risk gate di execute: una capability high/critical viene rifiutata (invalid_request) senza confirm_risk=True. Le chiavi di test validano e gate-ano per davvero ma non chiamano mai l'endpoint bound — la risposta porta test: true. La dichiarazione è una sostituzione completa: ri-dichiarare senza binding cancella quello memorizzato.

Flows — orchestrazione come API

Costruisci, esegui e supervisiona gli stessi flow del builder prodotto, white-label. Scope: flows:read (list/palette/stats/inbox/run detail) e flows:manage (create/update/delete/run/approve/reject).

palette = ando.flows_palette()          # ogni node type + campi di config
flows = ando.list_flows(page=1, page_size=20)
draft = ando.generate_flow(
    "Quando arriva una fattura, valida i dati e chiedi approvazione"
)  # copilot: NL -> draft flow (costo LLM, metered)

created = ando.create_flow(draft["flow"])
ando.run_flow(created["flow_id"], dry_run=True)
ando.flow_stats(created["flow_id"])
ando.update_flow(created["flow_id"], {"is_active": False})

# run manuale seminato con un PDF caricato: il suo content_item_id finisce
# nel trigger_data del run per i nodi che bindano
# {"$ref": "trigger.content_item_id"} (es. ai.cad_analyze)
run = ando.run_flow_with_upload(created["flow_id"], "fattura.pdf")

# polling di un run: stato, trigger, context accumulato, output per nodo
detail = ando.get_flow_run(run["run_id"])
print(detail["status"], detail["node_runs"])

# HITL: gli step `logic.approval` parcheggiano il run in `pending_approval`
inbox = ando.flows_pending_approvals()
ando.approve_flow_run("run_...", edited_value="testo corretto")
ando.approve_flow_run(
    "run_...",
    # correzione strutturata per-campo dello STESSO output editabile — se
    # arrivano entrambi, edited_fields vince sulle chiavi sovrapposte
    edited_fields={"to": "acme@example.com", "subject": "Ordine 42"},
)
ando.reject_flow_run("run_...", reason="importo errato")

run_flow_with_upload accetta solo PDF (invalid_request su tutto il resto). flow_templates() elenca la galleria di template; delete_flow(flow_id) rimuove un flow.

Usage — consumi e costi del project

usage() (usage:read) ritorna chiamate/token/costo aggregati del project — mese corrente di default, group_by="day" per la serie di fatturazione o "purpose" per capire cosa genera costo:

report = ando.usage(from_="2026-08-01", to="2026-08-07")
print(report["total_cost_usd"], report["usage"])

Sources — cosa può interrogare questa chiave

Prima di eseguire una /query (come fa la UI prodotto lasciando scegliere un connettore) bisogna scoprire cosa la chiave può interrogare. list_sources() lo dice, già intersecato con la resource policy della chiave.

for s in ando.list_sources()["sources"]:
    print(s["id"], s["kind"], s["queryable"], s.get("reason"))
    # id unstrutturati (fs/drive/…) -> filters.sources di /query
    # id strutturati (conn_<uuid>)  -> sources di /agentic/answer

Registra un database read-only del cliente (es. il back office SQL di un gestionale) come source strutturata — le credenziali sono cifrate at rest e mai ritornate; la risposta è white-label (capability, mai il vendor):

import time

source = ando.register_database_source(
    name="Gestionale produzione",
    connector_type="mysql",   # postgresql | mysql | sqlserver | oracle | ...
    credentials={"host": "db.example.com", "user": "ando_ro",
                 "password": "…", "database": "mexal"},
    config={"schema_filter": ["mexal"], "max_rows_per_query": 500},
)

# Excel/CSV analitici: ogni foglio diventa una tabella tipata nella stessa
# pipeline strutturata usata dalla Console (non un documento `fs`).
sheet = ando.upload_spreadsheet_source(
    "ordini.xlsx",
    name="Ordini 2026",
    sheet_names=["Ordini", "Clienti"],  # opzionale; ometti per tutti i fogli
)
ando.test_source(source["source_id"])   # {ok, error?} — probe, non un ApiError
ando.discover_source_schema(source["source_id"])
while True:
    profiling = ando.get_source_profiling_status(source["source_id"])
    if profiling["status"] == "complete":
        break
    time.sleep(2)

Per una source strutturata lo schema profilato si cura e poi si pubblica (il gate di queryabilità): finché non è pubblicato, queryable=false, reason="schema non pubblicato". Curare = sources:manage, leggere lo schema = query:read.

ando.get_source_schema("conn_…")                 # stato + objects (K-Schema)
ando.curate_source_schema(                        # override della proposta AI
    "conn_…", "public.orders",
    description="Ordini di vendita",
    column_descriptions={"total": "Importo totale, IVA inclusa"},
)
ando.publish_source_schema("conn_…")              # apre /query + /agentic

Webhooks

Endpoint HTTPS in uscita, firmati Standard Webhooks. Scope keys:manage. Il secret (whsec_…) è mostrato una sola volta alla creazione.

hook = ando.create_webhook("https://example.com/hooks", ["ingestion.completed"])
print(hook["secret"])        # salvalo ora — non torna più
ando.list_webhooks()         # mai il secret; cursor/limit opzionali
ando.test_webhook(hook["webhook_id"])            # test.ping sincrono

# storico deliveries + replay manuale (firma fresca, tentativo immediato)
deliveries = ando.list_webhook_deliveries(hook["webhook_id"], limit=100)
ando.replay_webhook_delivery(deliveries["deliveries"][0]["delivery_id"])

ando.delete_webhook(hook["webhook_id"])

Riferimento

Costruttore:

AndoPlatform(
    api_key,                                        # ando_sk_live_... / ando_sk_test_...
    base_url="https://api.ando-ai.com/platform/v1",
    timeout=30,                                     # secondi, per richiesta
)

Tutti i metodi ritornano il body JSON della risposta come dict, tranne download_generation / get_cad_result che ritornano un DownloadedFile (content bytes, content_type, filename, più .json() e .save(path)).

Costanti esportate: DEFAULT_K (10), MAX_K (50), FILTER_SOURCES, GENERATE_FORMATS, WEBHOOK_EVENTS, ACTION_RISK_LEVELS (tutte informative: i valori enum sono additivi, il SDK non li valida), AGENTIC_TIMEOUT (120s), DEFAULT_BATCH_SIZE (200), MAX_BATCH_SIZE (500), DEFAULT_BASE_URL. Tipi esportati: ActionBinding (il binding HTTP di una capability — un normale dict con quelle chiavi va bene ovunque).

CLI (ando)

Il pacchetto espone lo script ando (o python -m ando_ai):

ando kinds                                   # kind disponibili + contratto Record
ando validate records.jsonl                  # dry-run dei record (nessuna scrittura)
cat records.jsonl | ando validate -          # …da stdin (JSON array o JSONL)
ando sync acme-crm records.jsonl --mode full # push di un batch
ando sources                                 # cosa può interrogare questa chiave
ando me                                      # identità della chiave
ando mcp --client cursor                     # config MCP per il tuo agente

Auth: --api-key o $ANDO_API_KEY; base URL: --base-url o $ANDO_BASE_URL. ando validate consuma POST /connectors/validate e esce con codice 1 se un record è rifiutato — la CI di un connettore può fare da gate sul contratto. ando mcp stampa l'install one-command del server MCP hosted (Claude Code / Cursor / Claude Desktop); senza --client le stampa tutte.

Errori

Gerarchia (base comune AndoPlatformError):

• ApiError — errore dell'envelope {"error": {"code", "message"}}. Attributi: code (stabile: invalid_api_key, expired_api_key, revoked_api_key, insufficient_scope, rate_limited, invalid_request, not_found, internal_error), message, status e retry_after (secondi dall'header Retry-After, valorizzato sui 429).
• NetworkError — problema di trasporto (DNS, connessione, TLS, timeout); l'eccezione httpx originale è in __cause__. Su agentic_answer un read timeout arriva qui: di solito significa «alza timeout=», non «il server è rotto».
• JobFailed — job di ingestion (o CAD) terminato failed; porta error_code, error_message e il payload completo in job.
• JobTimeout — wait_for_job / wait_for_cad_job ha esaurito il timeout senza uno stato terminale.
• GenerationFailed — generazione terminata failed; porta error_code, error_message e il payload in generation.
• GenerationTimeout — wait_for_generation ha esaurito il timeout.

from ando_ai import AndoPlatform, ApiError

with AndoPlatform("ando_sk_live_...") as ando:
    try:
        ando.query("scadenze")
    except ApiError as exc:
        if exc.code == "rate_limited":
            print(f"Riprova tra {exc.retry_after}s")
        elif exc.code == "insufficient_scope":
            print(f"Scope mancante: {exc.message}")
        else:
            raise

Test

I test del SDK girano senza rete né backend (httpx.MockTransport):

uv run pytest tests/unit/test_platform_sdk_client.py

License

Proprietary — Copyright (c) 2026 Ando S.r.l. This SDK is licensed for use solely with the Ando Platform services; it is not open source. See the LICENSE file shipped in this package for the full terms.