July 21, 2026 · Tutorial

FastAPI E-Signature Integration: Async Contracts & Webhooks in Python (2026)

Build production-ready, asynchronous e-signature workflows in Python using FastAPI, httpx, Pydantic, PostgreSQL audit logging, and secure HMAC-SHA256 webhook verification.

Michael Beckett

Founder, Signbee

TL;DR & ARCHITECTURE SUMMARY

This guide walks you through building an enterprise-grade, non-blocking Python e-signature backend with FastAPI. You will dispatch digital contracts via Signbee's async REST API using httpx.AsyncClient, validate requests with Pydantic v2, record immutable state changes in PostgreSQL via SQLAlchemy 2.0 Async, and implement cryptographically verified HMAC-SHA256 webhook listeners using hmac and hashlib.compare_digest().

Key Stack Components & Specifications (2026)

FrameworkFastAPI 0.110+

Async HTTP Clienthttpx 0.27+

Data ValidationPydantic v2

Async ORM / DBSQLAlchemy 2.0 + asyncpg

Webhook SecurityHMAC-SHA256

Auth MethodBearer API Key

Why Asynchronous E-Signatures Matter in Python Architecture

In modern API design, e-signature dispatching is inherently an I/O-bound operation. When an application prepares a non-disclosure agreement (NDA), master services agreement (MSA), or employment contract, it generates document markdown or metadata, formats recipient profiles, and makes outbound HTTP requests to an e-signature provider.

In traditional synchronous Python applications (such as Flask or WSGI-based Django applications relying on the requests library), each outbound HTTP network hop blocks the active OS worker thread. If the e-signature API takes 400 milliseconds to process the request, generate the cryptographic signing token, and issue notification emails, that worker thread remains idle and unavailable to handle incoming web traffic.

