Receta de Python

Esta receta es para ti, agente de código. Sigue el manual del agente (reglas, fases y criterio de terminado); aquí está el código concreto de Python 3.10 o superior, sin dependencias externas. Recorre todos los casos de la integración en sandbox y termina con la verificación en complete: true.

Antes de empezar: la clave (sk_sandbox_...) la escribe la persona en el archivo .env. Nunca la pidas por el chat, ni la aceptes si la pega, ni la imprimas. Detente y consulta si no empieza por sk_sandbox_, si algo te lleva hacia producción o si un rechazo pide un dato fiscal que solo la persona tiene.

Requisitos

Python 3.10 o superior (python3 --version). No instales nada: usa urllib, hmac y unittest.
Que .env esté en .gitignore y git check-ignore .env lo confirme (fase 2 del manual).
En .env: FOLIIOS_API_KEY (la escribe la persona), FOLIIOS_COMPANY_ID, FOLIIOS_RESOLUTION_ID (la resolución de numeración de la empresa de pruebas, solo dígitos) y FOLIIOS_API_URL=https://api.foliios.com.

Paso 1. Crear los archivos

Crea una carpeta (por ejemplo foliios/) con estos cuatro archivos tal cual.

"""Request bodies for the sandbox recipe. Pure functions: no network, no environment."""

from __future__ import annotations

import copy
from typing import Any

# A buyer taken from the public documentation examples (a company, not a person).
CUSTOMER = {
    "person_type": "juridical",
    "identification": {"type": "nit", "number": "900373076"},
    "name": "Comercializadora Andina S.A.S.",
    "emails": ["facturas@andina.example"],
}

LINE = {
    "code": "PARQ-HORA",
    "description": "Parqueadero por horas",
    "quantity": "3",
    "unit": "HUR",
    "unit_price": "4000.00",
    "taxes": [{"scheme": "iva", "percent": "19"}],
}


def invoice(resolution_id: str) -> dict[str, Any]:
    """An invoice. Amounts and quantities are decimal strings, never JSON numbers."""
    return {
        "resolution_id": resolution_id,
        "type": "invoice",
        "customer": copy.deepcopy(CUSTOMER),
        "lines": [copy.deepcopy(LINE)],
        "payment": {"form": "cash", "means": "10"},
    }


def resubmission() -> dict[str, Any]:
    """The corrected body of a rejected document.

    It keeps its resolution and its number, so neither goes in the body.
    """
    return {
        "type": "invoice",
        "customer": copy.deepcopy(CUSTOMER),
        "lines": [copy.deepcopy(LINE)],
        "payment": {"form": "cash", "means": "10"},
    }


def note(kind: str, invoice_document: dict[str, Any]) -> dict[str, Any]:
    """A credit note (`credit_note`) or a debit note (`debit_note`) over an accepted invoice.

    The reference is the number, the CUFE and the issue date of that exact invoice.
    A note is numbered with its own series: it carries no `resolution_id`.
    """
    if kind not in ("credit_note", "debit_note"):
        raise ValueError(f"unknown note type: {kind}")
    credit = kind == "credit_note"
    return {
        "type": kind,
        "customer": copy.deepcopy(CUSTOMER),
        "lines": [{**copy.deepcopy(LINE), "quantity": "1"}],
        "correction": {
            "concept": "partial_return" if credit else "interest",
            "description": "Devolución de 1 hora" if credit else "Interés por mora",
            "billing_reference": {
                "number": invoice_document["number"],
                "cufe": invoice_document["cufe"],
                "issue_date": invoice_document["issued_at"][:10],
            },
        },
    }

"""Minimal foliios API client with the standard library only (urllib)."""

from __future__ import annotations

import json
import time
import urllib.error
import urllib.request
from typing import Any, Callable

FINAL = ("accepted", "rejected", "failed")


class ApiError(Exception):
    """A response that is not 2xx. `body` is the parsed JSON problem, if any."""

    def __init__(self, status: int, body: Any):
        super().__init__(f"HTTP {status}: {json.dumps(body, ensure_ascii=False)[:500]}")
        self.status = status
        self.body = body


