How to Automate Legal Compliance Monitoring

Automating legal compliance monitoring for regulatory changes catches the updates that affect your clients before they become problems. I built this for a firm that serves businesses in regulated industries. Instead of manually checking government websites for rule changes, the system monitors sources daily, uses AI to assess relevance, and alerts the right attorney when something matters.

What You Need Before Starting

• Python 3.8+
• Anthropic API key
• Web scraping capability (requests + BeautifulSoup)
• A list of regulatory sources to monitor

Step 1: Define Monitoring Sources

COMPLIANCE_SOURCES = [
    {
        "name": "Federal Register",
        "url": "https://www.federalregister.gov/api/v1/documents.json?conditions[type][]=RULE&per_page=20",
        "type": "api",
        "practice_areas": ["business_law", "employment", "finance"]
    },
    {
        "name": "State Legislature",
        "url": "https://your-state-legislature.gov/bills/recent",
        "type": "scrape",
        "practice_areas": ["business_law", "family_law"]
    }
]

Step 2: Build the Change Detector

import requests
import sqlite3
from datetime import datetime
from bs4 import BeautifulSoup
import hashlib

def init_compliance_db(db_path="compliance_monitor.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS regulatory_changes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source TEXT,
            title TEXT,
            url TEXT,
            content_hash TEXT UNIQUE,
            summary TEXT,
            relevance_score INTEGER,
            practice_areas TEXT,
            status TEXT DEFAULT 'new',
            detected_at TEXT
        )
    """)
    conn.commit()
    conn.close()

def fetch_regulatory_updates(source):
    updates = []
    
    if source["type"] == "api":
        response = requests.get(source["url"], timeout=30)
        if response.status_code == 200:
            data = response.json()
            for result in data.get("results", []):
                updates.append({
                    "title": result.get("title", ""),
                    "url": result.get("html_url", ""),
                    "content": result.get("abstract", ""),
                    "source": source["name"]
                })
    
    elif source["type"] == "scrape":
        try:
            response = requests.get(source["url"], timeout=30)
            soup = BeautifulSoup(response.text, "html.parser")
            for item in soup.select("article, .bill-item, .update-entry")[:20]:
                title_el = item.select_one("h2, h3, .title")
                link_el = item.select_one("a")
                updates.append({
                    "title": title_el.text.strip() if title_el else "",
                    "url": link_el["href"] if link_el else "",
                    "content": item.text.strip()[:500],
                    "source": source["name"]
                })
        except Exception as e:
            print(f"Error scraping {source['name']}: {e}")
    
    return updates

Step 3: AI Relevance Assessment

import anthropic
from dotenv import load_dotenv
load_dotenv()

def assess_relevance(update, practice_areas):
    client = anthropic.Anthropic()
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=512,
        system="""Assess this regulatory update for a law firm. Return JSON with:
- relevance_score: 1-10 (10 = directly impacts current clients)
- affected_areas: list of practice areas affected
- summary: 2-3 sentence plain English summary of what changed
- action_needed: what the firm should do (notify clients, update templates, review contracts, nothing)
- urgency: high/medium/low""",
        messages=[{
            "role": "user",
            "content": f"Practice areas we serve: {', '.join(practice_areas)}\n\nRegulatory update:\nTitle: {update['title']}\nContent: {update['content'][:2000]}"
        }]
    )
    
    import json
    try:
        return json.loads(response.content[0].text)
    except json.JSONDecodeError:
        return {"relevance_score": 3, "summary": response.content[0].text, "urgency": "low"}

Step 4: Process and Store Updates

def process_all_sources(db_path="compliance_monitor.db"):
    conn = sqlite3.connect(db_path)
    new_count = 0
    relevant_count = 0
    
    for source in COMPLIANCE_SOURCES:
        updates = fetch_regulatory_updates(source)
        
        for update in updates:
            content_hash = hashlib.md5(f"{update['title']}{update['url']}".encode()).hexdigest()
            
            existing = conn.execute("SELECT id FROM regulatory_changes WHERE content_hash = ?", (content_hash,)).fetchone()
            if existing:
                continue
            
            assessment = assess_relevance(update, source["practice_areas"])
            
            conn.execute("""
                INSERT OR IGNORE INTO regulatory_changes (source, title, url, content_hash, summary,
                    relevance_score, practice_areas, detected_at)
                VALUES (?,?,?,?,?,?,?,?)
            """, (update["source"], update["title"], update["url"], content_hash,
                  assessment.get("summary", ""), assessment.get("relevance_score", 0),
                  ",".join(assessment.get("affected_areas", [])), datetime.utcnow().isoformat()))
            
            new_count += 1
            if assessment.get("relevance_score", 0) >= 7:
                relevant_count += 1
                notify_compliance_team(update, assessment)
    
    conn.commit()
    conn.close()
    return {"new": new_count, "relevant": relevant_count}

def notify_compliance_team(update, assessment):
    message = f"Compliance Alert [{assessment.get('urgency', 'medium').upper()}]\n"
    message += f"Source: {update['source']}\n"
    message += f"Title: {update['title']}\n"
    message += f"Summary: {assessment.get('summary', 'N/A')}\n"
    message += f"Action: {assessment.get('action_needed', 'Review')}\n"
    message += f"Score: {assessment.get('relevance_score', 'N/A')}/10"
    
    requests.post("YOUR_SLACK_WEBHOOK", json={"text": message})

Step 5: Schedule Daily Monitoring

0 7 * * * python3 /path/to/compliance_monitor.py

if __name__ == "__main__":
    init_compliance_db()
    result = process_all_sources()
    print(f"Processed: {result['new']} new updates, {result['relevant']} relevant")

What to Build Next

Add client-specific monitoring. Map each client to the regulations that affect their industry, and only alert the responsible attorney about changes relevant to their specific clients.

Related Reading

• AI in Legal and Compliance - compliance automation
• Why Monitoring Is Not Optional - monitoring principles
• How to Build Automated Alerts That Actually Help - alert design