SaaS Contract Risk Scoring with LegalGuard AI API: A Developer's Guide

pip install requests pathlib PyPDF2

import requests
import base64
from pathlib import Path
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/legalguard/v1"
 
# SaaS-specific risk clauses to analyze
SAAS_RISK_CLAUSES = [
    "auto_renewal",
    "termination_for_convenience",
    "liability_cap",
    "indemnification",
    "data_processing",
    "data_deletion",
    "price_change_rights",
    "sla_credits",
    "intellectual_property",
    "governing_law",
    "dispute_resolution",
    "force_majeure",
    "assignment_rights"
]
 
def analyze_saas_contract(pdf_path: str, vendor_name: str = None) -> dict:
    """Submit a SaaS contract PDF to LegalGuard AI for risk scoring."""
    pdf_bytes = Path(pdf_path).read_bytes()
    encoded = base64.b64encode(pdf_bytes).decode("utf-8")
 
    payload = {
        "document": encoded,
        "document_type": "saas_agreement",
        "analysis_mode": "risk_scoring",
        "vendor_name": vendor_name,
        "clauses_to_analyze": SAAS_RISK_CLAUSES,
        "benchmark": "saas_market_standard_2026",
        "output_format": "structured"
    }
 
    response = requests.post(
        f"{BASE_URL}/analyze",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=90
    )
 
    if response.status_code != 200:
        raise ValueError(f"LegalGuard API error {response.status_code}: {response.text}")
 
    return response.json()

from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class ClauseRisk:
    clause_type: str
    risk_level: str          # low / medium / high / critical
    score: int               # 0-100 (higher = riskier)
    extracted_text: str
    standard_benchmark: str  # What market-standard looks like
    negotiation_priority: str
    recommendation: str
 
@dataclass
class ContractRiskReport:
    vendor_name: str
    overall_risk_score: int
    overall_risk_tier: str
    clause_findings: list[ClauseRisk]
    critical_clauses: list[ClauseRisk] = field(default_factory=list)
    high_risk_clauses: list[ClauseRisk] = field(default_factory=list)
    auto_renewal_risk: bool = False
    uncapped_liability: bool = False
    missing_data_deletion: bool = False
 
 
def parse_risk_report(api_response: dict, vendor_name: str) -> ContractRiskReport:
    """Parse LegalGuard AI response into a structured risk report."""
    overall = api_response.get("overall_risk", {})
    findings = api_response.get("clause_findings", [])
 
    clause_risks = []
    for finding in findings:
        clause_risks.append(ClauseRisk(
            clause_type=finding["clause_type"],
            risk_level=finding["risk_level"],
            score=finding["risk_score"],
            extracted_text=finding.get("extracted_text", "Not found"),
            standard_benchmark=finding.get("benchmark_text", ""),
            negotiation_priority=finding.get("negotiation_priority", "medium"),
            recommendation=finding.get("recommendation", "")
        ))
 
    report = ContractRiskReport(
        vendor_name=vendor_name or api_response.get("vendor_name", "Unknown Vendor"),
        overall_risk_score=overall.get("score", 0),
        overall_risk_tier=overall.get("tier", "unknown"),
        clause_findings=clause_risks,
        critical_clauses=[c for c in clause_risks if c.risk_level == "critical"],
        high_risk_clauses=[c for c in clause_risks if c.risk_level == "high"]
    )
 
    # Extract specific high-value risk flags
    auto_renewal = next((c for c in clause_risks if c.clause_type == "auto_renewal"), None)
    liability = next((c for c in clause_risks if c.clause_type == "liability_cap"), None)
    data_del = next((c for c in clause_risks if c.clause_type == "data_deletion"), None)
 
    report.auto_renewal_risk = bool(auto_renewal and auto_renewal.risk_level in ("high", "critical"))
    report.uncapped_liability = bool(liability and liability.risk_level == "critical")
    report.missing_data_deletion = bool(data_del and "not found" in data_del.extracted_text.lower())
 
    return report
 
 