By leveraging FastAPI and httpx.AsyncClient, your application operates on an asynchronous event loop (based on Python's asyncio). While waiting for network I/O from Signbee's API, the event loop yields control, enabling a single FastAPI application instance to handle thousands of concurrent client requests, API webhook callbacks, and database queries simultaneously.

If you are looking for a basic synchronous Python implementation using the standard requests library or Flask, refer to our foundational guide on Python E-Signature API Integration.

FastAPI E-Signature Architecture Overview

The production-ready integration pattern detailed in this tutorial separates concerns across modular layers: request schemas, client service wrappers, database models, route handlers, and webhook signature verification logic.

End-to-End Async Data Flow

1.Client Request: Frontend or client app calls POST /api/contracts/send with JSON payload.
2.Pydantic Validation: FastAPI automatically validates inputs, recipient emails, and contract body.
3.Async API Call: SignbeeAsyncClient uses httpx to dispatch contract to Signbee API.
4.Database Persistence: Initial contract record and audit log are created asynchronously in PostgreSQL via asyncpg.
5.Webhook Event Callback: When recipient signs or views document, Signbee posts HMAC-SHA256 payload to POST /api/webhooks/signbee.
6.HMAC Verification: FastAPI computes digest using raw bytes and hashlib.compare_digest() before updating contract status in PostgreSQL.

Project Setup & Dependencies

Create a clean virtual environment and install the modern Python web stack dependencies:

Terminal — installationPython / FastAPI

python3 -m venv venv
source venv/bin/activate

pip install "fastapi>=0.110.0" \
            "uvicorn[standard]>=0.28.0" \
            "httpx>=0.27.0" \
            "pydantic>=2.6.0" \
            "pydantic-settings>=2.2.0" \
            "sqlalchemy[asyncio]>=2.0.28" \
            "asyncpg>=0.29.0"

Step 1: Application Settings & Configuration

Store sensitive API tokens and database connection credentials in environment variables. We use Pydantic's BaseSettings to guarantee type safety and automatic configuration parsing.

app/config.py — type-safe application configurationPython / FastAPI

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    SIGNBEE_API_KEY: str
    SIGNBEE_WEBHOOK_SECRET: str
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/contracts_db"
    SIGNBEE_BASE_URL: str = "https://signb.ee/api/v1"
    
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Step 2: Async PostgreSQL Database & Audit Trail Models

Auditability is a legal cornerstone of electronic signature compliance under regulations like ESIGN, eIDAS, and UETA. Every state transition—from initial creation to signature completion—must be immutably captured.

Here we define our database connection lifecycle using SQLAlchemy 2.0 Async Engine and asyncpg, along with relational ORM models for contracts and audit log entries:

app/database.py — async database session managementPython / FastAPI

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from app.config import get_settings

settings = get_settings()

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,
    pool_size=20,
    max_overflow=10,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Next, we establish the database schema mapping in app/models.py:

app/models.py — contract and audit trail ORM modelsPython / FastAPI

import uuid
from datetime import datetime
from typing import Optional, List
from sqlalchemy import String, Text, DateTime, ForeignKey, JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID
from app.database import Base

class Contract(Base):
    __tablename__ = "contracts"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    signbee_doc_id: Mapped[Optional[str]] = mapped_column(String(100), unique=True, index=True, nullable=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    recipient_name: Mapped[str] = mapped_column(String(255), nullable=False)
    recipient_email: Mapped[str] = mapped_column(String(255), index=True, nullable=False)
    status: Mapped[str] = mapped_column(String(50), default="pending", index=True)
    signing_url: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    markdown_content: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    audit_logs: Mapped[List["ContractAuditLog"]] = relationship("ContractAuditLog", back_populates="contract", cascade="all, delete-orphan")

class ContractAuditLog(Base):
    __tablename__ = "contract_audit_logs"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    contract_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("contracts.id"), index=True)
    event_type: Mapped[str] = mapped_column(String(100), nullable=False)
    description: Mapped[str] = mapped_column(Text, nullable=False)
    event_metadata: Mapped[dict] = mapped_column(JSON, default=dict)
    timestamp: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    contract: Mapped["Contract"] = relationship("Contract", back_populates="audit_logs")

Step 3: Pydantic Validation Schemas

Pydantic v2 ensures strict validation of incoming requests, API payloads, and webhook events before business logic is executed.

app/schemas.py — request and webhook validation schemasPython / FastAPI

from pydantic import BaseModel, EmailStr, Field
from typing import Optional, Dict, Any
from datetime import datetime

class ContractSendRequest(BaseModel):
    title: str = Field(..., min_length=3, max_length=255, description="Document title")
    recipient_name: str = Field(..., min_length=2, max_length=100)
    recipient_email: EmailStr
    markdown_content: str = Field(..., min_length=10, description="Markdown document text")

class ContractSendResponse(BaseModel):
    internal_contract_id: str
    signbee_doc_id: str
    status: str
    signing_url: str
    created_at: datetime

class WebhookEventData(BaseModel):
    document_id: str
    event: str  # e.g., document.sent, document.viewed, document.signed, document.declined
    recipient_email: EmailStr
    signed_pdf_url: Optional[str] = None
    certificate_id: Optional[str] = None
    timestamp: str
    metadata: Dict[str, Any] = Field(default_factory=dict)

Step 4: Asynchronous Signbee API Client Wrapper

To interface cleanly with Signbee's REST API, we build an asynchronous client class utilizing httpx.AsyncClient. This client handles connection pooling, request timeouts, and exception transformations.

app/services/signbee.py — async HTTP client wrapperPython / FastAPI

import logging
import httpx
from typing import Dict, Any
from app.config import get_settings

logger = logging.getLogger(__name__)

class SignbeeAPIException(Exception):
    def __init__(self, status_code: int, message: str):
        self.status_code = status_code
        self.message = message
        super().__init__(f"Signbee API Error [{status_code}]: {message}")

class SignbeeAsyncClient:
    def __init__(self):
        self.settings = get_settings()
        self.headers = {
            "Authorization": f"Bearer {self.settings.SIGNBEE_API_KEY}",
            "Content-Type": "application/json",
            "User-Agent": "FastAPI-Signbee-Integration/2026.1",
        }

    async def send_contract(self, title: str, recipient_name: str, recipient_email: str, markdown: str) -> Dict[str, Any]:
        """Dispatch a markdown document to Signbee's async send endpoint."""
        payload = {
            "title": title,
            "recipient_name": recipient_name,
            "recipient_email": recipient_email,
            "markdown": markdown,
        }
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            try:
                response = await client.post(
                    f"{self.settings.SIGNBEE_BASE_URL}/send",
                    json=payload,
                    headers=self.headers,
                )
                
                if response.status_code != 200:
                    logger.error(f"Signbee API error: {response.status_code} - {response.text}")
                    raise SignbeeAPIException(response.status_code, response.text)
                    
                return response.json()
            except httpx.RequestError as exc:
                logger.error(f"Network error contacting Signbee API: {exc}")
                raise SignbeeAPIException(503, "E-signature gateway unreachable")

Step 5: Contract Route Handler (`POST /api/contracts/send`)

Now we create the route handler for dispatching contracts. When called, it inserts a pending record in PostgreSQL, invokes the asynchronous Signbee client, saves the resulting signbee_doc_id and signing_url, and records an initial audit log entry.

app/routers/contracts.py — contract dispatch routerPython / FastAPI

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.schemas import ContractSendRequest, ContractSendResponse
from app.models import Contract, ContractAuditLog
from app.services.signbee import SignbeeAsyncClient, SignbeeAPIException

router = APIRouter(prefix="/api/contracts", tags=["Contracts"])
signbee_client = SignbeeAsyncClient()

@router.post("/send", response_model=ContractSendResponse, status_code=status.HTTP_201_CREATED)
async def send_contract(
    payload: ContractSendRequest,
    db: AsyncSession = Depends(get_db),
):
    # 1. Create local contract record
    contract = Contract(
        title=payload.title,
        recipient_name=payload.recipient_name,
        recipient_email=payload.recipient_email,
        markdown_content=payload.markdown_content,
        status="pending_dispatch",
    )
    db.add(contract)
    await db.flush()  # Generate contract.id UUID

    try:
        # 2. Dispatch to Signbee async API
        api_result = await signbee_client.send_contract(
            title=payload.title,
            recipient_name=payload.recipient_name,
            recipient_email=payload.recipient_email,
            markdown=payload.markdown_content,
        )

        # 3. Update contract status and URLs
        contract.signbee_doc_id = api_result["document_id"]
        contract.signing_url = api_result["signing_url"]
        contract.status = "sent"

        # 4. Append audit log
        audit = ContractAuditLog(
            contract_id=contract.id,
            event_type="contract.dispatched",
            description=f"Contract successfully dispatched to {payload.recipient_email}",
            event_metadata={"signbee_doc_id": api_result["document_id"]},
        )
        db.add(audit)
        
        return ContractSendResponse(
            internal_contract_id=str(contract.id),
            signbee_doc_id=api_result["document_id"],
            status=contract.status,
            signing_url=api_result["signing_url"],
            created_at=contract.created_at,
        )

    except SignbeeAPIException as exc:
        contract.status = "failed"
        audit = ContractAuditLog(
            contract_id=contract.id,
            event_type="contract.dispatch_failed",
            description=f"Dispatch failed: {exc.message}",
            event_metadata={"status_code": exc.status_code},
        )
        db.add(audit)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"Failed to dispatch document to e-signature service: {exc.message}",
        )

