Build an Agentic Legal Contract Review Workflow with Obligation Extraction & Risk Scoring

Contract review is the bread and butter of legal practice, but it's also the biggest bottleneck. A standard commercial contract takes 4-6 hours of attorney time to review, and even experienced lawyers miss critical obligations buried in boilerplate. This changes everything.

By late 2026, agentic contract review systems can analyze a 50-page contract in 45 seconds, extracting every obligation, flagging every risky clause, and producing a structured risk report — at 1/50th the cost of human review.

This tutorial builds a complete contract review pipeline using LangGraph for orchestration and FastMCP for legal document processing.

The Contract Review Architecture

Four specialized agents handle the review lifecycle:

1. Document Parser Agent — Ingests contracts in PDF/DOCX format, identifies sections (definitions, obligations, indemnification, termination, governing law), and structures the document for downstream analysis.

2. Obligation Extractor Agent — Identifies every obligation, deadline, and commitment in the contract. Extracts who must do what, by when, and what happens if they don't.

3. Risk Scoring Agent — Compares each clause against a library of known risky patterns (unlimited liability, broad indemnification, overly aggressive IP assignment, non-standard governing law) and assigns risk scores.

4. Report Generator Agent — Produces a structured risk report with executive summary, obligation timeline, flagged clauses with recommended alternatives, and an overall risk score.

Implementation

Step 1: Document Parsing

from langgraph.graph import StateGraph
from typing import TypedDict, List, Optional, Dict


class ContractState(TypedDict):
    contract_id: str
    document_text: str
    sections: Optional[Dict[str, str]]
    obligations: Optional[List[dict]]
    risk_clauses: Optional[List[dict]]
    risk_score: Optional[float]
    report: Optional[str]
    status: str


async def parse_document(state: ContractState) -> ContractState:
    """Parse contract into structured sections."""
    text = state['document_text']

    # Use LLM to identify and segment contract sections
    sections_prompt = f"""Parse this legal contract into sections.
    Identify: Definitions, Obligations, Payment Terms, Indemnification,
    Limitation of Liability, IP Rights, Termination, Governing Law,
    Confidentiality, and any other relevant sections.

    For each section, provide the section name and full text.

    Contract:
    {text[:15000]}"""

    sections = await llm_structured_call(
        sections_prompt,
        output_schema={"sections": {"name": str, "text": str}}
    )

    return {
        **state,
        'sections': {s['name']: s['text'] for s in sections['sections']},
        'status': 'parsed'
    }

Step 2: Obligation Extraction

async def extract_obligations(state: ContractState) -> ContractState:
    """Extract all obligations, deadlines, and commitments."""
    obligations = []

    for section_name, section_text in state['sections'].items():
        extraction_prompt = f"""Extract all obligations from this contract section.
        For each obligation, identify:
        - party (who must perform)
        - action (what must be done)
        - deadline (when, if specified)
        - condition (triggers, if any)
        - consequence (what happens on breach)
        - section_ref (which section)

        Section: {section_name}
        Text: {section_text}"""

        extracted = await llm_structured_call(
            extraction_prompt,
            output_schema={
                "obligations": {
                    "party": str,
                    "action": str,
                    "deadline": str,
                    "condition": str,
                    "consequence": str,
                    "section_ref": str
                }
            }
        )
        obligations.extend(extracted['obligations'])

    return {
        **state,
        'obligations': obligations,
        'status': 'obligations_extracted'
    }

Step 3: Risk Scoring

RISK_LIBRARY = {
    'unlimited_liability': {
        'pattern': 'no limitation on liability',
        'severity': 'critical',
        'alternative': 'Cap liability at total contract value'
    },
    'broad_indemnification': {
        'pattern': 'indemnify for all losses.*including.*consequential',
        'severity': 'high',
        'alternative': 'Limit indemnification to direct damages only'
    },
    'ip_overreach': {
        'pattern': 'all intellectual property.*work.product.*assign',
        'severity': 'high',
        'alternative': 'Limit IP assignment to specific deliverables'
    },
    'non_standard_governing_law': {
        'pattern': 'governing law.*jurisdiction.*outside.*\b(home|default)\b',
        'severity': 'medium',
        'alternative': 'Negotiate governing law to home jurisdiction'
    }
}

async def score_risk(state: ContractState) -> ContractState:
    """Score each clause against the risk library."""
    risk_clauses = []

    for clause_text in state['sections'].values():
        for risk_type, risk_def in RISK_LIBRARY.items():
            if await check_risk_match(clause_text, risk_def['pattern']):
                risk_clauses.append({
                    'clause': clause_text[:200],
                    'risk_type': risk_type,
                    'severity': risk_def['severity'],
                    'alternative': risk_def['alternative']
                })

    # Calculate overall risk score (0-100)
    severity_weights = {'critical': 30, 'high': 15, 'medium': 5}
    risk_score = sum(
        severity_weights.get(r['severity'], 0) for r in risk_clauses
    )
    risk_score = min(100, risk_score)

    return {
        **state,
        'risk_clauses': risk_clauses,
        'risk_score': risk_score,
        'status': 'risk_scored'
    }

Step 4: Report Generation

async def generate_report(state: ContractState) -> ContractState:
    """Generate structured risk report."""
    report_prompt = f"""Generate a contract risk report:

    Obligations ({len(state['obligations'])} found):
    {format_obligations(state['obligations'])}

    Risk Clauses ({len(state['risk_clauses'])} flagged):
    {format_risks(state['risk_clauses'])}

    Overall Risk Score: {state['risk_score']}/100

    Produce:
    1. Executive Summary (3-4 sentences)
    2. Obligation Timeline (deadline-based)
    3. Risk Assessment by Severity
    4. Recommended Actions
    5. Negotiation Recommendations for flagged clauses"""

    report = await llm_call(report_prompt)
    return {**state, 'report': report, 'status': 'complete'}

The Complete Graph

graph = StateGraph(ContractState)
graph.add_node('parse', parse_document)
graph.add_node('obligations', extract_obligations)
graph.add_node('risks', score_risk)
graph.add_node('report', generate_report)

graph.set_entry_point('parse')
graph.add_edge('parse', 'obligations')
graph.add_edge('obligations', 'risks')
graph.add_edge('risks', 'report')
graph.add_edge('report', END)

contract_review = graph.compile()

Integration with Legal Practice

The FastMCP server exposes contract review tools to legal practice management systems. Attorneys can review a contract through a web interface that shows the AI-generated report side-by-side with the original document, with clickable risk flags that link to the relevant clause.

Key Metrics

Built by Deepak Bagada at DailyAIWorld.com. This workflow is part of our AI Workflows series — practical, buildable agent pipelines for real business problems.