How to Build an Agentic AI Invoice Fraud Detection System with FinAudit API

import requests
import base64
from pathlib import Path
from typing import Optional
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/api/finaudit"
 
def analyze_invoice_pdf(pdf_path: str, vendor_id: Optional[str] = None) -> dict:
    """
    Submit a PDF invoice for fraud analysis.
    
    Args:
        pdf_path: Path to the invoice PDF file
        vendor_id: Known vendor identifier for historical comparison
    
    Returns:
        Fraud analysis result with risk score and indicators
    """
    pdf_data = Path(pdf_path).read_bytes()
    encoded_pdf = base64.b64encode(pdf_data).decode("utf-8")
 
    payload = {
        "document": encoded_pdf,
        "document_type": "invoice",
        "vendor_id": vendor_id,
        "analysis_types": [
            "fraud_indicators",
            "duplicate_detection",
            "semantic_consistency",
            "historical_deviation",
            "metadata_analysis"
        ],
        "return_extracted_fields": True
    }
 
    response = requests.post(
        f"{BASE_URL}/analyze",
        json=payload,
        headers={"X-RapidAPI-Key": API_KEY},
        timeout=30
    )
    response.raise_for_status()
    return response.json()
 
def analyze_invoice_structured(invoice_data: dict) -> dict:
    """
    Analyze structured invoice data (pre-extracted fields).
    Use this when your AP system already extracts invoice fields.
    """
    payload = {
        "invoice": invoice_data,
        "analysis_types": [
            "fraud_indicators",
            "duplicate_detection",
            "semantic_consistency"
        ]
    }
 
    response = requests.post(
        f"{BASE_URL}/analyze-structured",
        json=payload,
        headers={"X-RapidAPI-Key": API_KEY},
        timeout=15
    )
    response.raise_for_status()
    return response.json()

from enum import Enum
from dataclasses import dataclass
from datetime import datetime
 
class RiskLevel(Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"
 
@dataclass
class InvoiceDecision:
    invoice_id: str
    risk_level: RiskLevel
    risk_score: float
    fraud_indicators: list
    action: str
    requires_review: bool
    timestamp: datetime
 
def classify_invoice_risk(analysis_result: dict, invoice_id: str) -> InvoiceDecision:
    """
    Translate FinAudit AI analysis result into a business action.
    Customize thresholds based on your risk tolerance.
    """
    risk_score = analysis_result["risk_score"]  # 0.0 to 1.0
    fraud_indicators = analysis_result.get("fraud_indicators", [])
    critical_flags = [f for f in fraud_indicators if f.get("severity") == "CRITICAL"]
 
    # Determine risk level
    if critical_flags or risk_score >= 0.85:
        risk_level = RiskLevel.CRITICAL
        action = "BLOCK_AND_ALERT"
        requires_review = True
    elif risk_score >= 0.65:
        risk_level = RiskLevel.HIGH
        action = "HOLD_FOR_REVIEW"
        requires_review = True
    elif risk_score >= 0.40:
        risk_level = RiskLevel.MEDIUM
        action = "QUEUE_FOR_REVIEW"
        requires_review = True
    else:
        risk_level = RiskLevel.LOW
        action = "APPROVE"
        requires_review = False
 
    return InvoiceDecision(
        invoice_id=invoice_id,
        risk_level=risk_level,
        risk_score=risk_score,
        fraud_indicators=fraud_indicators,
        action=action,
        requires_review=requires_review,
        timestamp=datetime.utcnow()
    )

import json
from typing import Callable
 
def process_invoice_batch(
    invoice_paths: list[str],
    notify_callback: Callable[[InvoiceDecision], None]
) -> dict:
    """
    Process a batch of invoices through the fraud detection pipeline.
    
    Args:
        invoice_paths: List of PDF file paths to process
        notify_callback: Function called for each decision (for alerting, logging)
    
    Returns:
        Summary statistics for the batch
    """
    results = {
        "total": len(invoice_paths),
        "approved": 0,
        "held_for_review": 0,
        "blocked": 0,
        "errors": 0,
        "decisions": []
    }
 
    for idx, pdf_path in enumerate(invoice_paths):
        invoice_id = f"INV-{idx:06d}"
        
        try:
            # Run fraud analysis
            analysis = analyze_invoice_pdf(pdf_path)
            decision = classify_invoice_risk(analysis, invoice_id)
 
            # Update counters
            if decision.action == "APPROVE":
                results["approved"] += 1
            elif decision.action in ("HOLD_FOR_REVIEW", "QUEUE_FOR_REVIEW"):
                results["held_for_review"] += 1
            elif decision.action == "BLOCK_AND_ALERT":
                results["blocked"] += 1
 
            results["decisions"].append({
                "invoice_id": invoice_id,
                "path": pdf_path,
                "risk_score": decision.risk_score,
                "action": decision.action,
                "indicators": decision.fraud_indicators
            })
 
            # Trigger notification (alert system, email, Slack, etc.)
            notify_callback(decision)
 
        except Exception as e:
            results["errors"] += 1
            print(f"Error processing {pdf_path}: {e}")
 
    return results
 
# Example usage
def alert_finance_team(decision: InvoiceDecision) -> None:
    if decision.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL):
        print(f"[ALERT] Invoice {decision.invoice_id} flagged: "
              f"Risk {decision.risk_score:.2f} — {decision.action}")
        # In production: send to Slack, email, or ticketing system
 
# Process a batch
invoice_files = [
    "invoices/vendor_acme_001.pdf",
    "invoices/vendor_globex_002.pdf",
    "invoices/vendor_initech_003.pdf"
]
 
summary = process_invoice_batch(invoice_files, alert_finance_team)
print(json.dumps(summary, indent=2, default=str))

def build_vendor_baseline(vendor_id: str, historical_invoices: list[dict]) -> dict:
    """
    Build a statistical baseline for a vendor from historical invoice data.
    Submit this to FinAudit AI to improve future anomaly detection for this vendor.
    """
    response = requests.post(
        f"{BASE_URL}/vendor-profile",
        json={
            "vendor_id": vendor_id,
            "historical_invoices": historical_invoices,
            "baseline_window_days": 365
        },
        headers={"X-RapidAPI-Key": API_KEY}
    )
    return response.json()