전체 예제

// 아래 PowerShell 준비 명령 후: node xdata-quickstart.mjs
import { readFileSync, writeFileSync, existsSync } from 'node:fs';
import { randomUUID } from 'node:crypto';
const BASE = 'https://api.xdata.kr';
const ENV_SCOPE = process.env.XDATA_ENV;
if (!['sandbox', 'real_test', 'production'].includes(ENV_SCOPE)) throw new Error('XDATA_ENV 확인');
if (!process.env.XDATA_CLIENT_ID || !process.env.XDATA_CLIENT_SECRET) {
  throw new Error('XDATA_CLIENT_ID와 XDATA_CLIENT_SECRET 을 설정하세요.');
}
const requestFile = './xdata-request.json';
const body = {
  provider: 'hometax', action: 'hometax.etax.invoice.search_list',
  // 등록된 자격증명이 하나뿐이면 생략해도 자동으로 선택됩니다. hometax 는 sandbox 에서도
  // 자동 선택 대상 자격증명이 없으면 실패합니다 - 준비 단계에서 비워 뒀다면 먼저
  // [자격증명 등록](/common-guide/account-link) 부터 하세요.
  ...(process.env.XDATA_ACCOUNT_LINK_ID ? { account_link_id: process.env.XDATA_ACCOUNT_LINK_ID } : {}),
  params: { from_date: '2026-01-01', to_date: '2026-01-31' },
};
const identity = JSON.stringify({ base: BASE, environment: ENV_SCOPE, client: process.env.XDATA_CLIENT_ID, body });
// 같은 파일로 재실행하면 새 작업이 아니라 기존 접수를 이어서 확인합니다.
if (!existsSync(requestFile)) writeFileSync(requestFile, JSON.stringify({ identity, key: randomUUID() }), { flag: 'wx', mode: 0o600 });
const request = JSON.parse(readFileSync(requestFile, 'utf8'));
if (request.identity !== identity || typeof request.key !== 'string' || !request.key) throw new Error('기존 요청 기록과 다릅니다. 파일을 지우지 말고 원래 환경, 키, 본문을 확인하세요.');

const token = Buffer.from(
  `${process.env.XDATA_CLIENT_ID}:${process.env.XDATA_CLIENT_SECRET}`,
).toString('base64');

const headers = {
  Authorization: `Basic ${token}`,
  'X-Env-Scope': ENV_SCOPE,
  'Content-Type': 'application/json',
};

async function readError(response) {
  // 오류 본문은 최상위 error 입니다. 없는 주소의 404 처럼 error 가 없는 응답은 detail 문장을 message 로 씁니다.
  const payload = await response.json().catch(() => null);
  return payload?.error ?? { message: typeof payload?.detail === 'string' ? payload.detail : undefined };
}

async function submitJob(body) {
  const response = await fetch(`${BASE}/v1/jobs`, {
    method: 'POST',
    // 재시도할 때는 이 키를 그대로 다시 씁니다. 새 키를 쓰면 작업이 한 번 더 실행됩니다.
    headers: { ...headers, 'Idempotency-Key': request.key },
    signal: AbortSignal.timeout(30000),
    body: JSON.stringify(body),
  });
  if (!response.ok) {
    const error = await readError(response);
    throw new Error(`제출 실패 ${response.status} ${error.code ?? '알 수 없음'}: ${error.message ?? ''}`);
  }
  return response.json();
}

async function waitForResult(jobId) {
  for (let attempt = 0; attempt < 20; attempt++) {
    const response = await fetch(`${BASE}/v1/jobs/${jobId}`, { headers, signal: AbortSignal.timeout(30000) });
    if (!response.ok) throw new Error(`상태 조회 HTTP ${response.status}. 같은 기록으로 재실행하세요.`);
    const job = await response.json();
    console.log(`${job.progress}% ${job.progress_message ?? job.status}`);

    if (job.status === 'SUCCEEDED') {
      const resultResponse = await fetch(`${BASE}/v1/jobs/${jobId}/result`, { headers, signal: AbortSignal.timeout(30000) });
      if (!resultResponse.ok) throw new Error(`결과 조회 HTTP ${resultResponse.status}`);
      return resultResponse.json();
    }
    if (job.status !== 'QUEUED' && job.status !== 'RUNNING') {
      throw new Error(`${job.status}: ${job.error?.code ?? ''}`);
    }
    await new Promise((resolve) => setTimeout(resolve, (job.retry_after_seconds ?? 2) * 1000));
  }
  throw new Error('조회 횟수 한도에 도달했습니다. 같은 기록으로 재실행하여 상태를 확인하세요.');
}

if (!request.job_id) {
  const submitted = await submitJob(body);
  if (typeof submitted.job_id !== 'string' || !submitted.job_id) throw new Error('Job 응답 형식 오류');
  request.job_id = submitted.job_id;
  writeFileSync(requestFile, JSON.stringify(request), { mode: 0o600 });
}
const jobId = request.job_id;

