Integración en Python

python3 -m venv .venv
source .venv/bin/activate
pip install requests
export ALCHEMY_API_KEY="ck_live_7f3a91c2m4p8q1w7x5n3v9b6t0"
export FIEL_PASSWORD="********"

"""Cliente mínimo de AlchemyCode API con requests.

Requiere Python 3.10+:
    python -m venv .venv && source .venv/bin/activate
    pip install requests
"""

from __future__ import annotations

import os
import time
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE = "https://api.434alchemycode.com/v1"


class AlchemyError(RuntimeError):
    def __init__(self, status: int, problem: dict[str, Any]) -> None:
        self.status = status
        self.code = problem.get("code", "UNKNOWN")
        self.retryable = bool(problem.get("retryable"))
        self.request_id = problem.get("request_id")
        super().__init__(f"[{self.code}] {problem.get('detail') or problem.get('title')}")


class AlchemyClient:
    def __init__(self, api_key: str, base_url: str = BASE) -> None:
        if not api_key:
            raise ValueError("Falta la API key de AlchemyCode.")
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
                "User-Agent": "OmnicomERP/2.4 (python-requests)",
            }
        )
        # Reintenta 429/5xx respetando Retry-After. Sólo métodos seguros o
        # POST con Idempotency-Key (el reintento no debe duplicar jobs).
        retry = Retry(
            total=3,
            backoff_factor=1.0,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset({"GET"}),
            respect_retry_after_header=True,
            raise_on_status=False,
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("https://", adapter)

    def request(self, method: str, path: str, **kwargs: Any) -> dict:
        response = self.session.request(
            method, self.base_url + path, timeout=(10, 45), **kwargs
        )
        if response.status_code >= 400:
            try:
                problem = response.json()
            except ValueError:
                problem = {"code": "UNKNOWN", "detail": response.text[:200]}
            raise AlchemyError(response.status_code, problem)
        return response.json()

    def upload_fiel(self, company_id: str, cer_path: str, key_path: str, password: str) -> dict:
        with open(cer_path, "rb") as cer, open(key_path, "rb") as key:
            return self.request(
                "POST",
                f"/companies/{company_id}/credentials/sat",
                files={
                    "cer": ("omnicom.cer", cer, "application/x-x509-ca-cert"),
                    "key": ("omnicom.key", key, "application/octet-stream"),
                },
                data={"password": password},
            )

    def wait_for_job(self, job_id: str, timeout: float = 180.0) -> dict:
        deadline = time.monotonic() + timeout
        interval = 2.0
        while time.monotonic() < deadline:
            job = self.request("GET", f"/jobs/{job_id}")
            if job["status"] not in ("queued", "running"):
                return job
            time.sleep(interval)
            interval = min(interval + 1, 5)
        raise TimeoutError(f"El job {job_id} no terminó en {timeout} s")

import os
from pathlib import Path

import requests

from alchemy_client import AlchemyClient, AlchemyError

client = AlchemyClient(os.environ["ALCHEMY_API_KEY"])

try:
    try:
        company = client.request(
            "POST",
            "/companies",
            json={
                "rfc": "OMX2503255F9",
                "legal_name": "OMNICONMX, S.A. de C.V.",
                "tax_regime": "601",
                "external_ref": "OMNICOM-001",
            },
        )
        company_id = company["id"]
    except AlchemyError as exc:
        if exc.code != "COMPANY_DUPLICATE":
            raise
        found = client.request("GET", "/companies", params={"q": "OMX2503255F9"})
        company_id = found["data"][0]["id"]

    credentials = client.request("GET", f"/companies/{company_id}/credentials")
    if not credentials["fiel"]["configured"]:
        fiel = client.upload_fiel(
            company_id, "omnicom.cer", "omnicom.key", os.environ["FIEL_PASSWORD"]
        )
        print(f"e.firma {fiel['status']}, vence {fiel['valid_to']}")

    job = client.request(
        "POST",
        "/sat/constancias",
        json={"company_id": company_id, "include_pdf": True, "wait": False},
        headers={"Idempotency-Key": "6f1c2c7a-3d0e-4a51-9b0e-5c2a7f9d1e44"},
    )
    print("Job", job["id"], job["status"])

    finished = client.wait_for_job(job["id"])
    if finished["status"] != "succeeded":
        raise SystemExit(f"Job {finished['status']}: {finished['error']}")

    print(finished["result"]["rfc"], finished["result"]["regimen_fiscal"]["descripcion"])

    for file in finished["files"]:
        response = requests.get(f"https://api.434alchemycode.com{file['url']}", timeout=60)
        response.raise_for_status()
        Path(file["name"]).write_bytes(response.content)
        print(file["name"], len(response.content), "bytes")

except AlchemyError as exc:
    print(f"{exc} (HTTP {exc.status}, request_id {exc.request_id})")
    raise SystemExit(2)

import time

from alchemy_client import AlchemyError

MAX_ATTEMPTS = 4

for attempt in range(1, MAX_ATTEMPTS + 1):
    try:
        job = client.request(
            "POST",
            "/sat/opiniones",
            json={"company_id": company_id},
            headers={"Idempotency-Key": f"opinion-{company_id}-2026-10"},
        )
        break
    except AlchemyError as exc:
        if not exc.retryable or attempt == MAX_ATTEMPTS:
            raise
        # backoff exponencial con jitter
        time.sleep((2 ** (attempt - 1)) + 0.3)

import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()
SECRET = os.environ["ALCHEMY_WEBHOOK_SECRET"]


@app.post("/webhooks/alchemy")
async def alchemy_webhook(request: Request) -> dict:
    raw = await request.body()
    header = request.headers.get("x-alchemy-signature", "")

    parts = dict(part.split("=", 1) for part in header.split(",") if "=" in part)
    timestamp = int(parts.get("t", "0"))
    signature = parts.get("v1", "")

    if not timestamp or not signature or abs(time.time() - timestamp) > 300:
        raise HTTPException(status_code=400, detail="firma expirada")

    expected = hmac.new(
        SECRET.encode(), f"{timestamp}.".encode() + raw, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="firma inválida")

    event = json.loads(raw)
    # Responde 200 rápido y procesa en una cola.
    if event.get("type") == "job.succeeded":
        print("Job listo:", event["data"]["job_id"])

    return {"received": True}