提交第一个脱敏任务。
使用 Python requests 完成上传、提交、轮询和下载。请配置服务地址与自己的 Bearer 密钥。
1. 配置地址与密钥
将 REDACTTOOL_BASE_URL 设为服务地址，REDACTTOOL_API_KEY 设为自己的 Bearer 密钥。实际远程文件应使用可信 HTTPS 入口。当前 HTTP 试用仅使用非保密样例，并明确设置 REDACTTOOL_ALLOW_HTTP_TRIAL=1。不要将凭据写入前端代码。
密钥绑定会话，最长 24 小时过期；重新生成后旧密钥失效。请将凭据保存在服务器，不要写入前端代码、网址或提示词。
Run an example
export REDACTTOOL_BASE_URL="https://redacttool.app" export REDACTTOOL_FILE="agreement.txt" # Supply REDACTTOOL_API_KEY securely in your environment. # The official endpoint uses HTTPS. # Python REST (pip install requests) python redact.py # Node.js 22+ node redact.mjs # cURL and jq bash redact.sh # Python MCP (pip install requests) python mcp_redact.py
2. 运行完整流程
可复制示例
"""RedactTool REST example. Supply your own endpoint and key.
No network activity occurs until this script is explicitly executed.
Requires: pip install requests
"""
import os
import time
import uuid
from pathlib import Path
from urllib.parse import urlparse
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
base = os.environ["REDACTTOOL_BASE_URL"].rstrip("/")
key = os.environ["REDACTTOOL_API_KEY"]
source = Path(os.environ.get("REDACTTOOL_FILE", "sample.txt"))
parsed = urlparse(base)
if parsed.scheme != "https" and parsed.hostname not in {"localhost", "127.0.0.1"} and os.environ.get("REDACTTOOL_ALLOW_HTTP_TRIAL") != "1":
raise ValueError("Use HTTPS, or explicitly set REDACTTOOL_ALLOW_HTTP_TRIAL=1 for non-confidential trial files")
output = source.with_name(source.stem + "-redacted" + source.suffix)
if output.exists():
raise FileExistsError(output)
session = requests.Session()
# Retry safe reads when an idle connection closes. Do not replay POST bodies.
retry = Retry(total=3, connect=3, read=3, backoff_factor=0.5,
allowed_methods=frozenset({"GET"}), status_forcelist=(502, 503, 504))
for scheme in ("https://", "http://"):
session.mount(scheme, HTTPAdapter(max_retries=retry))
session.headers["Authorization"] = "Bearer " + key
def checked(response):
response.raise_for_status()
return response.json()
with source.open("rb") as content:
uploaded = checked(session.post(base + "/v1/files",
files={"file": (source.name, content)}, timeout=(15, 120)))
# Keep this ID if you retry this same creation request.
request_id = uuid.uuid4().hex
print("Keep file_id and Idempotency-Key for recovery:", uploaded["file_id"], request_id)
job = checked(session.post(base + "/v1/redactions",
headers={"Idempotency-Key": request_id},
json={"file_id": uploaded["file_id"], "mode": "placeholder"},
timeout=(15, 60)))
print("Job:", job["id"])
# A local wait timeout does not cancel server processing.
deadline = time.monotonic() + 3600
while job["status"] not in {"completed", "failed"}:
if time.monotonic() >= deadline:
raise TimeoutError("Resume polling this job ID; do not create a new job")
time.sleep(5)
job = checked(session.get(base + "/v1/redactions/" + job["id"],
timeout=(15, 60)))
if job["status"] == "failed":
raise RuntimeError(job.get("error") or "job_failed")
# Use a known same-origin path, never an arbitrary returned host.
with session.get(base + "/v1/redactions/" + job["id"] + "/file",
stream=True, timeout=(15, 120)) as result:
result.raise_for_status()
with output.open("xb") as saved:
for chunk in result.iter_content(65536):
saved.write(chunk)
print("Saved:", output)
3. 处理连接中断
在应用中保存文件 ID、任务 ID 和幂等键。重试创建同一任务时，使用相同文件、模式和幂等键。本地等待结束后应继续查询原任务，不要重新处理文件。
此示例保持精简：无法恢复的 HTTP 错误会终止运行。生产接入还应针对 429 和临时故障设置有限退避重试，安全保存任务状态，并在需要时更新凭据。