Automate Contract Review with AI: A Developer's Integration Guide

pip install requests python-docx pypdf2

export APIVULT_API_KEY="YOUR_API_KEY"

import os
import requests
import base64
 
API_KEY = os.environ["APIVULT_API_KEY"]
BASE_URL = "https://apivult.com/api/v1/legalguard"
 
def analyze_contract(
    file_path: str,
    contract_type: str = "service_agreement",
    party_role: str = "recipient",  # or "provider"
) -> dict:
    """
    Analyze a contract file and return structured results.
 
    Args:
        file_path: Path to PDF or DOCX contract file
        contract_type: One of: service_agreement, nda, employment, vendor, lease
        party_role: Your role in the contract (affects risk perspective)
    """
    with open(file_path, "rb") as f:
        file_content = base64.b64encode(f.read()).decode()
 
    file_ext = file_path.split(".")[-1].lower()
 
    response = requests.post(
        f"{BASE_URL}/analyze",
        headers={"X-API-Key": API_KEY},
        json={
            "document": file_content,
            "format": file_ext,  # pdf, docx, or txt
            "contract_type": contract_type,
            "party_role": party_role,
            "options": {
                "extract_clauses": True,
                "risk_analysis": True,
                "generate_summary": True,
                "detect_missing": True,
            },
        },
    )
    response.raise_for_status()
    return response.json()
 
# Analyze a vendor contract
result = analyze_contract(
    "vendor_agreement.pdf",
    contract_type="vendor",
    party_role="recipient",
)
 
print(f"Contract type: {result['contract_type']}")
print(f"Overall risk level: {result['risk_level']}")  # low / medium / high / critical
print(f"Clauses found: {len(result['clauses'])}")
print(f"Risk flags: {len(result['risk_flags'])}")

def print_clause_summary(result: dict):
    """Display extracted clauses organized by category."""
    clauses = result["clauses"]
 
    # Group by category
    by_category = {}
    for clause in clauses:
        cat = clause["category"]
        by_category.setdefault(cat, []).append(clause)
 
    for category, items in sorted(by_category.items()):
        print(f"\n{'='*50}")
        print(f"  {category.upper().replace('_', ' ')}")
        print(f"{'='*50}")
 
        for item in items:
            print(f"\n  [{item['name']}]")
            print(f"  Plain English: {item['plain_english']}")
            print(f"  Original text: {item['text'][:200]}...")
            print(f"  Page: {item['page']}, Paragraph: {item['paragraph']}")
 
print_clause_summary(result)
# ══════════════════════════════════════════════════
#   PAYMENT TERMS
# ══════════════════════════════════════════════════
#
#   [Net Payment Period]
#   Plain English: You must pay invoices within 90 days of receipt.
#   Original text: Invoices shall be payable within ninety (90) calendar days...
#   Page: 3, Paragraph: 12
#
#   [Late Payment Penalty]
#   Plain English: Late payments accrue 1.5% interest per month.
#   Original text: Any amounts not paid by the due date shall accrue interest...

def review_risk_flags(result: dict, min_severity: str = "medium"):
    """Print risk flags above a minimum severity threshold."""
    severity_order = {"low": 0, "medium": 1, "high": 2, "critical": 3}
    threshold = severity_order.get(min_severity, 1)
 
    flags = [
        f for f in result["risk_flags"]
        if severity_order.get(f["severity"], 0) >= threshold
    ]
 
    if not flags:
        print("No significant risk flags found.")
        return
 
    print(f"\nFound {len(flags)} risk flags (severity >= {min_severity}):\n")
 
    for flag in sorted(flags, key=lambda x: severity_order.get(x["severity"], 0), reverse=True):
        severity_emoji = {
            "critical": "🔴",
            "high": "🟠",
            "medium": "🟡",
            "low": "🟢",
        }.get(flag["severity"], "⚪")
 
        print(f"{severity_emoji} [{flag['severity'].upper()}] {flag['title']}")
        print(f"   Issue: {flag['description']}")
        print(f"   Location: {flag['clause_name']} (Page {flag['page']})")
        print(f"   Recommendation: {flag['recommendation']}\n")
 
