Autonomous Legal AI Agents: Negotiating & Signing in 2026

from typing import TypedDict, List, Optional, Dict, Any
from pydantic import BaseModel, Field

class ClauseAnalysis(BaseModel):
    clause_title: str
    original_text: str
    risk_level: str = Field(description="LOW, MEDIUM, HIGH, or CRITICAL")
    suggested_redline: Optional[str] = None
    rationale: Optional[str] = None

class ContractEvaluation(BaseModel):
    contract_title: str
    contract_value_usd: float
    governing_law: str
    liability_cap_multiplier: float
    high_risk_clauses: List[ClauseAnalysis]
    overall_risk_score: int = Field(description="0 to 100 scale")
    requires_human_signoff: bool

class LegalAgentState(TypedDict):
    raw_contract_md: str
    counterparty_email: str
    counterparty_name: str
    principal_email: str
    evaluation: Optional[ContractEvaluation]
    redlined_contract_md: Optional[str]
    human_approved: bool
    signbee_document_id: Optional[str]
    signing_url: Optional[str]
    execution_status: str

import json
from typing import Dict, Any
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage
from agent_state import LegalAgentState, ContractEvaluation

llm = ChatAnthropic(model="claude-3-7-sonnet-20250219", temperature=0.0)

LEGAL_PLAYBOOK = """
CORPORATE LEGAL PLAYBOOK POLICIES (2026):
1. Governing Law: Must be Delaware, New York, or England & Wales. Reject others.
2. Liability Cap: Must not exceed 12 months of contract fees (1.0x multiplier).
3. Indemnification: Must be mutual. Uncapped IP indemnification is STRICTLY FORBIDDEN.
4. Payment Terms: Maximum Net 30 days. Reject Net 60/90.
5. Max Auto-Approval Threshold: $25,000 USD and Risk Score <= 25.
"""

def evaluate_contract_node(state: LegalAgentState) -> Dict[str, Any]:
    """Analyzes the contract against company legal playbook."""
    structured_evaluator = llm.with_structured_output(ContractEvaluation)
    
    prompt = f"""
    You are an autonomous corporate legal counsel agent.
    Analyze the following Markdown contract against our company playbook.
    
    PLAYBOOK:
    {LEGAL_PLAYBOOK}
    
    INBOUND CONTRACT:
    {state['raw_contract_md']}
    """
    
    evaluation: ContractEvaluation = structured_evaluator.invoke([
        SystemMessage(content="You are an expert enterprise legal AI agent."),
        HumanMessage(content=prompt)
    ])
    
    # Deterministic risk check: enforce hard thresholds
    if evaluation.contract_value_usd > 25000 or evaluation.overall_risk_score > 25:
        evaluation.requires_human_signoff = True
        
    return {"evaluation": evaluation}

def generate_redlines_node(state: LegalAgentState) -> Dict[str, Any]:
    """Generates standardized redline contract if playbook deviations exist."""
    eval_data = state["evaluation"]
    if not eval_data or not eval_data.high_risk_clauses:
        return {"redlined_contract_md": state["raw_contract_md"]}
    
    prompt = f"""
    The following contract has non-compliant clauses. Replace non-compliant terms
    with our standard playbook fallback clauses while preserving all other terms.
    
    ORIGINAL CONTRACT:
    {state['raw_contract_md']}
    
    ISSUES DETECTED:
    {json.dumps([c.dict() for c in eval_data.high_risk_clauses], indent=2)}
    
    Output ONLY the updated, fully redlined contract in clean Markdown format.
    """
    
    response = llm.invoke([
        SystemMessage(content="You are an automated contract drafting engine. Output only valid Markdown."),
        HumanMessage(content=prompt)
    ])
    
    return {"redlined_contract_md": response.content}

import os
import requests
from typing import Dict, Any
from agent_state import LegalAgentState

SIGNBEE_API_KEY = os.environ.get("SIGNBEE_API_KEY")
SIGNBEE_API_URL = "https://api.signb.ee/v1/documents"