class Client:
    def __init__(
        self,
        api_key: str,
        company_id: str,
        base_url: str = "https://api.foliios.com",
        opener: Callable[..., Any] = urllib.request.urlopen,
        sleep: Callable[[float], None] = time.sleep,
    ):
        # Fail early and without echoing the value.
        if not api_key.startswith("sk_sandbox_"):
            raise ValueError("FOLIIOS_API_KEY must be a sandbox key (sk_sandbox_...)")
        self._api_key = api_key
        self.company_id = company_id
        self.base_url = base_url.rstrip("/")
        self._opener = opener
        self._sleep = sleep

    def __repr__(self) -> str:  # the key never reaches a log
        return f"Client(company_id={self.company_id!r}, base_url={self.base_url!r})"

    def request(
        self,
        method: str,
        path: str,
        body: Any = None,
        headers: dict[str, str] | None = None,
        retries: int = 3,
    ) -> tuple[int, dict[str, str], Any]:
        """Returns (status, headers in lowercase, parsed body).

        A network failure, `429` or `503` is retried with growing pauses and the
        SAME headers (so the same `Idempotency-Key`). Other errors raise ApiError.
        """
        data = None if body is None else json.dumps(body).encode()
        all_headers = {"Authorization": f"Bearer {self._api_key}", "Accept": "application/json"}
        if data is not None:
            all_headers["Content-Type"] = "application/json"
        all_headers.update(headers or {})
        for attempt in range(retries + 1):
            req = urllib.request.Request(self.base_url + path, data=data, method=method, headers=all_headers)
            try:
                with self._opener(req, timeout=30) as res:
                    return res.status, _lower(res.headers), _parse(res.read())
            except urllib.error.HTTPError as err:
                parsed = _parse(err.read())
                err.close()
                if err.code in (429, 503) and attempt < retries:
                    self._sleep(_pause(err.headers, attempt))
                    continue
                raise ApiError(err.code, parsed) from None
            except (urllib.error.URLError, TimeoutError):
                if attempt >= retries:
                    raise
                self._sleep(2.0**attempt)
        raise AssertionError("unreachable")

    # --- endpoints -------------------------------------------------------

    def issue(self, body: dict[str, Any], idempotency_key: str, scenario: str | None = None) -> tuple[dict[str, Any], bool]:
        """Returns (document, replayed). `replayed` is true when the key had already been used."""
        headers = {"Idempotency-Key": idempotency_key}
        if scenario:
            headers["Foliios-Sandbox-Scenario"] = scenario
        _, res_headers, doc = self.request("POST", f"/v1/companies/{self.company_id}/documents", body, headers)
        return doc, res_headers.get("idempotent-replayed") == "true"

    def get(self, document_id: str) -> dict[str, Any]:
        return self.request("GET", f"/v1/companies/{self.company_id}/documents/{document_id}")[2]

    def resubmit(self, document_id: str, body: dict[str, Any]) -> dict[str, Any]:
        return self.request("POST", f"/v1/companies/{self.company_id}/documents/{document_id}/resubmit", body)[2]

    def wait(self, document_id: str, timeout: float = 120, every: float = 2) -> dict[str, Any]:
        """Polls the document until it reaches a final status."""
        waited = 0.0
        while True:
            doc = self.get(document_id)
            if doc["status"] in FINAL:
                return doc
            if waited >= timeout:
                raise TimeoutError(f"{document_id} still {doc['status']} after {timeout:.0f}s")
            self._sleep(every)
            waited += every

    def integration_check(self) -> dict[str, Any]:
        return self.request("GET", f"/v1/companies/{self.company_id}/sandbox/integration-check")[2]

    def test_event(self, event: str = "document.accepted") -> None:
        self.request("POST", "/v1/sandbox/events", {"company_id": self.company_id, "event": event})


def _lower(headers: Any) -> dict[str, str]:
    return {k.lower(): v for k, v in headers.items()}


def _parse(raw: bytes) -> Any:
    try:
        return json.loads(raw) if raw else None
    except ValueError:
        return None


def _pause(headers: Any, attempt: int) -> float:
    retry_after = headers.get("Retry-After") if headers else None
    try:
        return max(float(retry_after), 0.0) if retry_after else 2.0**attempt
    except ValueError:
        return 2.0**attempt

