Automate Employment Contract Risk Analysis with LegalGuard AI API

pip install httpx python-dotenv PyPDF2

# .env
LEGALGUARD_API_KEY=YOUR_API_KEY

import httpx
import os
import base64
from pathlib import Path
 
LEGALGUARD_BASE_URL = "https://apivult.com/api/legalguard"
 
 
async def analyze_employment_contract(
    contract_path: str,
    employee_state: str,
    contract_type: str = "employment_agreement"
) -> dict:
    """
    Analyze an employment contract for risk clauses.
 
    contract_type: employment_agreement | offer_letter | contractor_agreement
    employee_state: US state where employee will work (enforceability varies by state)
    """
    with open(contract_path, "rb") as f:
        content = base64.b64encode(f.read()).decode()
 
    file_ext = Path(contract_path).suffix.lower().lstrip(".")
 
    async with httpx.AsyncClient(timeout=45.0) as client:
        response = await client.post(
            f"{LEGALGUARD_BASE_URL}/analyze",
            headers={
                "X-RapidAPI-Key": os.getenv("LEGALGUARD_API_KEY"),
                "Content-Type": "application/json"
            },
            json={
                "document": content,
                "document_type": file_ext,
                "contract_category": contract_type,
                "jurisdiction": employee_state,
                "analysis_focus": [
                    "non_compete",
                    "ip_assignment",
                    "compensation",
                    "termination",
                    "arbitration",
                    "confidentiality",
                    "governing_law"
                ],
                "include_plain_english_summary": True,
                "include_clause_text": True
            }
        )
        response.raise_for_status()
        return response.json()
 
 
async def extract_key_terms(contract_path: str) -> dict:
    """
    Extract structured key terms from an employment contract.
    """
    with open(contract_path, "rb") as f:
        content = base64.b64encode(f.read()).decode()
 
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            f"{LEGALGUARD_BASE_URL}/extract",
            headers={
                "X-RapidAPI-Key": os.getenv("LEGALGUARD_API_KEY"),
                "Content-Type": "application/json"
            },
            json={
                "document": content,
                "document_type": "pdf",
                "extract_fields": [
                    "start_date",
                    "salary",
                    "equity_grant",
                    "vesting_schedule",
                    "non_compete_duration",
                    "non_compete_geography",
                    "notice_period",
                    "severance_terms",
                    "governing_law",
                    "arbitration_venue"
                ]
            }
        )
        response.raise_for_status()
        return response.json()

from dataclasses import dataclass
from typing import List
 
@dataclass
class ContractRisk:
    clause_type: str
    severity: str  # HIGH | MEDIUM | LOW
    issue: str
    clause_text: str
    plain_english: str
    recommendation: str
    state_specific: bool = False
 
 
def classify_employment_risks(analysis_result: dict, employee_state: str) -> List[ContractRisk]:
    """
    Parse analysis results into structured risk objects, applying state-specific rules.
    """
    risks = []
    clauses = analysis_result.get("flagged_clauses", [])
 
    # States with strong non-compete restrictions
    non_compete_restricted_states = {
        "CA", "ND", "MN", "OK", "DC"
    }
 
    for clause in clauses:
        # Elevate non-compete severity in restricted states
        severity = clause.get("severity", "MEDIUM")
 
        if (clause.get("clause_type") == "non_compete"
                and employee_state.upper() in non_compete_restricted_states):
            severity = "HIGH"
            clause["recommendation"] = (
                f"Non-compete clauses are generally unenforceable in {employee_state}. "
                "Recommend removal or limiting to trade secret protection only."
            )
 
        risks.append(ContractRisk(
            clause_type=clause.get("clause_type"),
            severity=severity,
            issue=clause.get("issue_description"),
            clause_text=clause.get("original_text", ""),
            plain_english=clause.get("plain_english_explanation", ""),
            recommendation=clause.get("recommendation", ""),
            state_specific=employee_state.upper() in non_compete_restricted_states
        ))
 
    # Sort by severity
    severity_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
    risks.sort(key=lambda r: severity_order.get(r.severity, 3))
 
    return risks

from fastapi import FastAPI, UploadFile, File, Form
import tempfile
import os
 