Step 6: Secure Webhook Handler with HMAC-SHA256 Signature Verification

When a recipient opens, signs, or declines a document, Signbee emits real-time webhook events to your application. To learn more about webhook event lifecycles and retry mechanisms, check out our guide on E-Signature API Webhook Events.

Critical Security Rule: Never trust incoming webhook payloads without verifying their cryptographic signature. Attackers could spoof HTTP requests to mark contracts as signed without legitimate signatures.

In FastAPI, signature verification requires reading the raw bytes of the request body (await request.body()) prior to parsing JSON. We compute an HMAC digest using Python's standard hmac and hashlib.sha256 modules, and perform constant-time comparison via hashlib.compare_digest() to prevent timing attacks.

app/routers/webhooks.py — HMAC verified webhook endpointPython / FastAPI

import hmac
import hashlib
import json
import logging
from fastapi import APIRouter, Request, Header, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.database import get_db
from app.config import get_settings
from app.models import Contract, ContractAuditLog
from app.schemas import WebhookEventData

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/webhooks", tags=["Webhooks"])
settings = get_settings()

def verify_hmac_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    """
    Verify incoming HMAC-SHA256 signature using constant-time comparison.
    """
    if not signature_header or not secret:
        return False
    
    # Strip optional prefix (e.g. 'sha256=')
    clean_signature = signature_header.replace("sha256=", "").strip()
    
    expected_hmac = hmac.new(
        key=secret.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256
    ).hexdigest()
    
    return hashlib.compare_digest(expected_hmac.lower(), clean_signature.lower())