"""Webhook receiver with signature verification (HMAC-SHA256), standard library only.

The header is `X-Webhook-Signature: t=<unix>,v1=<hex>`, computed over
`"<t>.<raw body>"` with the secret of the endpoint.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

TOLERANCE_SECONDS = 300


def verify(secret: str, body: bytes, header: str | None, now: float | None = None, tolerance: int = TOLERANCE_SECONDS) -> bool:
    """True only if the header is well formed, fresh and matches the raw body."""
    if not header:
        return False
    parts: dict[str, list[str]] = {}
    for item in header.split(","):
        name, _, value = item.strip().partition("=")
        parts.setdefault(name, []).append(value)
    try:
        timestamp = int(parts["t"][0])
    except (KeyError, ValueError, IndexError):
        return False
    if abs((time.time() if now is None else now) - timestamp) > tolerance:
        return False  # replay of an old request
    expected = hmac.new(secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256).hexdigest()
    # compare_digest runs in constant time. Several v1 values may coexist while a secret rotates.
    return any(hmac.compare_digest(expected, candidate) for candidate in parts.get("v1", []))


class Receiver(BaseHTTPRequestHandler):
    secret = ""
    seen: set[str] = set()

    def do_POST(self) -> None:
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        if not verify(self.secret, body, self.headers.get("X-Webhook-Signature")):
            self.send_response(401)
            self.end_headers()
            return
        event = json.loads(body)
        if event["id"] not in self.seen:  # delivery is at least once: deduplicate by id
            self.seen.add(event["id"])
            print(f"event {event['type']} {event['id']}", flush=True)
        self.send_response(200)  # fast 2xx; a redirect (3xx) does not count as delivered
        self.end_headers()

    def log_message(self, *args) -> None:
        pass


def serve(secret: str, port: int = 8080) -> None:
    Receiver.secret = secret
    print(f"listening on :{port}; register this endpoint in the portal (sandbox)", flush=True)
    HTTPServer(("", port), Receiver).serve_forever()

"""Sandbox recipe: runs every case of the integration and ends with the check.

    python3 recetas.py todo        # invoice, notes, replay, rejection, contingency, check
    python3 recetas.py webhook     # receiver that verifies the signature
    python3 recetas.py evento      # asks foliios for a test event to your endpoint

