LangGraph & CrewAI Contract Agents: Multi-Agent Signing (2026)

from typing import TypedDict, List, Literal, Optional
from pydantic import BaseModel, Field
import os
import requests
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

# ------------------------------------------------------------------
# 1. State Definition
# ------------------------------------------------------------------
class ContractState(TypedDict):
    contract_type: str
    parties: dict  # {"sender": {"name": "...", "email": "..."}, "recipient": {...}}
    commercial_terms: dict
    draft_markdown: str
    risk_score: int
    risk_flags: List[str]
    revision_count: int
    human_approved: bool
    signbee_doc_id: Optional[str]
    signing_url: Optional[str]
    sha256_hash: Optional[str]
    status: str

# ------------------------------------------------------------------
# 2. Signbee Dispatch Execution Tool
# ------------------------------------------------------------------
def execute_signbee_dispatch(state: ContractState) -> dict:
    """Invokes Signbee API to compile Markdown and initiate cryptographic signing."""
    api_key = os.getenv("SIGNBEE_API_KEY")
    endpoint = "https://signb.ee/api/v1/send"
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "recipient_name": state["parties"]["recipient"]["name"],
        "recipient_email": state["parties"]["recipient"]["email"],
        "sender_name": state["parties"]["sender"]["name"],
        "sender_email": state["parties"]["sender"]["email"],
        "title": f"{state['contract_type']} - Execution Draft",
        "markdown": state["draft_markdown"],
        "expires_in_days": 14,
        "webhook_url": "https://api.enterprise.internal/webhooks/signbee"
    }
    
    response = requests.post(endpoint, json=payload, headers=headers, timeout=15)
    response.raise_for_status()
    data = response.json()
    
    # Extract Signbee document ID, status, and preliminary hash metadata
    return {
        "doc_id": data.get("id"),
        "signing_url": data.get("signing_url"),
        "status": data.get("status", "pending_recipient")
    }

# ------------------------------------------------------------------
# 3. Node Definitions
# ------------------------------------------------------------------
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

def drafter_node(state: ContractState) -> dict:
    """Generates or updates contract Markdown based on terms and critique feedback."""
    revision = state.get("revision_count", 0) + 1
    feedback = "\n".join(state.get("risk_flags", []))
    
    prompt = f"""You are an elite enterprise commercial contract counsel.
Draft an authoritative, highly professional B2B {state['contract_type']} in clean GitHub-flavored Markdown.
Parties:
- Provider / Sender: {state['parties']['sender']['name']} ({state['parties']['sender']['email']})
- Client / Recipient: {state['parties']['recipient']['name']} ({state['parties']['recipient']['email']})

Commercial Terms: {state['commercial_terms']}

Previous Critic Feedback (if any):
{feedback if feedback else 'None. First draft.'}

Requirements:
1. Clear section headings (1. Services, 2. Fees, 3. Liability, 4. IP, 5. Governing Law).
2. Mutual standard confidentiality and 12-month aggregate fee liability cap.
3. Explicit two-party signature block at the bottom using Markdown formatting.
"""
    response = llm.invoke([
        SystemMessage(content="You draft precise commercial legal agreements in Markdown."),
        HumanMessage(content=prompt)
    ])
    
    return {
        "draft_markdown": response.content,
        "revision_count": revision,
        "status": "drafted"
    }

class ComplianceAuditOutput(BaseModel):
    risk_score: int = Field(description="Risk score from 0 (safe) to 100 (critical risk)")
    risk_flags: List[str] = Field(description="List of detected policy violations or dangerous clauses")
    approved_without_escalation: bool = Field(description="True if score <= 20 and no critical redlines exist")

def compliance_critic_node(state: ContractState) -> dict:
    """Scrutinizes Markdown draft against corporate risk playbooks."""
    evaluator_llm = llm.with_structured_output(ComplianceAuditOutput)
    
    audit_prompt = f"""Audit this commercial agreement against corporate legal risk policy:
Agreement Type: {state['contract_type']}
Terms: {state['commercial_terms']}

Draft Content:
{state['draft_markdown']}

Policy Constraints:
- Aggregate liability MUST NOT exceed 12 months of paid fees or $100,000.
- Indemnification must be mutual and strictly limited to third-party IP claims.
- Governing law must be Delaware or England & Wales.
- No assignment of existing background IP.

Score the risk from 0-100 and identify any violations.
"""
    audit = evaluator_llm.invoke(audit_prompt)
    return {
        "risk_score": audit.risk_score,
        "risk_flags": audit.risk_flags,
        "status": "audited"
    }

