一、场景概述

在企业合同管理场景中，批量发送合同并追踪签署状态是高频需求。DocuSign eSignature REST API提供了完善的解决方案，支持通过编程方式创建信封（Envelope）、指定签署人、发送签名请求，并通过Webhook实时接收签署状态变更通知。

本文使用Python 3.10+实现以下完整流程：

• OAuth 2.0 JWT模式认证
• 批量创建并发送签署信封
• 配置Connect Webhook接收回调
• 查询签署状态及异常处理

---

二、环境准备与认证配置

2.1 前期准备

在开始编码前，需在DocuSign管理后台完成以下配置：

1. 登录DocuSign Admin，进入 API and Keys 页面
2. 创建新的应用（App），获取 Integration Key
3. 在应用配置中设置 RSA Key Pair，下载私钥文件
4. 配置 Redirect URI（可使用 https://example.com/callback 作为占位）
5. 记录 Account ID 和 Base URL

2.2 安装依赖

pip install docusign-esign requests PyJWT cryptography

2.3 JWT认证实现

DocuSign推荐在生产环境中使用JWT OAuth 2.0流程，无需用户交互即可获取Access Token。

import json
import time
import requests
import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend

class DocuSignAuth:
    """DocuSign JWT OAuth认证客户端"""

    def __init__(self, integration_key, user_id, rsa_private_key_path, auth_server="account-d.docusign.com"):
        self.integration_key = integration_key
        self.user_id = user_id  # API Username (GUID)
        self.auth_server = auth_server
        self.base_url = f"https://{auth_server}"
        self.token = None
        self.expires_at = 0

        # 加载RSA私钥
        with open(rsa_private_key_path, "rb") as key_file:
            self.private_key = serialization.load_pem_private_key(
                key_file.read(),
                password=None,
                backend=default_backend()
            )

    def _generate_jwt_assertion(self):
        """生成JWT断言，用于换取Access Token"""
        now = int(time.time())
        payload = {
            "iss": self.integration_key,         # 签发者：Integration Key
            "sub": self.user_id,                  # 主题：API Username
            "aud": f"{self.base_url}/oauth/token",
            "iat": now,
            "exp": now + 3600,                     # JWT有效期1小时
            "scope": "signature impersonation"     # 请求的权限范围
        }
        return jwt.encode(payload, self.private_key, algorithm="RS256")

    def get_access_token(self):
        """获取Access Token（带缓存）"""
        if self.token and time.time() < self.expires_at - 60:
            return self.token

        assertion = self._generate_jwt_assertion()
        url = f"{self.base_url}/oauth/token"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": assertion
        }

        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status()
        result = response.json()

        self.token = result["access_token"]
        self.expires_at = time.time() + result.get("expires_in", 3600)
        return self.token

# 初始化认证（使用占位参数）
auth = DocuSignAuth(
    integration_key="YOUR_INTEGRATION_KEY",
    user_id="YOUR_API_USER_GUID",
    rsa_private_key_path="./private.key"
)
access_token = auth.get_access_token()
print(f"Access Token获取成功: {access_token[:20]}...")

参数说明：YOUR_INTEGRATION_KEY替换为DocuSign管理后台获取的Integration Key，YOUR_API_USER_GUID替换为API Username。account-d.docusign.com是Demo环境地址，生产环境使用account.docusign.com。

---

三、批量信封创建与发送

3.1 创建单个信封

DocuSign中的"信封"对应一份待签署的合同文档。以下代码展示如何创建包含一份PDF文档和一个签署人的信封。

import base64

