레시피: 지급 전에 거래 상대를 세 번의 호출로 검증하기

세 번의 호출을 이 순서로 합니다. 판정과 그 쌍이 새로운지 여부를 위해 from=에 자신의 주소를 넣은 /risk, 등급이 none이 아닐 때 사실 확인을 위한 개요, 순회가 중단되어 판정이 unknown일 때 어디까지 진행했는지 보기 위한 /exposure입니다. 프로그램은 refuse, allow, escalate 세 단어 중 하나로 답하고, 요청 ID와 computed_at을 로그에 기록합니다. 하드코딩된 유일한 정책은 스캐너가 고수하는 것입니다. unknown은 상향 검토하며, 결코 허용하지 않습니다.

결정

new_counterparty: "new"는 점수에 반영하지 않고 로그에만 남깁니다. 한 주소로의 첫 지급은 검토자를 위한 맥락입니다.

curl

#!/bin/sh
# vet.sh <chain> <counterparty> <my-address>  → prints refuse | allow | escalate
B=https://stablescan.achivx.com; CHAIN=$1; ADDR=$2; ME=$3
risk=$(curl -sf "$B/v1/addresses/$ADDR/risk?chain=$CHAIN&from=$ME") || { echo "escalate (no verdict)"; exit 2; }
tier=$(echo "$risk" | jq -r .tier); reasons=$(echo "$risk" | jq -c .incomplete_reasons)
if [ "$tier" = unknown ] && echo "$reasons" | grep -q time_budget; then
  sleep 15; risk=$(curl -sf "$B/v1/addresses/$ADDR/risk?chain=$CHAIN&from=$ME") && tier=$(echo "$risk" | jq -r .tier)
fi
echo "$risk" | jq -r '"tier=\(.tier) flags=\(.flags) new=\(.new_counterparty // "n/a") computed_at=\(.computed_at) reasons=\(.incomplete_reasons)"' >&2
case "$tier" in
  severe|high) echo refuse; exit 1 ;;
  none)        echo allow ;;
  *)           echo "escalate ($tier)"; exit 2 ;;
esac

Python

import json, sys, time, urllib.error, urllib.parse, urllib.request

BASE = "https://stablescan.achivx.com"

def scan(path, **params):
    url = BASE + path + ("?" + urllib.parse.urlencode(params) if params else "")
    req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "vet/1.0"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r), r.headers.get("X-Request-Id")

def vet(chain, addr, me):
    '''returns ("refuse"|"allow"|"escalate", detail dict)'''
    try:
        risk, rid = scan(f"/v1/addresses/{addr}/risk", chain=chain, **{"from": me})
        if risk["tier"] == "unknown" and "time_budget" in risk["incomplete_reasons"]:
            time.sleep(15)                                    # the cut verdict is reused for ~10 s
            risk, rid = scan(f"/v1/addresses/{addr}/risk", chain=chain, **{"from": me})
    except urllib.error.HTTPError as e:
        err = json.loads(e.read()).get("error", {})
        return "escalate", {"error": err.get("code"), "request_id": err.get("request_id")}
    detail = {"tier": risk["tier"], "flags": risk["flags"], "reasons": risk["incomplete_reasons"],
              "new_counterparty": risk.get("new_counterparty"), "computed_at": risk["computed_at"], "request_id": rid}
    if risk["tier"] in ("severe", "high"):
        facts, _ = scan(f"/v1/addresses/{addr}", chain=chain)      # the facts for the record
        detail["sanctioned"] = facts["sanctioned"]
        detail["frozen"] = [r["token"] for r in facts["freeze"] if r["status"] in ("frozen", "seized")]
        return "refuse", detail
    if risk["tier"] == "none":
        return "allow", detail
    if risk["tier"] == "unknown":
        exp, _ = scan(f"/v1/addresses/{addr}/exposure", chain=chain)   # how far the walk got, for the reviewer
        detail["walk"] = {"exposure": exp["exposure"], "hops_completed": exp["hops_completed"], "truncation_reason": exp.get("truncation_reason")}
    return "escalate", detail

