How to Automate Customer Onboarding Documents with DocForge API

onboarding_template = {
    "template_id": "customer-onboarding-v2",
    "sections": [
        "welcome_header",
        "account_details",
        "getting_started_checklist",
        "support_contacts",
        "terms_summary"
    ],
    "branding": {
        "logo_url": "https://yourcompany.com/logo.png",
        "primary_color": "#1a56db",
        "font": "Inter"
    }
}

import requests
import json
from datetime import datetime, timedelta
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/docforge/v1"
 
def build_onboarding_payload(customer: dict) -> dict:
    """Build the document generation payload for a new customer."""
    account_start = datetime.now()
    renewal_date = account_start + timedelta(days=365)
 
    return {
        "template_id": "customer-onboarding-v2",
        "output_format": "pdf",
        "data": {
            "customer_name": customer["name"],
            "company_name": customer.get("company", ""),
            "account_id": customer["account_id"],
            "plan": customer["plan"],
            "account_manager": customer.get("account_manager", "Support Team"),
            "support_email": "[email protected]",
            "account_start_date": account_start.strftime("%B %d, %Y"),
            "renewal_date": renewal_date.strftime("%B %d, %Y"),
            "api_endpoint": f"https://api.yourcompany.com/v1",
            "onboarding_checklist": [
                "Complete account profile",
                "Configure API credentials",
                "Run your first API call",
                "Set up webhook endpoints",
                "Review SLA terms"
            ],
            "sla_uptime": "99.9%",
            "support_response_time": "4 hours",
            "jurisdiction": customer.get("jurisdiction", "US")
        },
        "metadata": {
            "reference_id": customer["account_id"],
            "generated_for": customer["name"]
        }
    }

def generate_onboarding_doc(customer: dict) -> bytes:
    """Call DocForge API to generate the onboarding PDF."""
    payload = build_onboarding_payload(customer)
 
    response = requests.post(
        f"{BASE_URL}/generate",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=30
    )
 
    if response.status_code != 200:
        raise ValueError(f"DocForge API error: {response.status_code} - {response.text}")
 
    return response.content  # Raw PDF bytes
 
 
def save_onboarding_doc(customer: dict, output_dir: str = "./onboarding_docs") -> str:
    """Generate and save the onboarding PDF locally."""
    import os
    os.makedirs(output_dir, exist_ok=True)
 
    pdf_bytes = generate_onboarding_doc(customer)
    filename = f"{output_dir}/{customer['account_id']}_onboarding.pdf"
 
    with open(filename, "wb") as f:
        f.write(pdf_bytes)
 
    print(f"Generated: {filename} ({len(pdf_bytes):,} bytes)")
    return filename

import concurrent.futures
 
def batch_generate_onboarding_docs(customers: list[dict]) -> dict:
    """Generate onboarding documents for multiple customers in parallel."""
    results = {"success": [], "failed": []}
 
    def process_one(customer):
        try:
            path = save_onboarding_doc(customer)
            return {"account_id": customer["account_id"], "path": path, "status": "ok"}
        except Exception as e:
            return {"account_id": customer["account_id"], "error": str(e), "status": "error"}
 
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(process_one, c) for c in customers]
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result["status"] == "ok":
                results["success"].append(result)
            else:
                results["failed"].append(result)
 
    print(f"Generated {len(results['success'])} docs, {len(results['failed'])} failed")
    return results
 
 
# Example usage
new_customers = [
    {
        "name": "Alice Chen",
        "company": "Acme Corp",
        "account_id": "ACC-10042",
        "plan": "Enterprise",
        "account_manager": "James Liu",
        "jurisdiction": "US"
    },
    {
        "name": "Marco Silva",
        "company": "TechStart GmbH",
        "account_id": "ACC-10043",
        "plan": "Pro",
        "jurisdiction": "DE"
    }
]
 
batch_results = batch_generate_onboarding_docs(new_customers)

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders
 
def send_onboarding_email(customer: dict, pdf_path: str):
    """Send the onboarding PDF to the new customer."""
    msg = MIMEMultipart()
    msg["From"] = "[email protected]"
    msg["To"] = customer["email"]
    msg["Subject"] = f"Welcome to YourProduct, {customer['name'].split()[0]}!"
 
    body = f"""
    Hi {customer['name'].split()[0]},
 
    Welcome aboard! Please find your personalized onboarding guide attached.
 
    Your Account ID: {customer['account_id']}
    Plan: {customer['plan']}
 
    If you have any questions, reply to this email or reach us at [email protected].
 
    Best,
    The YourProduct Team
    """
    msg.attach(MIMEText(body, "plain"))
 
    with open(pdf_path, "rb") as f:
        attachment = MIMEBase("application", "octet-stream")
        attachment.set_payload(f.read())
        encoders.encode_base64(attachment)
        attachment.add_header(
            "Content-Disposition",
            f"attachment; filename=welcome_{customer['account_id']}.pdf"
        )
        msg.attach(attachment)
 
    # Connect to your SMTP server
    with smtplib.SMTP("smtp.yourprovider.com", 587) as server:
        server.starttls()
        server.login("[email protected]", "YOUR_SMTP_PASSWORD")
        server.send_message(msg)
 
    print(f"Sent onboarding email to {customer['email']}")

from fastapi import FastAPI, Request
import asyncio
 
app = FastAPI()
 
@app.post("/webhooks/new-customer")
async def handle_new_customer(request: Request):
    """Triggered when a new customer completes signup."""
    event = await request.json()
 
    if event.get("type") != "customer.created":
        return {"status": "ignored"}
 
    customer = event["data"]
 
    # Run document generation in the background
    asyncio.create_task(
        generate_and_send_onboarding(customer)
    )
 
    return {"status": "processing"}
 
 
async def generate_and_send_onboarding(customer: dict):
    """Background task: generate PDF and send email."""
    try:
        pdf_path = save_onboarding_doc(customer)
        send_onboarding_email(customer, pdf_path)
        print(f"Onboarding complete for {customer['account_id']}")
    except Exception as e:
        print(f"Onboarding failed for {customer.get('account_id')}: {e}")
        # Add your error alerting logic here

payload["data"]["locale"] = customer.get("locale", "en-US")
# Supports: en-US, en-GB, de-DE, fr-FR, es-ES, ja-JP, zh-CN, pt-BR