def human_checkpoint_node(state: ContractState) -> dict:
    """Pauses graph execution for general counsel approval if risk exceeds tolerance."""
    print(f"\n⚠️ HIGH RISK DETECTED (Score: {state['risk_score']}). Pausing execution for human approval...")
    
    # LangGraph v0.2+ persistent interrupt primitive
    human_decision = interrupt({
        "question": "Do you approve dispatching this agreement to counterparty via Signbee?",
        "risk_score": state["risk_score"],
        "risk_flags": state["risk_flags"],
        "draft_preview": state["draft_markdown"][:500] + "... [truncated]"
    })
    
    return {
        "human_approved": human_decision.get("approved", False),
        "status": "human_approved" if human_decision.get("approved") else "rejected_by_human"
    }

def signbee_execution_node(state: ContractState) -> dict:
    """Final deterministic execution: sends Markdown to Signbee for SHA-256 sealed signing."""
    print("\n🚀 Initiating cryptographic execution via Signbee...")
    result = execute_signbee_dispatch(state)
    return {
        "signbee_doc_id": result["doc_id"],
        "signing_url": result["signing_url"],
        "status": "dispatched"
    }

# ------------------------------------------------------------------
# 4. Conditional Edge Routing
# ------------------------------------------------------------------
def evaluate_next_step(state: ContractState) -> Literal["drafter_node", "human_checkpoint_node", "signbee_execution_node", "__end__"]:
    # If drafted and revised more than 3 times without consensus, halt
    if state["revision_count"] > 3 and state["risk_score"] > 20:
        return "human_checkpoint_node"
    
    # If critic found high risk (> 20), trigger redline revision or human gate
    if state["risk_score"] > 20:
        if state["revision_count"] < 2:
            return "drafter_node"  # Let drafter try fixing critique flags
        return "human_checkpoint_node"
    
    return "signbee_execution_node"

def evaluate_human_gate(state: ContractState) -> Literal["signbee_execution_node", "__end__"]:
    if state.get("human_approved"):
        return "signbee_execution_node"
    return END

# ------------------------------------------------------------------
# 5. Graph Compilation
# ------------------------------------------------------------------
workflow = StateGraph(ContractState)

workflow.add_node("drafter_node", drafter_node)
workflow.add_node("compliance_critic_node", compliance_critic_node)
workflow.add_node("human_checkpoint_node", human_checkpoint_node)
workflow.add_node("signbee_execution_node", signbee_execution_node)

workflow.set_entry_point("drafter_node")
workflow.add_edge("drafter_node", "compliance_critic_node")
workflow.add_conditional_edges("compliance_critic_node", evaluate_next_step)
workflow.add_conditional_edges("human_checkpoint_node", evaluate_human_gate)
workflow.add_edge("signbee_execution_node", END)

checkpointer = MemorySaver()
contract_app = workflow.compile(checkpointer=checkpointer)

from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import requests
import os

# ------------------------------------------------------------------
# 1. Custom CrewAI Signbee Tool
# ------------------------------------------------------------------
class SignbeeInputSchema(BaseModel):
    title: str = Field(..., description="The official legal title of the agreement.")
    recipient_name: str = Field(..., description="Full legal name of the counterparty signatory.")
    recipient_email: str = Field(..., description="Verified email address of the counterparty signatory.")
    markdown_content: str = Field(..., description="The complete, finalized agreement formatted in Markdown.")

class SignbeeDispatchTool(BaseTool):
    name: str = "signbee_dispatch_tool"
    description: str = (
        "Dispatches a finalized, legally audited Markdown contract to Signbee. "
        "Signbee compiles the Markdown into an immutable PDF/A, creates a cryptographic "
        "SHA-256 seal, and delivers electronic signature links to all parties."
    )
    args_schema: type[BaseModel] = SignbeeInputSchema

    def _run(self, title: str, recipient_name: str, recipient_email: str, markdown_content: str) -> str:
        api_key = os.getenv("SIGNBEE_API_KEY")
        if not api_key:
            return "Error: Missing SIGNBEE_API_KEY environment variable."

        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "title": title,
            "recipient_name": recipient_name,
            "recipient_email": recipient_email,
            "sender_name": "Acme Legal Ops",
            "sender_email": "legal-ops@acmecorp.com",
            "markdown": markdown_content,
            "expires_in_days": 7
        }

        try:
            res = requests.post("https://signb.ee/api/v1/send", json=payload, headers=headers, timeout=15)
            res.raise_for_status()
            data = res.json()
            return f"Success! Document ID: {data.get('id')} - Signing URL: {data.get('signing_url')} - Status: {data.get('status')}"
        except Exception as e:
            return f"Signbee API Error: {str(e)}"