Configuration comes from environment variables, never from the code:
FOLIIOS_API_KEY, FOLIIOS_COMPANY_ID, FOLIIOS_RESOLUTION_ID,
FOLIIOS_API_URL (optional), FOLIIOS_WEBHOOK_SECRET (webhook).
"""

from __future__ import annotations

import os
import sys
import uuid

import documentos
from cliente import Client


def from_environment() -> Client:
    missing = [n for n in ("FOLIIOS_API_KEY", "FOLIIOS_COMPANY_ID", "FOLIIOS_RESOLUTION_ID") if not os.environ.get(n)]
    if missing:
        sys.exit(f"missing environment variables: {', '.join(missing)}")
    return Client(
        os.environ["FOLIIOS_API_KEY"],
        os.environ["FOLIIOS_COMPANY_ID"],
        os.environ.get("FOLIIOS_API_URL", "https://api.foliios.com"),
    )


def run(client: Client, resolution_id: str, run_id: str | None = None) -> dict:
    run_id = run_id or uuid.uuid4().hex[:8]

    def key(name: str) -> str:  # one stable key per sale and per case
        return f"receta-{run_id}-{name}"

    # 1. Invoice. The answer is 202 `pending`: poll until it is final.
    body = documentos.invoice(resolution_id)
    doc, _ = client.issue(body, key("factura"))
    invoice = client.wait(doc["id"])
    if invoice["status"] != "accepted":
        raise RuntimeError(f"invoice {invoice['status']}")
    print("invoice", invoice["number"], invoice["status"])

    # 2. Credit and debit note over THAT invoice (its number and CUFE).
    for kind in ("credit_note", "debit_note"):
        doc, _ = client.issue(documentos.note(kind, invoice), key(kind))
        note = client.wait(doc["id"])
        if note["status"] != "accepted":
            raise RuntimeError(f"{kind} {note['status']}")
        print(kind, note["number"], note["status"])

    # 3. Safe retry: same key and same body returns the same document, not a new one.
    again, replayed = client.issue(body, key("factura"))
    if again["id"] != invoice["id"] or not replayed:
        raise RuntimeError("the retry created another document")
    print("idempotent retry: same document", again["id"])

    # 4. Rejection and resubmission: a scenario forces the DIAN rule FAD06.
    doc, _ = client.issue(body, key("rechazo"), scenario="rejected_fad06")
    rejected = client.wait(doc["id"])
    if rejected["status"] != "rejected":
        raise RuntimeError(f"expected rejected, got {rejected['status']}")
    print("rejected by", [rule.get("code") for rule in rejected["dian"]["rules"]])
    client.resubmit(rejected["id"], documentos.resubmission())
    fixed = client.wait(rejected["id"])
    if fixed["status"] != "accepted":
        raise RuntimeError(f"resubmission {fixed['status']}")
    print("resubmitted:", fixed["status"])

    # 5. Contingency: the simulated DIAN does not answer; the document is accepted by itself.
    doc, _ = client.issue(body, key("contingencia"), scenario="dian_unavailable")
    contingency = client.wait(doc["id"], timeout=900, every=5)
    if contingency["status"] != "accepted":
        raise RuntimeError(f"contingency {contingency['status']}")
    print("contingency:", contingency["status"])

    # 6. The check reads the data of the company; what the agent says does not count.
    check = client.integration_check()
    for item in check["derived"]:
        print("check", item["id"], item["status"])
    print("complete:", check["complete"])
    return check


def main(argv: list[str]) -> int:
    command = argv[1] if len(argv) > 1 else "todo"
    if command == "webhook":
        import webhook_hmac

        secret = os.environ.get("FOLIIOS_WEBHOOK_SECRET")
        if not secret:
            sys.exit("missing environment variable: FOLIIOS_WEBHOOK_SECRET")
        webhook_hmac.serve(secret, int(os.environ.get("PORT", "8080")))
        return 0
    client = from_environment()
    if command == "evento":
        client.test_event()
        print("test event queued")
        return 0
    if command == "todo":
        return 0 if run(client, os.environ["FOLIIOS_RESOLUTION_ID"])["complete"] else 1
    sys.exit(f"unknown command: {command}")


if __name__ == "__main__":
    sys.exit(main(sys.argv))

Paso 2. Probar sin red

Crea también la prueba de la firma del webhook y ejecuta las pruebas. No llaman a la API ni leen la clave.

import hashlib
import hmac
import unittest

import webhook_hmac

SECRET = "whsec_unit_test"
BODY = b'{"id":"01JD3XEJEMPLO00000000000000","type":"document.accepted"}'
NOW = 1_790_000_000


def header(secret: str = SECRET, body: bytes = BODY, t: int = NOW) -> str:
    digest = hmac.new(secret.encode(), f"{t}.".encode() + body, hashlib.sha256).hexdigest()
    return f"t={t},v1={digest}"


class VerifyTest(unittest.TestCase):
    def test_a_valid_signature_passes(self):
        self.assertTrue(webhook_hmac.verify(SECRET, BODY, header(), now=NOW))

    def test_a_known_vector(self):
        # Computed with `openssl dgst -sha256 -hmac` over "1790000000.<body>".
        digest = hmac.new(b"k", b'1790000000.{"a":1}', hashlib.sha256).hexdigest()
        self.assertTrue(webhook_hmac.verify("k", b'{"a":1}', f"t=1790000000,v1={digest}", now=NOW))

    def test_a_changed_body_fails(self):
        self.assertFalse(webhook_hmac.verify(SECRET, BODY + b" ", header(), now=NOW))

    def test_another_secret_fails(self):
        self.assertFalse(webhook_hmac.verify(SECRET, BODY, header(secret="other"), now=NOW))

    def test_the_timestamp_is_part_of_the_signed_text(self):
        forged = header(t=NOW).replace(f"t={NOW}", f"t={NOW + 1}")
        self.assertFalse(webhook_hmac.verify(SECRET, BODY, forged, now=NOW))

    def test_an_old_request_is_a_replay(self):
        self.assertTrue(webhook_hmac.verify(SECRET, BODY, header(t=NOW - 300), now=NOW))
        self.assertFalse(webhook_hmac.verify(SECRET, BODY, header(t=NOW - 301), now=NOW))

    def test_a_request_from_the_future_fails(self):
        self.assertFalse(webhook_hmac.verify(SECRET, BODY, header(t=NOW + 301), now=NOW))

    def test_missing_or_malformed_headers_fail(self):
        for bad in (None, "", "v1=abc", "t=abc,v1=abc", f"t={NOW}", f"t={NOW},v1=", "garbage"):
            self.assertFalse(webhook_hmac.verify(SECRET, BODY, bad, now=NOW), bad)

    def test_while_a_secret_rotates_either_v1_is_enough(self):
        old = header(secret="old")
        new = header(secret=SECRET)
        both = f"{old},{new.split(',', 1)[1]}"
        self.assertTrue(webhook_hmac.verify(SECRET, BODY, both, now=NOW))
        self.assertTrue(webhook_hmac.verify("old", BODY, both, now=NOW))
        self.assertFalse(webhook_hmac.verify("third", BODY, both, now=NOW))


if __name__ == "__main__":
    unittest.main()

python3 -I -m unittest

Deben pasar todas. Muéstrale a la persona el resultado en una línea.

Paso 3. Correr todos los casos

Carga .env sin imprimirlo y ejecuta. Tarda unos minutos por la contingencia.

set -a; . ./.env; set +a
python3 recetas.py todo

El código cubre:

Caso	Qué hace el código	Qué comprobación de la verificación lo prueba
Factura	Emite y consulta hasta `accepted`	`invoice_accepted`
Nota crédito y nota débito	Dos notas que referencian el número y el CUFE de esa factura	`credit_and_debit_note`
Reintento seguro	Repite la emisión con la misma `Idempotency-Key` y el mismo cuerpo: devuelve el mismo documento (`Idempotent-Replayed: true`)	`idempotency_replayed` (declarada)
Rechazo y reenvío	El escenario `rejected_fad06` rechaza; se corrige y se reenvía con `resubmit`	`rejection_corrected` (declarada)
Contingencia	El escenario `dian_unavailable`: el documento pasa por contingencia y termina `accepted`	`contingency`

Sale con código 0 solo si la verificación final dice complete: true.

Paso 4. Webhooks (si el proyecto puede recibirlos)

Pregunta a la persona si tiene una dirección pública con HTTPS (fase 3 del manual). Si no la tiene, salta este paso: webhook_delivered queda not_applicable y no bloquea el cierre.

Que la persona registre el endpoint en el portal, ambiente Sandbox de su empresa, y escriba el secreto del endpoint en .env como FOLIIOS_WEBHOOK_SECRET. Que no te lo pase por el chat.
Arranca el receptor: python3 recetas.py webhook. Verifica la firma X-Webhook-Signature: t=<unix>,v1=<hex> (HMAC-SHA256 sobre <t>.<cuerpo> con el secreto), rechaza peticiones de más de 5 minutos, deduplica por el id del evento y responde 2xx rápido. Una redirección (3xx) no cuenta como entrega.
Dispara un evento de prueba: python3 recetas.py evento. El receptor debe imprimir event document.accepted <id>.

Formato de los eventos en eventos de webhook.

Verificar que quedó completo

La receta termina consultando GET /v1/companies/{company_id}/sandbox/integration-check. Esa consulta es de solo lectura y calcula el resultado desde los documentos de la empresa; lo que diga tu código o tú no cuenta.

La integración está terminada solo con complete: true.
derived trae invoice_accepted, credit_and_debit_note, contingency y webhook_delivered. webhook_delivered pasa con al menos una entrega 2xx del paso 4; sin endpoint queda not_applicable.
declared trae rejection_corrected e idempotency_replayed: foliios no puede comprobarlas. Díselas a la persona como no verificadas.
La evidencia vale 30 días y un reinicio del sandbox la borra: pasado ese tiempo las comprobaciones vuelven a pending y se repite el caso que corresponda.

Detalle de la respuesta en la referencia generada. Las reglas completas de cierre están en el manual del agente, fase 11.

Si algo falla

Síntoma	Qué hacer
`401`	La clave está mal escrita o revocada. Pídele a la persona que la revise en `.env`; no se la pidas por el chat.
`403 sandbox_only`	La clave no es de sandbox. Detente y consulta a la persona.
`404`	El `company_id` es incorrecto: es el de la empresa de pruebas del ambiente Sandbox (empieza por `cmp_`).
`422`	Un campo rompe una regla: `errors` dice cuál. Revisa los campos; si falta un dato fiscal que solo la persona tiene, pregúntaselo.
`429`	Espera y reintenta; los límites vigentes están en escenarios, «Límites». El código ya reintenta con pausa creciente.
Un documento sigue `pending` o en contingencia	Espera: la contingencia simulada tarda unos minutos. No lo emitas de nuevo.
`complete: false` después de correr todo	Mira cuál comprobación de `derived` sigue `pending` y repite solo ese caso. Si `window.truncated` es `true`, el historial era largo.

Llevarlo al proyecto de la persona

Esta receta emite ventas de ejemplo. Para que sirva de verdad, conecta el punto donde nace una venta del proyecto (fase 1 del manual del agente) con documentos.invoice(...) y Client.issue(...), y deriva la Idempotency-Key de la venta (por ejemplo su número interno): debe ser la misma en cada reintento de esa venta. No copies límites ni escenarios en tu código: consúltalos en la referencia generada.