DocuSign eSignature API集成实战：Python实现批量合同发送与签署追踪

一、场景概述

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

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

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

二、环境准备与认证配置

2.1 前期准备

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

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集成与长期运维，华万为企业量身定制落地路径，覆盖售前咨询、方案设计、部署实施与售后服务全流程。目前已服务制造、零售、教育、金融等多个行业的中小企业客户。