app = FastAPI()
 
 
@app.post("/api/contracts/review")
async def review_employment_contract(
    contract_file: UploadFile = File(...),
    employee_name: str = Form(...),
    employee_state: str = Form(...),
    contract_type: str = Form(default="employment_agreement"),
    position: str = Form(...)
):
    """
    HR portal endpoint for employment contract review.
    Returns risk summary and plain-English explanation.
    """
    with tempfile.NamedTemporaryFile(
        delete=False,
        suffix=f".{contract_file.filename.split('.')[-1]}"
    ) as tmp:
        tmp.write(await contract_file.read())
        tmp_path = tmp.name
 
    try:
        analysis = await analyze_employment_contract(
            contract_path=tmp_path,
            employee_state=employee_state,
            contract_type=contract_type
        )
 
        key_terms = await extract_key_terms(tmp_path)
        risks = classify_employment_risks(analysis, employee_state)
 
        high_risks = [r for r in risks if r.severity == "HIGH"]
        medium_risks = [r for r in risks if r.severity == "MEDIUM"]
 
        # Determine overall status
        if high_risks:
            review_status = "LEGAL_REVIEW_REQUIRED"
        elif medium_risks:
            review_status = "HR_REVIEW_RECOMMENDED"
        else:
            review_status = "APPROVED_STANDARD"
 
        return {
            "employee": employee_name,
            "position": position,
            "contract_type": contract_type,
            "employee_state": employee_state,
            "review_status": review_status,
            "overall_risk_score": analysis.get("overall_risk_score"),
            "key_terms": key_terms.get("extracted_data", {}),
            "risk_summary": {
                "high_risk_clauses": len(high_risks),
                "medium_risk_clauses": len(medium_risks),
                "low_risk_clauses": len([r for r in risks if r.severity == "LOW"])
            },
            "flagged_clauses": [
                {
                    "type": r.clause_type,
                    "severity": r.severity,
                    "issue": r.issue,
                    "plain_english": r.plain_english,
                    "recommendation": r.recommendation,
                    "state_specific_flag": r.state_specific
                }
                for r in risks
                if r.severity in ("HIGH", "MEDIUM")
            ],
            "executive_summary": analysis.get("executive_summary"),
            "reviewed_at": get_utc_timestamp()
        }
 
    finally:
        os.unlink(tmp_path)

async def compare_against_standard_template(
    submitted_contract_path: str,
    standard_template_path: str
) -> dict:
    """
    Compare a submitted contract against your company's standard template.
    Flags deviations that were not in the approved template.
    """
    with open(submitted_contract_path, "rb") as f:
        submitted = base64.b64encode(f.read()).decode()
 
    with open(standard_template_path, "rb") as f:
        standard = base64.b64encode(f.read()).decode()
 
    async with httpx.AsyncClient(timeout=45.0) as client:
        response = await client.post(
            f"{LEGALGUARD_BASE_URL}/compare",
            headers={
                "X-RapidAPI-Key": os.getenv("LEGALGUARD_API_KEY"),
                "Content-Type": "application/json"
            },
            json={
                "document_a": submitted,
                "document_a_type": "pdf",
                "document_b": standard,
                "document_b_type": "pdf",
                "comparison_mode": "clause_by_clause",
                "flag_additions": True,
                "flag_modifications": True,
                "flag_deletions": True
            }
        )
        response.raise_for_status()
        return response.json()

import asyncio
import glob
 
async def batch_review_contracts(
    contracts_directory: str,
    employee_state: str,
    contract_type: str = "employment_agreement"
) -> dict:
    """
    Review all contracts in a directory — useful during seasonal hiring sprints.
    """
    contract_files = glob.glob(f"{contracts_directory}/*.pdf")
 
    tasks = [
        analyze_employment_contract(
            contract_path=path,
            employee_state=employee_state,
            contract_type=contract_type
        )
        for path in contract_files
    ]
 
    results = await asyncio.gather(*tasks)
 
    high_risk_contracts = [
        {"file": contract_files[i], "result": results[i]}
        for i, r in enumerate(results)
        if r.get("overall_risk_score", 0) >= 0.7
    ]
 
    return {
        "total_contracts": len(results),
        "high_risk": len(high_risk_contracts),
        "requires_legal_review": high_risk_contracts,
        "clean_contracts": len(results) - len(high_risk_contracts)
    }