if __name__ == "__main__":
    chain, addr, me = sys.argv[1:4]
    decision, detail = vet(chain, addr, me)
    print(decision); print(json.dumps(detail, indent=1), file=sys.stderr)
    sys.exit({"allow": 0, "refuse": 1, "escalate": 2}[decision])

Go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"time"
)

const base = "https://stablescan.achivx.com"

type risk struct {
	Tier              string          `json:"tier"`
	Flags             map[string]bool `json:"flags"`
	IncompleteReasons []string        `json:"incomplete_reasons"`
	NewCounterparty   string          `json:"new_counterparty"`
	ComputedAt        string          `json:"computed_at"`
}

func get(path string, q url.Values, out any) (string, error) {
	req, _ := http.NewRequest(http.MethodGet, base+path+"?"+q.Encode(), nil)
	req.Header.Set("Accept", "application/json")
	res, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
	if err != nil {
		return "", err
	}
	defer res.Body.Close()
	rid := res.Header.Get("X-Request-Id")
	if res.StatusCode >= 400 {
		var env struct {
			Error struct{ Code, Message, RequestID string } `json:"error"`
		}
		_ = json.NewDecoder(res.Body).Decode(&env)
		return rid, fmt.Errorf("%d %s: %s", res.StatusCode, env.Error.Code, env.Error.Message)
	}
	return rid, json.NewDecoder(res.Body).Decode(out)
}

// vet answers refuse, allow or escalate for one counterparty.
func vet(chain, addr, me string) (string, map[string]any) {
	q := url.Values{"chain": {chain}, "from": {me}}
	var r risk
	rid, err := get("/v1/addresses/"+addr+"/risk", q, &r)
	if err == nil && r.Tier == "unknown" && contains(r.IncompleteReasons, "time_budget") {
		time.Sleep(15 * time.Second)
		rid, err = get("/v1/addresses/"+addr+"/risk", q, &r)
	}
	if err != nil {
		return "escalate", map[string]any{"error": err.Error(), "request_id": rid}
	}
	detail := map[string]any{"tier": r.Tier, "flags": r.Flags, "reasons": r.IncompleteReasons, "new_counterparty": r.NewCounterparty, "computed_at": r.ComputedAt, "request_id": rid}
	switch r.Tier {
	case "severe", "high":
		return "refuse", detail
	case "none":
		return "allow", detail
	}
	if r.Tier == "unknown" {
		var exp struct {
			Exposure         string `json:"exposure"`
			HopsCompleted    int    `json:"hops_completed"`
			TruncationReason string `json:"truncation_reason"`
		}
		if _, err := get("/v1/addresses/"+addr+"/exposure", url.Values{"chain": {chain}}, &exp); err == nil {
			detail["walk"] = exp
		}
	}
	return "escalate", detail
}

func contains(xs []string, s string) bool {
	for _, x := range xs {
		if x == s {
			return true
		}
	}
	return false
}

func main() {
	decision, detail := vet(os.Args[1], os.Args[2], os.Args[3])
	fmt.Println(decision)
	json.NewEncoder(os.Stderr).Encode(detail)
	os.Exit(map[string]int{"allow": 0, "refuse": 1, "escalate": 2}[decision])
}

TypeScript

// vet.ts — run with: node --experimental-strip-types vet.ts <chain> <counterparty> <my-address>  (Node 22+), or compile with tsc
const BASE = "https://stablescan.achivx.com";
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

type Risk = { tier: "severe" | "high" | "medium" | "low" | "none" | "unknown"; flags: Record<string, boolean>; incomplete_reasons: string[]; new_counterparty?: string; computed_at: string };
type Exposure = { exposure: string; hops_completed: number; truncation_reason?: string };

async function scan<T>(path: string, params: Record<string, string>): Promise<{ body: T; requestId: string | null }> {
  const res = await fetch(`${BASE}${path}?${new URLSearchParams(params)}`, { headers: { accept: "application/json", "user-agent": "vet/1.0" } });
  const requestId = res.headers.get("x-request-id");
  if (!res.ok) {
    const { error } = (await res.json().catch(() => ({ error: { code: String(res.status), message: res.statusText } }))) as { error: { code: string; message: string } };
    throw new Error(`${res.status} ${error.code}: ${error.message} (request ${requestId})`);
  }
  return { body: (await res.json()) as T, requestId };
}