def execute_signbee_dispatch_node(state: LegalAgentState) -> Dict[str, Any]:
    """
    Sends the negotiated Markdown contract for legal e-signing in a single API call.
    Signbee automatically renders PDF/A, sets signature fields, and seals with SHA-256.
    """
    contract_content = state.get("redlined_contract_md") or state["raw_contract_md"]
    doc_title = state["evaluation"].contract_title if state.get("evaluation") else "Commercial Agreement"

    payload = {
        "title": doc_title,
        "content": contract_content,  # Clean semantic Markdown
        "recipients": [
            {
                "name": state["counterparty_name"],
                "email": state["counterparty_email"],
                "role": "signer",
                "authentication": "email_otp"
            },
            {
                "name": "Michael Beckett (via Legal AI Agent)",
                "email": state["principal_email"],
                "role": "signer",
                "authentication": "email_otp"
            }
        ],
        "metadata": {
            "agent_id": "legal-agent-v4",
            "risk_score": state["evaluation"].overall_risk_score if state.get("evaluation") else 0,
            "contract_value_usd": state["evaluation"].contract_value_usd if state.get("evaluation") else 0,
            "automated_execution": True
        }
    }

    headers = {
        "Authorization": f"Bearer {SIGNBEE_API_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(SIGNBEE_API_URL, json=payload, headers=headers)
    response.raise_for_status()
    result = response.json()

    return {
        "signbee_document_id": result["id"],
        "signing_url": result["recipients"][0]["signing_url"],
        "execution_status": "dispatched"
    }

from typing import Dict, Any
from langgraph.graph import StateGraph, END
from agent_state import LegalAgentState
from legal_nodes import evaluate_contract_node, generate_redlines_node
from signbee_executor import execute_signbee_dispatch_node

def route_after_evaluation(state: LegalAgentState) -> str:
    """Decides whether to redline, ask for human approval, or execute immediately."""
    evaluation = state["evaluation"]
    
    if evaluation.requires_human_signoff and not state.get("human_approved", False):
        return "human_approval_checkpoint"
    
    if evaluation.high_risk_clauses:
        return "generate_redlines"
        
    return "execute_signbee"

def human_approval_checkpoint(state: LegalAgentState) -> Dict[str, Any]:
    """Pauses execution until internal Slack/Teams webhook approves."""
    print(f"🚨 [HITL ALERT] Contract '{state['evaluation'].contract_title}' requires human review!")
    print(f"Risk Score: {state['evaluation'].overall_risk_score}/100 | Value: ${state['evaluation'].contract_value_usd:,.2f}")
    # In production, this emits a webhook to Slack with interactive 'Approve' / 'Reject' buttons
    return {"execution_status": "pending_human_review"}

# Build Workflow Graph
workflow = StateGraph(LegalAgentState)

workflow.add_node("evaluate", evaluate_contract_node)
workflow.add_node("generate_redlines", generate_redlines_node)
workflow.add_node("human_approval_checkpoint", human_approval_checkpoint)
workflow.add_node("execute_signbee", execute_signbee_dispatch_node)

workflow.set_entry_point("evaluate")

workflow.add_conditional_edges(
    "evaluate",
    route_after_evaluation,
    {
        "human_approval_checkpoint": "human_approval_checkpoint",
        "generate_redlines": "generate_redlines",
        "execute_signbee": "execute_signbee"
    }
)

workflow.add_edge("generate_redlines", "execute_signbee")
workflow.add_edge("execute_signbee", END)

legal_agent_app = workflow.compile()

import hmac
import hashlib
import os
from fastapi import FastAPI, Request, HTTPException, Header

app = FastAPI()
WEBHOOK_SECRET = os.environ.get("SIGNBEE_WEBHOOK_SECRET")

@app.post("/webhooks/signbee")
async def handle_signbee_webhook(
    request: Request,
    x_signbee_signature: str = Header(None)
):
    body_bytes = await request.body()
    
    # 1. Verify HMAC SHA-256 signature for security
    computed_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        body_bytes,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(computed_signature, x_signbee_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")
        
    payload = await request.json()
    event_type = payload.get("event")
    
    if event_type == "document.completed":
        data = payload["data"]
        doc_id = data["id"]
        sha256_digest = data["sha256"]
        certified_pdf_url = data["completed_pdf_url"]
        
        print(f"✅ Agreement finalized! Document ID: {doc_id}")
        print(f"🔒 SHA-256 Cryptographic Hash: {sha256_digest}")
        print(f"📄 Certified PDF Download: {certified_pdf_url}")
        
        # Trigger autonomous downstream actions
        # 1. Provision enterprise API keys
        # 2. Update CRM contract status to 'Signed'
        # 3. Release escrow payments via Stripe / MPP
        
    return {"status": "success"}