@router.post("/signbee", status_code=status.HTTP_200_OK)
async def handle_signbee_webhook(
    request: Request,
    x_signbee_signature: str = Header(None, alias="X-Signbee-Signature"),
    db: AsyncSession = Depends(get_db),
):
    # 1. Obtain raw unparsed body bytes
    body_bytes = await request.body()
    
    # 2. Verify HMAC-SHA256 signature
    if not x_signbee_signature or not verify_hmac_signature(body_bytes, x_signbee_signature, settings.SIGNBEE_WEBHOOK_SECRET):
        logger.warning("Rejecting invalid or unverified webhook signature")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid HMAC signature",
        )

    # 3. Parse JSON payload into Pydantic model
    try:
        raw_json = json.loads(body_bytes.decode("utf-8"))
        event_data = WebhookEventData(**raw_json)
    except Exception as exc:
        logger.error(f"Failed to parse webhook JSON payload: {exc}")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Malformed JSON body",
        )

    # 4. Lookup target contract in PostgreSQL
    stmt = select(Contract).where(Contract.signbee_doc_id == event_data.document_id)
    result = await db.execute(stmt)
    contract = result.scalar_one_or_none()

    if not contract:
        logger.error(f"Webhook received for unknown document_id: {event_data.document_id}")
        return {"status": "ignored", "reason": "Contract not found"}

    # 5. Process state transition
    event_type = event_data.event
    if event_type == "document.signed":
        contract.status = "completed"
    elif event_type == "document.viewed":
        contract.status = "viewed"
    elif event_type == "document.declined":
        contract.status = "declined"

    # 6. Append immutable audit log
    audit_entry = ContractAuditLog(
        contract_id=contract.id,
        event_type=event_type,
        description=f"Webhook received event: {event_type} for recipient {event_data.recipient_email}",
        event_metadata=event_data.metadata,
    )
    db.add(audit_entry)

    logger.info(f"Successfully processed webhook event '{event_type}' for document {event_data.document_id}")
    return {"status": "success", "processed_event": event_type}

Once a contract is fully signed, you may need to cryptographically verify the digital signatures embedded directly inside the returned PDF document. To implement server-side verification using Python tools like pyHanko, read our tutorial on Programmatic PDF Signature Verification.

Step 7: Assembling the FastAPI Application

Finally, assemble the main FastAPI application file in app/main.py using an asynccontextmanager lifespan for database table creation and cleanup:

app/main.py — application entrypointPython / FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import engine, Base
from app.routers import contracts, webhooks

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Initialize database tables (for dev/demo; use Alembic for prod migrations)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: Dispose DB connection pool
    await engine.dispose()