review_risk_flags(result, min_severity="medium")
# 🔴 [CRITICAL] Uncapped Liability Exposure
#    Issue: No liability cap is specified. You could be liable for unlimited damages.
#    Location: Liability clause (Page 7)
#    Recommendation: Negotiate a cap at 12 months of fees paid
#
# 🟠 [HIGH] Unilateral Termination Rights
#    Issue: Vendor can terminate with 5 days notice; you need 90 days.
#    Location: Termination clause (Page 11)
#    Recommendation: Negotiate mutual 30-day notice period
#
# 🟡 [MEDIUM] Broad IP Assignment
#    Issue: All work product created during engagement is assigned to vendor.
#    Location: IP ownership (Page 9)
#    Recommendation: Carve out pre-existing IP and specify ownership of deliverables

def check_missing_clauses(result: dict):
    missing = result.get("missing_clauses", [])
 
    if not missing:
        print("All expected clauses are present.")
        return
 
    print(f"\nMissing clauses ({len(missing)}):\n")
    for clause in missing:
        importance = "⚠️ Required" if clause["required"] else "ℹ️ Recommended"
        print(f"  {importance}: {clause['name']}")
        print(f"   Why it matters: {clause['importance']}\n")
 
check_missing_clauses(result)
# ⚠️ Required: Data Processing Agreement (DPA)
#    Why it matters: Vendor will process personal data. A DPA is required under GDPR.
#
# ℹ️ Recommended: Dispute Resolution Clause
#    Why it matters: Without this, disputes default to litigation. Consider adding
#    arbitration to reduce cost and time of resolution.

def get_executive_summary(result: dict) -> str:
    summary = result["summary"]
 
    output = []
    output.append(f"CONTRACT SUMMARY")
    output.append(f"{'='*60}")
    output.append(f"Type: {summary['contract_type']}")
    output.append(f"Duration: {summary['term']}")
    output.append(f"Total Value: {summary['contract_value']}")
    output.append(f"Governing Law: {summary['governing_law']}")
    output.append(f"")
    output.append(f"KEY TERMS:")
    for term in summary["key_terms"]:
        output.append(f"  • {term}")
    output.append(f"")
    output.append(f"OVERALL RISK ASSESSMENT:")
    output.append(f"  {summary['risk_assessment']}")
    output.append(f"")
    output.append(f"RECOMMENDED ACTIONS:")
    for action in summary["recommended_actions"]:
        output.append(f"  {action['priority']}. {action['action']}")
 
    return "\n".join(output)
 
print(get_executive_summary(result))

import json
from pathlib import Path
 
def process_contract_queue(input_dir: str, output_dir: str):
    """Process all contracts in a directory and save analysis results."""
    Path(output_dir).mkdir(exist_ok=True)
    client_headers = {"X-API-Key": API_KEY}
 
    contracts = list(Path(input_dir).glob("*.pdf")) + list(Path(input_dir).glob("*.docx"))
    print(f"Processing {len(contracts)} contracts...")
 
    results = []
    for contract_path in contracts:
        print(f"  Analyzing: {contract_path.name}")
        try:
            result = analyze_contract(str(contract_path))
 
            # Save detailed result
            output_path = Path(output_dir) / f"{contract_path.stem}_analysis.json"
            with open(output_path, "w") as f:
                json.dump(result, f, indent=2)
 
            results.append({
                "file": contract_path.name,
                "risk_level": result["risk_level"],
                "critical_flags": sum(
                    1 for f in result["risk_flags"] if f["severity"] == "critical"
                ),
                "high_flags": sum(
                    1 for f in result["risk_flags"] if f["severity"] == "high"
                ),
                "missing_required": sum(
                    1 for c in result.get("missing_clauses", []) if c["required"]
                ),
                "status": "analyzed",
            })
        except Exception as e:
            results.append({"file": contract_path.name, "status": "error", "error": str(e)})
 
    # Print summary table
    print(f"\n{'File':<40} {'Risk':<10} {'Critical':<10} {'High':<8} {'Missing':<10}")
    print("-" * 80)
    for r in results:
        if r["status"] == "analyzed":
            print(
                f"{r['file']:<40} {r['risk_level']:<10} "
                f"{r['critical_flags']:<10} {r['high_flags']:<8} {r['missing_required']:<10}"
            )
        else:
            print(f"{r['file']:<40} ERROR: {r['error']}")
 
process_contract_queue("contracts/pending/", "contracts/analyzed/")