export async function vet(chain: string, addr: string, me: string): Promise<["refuse" | "allow" | "escalate", Record<string, unknown>]> {
  let risk: Risk, requestId: string | null;
  try {
    ({ body: risk, requestId } = await scan<Risk>(`/v1/addresses/${addr}/risk`, { chain, from: me }));
    if (risk.tier === "unknown" && risk.incomplete_reasons.includes("time_budget")) {
      await sleep(15_000);
      ({ body: risk, requestId } = await scan<Risk>(`/v1/addresses/${addr}/risk`, { chain, from: me }));
    }
  } catch (e) {
    return ["escalate", { error: String(e) }];
  }
  const detail: Record<string, unknown> = { tier: risk.tier, flags: risk.flags, reasons: risk.incomplete_reasons, new_counterparty: risk.new_counterparty, computed_at: risk.computed_at, requestId };
  if (risk.tier === "severe" || risk.tier === "high") return ["refuse", detail];
  if (risk.tier === "none") return ["allow", detail];
  if (risk.tier === "unknown") {
    const { body: exp } = await scan<Exposure>(`/v1/addresses/${addr}/exposure`, { chain });
    detail.walk = { exposure: exp.exposure, hops_completed: exp.hops_completed, truncation_reason: exp.truncation_reason };
  }
  return ["escalate", detail];
}

const [chain, addr, me] = process.argv.slice(2);
const [decision, detail] = await vet(chain, addr, me);
console.log(decision); console.error(JSON.stringify(detail, null, 1));
process.exit({ allow: 0, refuse: 1, escalate: 2 }[decision]);

예시에 대해 실행하기

2026-10-24에 vet 1 0x098B716B8Aaf21512996dC57EB0615e2383E2f96 <me>는 sanctioned: true와 동결된 토큰 세 개와 함께 refuse로 답했습니다. vet 1 0x28C6c06298d514Db089934071355E5743bf21d60 <me>는 두 번째 시도 후 tier: "unknown", reasons: ["time_budget"], flags: {"service": true}와 함께 escalate로 답했습니다 — 바이낸스(Binance) 핫월렛은 순회가 거의 완료되지 않는 허브이며, 검토자는 detail.walk에서 3홉 중 0홉이 완료되었음을 봅니다. 기록이 없는 새 지갑은 short_history와 함께 escalate로 답합니다. 어느 것도 프로그램의 버그가 아닙니다. 이것이 세 가지 결과입니다.

운영 환경에서 추가할 것

단순한 헬퍼 대신 오류와 한도의 재시도 클라이언트를 사용하고, 실행마다 /v1/sync/status를 한 번 호출해 오래된 네트워크를 판정 옆에 기록하며, 지급 후에는 /v1/verify?tx=를 호출합니다 — 다음 레시피입니다.

Stablecoin scanner ACHIVX에서

• from=이 있는 판정: /risk?chain=1&from=0x28C6…1d60.
• 매뉴얼이 직접 쓴 A2A 기반 루프: /agents.md §5.

출처

• 에이전트 매뉴얼 — agents.md(버전 2026-10-05)
• REST 인터페이스의 OpenAPI 3.1 명세(API 1.4.0)

더 알아보기

• 사람이 개입하지 않는 일반적인 루프는 무엇입니까?
• 주소 읽기
• 스크리닝 루프 처음부터 끝까지

관련 질문

• 주소에 대한 단일 위험 판정은 어떻게 얻고, 언제 unknown이 됩니까?
• 응답은 어떻게 읽습니까: unknown, null, 빈 목록, no-store
• 레시피: 결제가 도착했고 동결되지 않았는지 확인하기

레시피 · risk → overview → exposure · 2026-10-24 작성, openapi.json, agents.md 및 당일 실제 호출 기반. 구조는 인용한 것이며, 수치는 날짜 기준이고 인덱스에 따라 바뀝니다.