# ------------------------------------------------------------------
# 2. Agent Persona Definitions
# ------------------------------------------------------------------
contract_drafter = Agent(
    role="Senior Commercial Counsel & Contract Architect",
    goal="Draft balanced, comprehensive, and legally robust B2B agreements from raw deal parameters.",
    backstory=(
        "With 15 years of experience in enterprise technology licensing, you draft crystal-clear "
        "agreements adhering to modern commercial norms. You write in pristine Markdown."
    ),
    verbose=True,
    memory=True
)

compliance_officer = Agent(
    role="Enterprise Risk & Regulatory Auditor",
    goal="Identify non-standard clauses, liability exposure, and regulatory compliance gaps in commercial drafts.",
    backstory=(
        "Former GC at a Fortune 500 SaaS company. You meticulously examine indemnification scopes, "
        "ensure liability is capped at 12 months of fees, and verify intellectual property protections."
    ),
    verbose=True,
    memory=True
)

signee_clerk = Agent(
    role="Digital Execution & Escrow Dispatcher",
    goal="Verify that compliance audits are passed, format the final document, and invoke Signbee execution tools.",
    backstory=(
        "A precision-focused legal operations specialist who ensures that only 100% compliant contracts "
        "reach the execution phase. You hold the cryptographic signing keys."
    ),
    tools=[SignbeeDispatchTool()],
    verbose=True
)

# ------------------------------------------------------------------
# 3. Task Pipeline Definition
# ------------------------------------------------------------------
drafting_task = Task(
    description=(
        "Draft a Master Services Agreement (MSA) between Acme Corp (Provider) and Horizon Labs (Client). "
        "Scope: Cloud infrastructure deployment and maintenance. Monthly retainer: $12,500. "
        "Output the complete agreement in clean, valid Markdown with formal section numbering."
    ),
    expected_output="Complete MSA agreement text formatted in Markdown.",
    agent=contract_drafter
)

compliance_task = Task(
    description=(
        "Review the drafted MSA from the drafting task. Validate that: "
        "1. Aggregate liability is capped at exactly 12 months of paid fees ($150,000 max). "
        "2. Confidentiality is mutual and survives 3 years. "
        "3. Governing law is Delaware. "
        "If satisfactory, produce the finalized Markdown draft with an approved audit header."
    ),
    expected_output="Audited and approved Markdown agreement ready for execution.",
    agent=compliance_officer,
    context=[drafting_task]
)

execution_task = Task(
    description=(
        "Take the audited Markdown agreement approved by the compliance officer. "
        "Invoke the signbee_dispatch_tool with title 'Acme Corp & Horizon Labs - Master Services Agreement', "
        "recipient_name 'Sarah Jenkins', recipient_email 'sjenkins@horizonlabs.ai', and the finalized markdown."
    ),
    expected_output="Confirmation of dispatch including Signbee Document ID and signing link.",
    agent=signee_clerk,
    context=[compliance_task]
)

# ------------------------------------------------------------------
# 4. Crew Assembly & Execution
# ------------------------------------------------------------------
legal_crew = Crew(
    agents=[contract_drafter, compliance_officer, signee_clerk],
    tasks=[drafting_task, compliance_task, execution_task],
    process=Process.sequential,
    verbose=True
)

if __name__ == "__main__":
    result = legal_crew.kickoff()

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

app = FastAPI()
SIGNBEE_WEBHOOK_SECRET = os.getenv("SIGNBEE_WEBHOOK_SECRET", "whsec_sample_secret_key")

@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-SHA256 signature to prevent spoofing
    computed_signature = hmac.new(
        SIGNBEE_WEBHOOK_SECRET.encode("utf-8"),
        body_bytes,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(computed_signature, x_signbee_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid cryptographic webhook signature")

    payload = await request.json()
    event = payload.get("event")
    
    if event == "document.signed":
        doc_data = payload.get("data", {})
        doc_id = doc_data.get("id")
        sha256_hash = doc_data.get("sha256")
        pdf_url = doc_data.get("completed_pdf_url")
        
        print(f"✅ Document {doc_id} successfully signed by all parties!")
        print(f"🔒 SHA-256 Digest: {sha256_hash}")
        print(f"📄 Downloadable Signed PDF: {pdf_url}")
        
        # Trigger autonomous post-contract downstream actions:
        # - Provision tenant database in AWS/GCP
        # - Issue OAuth2 credentials
        # - Dispatch first retainer invoice
        
    return {"status": "processed"}