class DocuSignEnvelopeClient:
    """DocuSign信封操作客户端"""

    def __init__(self, auth_client, account_id, base_url="https://demo.docusign.net/restapi"):
        self.auth = auth_client
        self.account_id = account_id
        self.base_url = base_url

    def _get_headers(self):
        """生成API请求头"""
        token = self.auth.get_access_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def create_envelope(self, document_path, signer_name, signer_email, envelope_subject):
        """
        创建单个签署信封

        :param document_path: 待签署的PDF文档路径
        :param signer_name: 签署人姓名
        :param signer_email: 签署人邮箱
        :param envelope_subject: 信封主题
        :return: 信封ID
        """
        # 读取文档并转为Base64编码
        with open(document_path, "rb") as f:
            doc_b64 = base64.b64encode(f.read()).decode("utf-8")

        # 构建信封请求体
        envelope_request = {
            "emailSubject": envelope_subject,
            "documents": [{
                "documentId": "1",
                "name": document_path.split("/")[-1],
                "documentBase64": doc_b64
            }],
            "recipients": {
                "signers": [{
                    "email": signer_email,
                    "name": signer_name,
                    "recipientId": "1",
                    "routingOrder": "1",
                    # 在文档第1页指定签署位置（坐标单位：pt，从左上角起算）
                    "tabs": {
                        "signHereTabs": [{
                            "documentId": "1",
                            "pageNumber": "1",
                            "xPosition": "200",
                            "yPosition": "600"
                        }],
                        "dateSignedTabs": [{
                            "documentId": "1",
                            "pageNumber": "1",
                            "xPosition": "200",
                            "yPosition": "700"
                        }]
                    }
                }]
            },
            "status": "sent"  # "created"为草稿，"sent"为直接发送
        }

        url = f"{self.base_url}/v2.1/accounts/{self.account_id}/envelopes"
        response = requests.post(url, headers=self._get_headers(), json=envelope_request)
        response.raise_for_status()

        envelope_id = response.json()["envelopeId"]
        print(f"信封已创建并发送: {envelope_id}")
        return envelope_id

3.2 批量发送实现

实际业务中往往需要同时向多个签署方发送合同。以下代码展示批量处理逻辑：

def batch_send_envelopes(self, contracts, envelope_subject_prefix):
    """
    批量发送签署信封

    :param contracts: 合同列表，每项包含 {document_path, signer_name, signer_email}
    :param envelope_subject_prefix: 信封主题前缀
    :return: 包含成功和失败列表的字典
    """
    results = {"success": [], "failed": []}

    for idx, contract in enumerate(contracts, 1):
        try:
            subject = f"{envelope_subject_prefix} - 合同{idx:03d}"
            envelope_id = self.create_envelope(
                document_path=contract["document_path"],
                signer_name=contract["signer_name"],
                signer_email=contract["signer_email"],
                envelope_subject=subject
            )
            results["success"].append({
                "index": idx,
                "envelope_id": envelope_id,
                "signer": contract["signer_name"]
            })
        except requests.exceptions.RequestException as e:
            results["failed"].append({
                "index": idx,
                "signer": contract["signer_name"],
                "error": str(e)
            })
            print(f"合同{idx:03d}发送失败: {contract['signer_name']} - {e}")

    return results

3.3 使用示例

# 初始化客户端
client = DocuSignEnvelopeClient(
    auth_client=auth,
    account_id="YOUR_ACCOUNT_ID"
)

# 准备批量合同数据
contracts = [
    {"document_path": "./contracts/agreement_001.pdf", "signer_name": "签署方A", "signer_email": "signer_a@example.com"},
    {"document_path": "./contracts/agreement_002.pdf", "signer_name": "签署方B", "signer_email": "signer_b@example.com"},
    {"document_path": "./contracts/agreement_003.pdf", "signer_name": "签署方C", "signer_email": "signer_c@example.com"},
]

# 执行批量发送
results = client.batch_send_envelopes(
    contracts=contracts,
    envelope_subject_prefix="服务协议签署通知"
)

print(f"成功: {len(results['success'])} 份, 失败: {len(results['failed'])} 份")

---

四、Webhook回调配置

DocuSign Connect提供了实时事件通知机制。当信封状态发生变化时，DocuSign向指定的回调URL发送Webhook请求。

4.1 创建Connect配置