def print_risk_report(report: ContractRiskReport):
    """Print a formatted risk report to the console."""
    tier_emoji = {"low": "✅", "medium": "⚠", "high": "🔴", "critical": "🚨"}.get(
        report.overall_risk_tier.lower(), "❓"
    )
 
    print(f"\n{'═'*60}")
    print(f"CONTRACT RISK REPORT: {report.vendor_name}")
    print(f"{'═'*60}")
    print(f"Overall Risk: {tier_emoji} {report.overall_risk_tier.upper()} (Score: {report.overall_risk_score}/100)")
    print()
 
    if report.auto_renewal_risk:
        print("⚠ AUTO-RENEWAL TRAP: Short notice window or difficult exit provisions detected")
    if report.uncapped_liability:
        print("🚨 UNCAPPED LIABILITY: No liability limitation clause found or inadequate cap")
    if report.missing_data_deletion:
        print("⚠ DATA DELETION: No clear data deletion timeline specified on termination")
 
    if report.critical_clauses:
        print(f"\nCRITICAL CLAUSES ({len(report.critical_clauses)}) — Requires legal review before signing:")
        for clause in report.critical_clauses:
            print(f"  • {clause.clause_type.replace('_', ' ').title()}")
            print(f"    Found: \"{clause.extracted_text[:120]}...\"")
            print(f"    ⚡ {clause.recommendation}")
 
    if report.high_risk_clauses:
        print(f"\nHIGH RISK CLAUSES ({len(report.high_risk_clauses)}) — Negotiate if possible:")
        for clause in report.high_risk_clauses:
            print(f"  • {clause.clause_type.replace('_', ' ').title()}: {clause.recommendation}")

import json
from datetime import datetime
 
 
def batch_score_contracts(contract_files: list[dict]) -> list[ContractRiskReport]:
    """
    Score multiple contracts in sequence.
    contract_files: [{"path": "...", "vendor_name": "..."}, ...]
    """
    reports = []
 
    for contract in contract_files:
        print(f"Analyzing: {contract['vendor_name']}...")
        try:
            api_result = analyze_saas_contract(
                contract["path"],
                contract.get("vendor_name")
            )
            report = parse_risk_report(api_result, contract.get("vendor_name", "Unknown"))
            reports.append(report)
        except Exception as e:
            print(f"  Error analyzing {contract['vendor_name']}: {e}")
 
    return sorted(reports, key=lambda r: r.overall_risk_score, reverse=True)
 
 
def export_risk_summary(reports: list[ContractRiskReport], output_path: str = None) -> dict:
    """Export risk summary as JSON for integration with your procurement system."""
    summary = {
        "generated_at": datetime.now().isoformat(),
        "total_contracts": len(reports),
        "risk_distribution": {
            "critical": sum(1 for r in reports if r.overall_risk_tier == "critical"),
            "high": sum(1 for r in reports if r.overall_risk_tier == "high"),
            "medium": sum(1 for r in reports if r.overall_risk_tier == "medium"),
            "low": sum(1 for r in reports if r.overall_risk_tier == "low")
        },
        "contracts": [
            {
                "vendor": r.vendor_name,
                "risk_score": r.overall_risk_score,
                "risk_tier": r.overall_risk_tier,
                "auto_renewal_risk": r.auto_renewal_risk,
                "uncapped_liability": r.uncapped_liability,
                "missing_data_deletion": r.missing_data_deletion,
                "critical_clause_count": len(r.critical_clauses),
                "high_risk_clause_count": len(r.high_risk_clauses)
            }
            for r in reports
        ]
    }
 
    if output_path:
        with open(output_path, "w") as f:
            json.dump(summary, f, indent=2)
        print(f"Risk summary exported to {output_path}")
 
    return summary

from fastapi import FastAPI, UploadFile, File, Form
import tempfile, os
 
app = FastAPI()
 
@app.post("/procurement/screen-contract")
async def screen_contract(
    contract: UploadFile = File(...),
    vendor_name: str = Form(...),
    contract_value: float = Form(None),
    urgency: str = Form("normal")  # normal / urgent
):
    """
    Procurement screening endpoint: returns risk tier for routing decisions.
    Low/Medium risk → auto-approve for legal light-touch review
    High/Critical risk → route to full legal review queue
    """
    with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
        tmp.write(await contract.read())
        tmp_path = tmp.name
 
    try:
        api_result = analyze_saas_contract(tmp_path, vendor_name)
        report = parse_risk_report(api_result, vendor_name)
 
        # Escalate high-value contracts automatically
        if contract_value and contract_value > 100000 and report.overall_risk_tier == "medium":
            report.overall_risk_tier = "high"
 
        route = "auto_approve" if report.overall_risk_tier == "low" else (
            "light_review" if report.overall_risk_tier == "medium" else "full_legal_review"
        )
 
        return {
            "vendor": vendor_name,
            "risk_score": report.overall_risk_score,
            "risk_tier": report.overall_risk_tier,
            "routing_decision": route,
            "critical_count": len(report.critical_clauses),
            "high_risk_count": len(report.high_risk_clauses),
            "key_flags": {
                "auto_renewal_risk": report.auto_renewal_risk,
                "uncapped_liability": report.uncapped_liability,
                "missing_data_deletion": report.missing_data_deletion
            },
            "top_findings": [
                {"clause": c.clause_type, "recommendation": c.recommendation}
                for c in (report.critical_clauses + report.high_risk_clauses)[:5]
            ]
        }
    finally:
        os.unlink(tmp_path)