AI Agentic Contract Signing Workflows: How Autonomous Agents Execute Agreements (2026)

langchain>=0.2.0
langchain-openai>=0.1.0
pydantic>=2.0.0
requests>=2.31.0
fastapi>=0.110.0
uvicorn>=0.28.0

import os
import hmac
import hashlib
import json
import requests
from typing import List, Optional
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# Configuration & Credentials
SIGNBEE_API_KEY = os.getenv("SIGNBEE_API_KEY", "sb_live_secret_key_example")
SIGNBEE_API_URL = "https://signb.ee/api/v1/documents"
MAX_AUTONOMOUS_BUDGET_USD = 10000.0

# ---------------------------------------------------------------------------
# 1. Pydantic Models for Structured Agent Tool Arguments
# ---------------------------------------------------------------------------

class Signer(BaseModel):
    name: str = Field(description="Full legal name of the signing individual")
    email: str = Field(description="Verified email address of the signer")
    role: str = Field(description="Role or title of the signer, e.g. Client or Service Provider")

class CreateContractInput(BaseModel):
    title: str = Field(description="Descriptive title of the agreement")
    markdown_content: str = Field(description="Complete contract body formatted in valid Markdown")
    signers: List[Signer] = Field(description="List of required contract signers")
    contract_value_usd: float = Field(description="Total financial value of the contract in USD")
    webhook_url: Optional[str] = Field(default=None, description="Callback URL for signature status updates")

# ---------------------------------------------------------------------------
# 2. Custom LangChain Tool for Signbee Execution
# ---------------------------------------------------------------------------

@tool("dispatch_signbee_contract", args_schema=CreateContractInput)
def dispatch_signbee_contract(
    title: str,
    markdown_content: str,
    signers: List[Signer],
    contract_value_usd: float,
    webhook_url: Optional[str] = None
) -> str:
    """
    Evaluates safety thresholds and dispatches a Markdown document for legally binding 
    e-signature via the Signbee REST API.
    """
    # Safety Check: Budget Threshold Enforcement
    if contract_value_usd > MAX_AUTONOMOUS_BUDGET_USD:
        return json.dumps({
            "status": "ESCALATED_TO_HUMAN",
            "reason": f"Contract value ${contract_value_usd:,.2f} exceeds max autonomous threshold of ${MAX_AUTONOMOUS_BUDGET_USD:,.2f}.",
            "action_required": "Routing contract draft to Human Admin review portal."
        })

    # Prepare Signbee API Payload
    payload = {
        "title": title,
        "markdown": markdown_content,
        "signers": [signer.model_dump() for signer in signers],
        "webhook_url": webhook_url or "https://api.yourcompany.com/webhooks/signbee"
    }

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

    try:
        response = requests.post(SIGNBEE_API_URL, json=payload, headers=headers, timeout=15)
        response.raise_for_status()
        data = response.json()
        
        return json.dumps({
            "status": "SUCCESS",
            "document_id": data.get("id"),
            "signing_url": data.get("signing_url"),
            "sha256_checksum": data.get("sha256"),
            "message": "Contract compiled and signature requests dispatched successfully."
        })
    except requests.exceptions.RequestException as e:
        return json.dumps({
            "status": "ERROR",
            "error_details": str(e),
            "response_body": e.response.text if hasattr(e, 'response') and e.response else None
        })

# ---------------------------------------------------------------------------
# 3. Agent Execution Setup
# ---------------------------------------------------------------------------

def run_procurement_agent():
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    tools = [dispatch_signbee_contract]

    prompt = ChatPromptTemplate.from_messages([
        ("system", 
         "You are an autonomous Procurement Agent for Acme Corp. "
         "Your goal is to draft a Master Services Agreement (MSA) for a vendor, "
         "format it in clear Markdown, and execute it using the dispatch_signbee_contract tool. "
         "Always ensure the contract includes clear scope, payment terms, legal governing law, "
         "and proper signature blocks for both parties."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])

    agent = create_tool_calling_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    task_description = (
        "Draft and dispatch an MSA with 'Apex AI Consulting' for custom LLM fine-tuning services. "
        "The project fee is $7,500 USD with a completion deadline of August 30, 2026. "
        "Signer 1: Jane Doe (Client, jane@acmecorp.com). "
        "Signer 2: Alex Vance (Vendor, alex@apexai.io)."
    )

    print("--- Starting Autonomous Agent Contract Workflow ---")
    result = agent_executor.invoke({"input": task_description})
    print("
--- Final Agent Execution Output ---")
    print(result["output"])

if __name__ == "__main__":
    run_procurement_agent()

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

app = FastAPI()
WEBHOOK_SECRET = "whsec_example_secret_key_123"

@app.post("/webhooks/signbee")
async def handle_signbee_webhook(
    request: Request,
    x_signbee_signature: str = Header(None)
):
    body = await request.body()
    
    # 1. Cryptographic HMAC Verification of Webhook Payload
    computed_sig = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        body,
        hashlib.sha256
    ).hexdigest()

    if not x_signbee_signature or not hmac.compare_digest(computed_sig, x_signbee_signature):
        raise HTTPException(status_code=401, detail="Invalid webhook signature signature failure")

    event = json.loads(body)
    event_type = event.get("type")
    payload = event.get("data", {})

    # 2. Process Contract Completion Event
    if event_type == "document.completed":
        doc_id = payload.get("id")
        sha256_hash = payload.get("sha256")
        pdf_url = payload.get("completed_pdf_url")
        
        print(f"Contract Completed! ID: {doc_id}")
        print(f"SHA-256 Audit Digest: {sha256_hash}")
        print(f"Download Certified PDF: {pdf_url}")
        
        # Trigger internal automated downstream actions (e.g. release escrow payment)
        
    return {"status": "received"}