def create_connect_configuration(self, webhook_url, include_documents=False):
    """
    创建Connect Webhook配置

    :param webhook_url: 接收回调的URL，例如 https://yourapp.com/api/docusign/callback
    :param include_documents: 是否在回调中包含文档内容
    """
    config = {
        "urlToPublishTo": webhook_url,
        "eventData": {
            "version": "v2.1",
            "format": "json",
            "includeDocuments": include_documents,
            # 订阅所有信封相关事件
            "envelopeEvents": [
                {"envelopeEventStatusCode": "sent"},
                {"envelopeEventStatusCode": "delivered"},
                {"envelopeEventStatusCode": "completed"},
                {"envelopeEventStatusCode": "declined"},
                {"envelopeEventStatusCode": "voided"}
            ],
            "recipientEvents": [
                {"recipientEventStatusCode": "Sent"},
                {"recipientEventStatusCode": "Delivered"},
                {"recipientEventStatusCode": "Completed"}
            ]
        },
        "name": "Production Connect - YourApp",
        "status": "active"
    }

    url = f"{self.base_url}/v2.1/accounts/{self.account_id}/connect"
    response = requests.post(url, headers=self._get_headers(), json=config)
    response.raise_for_status()

    connect_id = response.json()["connectId"]
    print(f"Connect配置已创建: {connect_id}")
    return connect_id

4.2 Flask回调服务示例

from flask import Flask, request, jsonify
import hashlib
import hmac

app = Flask(__name__)

# DocuSign Connect HMAC密钥（从DocuSign管理后台获取）
HMAC_SECRET = "YOUR_HMAC_SECRET".encode("utf-8")