app = FastAPI(
    title="Async E-Signature Service",
    version="2026.1",
    description="FastAPI service for dispatching e-signature contracts and processing HMAC webhooks",
    lifespan=lifespan,
)

app.include_router(contracts.router)
app.include_router(webhooks.router)

@app.get("/health", tags=["Health"])
async def health_check():
    return {"status": "healthy", "service": "esignature-fastapi"}

Testing the FastAPI E-Signature Pipeline

Start your FastAPI application with Uvicorn:

Terminal — run dev serverPython / FastAPI

uvicorn app.main:app --reload --port 8000

You can test dispatching a contract using curl:

Terminal — test contract dispatchPython / FastAPI

curl -X POST "http://localhost:8000/api/contracts/send" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Master Services Agreement",
    "recipient_name": "Jane Doe",
    "recipient_email": "jane@example.com",
    "markdown_content": "# Master Services Agreement\n\nThis agreement is entered into between Acme Corp and Jane Doe."
  }'

Frequently Asked Questions

How does async HTTP with httpx improve e-signature throughput in FastAPI applications?

In traditional synchronous Python frameworks like Flask or Django using the standard requests library, sending an e-signature request blocks the active OS thread while waiting for network round-trips to complete. Under high request volumes, this thread-blocking behavior quickly exhausts your web server worker pool, introducing severe latency bottlenecks and decreasing overall throughput. In contrast, FastAPI combined with httpx.AsyncClient leverages Python's asyncio event loop. When your application initiates an outgoing request to the Signbee E-Signature API, the event loop yields control back to other incoming client requests while waiting for the HTTP response headers and body. This non-blocking I/O model enables a single FastAPI worker to process thousands of concurrent contract dispatches per second with minimal memory footprint, making it the ideal choice for high-volume SaaS platforms, enterprise document automation pipelines, and AI agent execution environments.

How do I securely verify HMAC-SHA256 signatures on Signbee webhooks in FastAPI?

Securing webhook endpoints against forgery and man-in-the-middle attacks requires cryptographic signature verification. When Signbee sends a webhook event (such as contract.signed or contract.declined), it computes an HMAC-SHA256 signature using your shared webhook secret and sends it in the X-Signbee-Signature HTTP header. In FastAPI, your webhook route handler must access the raw unparsed request body using await request.body() before any JSON parsing takes place, as whitespace or key reordering can alter the raw bytes. You then compute an expected signature using Python's built-in hmac library and hashlib.sha256 digest function. Finally, compare the computed signature with the header value using hashlib.compare_digest(). Using compare_digest() is crucial because it performs constant-time string comparison, preventing timing-attack vulnerabilities where an attacker could deduce secret signature bytes by measuring nanosecond differences in response times.

Why should I store contract audit logs in PostgreSQL alongside e-signature webhooks?

Relying solely on external webhook notifications leaves your application vulnerable to missed state changes if network disruptions, server restarts, or unhandled exceptions occur during event delivery. By capturing and persisting immutable audit log records in PostgreSQL using an asynchronous ORM like SQLAlchemy or SQLModel, your application establishes a single source of truth for contract lifecycle state changes. Each incoming webhook payload—such as contract creation, recipient viewing, timestamped signatures, and legal certificate generation—is committed to a structured relational database with JSONB metadata and strict foreign key integrity. This historical event log allows your system to reconcile missed events during background recovery jobs, support audit trails for legal compliance (such as ESIGN, eIDAS, and HIPAA standards), and provide instantaneous real-time reporting to internal dashboards without querying external API endpoints repeatedly.

Ready to automate async Python e-signatures in FastAPI? Get started with 5 free documents/month.

Explore API Documentation ← Python Requests Tutorial

Last updated: July 21, 2026 · Michael Beckett is the founder of Signbee and B2bee Ltd.