const result = await waitForResult(jobId);
console.log('result_received:', typeof result === 'object' && result !== null);
// 결과 원문은 로그에 출력하지 않습니다. 실제 값은 응답을 그대로 저장해 사용하세요.

# 아래 PowerShell 준비 명령 후: python xdata_quickstart.py
# pip install requests
import base64, json, os, time, uuid, requests
from pathlib import Path

BASE = 'https://api.xdata.kr'
ENV_SCOPE = os.environ['XDATA_ENV']
if ENV_SCOPE not in ('sandbox', 'real_test', 'production'):
    raise ValueError('XDATA_ENV 확인')
body = {
    'provider': 'hometax', 'action': 'hometax.etax.invoice.search_list',
    'params': {'from_date': '2026-01-01', 'to_date': '2026-01-31'},
}
# 등록된 자격증명이 하나뿐이면 생략해도 자동으로 선택됩니다. hometax 는 sandbox 에서도
# 자동 선택 대상 자격증명이 없으면 실패합니다 - 준비 단계에서 비워 뒀다면 먼저
# [자격증명 등록](/common-guide/account-link) 부터 하세요.
if os.environ.get('XDATA_ACCOUNT_LINK_ID'):
    body['account_link_id'] = os.environ['XDATA_ACCOUNT_LINK_ID']
identity = dict(base=BASE, environment=ENV_SCOPE, client=os.environ['XDATA_CLIENT_ID'], body=body)
request_file = Path('./xdata-request.json')
# 같은 파일로 재실행하면 새 작업이 아니라 기존 접수를 이어서 확인합니다.
if not request_file.exists():
    with request_file.open('x', encoding='utf-8') as stream:
        json.dump(dict(identity=identity, key=str(uuid.uuid4())), stream)
        stream.flush()
        os.fsync(stream.fileno())
record = json.loads(request_file.read_text(encoding='utf-8'))
if record.get('identity') != identity or not isinstance(record.get('key'), str) or not record['key']:
    raise ValueError('기존 요청 기록과 다릅니다. 파일을 지우지 말고 원래 환경, 키, 본문을 확인하세요.')

token = base64.b64encode(
    f"{os.environ['XDATA_CLIENT_ID']}:{os.environ['XDATA_CLIENT_SECRET']}".encode()
).decode()

HEADERS = {
    'Authorization': f'Basic {token}',
    'X-Env-Scope': ENV_SCOPE,
    'Content-Type': 'application/json',
}


def read_error(response) -> dict:
    # 오류 본문은 최상위 error 입니다. 없는 주소의 404 처럼 error 가 없는 응답은 detail 문장을 message 로 씁니다.
    try:
        payload = response.json()
    except ValueError:
        return {}
    if not isinstance(payload, dict):
        return {}
    detail = payload.get('detail')
    return payload.get('error') or ({'message': detail} if isinstance(detail, str) else {})


def submit_job(body: dict) -> dict:
    response = requests.post(
        f'{BASE}/v1/jobs',
        # 재시도할 때는 이 키를 그대로 다시 씁니다. 새 키를 쓰면 작업이 한 번 더 실행됩니다.
        headers={**HEADERS, 'Idempotency-Key': record['key']},
        timeout=30,
        json=body,
    )
    if not response.ok:
        error = read_error(response)
        raise RuntimeError(f"제출 실패 {response.status_code} {error.get('code', '알 수 없음')}: {error.get('message', '')}")
    return response.json()


def wait_for_result(job_id: str) -> dict:
    for attempt in range(20):
        response = requests.get(f'{BASE}/v1/jobs/{job_id}', headers=HEADERS, timeout=30)
        response.raise_for_status()
        job = response.json()
        print(f"{job['progress']}% {job.get('progress_message') or job['status']}")

        if job['status'] == 'SUCCEEDED':
            response = requests.get(f'{BASE}/v1/jobs/{job_id}/result', headers=HEADERS, timeout=30)
            response.raise_for_status()
            return response.json()
        if job['status'] not in ('QUEUED', 'RUNNING'):
            raise RuntimeError(f"{job['status']}: {(job.get('error') or {}).get('code')}")

        time.sleep(job.get('retry_after_seconds') or 2)

    raise TimeoutError('조회 횟수 한도에 도달했습니다. 같은 기록으로 재실행하세요.')


if not record.get('job_id'):
    submitted = submit_job(body)
    if not isinstance(submitted.get('job_id'), str) or not submitted['job_id']:
        raise ValueError('Job 응답 형식 오류')
    record['job_id'] = submitted['job_id']
    with request_file.open('w', encoding='utf-8') as stream:
        json.dump(record, stream)
        stream.flush()
        os.fsync(stream.fileno())
result = wait_for_result(record['job_id'])
print('result_received:', isinstance(result, dict))
# 결과 원문은 로그에 출력하지 않습니다. 실제 값은 응답을 그대로 저장해 사용하세요.