109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""
|
|
MCP Indy custom — serveur FastMCP qui appelle l'API privee d'Indy avec ton token,
|
|
recupere tes transactions et les convertit au format d'import de JB Compta.
|
|
|
|
API non officielle d'Indy : pour TES donnees uniquement. Peut casser si Indy
|
|
change son interface ; le token JWT expire et doit etre renouvele.
|
|
"""
|
|
import os
|
|
|
|
import httpx
|
|
from fastmcp import FastMCP
|
|
|
|
import indy_mapping
|
|
|
|
BASE_URL = os.getenv("INDY_BASE_URL", "https://app.indy.fr").rstrip("/")
|
|
TOKEN = os.getenv("INDY_TOKEN", "")
|
|
AUTH_HEADER = os.getenv("INDY_AUTH_HEADER", "Authorization")
|
|
AUTH_PREFIX = os.getenv("INDY_AUTH_PREFIX", "Bearer ")
|
|
COOKIE = os.getenv("INDY_COOKIE", "")
|
|
TRANSACTIONS_PATH = os.getenv("INDY_TRANSACTIONS_PATH",
|
|
"/api/transactions/transactions-list")
|
|
|
|
mcp = FastMCP("indy-compta-custom")
|
|
|
|
|
|
def _client():
|
|
headers = {
|
|
"Accept": "application/json, text/plain, */*",
|
|
"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0 Safari/537.36"),
|
|
"x-client-app-type": "web",
|
|
}
|
|
if TOKEN:
|
|
headers[AUTH_HEADER] = f"{AUTH_PREFIX}{TOKEN}"
|
|
if COOKIE:
|
|
headers["Cookie"] = COOKIE
|
|
return httpx.Client(base_url=BASE_URL, headers=headers, timeout=30,
|
|
follow_redirects=True)
|
|
|
|
|
|
def _appel_api(path, method="GET", params=None, json_body=None):
|
|
if not TOKEN and not COOKIE:
|
|
return {"erreur": "INDY_TOKEN (ou INDY_COOKIE) non configure dans .env"}
|
|
try:
|
|
with _client() as c:
|
|
r = c.request(method.upper(), path, params=params, json=json_body)
|
|
ct = r.headers.get("content-type", "")
|
|
body = r.json() if "json" in ct else r.text[:3000]
|
|
return {"status": r.status_code, "url": str(r.request.url), "data": body}
|
|
except Exception as ex: # noqa: BLE001
|
|
return {"erreur": f"{type(ex).__name__}: {ex}"}
|
|
|
|
|
|
def _fetch_transactions(search="", date_debut=None, date_fin=None, page=1):
|
|
params = {"search": search or "", "page": page}
|
|
if date_debut:
|
|
params["dateFrom"] = date_debut
|
|
if date_fin:
|
|
params["dateTo"] = date_fin
|
|
res = _appel_api(TRANSACTIONS_PATH, "GET", params=params)
|
|
if "erreur" in res:
|
|
return res, []
|
|
data = res.get("data") or {}
|
|
txs = data.get("transactions", []) if isinstance(data, dict) else []
|
|
return res, txs
|
|
|
|
|
|
@mcp.tool
|
|
def etat() -> dict:
|
|
"""Verifie la configuration : URL de base, presence du token, endpoint."""
|
|
return {"base_url": BASE_URL, "token_present": bool(TOKEN),
|
|
"auth_header": AUTH_HEADER, "transactions_path": TRANSACTIONS_PATH}
|
|
|
|
|
|
@mcp.tool
|
|
def appel_api(path: str, method: str = "GET", params: dict | None = None,
|
|
json_body: dict | None = None) -> dict:
|
|
"""Explorateur generique : appelle n'importe quel endpoint de l'API Indy."""
|
|
return _appel_api(path, method, params, json_body)
|
|
|
|
|
|
@mcp.tool
|
|
def lister_transactions(search: str = "", date_debut: str | None = None,
|
|
date_fin: str | None = None, page: int = 1) -> dict:
|
|
"""Liste brute des transactions Indy (filtres : search, dates AAAA-MM-JJ, page)."""
|
|
res, _ = _fetch_transactions(search, date_debut, date_fin, page)
|
|
return res
|
|
|
|
|
|
@mcp.tool
|
|
def transactions_pour_import(date_debut: str | None = None, date_fin: str | None = None,
|
|
search: str = "", page: int = 1) -> dict:
|
|
"""Recupere les transactions Indy au format CSV de JB Compta (a coller dans /import/).
|
|
Colonnes : date;libelle;categorie;activite;taux_tva;regime;montant
|
|
"""
|
|
res, txs = _fetch_transactions(search, date_debut, date_fin, page)
|
|
if "erreur" in res:
|
|
return res
|
|
data = res.get("data") or {}
|
|
return {
|
|
"nb_transactions_page": len(txs),
|
|
"nb_total": data.get("nbTransactions") if isinstance(data, dict) else None,
|
|
"csv": indy_mapping.vers_csv(txs),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|