五分钟快速开始

本页用最少的参数在测试环境跑通一次完整链路。生产接入前请补齐 回调与错误处理。

准备

向平台索取测试环境的两项凭据，并把它们放进服务端环境变量：

export RPA_BASE_URL="https://rpa-test-api.xiaowu005.xyz"
export RPA_CHANNEL_KEY="rpa_channel_xxx"      # X-Channel-Key
export RPA_CALLBACK_SECRET="xxx"              # 签名与回调验签用，本页暂不需要
export RPA_CHANNEL_CODE="BANK"                # 渠道编码

确认网络连通：

curl -s "$RPA_BASE_URL/health"

第 1 步 · 创建授权链接

curl -sS -X POST "$RPA_BASE_URL/api/v1/channel/authorization-links" \
  -H "X-Channel-Key: $RPA_CHANNEL_KEY" \
  -H "Content-Type: application/json; charset=utf-8" \
  -d '{
    "external_order_id": "DEMO-20260822-0001",
    "enterprise_name": "示例企业有限公司",
    "taxpayer_id": "91420100XXXXXXXXXX",
    "collect_tax": true,
    "collect_invoice": true,
    "tax_data_types": ["ALL", "STANDARD"],
    "invoice_data_types": ["ALL", "STANDARD"],
    "invoice_period": "LAST_24_MONTHS"
  }'

响应中的两个值要记下来：

{
  "authorization_code": "550e8400-e29b-41d4-a716-446655440000",
  "authorization_url": "https://rpa-test-auth.xiaowu005.xyz/user/authorization/taxXyd?...",
  "external_order_id": "DEMO-20260822-0001",
  "expires_at": "2026-08-22T10:30:00Z"
}

直接跳转，不要拼接

authorization_url 由平台按本次申请动态生成。客户前端必须原样打开它， 自行拼接授权页地址会导致授权失败。

第 2 步 · 完成授权

在浏览器打开 authorization_url，按页面提示完成税局登录与授权确认。 链接默认 30 分钟有效（expires_in_seconds）。

第 3 步 · 等待采集完成

轮询订单状态，间隔不低于 15 秒：

curl -sS "$RPA_BASE_URL/api/v1/channel/orders/DEMO-20260822-0001" \
  -H "X-Channel-Key: $RPA_CHANNEL_KEY"

看两个字段：

• state —— SUCCESS 或 ERROR 表示结束；CONNECT / PROCESS / INTERACTION 继续等。
• job_id —— 下一步取数要用，它就是数据接口里的 orderNo。

生产环境应以回调通知为主链路，轮询只作兜底。

第 4 步 · 确认产品可取数

curl -sS "$RPA_BASE_URL/api/v1/channel/orders/DEMO-20260822-0001/products/TAX" \
  -H "X-Channel-Key: $RPA_CHANNEL_KEY"

取数前提

必须 state == "SUCCEEDED" 且 available == true。 不要用页面进度百分比或整单 state 代替这个判断。

第 5 步 · 取数

把上一步的 job_id 填进 orderNo：

curl -sS -X POST "$RPA_BASE_URL/api/channel/data/tax/standardTable" \
  -H "X-Channel-Key: $RPA_CHANNEL_KEY" \
  -H "Content-Type: application/json; charset=utf-8" \
  -d '{
    "channelCode": "'"$RPA_CHANNEL_CODE"'",
    "channelOrderNo": "DEMO-20260822-0001",
    "orderNo": "1e3f2f55-8da4-4ba9-b941-84860eb1d243"
  }'

成功时 success 为 true：

{
  "code": 200,
  "msg": "操作成功",
  "success": true,
  "data": { "schemaVersion": "tax-standard.v2", "selectedModules": ["..."], "data": {} }
}

发票原始数据换成 /api/channel/data/invoice/all 并带上分页参数， 注意它成功码是 0：

curl -sS -X POST "$RPA_BASE_URL/api/channel/data/invoice/all" \
  -H "X-Channel-Key: $RPA_CHANNEL_KEY" \
  -H "Content-Type: application/json; charset=utf-8" \
  -d '{
    "channelCode": "'"$RPA_CHANNEL_CODE"'",
    "channelOrderNo": "DEMO-20260822-0001",
    "orderNo": "1e3f2f55-8da4-4ba9-b941-84860eb1d243",
    "pageNum": 1,
    "pageSize": 500
  }'

一个可运行的最小客户端

import time
import requests

BASE = "https://rpa-test-api.xiaowu005.xyz"
KEY = "rpa_channel_xxx"
CHANNEL_CODE = "BANK"
ORDER_ID = "DEMO-20260822-0001"

session = requests.Session()
session.headers["X-Channel-Key"] = KEY


def create_link() -> dict:
    response = session.post(
        f"{BASE}/api/v1/channel/authorization-links",
        json={
            "external_order_id": ORDER_ID,
            "enterprise_name": "示例企业有限公司",
            "taxpayer_id": "91420100XXXXXXXXXX",
            "tax_data_types": ["ALL", "STANDARD"],
            "invoice_data_types": ["ALL"],
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


def wait_for_order() -> dict:
    """轮询兜底；生产环境应以回调为主链路。"""
    while True:
        order = session.get(
            f"{BASE}/api/v1/channel/orders/{ORDER_ID}", timeout=30
        ).json()
        if order["state"] in {"SUCCESS", "ERROR"}:
            return order
        time.sleep(20)  # 间隔不得低于 15 秒


def fetch_tax_standard(job_id: str) -> dict:
    payload = {
        "channelCode": CHANNEL_CODE,
        "channelOrderNo": ORDER_ID,
        "orderNo": job_id,
    }
    body = session.post(
        f"{BASE}/api/channel/data/tax/standardTable", json=payload, timeout=60
    ).json()
    if not body.get("success"):
        raise RuntimeError(f"取数失败 code={body.get('code')} msg={body.get('msg')}")
    return body["data"]


link = create_link()
print("请打开：", link["authorization_url"])

order = wait_for_order()
if order["state"] != "SUCCESS":
    raise SystemExit(f"采集失败：{order.get('result')}")

product = session.get(
    f"{BASE}/api/v1/channel/orders/{ORDER_ID}/products/TAX", timeout=30
).json()
if product["state"] == "SUCCEEDED" and product["available"]:
    print(fetch_tax_standard(order["job_id"]))

接下来

你要做的事	去哪一篇
用回调代替轮询	回调通知
发票量很大	大批量发票交付
需要 PDF / HTML 报告	PDF 与 HTML 报告
渠道被要求强制签名	鉴权与签名
准备上生产	上线检查清单