@app.route("/api/docusign/callback", methods=["POST"])
def docusign_webhook():
    """
    处理DocuSign Connect回调
    验证HMAC签名后处理信封状态变更
    """
    # 验证HMAC签名（防止伪造请求）
    signature = request.headers.get("X-DocuSign-Signature-1", "")
    payload = request.data
    expected_sig = hmac.new(HMAC_SECRET, payload, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(signature, expected_sig):
        return jsonify({"error": "Invalid signature"}), 401

    data = request.get_json()

    # 提取信封ID和事件状态
    envelope_id = data.get("envelopeId", "")
    status = data.get("envelopeStatus", "")

    print(f"信封状态变更: {envelope_id} -> {status}")

    # 根据不同状态执行业务逻辑
    if status == "Completed":
        handle_completed_envelope(envelope_id, data)
    elif status == "Declined":
        handle_declined_envelope(envelope_id, data)
    elif status == "Voided":
        handle_voided_envelope(envelope_id, data)

    return jsonify({"status": "received"}), 200

def handle_completed_envelope(envelope_id, data):
    """处理签署完成的信封"""
    # 获取签署人信息
    recipients = data.get("envelopeSummary", {}).get("recipients", {})
    signers = recipients.get("signers", [])
    for signer in signers:
        signed_time = signer.get("signedDateTime", "")
        print(f"签署人 {signer['name']} 于 {signed_time} 完成签署")

def handle_declined_envelope(envelope_id, data):
    """处理被拒绝的信封"""
    print(f"信封 {envelope_id} 已被拒签，需人工跟进")

def handle_voided_envelope(envelope_id, data):
    """处理已作废的信封"""
    print(f"信封 {envelope_id} 已作废")

---

五、签署状态查询与异常处理

5.1 查询信封状态

def get_envelope_status(self, envelope_id):
    """
    查询指定信封的签署状态

    :param envelope_id: 信封ID
    :return: 信封状态详情字典
    """
    url = f"{self.base_url}/v2.1/accounts/{self.account_id}/envelopes/{envelope_id}"
    response = requests.get(url, headers=self._get_headers())
    response.raise_for_status()

    envelope = response.json()
    return {
        "envelope_id": envelope_id,
        "status": envelope.get("status", "unknown"),
        "sent_date": envelope.get("sentDateTime", ""),
        "completed_date": envelope.get("completedDateTime", ""),
        "email_subject": envelope.get("emailSubject", "")
    }

def batch_get_status(self, envelope_ids):
    """批量查询多个信封的状态"""
    batch_status = []
    for eid in envelope_ids:
        try:
            status = self.get_envelope_status(eid)
            batch_status.append(status)
        except requests.exceptions.RequestException as e:
            batch_status.append({
                "envelope_id": eid,
                "status": "query_failed",
                "error": str(e)
            })
    return batch_status

5.2 常见异常处理策略

import time
from requests.exceptions import RequestException, Timeout, HTTPError

class DocuSignExceptionHandler:
    """DocuSign API异常处理与重试机制"""

    def __init__(self, max_retries=3, base_delay=2):
        self.max_retries = max_retries
        self.base_delay = base_delay

    def call_with_retry(self, api_func, *args, **kwargs):
        """
        带指数退避重试的API调用包装

        适用场景：网络抖动、API限流（429状态码）
        """
        last_exception = None

        for attempt in range(1, self.max_retries + 1):
            try:
                return api_func(*args, **kwargs)
            except HTTPError as e:
                status_code = e.response.status_code

                # 401: Token过期，刷新并重试
                if status_code == 401:
                    print(f"Token过期，重新认证后重试 (尝试 {attempt}/{self.max_retries})")
                    auth.get_access_token()  # 刷新Token
                    continue

                # 429: 请求限流，等待后重试
                elif status_code == 429:
                    retry_after = int(e.response.headers.get("Retry-After", self.base_delay))
                    wait = retry_after * attempt
                    print(f"触发限流，等待 {wait} 秒后重试 (尝试 {attempt}/{self.max_retries})")
                    time.sleep(wait)
                    continue

                # 400/403/500: 业务错误，记录日志后终止
                else:
                    print(f"API返回错误 {status_code}: {e.response.text}")
                    raise

            except Timeout:
                wait = self.base_delay * (2 ** (attempt - 1))
                print(f"请求超时，等待 {wait} 秒后重试 (尝试 {attempt}/{self.max_retries})")
                time.sleep(wait)

            except RequestException as e:
                wait = self.base_delay * (2 ** (attempt - 1))
                print(f"网络异常: {e}，等待 {wait} 秒后重试 (尝试 {attempt}/{self.max_retries})")
                time.sleep(wait)

        raise Exception(f"API调用在 {self.max_retries} 次重试后仍失败: {last_exception}")

---

六、完整调用示例

if __name__ == "__main__":
    # 1. 认证
    auth = DocuSignAuth(
        integration_key="YOUR_INTEGRATION_KEY",
        user_id="YOUR_API_USER_GUID",
        rsa_private_key_path="./private.key"
    )

    # 2. 初始化客户端
    client = DocuSignEnvelopeClient(
        auth_client=auth,
        account_id="YOUR_ACCOUNT_ID"
    )

    # 3. 配置Webhook
    client.create_connect_configuration(
        webhook_url="https://yourapp.com/api/docusign/callback"
    )

    # 4. 批量发送合同
    contracts = [
        {"document_path": "./contracts/sla_001.pdf", "signer_name": "签署方A", "signer_email": "signer_a@example.com"},
        {"document_path": "./contracts/sla_002.pdf", "signer_name": "签署方B", "signer_email": "signer_b@example.com"},
    ]
    results = client.batch_send_envelopes(contracts, "SLA服务协议")

    # 5. 查询已发送的信封状态
    envelope_ids = [item["envelope_id"] for item in results["success"]]
    statuses = client.batch_get_status(envelope_ids)
    for s in statuses:
        print(f"{s['envelope_id']}: {s['status']}")

---

七、注意事项

• Demo环境与生产环境：开发测试使用 demo.docusign.net 和 account-d.docusign.com，上线切换为 account.docusign.com
• Token缓存：Access Token有效期通常为1小时，务必实现缓存机制避免频繁请求
• HMAC安全验证：生产环境中Connect回调必须校验X-DocuSign-Signature-1签名
• 重试策略：对429和5xx错误实施指数退避重试，对4xx业务错误立即终止
• 日志记录：所有API调用和回调务必记录详细日志，便于排查问题

---

上海华万，专注为企业提供SaaS产品的一站式选型与集成服务。国内产品线涵盖腾讯会议、企业微信、腾讯电子签等腾讯生态产品，国际产品线包括Microsoft Teams、Zoom、DocuSign等协作与签约工具。从需求诊断、产品选型到系统部署、API集成与长期运维，华万为企业量身定制落地路径，覆盖售前咨询、方案设计、部署实施与售后服务全流程。目前已服务制造、零售、教育、金融等多个行业的中小企业客户。