#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Facturolia MCP-server: laseb AI-assistendil lugeda ja kontrollida elektroonilisi arveid (Factur-X, ZUGFeRD, XRechnung, UBL, CII). MIKS SEE OLEMAS ON ------------------ Logianalüüs näitas, et ChatGPT on meie päris liikluse allikas ja konkurent lekteo.eu pakub MCP-liidest oma lugejale. Meil on midagi paremat — päris REST API — ja see fail teeb selle assistentidele kättesaadavaks. SÕLTUVUSTETA TEADLIKULT ----------------------- Ainult Pythoni standardteek. Kasutaja ei pea midagi paigaldama peale Pythoni (3.10+), mis alandab proovimise künnise nullilähedaseks. MCP SDK lisamine tähendaks pip-install-sammu iga kasutaja jaoks — see maksaks rohkem, kui võidaks. PRIVAATSUS — ÜTLE SEE KASUTAJALE VÄLJA -------------------------------------- See server SAADAB arvefaili facturolia.fr API-le analüüsiks. Fail analüüsitakse mälus ja kustutatakse kohe (sama lubadus mis veebiliideses), aga see EI OLE kohalik töötlus. Kes ei taha faili välja saata, kasutagu veebiliidest või API-t ise. KASUTAMINE (Claude Desktop / muu MCP-klient), claude_desktop_config.json: { "mcpServers": { "facturolia": { "command": "python", "args": ["/chemin/vers/facturolia_mcp.py"], "env": {"FACTUROLIA_API_KEY": "your-key"} } } } Votme saab https://facturolia.fr/en/pricing (tasuta tase olemas). Vastuste keel: vaikimisi INGLISE. Prantsuse keeleks lisa env-i "FACTUROLIA_LANG": "fr". Sama loogika mis /v1 API-l. """ from __future__ import annotations import json import mimetypes import os import sys import urllib.error import urllib.request import uuid from pathlib import Path NAME = "facturolia" VERSION = "1.0.0" PROTOCOL = "2024-11-05" BASE_URL = os.environ.get("FACTUROLIA_BASE_URL", "https://facturolia.fr").rstrip("/") API_KEY = os.environ.get("FACTUROLIA_API_KEY", "") TIMEOUT = 60 MAX_BYTES = 10 * 1024 * 1024 ALLOWED = {".pdf", ".xml"} # Keel: INGLISE vaikimisi, tapselt nagu /v1 API-l. Prantsuse kasutaja lisab # konfiguratsiooni "env": {"FACTUROLIA_LANG": "fr"}. Varem oli koik kovasti # prantsuse keeles: ingliskeelne kasutaja sai prantsuskeelse vastuse ega # saanud sellest kuidagi ule. _raw_lang = os.environ.get("FACTUROLIA_LANG", "en").strip().lower()[:2] LANG = _raw_lang if _raw_lang in ("en", "fr") else "en" BLANK = chr(10) # tuhi rida ploki ette SEP = chr(10) # ridade eraldaja valjundis T = { "en": { "error": "Error: ", "no_key": ("FACTUROLIA_API_KEY is not set. Get a key at " "https://facturolia.fr/en/pricing and add it to the MCP " "configuration, in the \"env\" field."), "no_connect": "Cannot reach {base}: {reason}", "not_found": "File not found: {p}", "bad_ext": ("Unsupported extension ({ext}). Expected .pdf " "(Factur-X/ZUGFeRD) or .xml (UBL/CII/XRechnung)."), "too_big": "File too large (max {mb} MB).", "format": "Format: {f}", "profile": " ({p} profile)", "invoice_no": "Invoice {n}", "issued": " · issued {d}", "due": " · due {d}", "seller": "Supplier: {n}", "buyer": "Customer: {n}", "totals": "Total excl. VAT {ht} · VAT {va} · Total incl. VAT {tc}", "lines": "Lines ({n}):", "more_lines": " ... and {n} more lines", "vat_rate": " (VAT {r} %)", "check_ok": "EN 16931 check: no anomalies found.", "check_bad": "EN 16931 check: {e} error(s), {w} warning(s)", "cmp": "PDF to XML comparison: {s}", "cmp_done": "done", "warning": "Warning: {t}", }, "fr": { "error": "Erreur : ", "no_key": ("FACTUROLIA_API_KEY n'est pas défini. Obtenez une clé sur " "https://facturolia.fr/tarifs puis ajoutez-la dans la " "configuration MCP, champ \"env\"."), "no_connect": "Connexion impossible à {base} : {reason}", "not_found": "Fichier introuvable : {p}", "bad_ext": ("Extension non prise en charge ({ext}). Attendu : .pdf " "(Factur-X/ZUGFeRD) ou .xml (UBL/CII/XRechnung)."), "too_big": "Fichier trop volumineux (max {mb} Mo).", "format": "Format : {f}", "profile": " (profil {p})", "invoice_no": "Facture n° {n}", "issued": " · émise le {d}", "due": " · échéance {d}", "seller": "Fournisseur : {n}", "buyer": "Client : {n}", "totals": "Total HT {ht} · TVA {va} · TTC {tc}", "lines": "Lignes ({n}) :", "more_lines": " ... et {n} autres lignes", "vat_rate": " (TVA {r} %)", "check_ok": "Contrôle EN 16931 : aucune anomalie détectée.", "check_bad": "Contrôle EN 16931 : {e} erreur(s), {w} avertissement(s)", "cmp": "Comparaison PDF vers XML : {s}", "cmp_done": "effectuée", "warning": "Avertissement : {t}", }, }[LANG] # --------------------------------------------------------------- HTTP-kiht def _multipart(path: Path) -> tuple[bytes, str]: """Ehitab multipart/form-data keha ilma väliste teekideta.""" boundary = f"----facturolia{uuid.uuid4().hex}" mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" body = b"".join([ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n'.encode(), f"Content-Type: {mime}\r\n\r\n".encode(), path.read_bytes(), f"\r\n--{boundary}--\r\n".encode(), ]) return body, f"multipart/form-data; boundary={boundary}" def call_api(path: Path) -> dict: """Saadab faili /v1/invoices/parse'ile. Tagastab API vastuse dict'ina.""" if not API_KEY: raise RuntimeError(T["no_key"]) body, content_type = _multipart(path) req = urllib.request.Request( f"{BASE_URL}/v1/invoices/parse?lang={LANG}", data=body, headers={"Content-Type": content_type, "X-API-Key": API_KEY, "Accept": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: # API veakered on JSON — anname assistendile sisulise põhjuse edasi, # mitte lihtsalt "HTTP 422". raw = e.read().decode("utf-8", "replace") try: err = json.loads(raw).get("error", {}) raise RuntimeError( f"{err.get('code', e.code)}: {err.get('message', raw[:200])}") from None except (ValueError, AttributeError): raise RuntimeError(f"HTTP {e.code}: {raw[:200]}") from None except urllib.error.URLError as e: raise RuntimeError(T["no_connect"].format(base=BASE_URL, reason=e.reason)) from None def _checked_path(raw: str) -> Path: p = Path(raw).expanduser() if not p.is_file(): raise RuntimeError(T["not_found"].format(p=p)) if p.suffix.lower() not in ALLOWED: raise RuntimeError(T["bad_ext"].format(ext=p.suffix or "none")) if p.stat().st_size > MAX_BYTES: raise RuntimeError(T["too_big"].format(mb=MAX_BYTES // 1024 // 1024)) return p # ------------------------------------------------------------ vormindamine def _money(v, cur) -> str: return "—" if v in (None, "") else f"{v} {cur or ''}".strip() def _date(v) -> str: """CII kirjutab kuupäeva kujul 20260715. Assistent loeb selle kasutajale ETTE — toores number kõlaks arusaamatult, seega vormindame.""" if not v: return "—" s = str(v).strip().replace("-", "")[:8] if len(s) == 8 and s.isdigit(): # EN saab ISO: 07/08 on britile ja ameeriklasele eri paev return (f"{s[0:4]}-{s[4:6]}-{s[6:8]}" if LANG == "en" else f"{s[6:8]}/{s[4:6]}/{s[0:4]}") return str(v) def summarize(res: dict) -> str: """API vastus -> kompaktne tekst assistendile. Miks tekst, mitte toores JSON: assistent peab vastuse kasutajale ette lugema. Täis-JSON on müra; siin on see, mis arve juures loeb. """ inv = res.get("invoice") or {} val = res.get("validation") or {} cur = inv.get("currency") seller = (inv.get("seller") or {}).get("name") or "—" buyer = (inv.get("buyer") or {}).get("name") or "—" out = [ T["format"].format(f=res.get("format") or "-") + (T["profile"].format(p=res["profile"]) if res.get("profile") else ""), T["invoice_no"].format(n=inv.get("invoice_number") or "-") + T["issued"].format(d=_date(inv.get("issue_date"))) + (T["due"].format(d=_date(inv["due_date"])) if inv.get("due_date") else ""), T["seller"].format(n=seller), T["buyer"].format(n=buyer), T["totals"].format(ht=_money(inv.get("total_excl_vat"), cur), va=_money(inv.get("total_vat"), cur), tc=_money(inv.get("total_incl_vat"), cur)), ] lines = inv.get("lines") or [] if lines: out.append(BLANK + T["lines"].format(n=len(lines))) for ln in lines[:20]: out.append( f" - {ln.get('description') or '-'} : " f"{ln.get('quantity') or '?'} x {_money(ln.get('unit_price'), cur)}" f" = {_money(ln.get('line_total'), cur)}" + (T["vat_rate"].format(r=ln["vat_rate"]) if ln.get("vat_rate") else "")) if len(lines) > 20: out.append(T["more_lines"].format(n=len(lines) - 20)) issues = val.get("issues") or [] if val.get("valid") and not issues: out.append(BLANK + T["check_ok"]) else: out.append(BLANK + T["check_bad"].format(e=val.get("error_count", 0), w=val.get("warning_count", 0))) for i in issues: out.append(f" [{i.get('severity', '?')}] {i.get('field', '')} : " f"{i.get('text') or i.get('code')}") # PDF vs XML: meie eristaja. Ilma selleta jaaks koige vaartuslikum vaikseks. cmp_ = inv.get("pdf_comparison") if isinstance(cmp_, dict) and cmp_: status = cmp_.get("status") or cmp_.get("result") or T["cmp_done"] out.append(BLANK + T["cmp"].format(s=status)) for k in ("mismatches", "differences"): for d in (cmp_.get(k) or [])[:10]: out.append(f" - {d}") for w in res.get("warnings") or []: out.append(BLANK + T["warning"].format(t=w.get("text") or w.get("code"))) return SEP.join(out) # ------------------------------------------------------------------ tööriistad # Neid kirjeldusi loeb ASSISTENT, mitte inimene, ja MCP-okosusteem on # ingliskeelne. Prantsuse kasutaja saab vastused prantsuse keeles (FACTUROLIA_LANG), # aga tooriista kirjeldus jaab inglise keelde, et iga klient seda moistaks. TOOLS = [ { "name": "read_invoice", "title": "Read an e-invoice", "description": ( "Extract the full contents of one electronic invoice file (Factur-X, " "ZUGFeRD, XRechnung, UBL or CII) as plain text: invoice number and dates, " "supplier and customer, every line, the VAT breakdown per rate, totals, " "IBAN, the EN 16931 consistency check, and whether the amounts on a " "Factur-X PDF's visible page match its embedded XML. " "Use it when the user wants to know what an invoice contains. If they only " "ask whether the invoice is correct, use check_invoice instead, which " "returns the verdict without the contents. " "Side effects: the file is uploaded over HTTPS to the Facturolia API, " "processed in memory and discarded; nothing is written locally. Requires " "FACTUROLIA_API_KEY; files over 10 MB are refused; requests are rate " "limited per key tier and a 429 error means wait and retry." ), "inputSchema": { "type": "object", "properties": { "path": { "type": "string", "description": ( "Absolute path on this machine to one invoice file. Must end " "in .pdf (a Factur-X or ZUGFeRD PDF with embedded XML) or .xml " "(UBL, CII or XRechnung). A plain PDF without embedded data " "returns an error explaining that it is not a structured invoice." ), }, }, "required": ["path"], "additionalProperties": False, }, "annotations": { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, }, { "name": "check_invoice", "title": "Check an e-invoice against EN 16931", "description": ( "Validate one electronic invoice file (Factur-X, ZUGFeRD, XRechnung, UBL " "or CII) against the EN 16931 standard and return only the verdict and the " "list of anomalies: totals that do not add up, VAT miscalculated for a " "rate, missing required fields, each with the rule code and the figures " "involved. " "Use it to answer 'is this invoice correct?' or to screen several files " "quickly. It does not list the invoice contents; for parties, lines and " "totals use read_invoice. " "This is a consistency check, not a legal compliance certificate, and it " "does not transmit the invoice to any tax platform. " "Side effects: the file is uploaded over HTTPS to the Facturolia API, " "processed in memory and discarded. Requires FACTUROLIA_API_KEY; 10 MB " "limit; rate limited per key tier." ), "inputSchema": { "type": "object", "properties": { "path": { "type": "string", "description": ( "Absolute path on this machine to one invoice file, .pdf " "(Factur-X or ZUGFeRD) or .xml (UBL, CII or XRechnung)." ), }, }, "required": ["path"], "additionalProperties": False, }, "annotations": { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, }, ] def run_tool(name: str, args: dict) -> str: path = _checked_path(str(args.get("path") or "")) res = call_api(path) if name == "read_invoice": return summarize(res) if name == "check_invoice": val = res.get("validation") or {} issues = val.get("issues") or [] head = (f"{path.name} : format {res.get('format') or '—'}" + (f" (profil {res['profile']})" if res.get("profile") else "")) if val.get("valid") and not issues: return f"{head}\n\nConforme : aucune anomalie EN 16931 détectée." body = "\n".join( f" [{i.get('severity', '?')}] {i.get('field', '')} : " f"{i.get('text') or i.get('code')}" for i in issues) return (f"{head}\n\n{val.get('error_count', 0)} erreur(s), " f"{val.get('warning_count', 0)} avertissement(s) :\n{body}") raise RuntimeError(f"Outil inconnu : {name}") # --------------------------------------------------------------- JSON-RPC def reply(msg_id, result=None, error=None): out = {"jsonrpc": "2.0", "id": msg_id} if error is not None: out["error"] = error else: out["result"] = result sys.stdout.write(json.dumps(out, ensure_ascii=False) + "\n") sys.stdout.flush() def handle(msg: dict) -> None: method = msg.get("method") msg_id = msg.get("id") # Teavitustel (id puudub) EI TOHI vastata — klient loeb seda protokollirikkeks. if msg_id is None: return if method == "initialize": reply(msg_id, { "protocolVersion": PROTOCOL, "capabilities": {"tools": {}}, "serverInfo": {"name": NAME, "version": VERSION}, }) elif method == "tools/list": reply(msg_id, {"tools": TOOLS}) elif method == "tools/call": params = msg.get("params") or {} try: text = run_tool(params.get("name", ""), params.get("arguments") or {}) reply(msg_id, {"content": [{"type": "text", "text": text}]}) except Exception as e: # Viga tööriista SEES tuleb tagastada isError-tulemusena, mitte # JSON-RPC veana: nii näeb assistent põhjust ja saab kasutajale # selgitada (nt puuduv API-võti), selle asemel et lihtsalt kukkuda. reply(msg_id, {"content": [{"type": "text", "text": T["error"] + f"{e}"}], "isError": True}) elif method == "ping": reply(msg_id, {}) else: reply(msg_id, error={"code": -32601, "message": f"Method not found: {method}"}) def main() -> None: for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except ValueError: continue try: handle(msg) except Exception as e: # server ei tohi ühe vigase sõnumi peale surra print(f"[facturolia-mcp] {e!r}", file=sys.stderr) if __name__ == "__main__": main()