infinihash-kyc 0.1.0: Python SDK for the KYC API

.github/workflows/ci.yml

@@ -0,0 +1,32 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.10", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install ".[test]"
+      - name: Tests
+        run: pytest -q
+      - name: Examples import cleanly
+        run: |
+          python -m py_compile examples/*.py
+          python -c "import infinihash_kyc as k; print(k.__version__)"

README.md

@@ -1,2 +1,77 @@
-# kyc-python
-Python SDK for the Infinihash KYC API
+# infinihash-kyc
+
+Python SDK for the [Infinihash KYC API](https://kyc.infinihash.com): KYC and KYB cases,
+sanctions screening, and case reports.
+
+```bash
+pip install "git+https://github.com/Infinihash/kyc-python@v0.1.0"
+export INFINIHASH_KYC_KEY=...        # your API key
+```
+
+## Screen a name
+
+```python
+from infinihash_kyc import KYC, ScreeningUnavailable
+
+kyc = KYC()
+try:
+    r = kyc.screen("Vladimir Putin", country="RU")
+    print(r["status"], r["totalHits"])        # HIT 6
+except ScreeningUnavailable:
+    ...                                       # lists were NOT searched -- never treat as clear
+```
+
+## Onboard a company
+
+```python
+from infinihash_kyc import KYC, is_unscreened
+
+kyc = KYC()
+case = kyc.submissions.create(
+    entity_type="private_corp",
+    entity_name="Acme Widgets Ltd",
+    country="GB",
+    registration_number="01234567",
+)
+kyc.submissions.run(case["id"])
+result = kyc.submissions.wait(case["id"])      # complete | review | failed
+
+if is_unscreened(result):
+    print("manual review:", result["summary"]["error"])
+else:
+    print(result["summary"]["action"])         # approve | enhanced_review | manual_review | block_report
+pdf = kyc.submissions.export_pdf(case["id"])
+```
+
+Runnable versions are in [`examples/`](examples/).
+
+## Fail-closed by design
+
+- **`screen()` never returns an unscreened result.** It raises `ScreeningUnavailable`
+  on HTTP 503 or when the server reports `screened: false`.
+- **A case the API could not screen ends in `review` with `summary.error` set.** It
+  never ends in `complete`. `is_unscreened(case)` tells you which happened.
+- **`submissions.create()` rejects unknown fields locally,** because the API would
+  silently drop them. It also validates `entity_type` and `risk_tier` against the
+  same lists the API uses.
+
+## Reference
+
+| Method | API |
+|---|---|
+| `submissions.create(entity_type, entity_name, risk_tier="low", **fields)` | `POST /api/kyc` |
+| `submissions.run(id)` | `POST /api/kyc/{id}/run` |
+| `submissions.get(id)` / `wait(id, timeout=300)` | `GET /api/kyc/{id}` |
+| `submissions.list()` | `GET /api/kyc` |
+| `submissions.export_pdf(id)` | `GET /api/kyc/{id}/export/pdf` |
+| `screen(name, country=None, dob=None)` | `POST /api/kyc/screen` |
+
+Entity types: `individual, listed_public, wholly_owned_sub, pension_erisa, governmental,
+nonprofit, trust, spv, complex, private_corp`. The full schema is the
+[OpenAPI spec](https://kyc-api.infinihash.com/api/v1/openapi.json).
+
+Errors: any non-2xx raises `KYCError(status, message, body)`.
+
+## License
+
+MIT

examples/onboard_company.py

@@ -0,0 +1,30 @@
+"""Onboard a company end to end: create the case, run the checks, act on the decision.
+
+    INFINIHASH_KYC_KEY=... python examples/onboard_company.py
+"""
+from infinihash_kyc import KYC, is_unscreened
+
+kyc = KYC()
+
+case = kyc.submissions.create(
+    entity_type="private_corp",
+    entity_name="Acme Widgets Ltd",
+    country="GB",
+    registration_number="01234567",
+    address_line1="1 High Street",
+    city="London",
+    postal_code="EC1A 1AA",
+)
+print("created", case["id"], "required checks:", ", ".join(case["requiredChecks"]))
+
+kyc.submissions.run(case["id"])
+result = kyc.submissions.wait(case["id"], timeout=300)
+
+summary = result.get("summary") or {}
+if is_unscreened(result):
+    print("NOT SCREENED -- manual review required:", summary["error"])
+else:
+    print(f"decision: {summary.get('action')}  (status {result['status']}, "
+          f"risk {summary.get('riskScore')} / {summary.get('riskTier')})")
+    for c in result["checks"]:
+        print(f"  {c['checkType']:<24} {c['status']}")

examples/screen_name.py

@@ -0,0 +1,21 @@
+"""Screen one name against the sanctions lists.
+
+    INFINIHASH_KYC_KEY=... python examples/screen_name.py "Vladimir Putin" RU
+"""
+import sys
+
+from infinihash_kyc import KYC, ScreeningUnavailable
+
+name = sys.argv[1] if len(sys.argv) > 1 else "Vladimir Putin"
+country = sys.argv[2] if len(sys.argv) > 2 else None
+
+kyc = KYC()
+try:
+    r = kyc.screen(name, country=country)
+except ScreeningUnavailable as e:
+    # Never treat this as clear: the lists were not searched.
+    sys.exit(f"NOT SCREENED -- retry later ({e})")
+
+print(f"{r['name']}: {r['status']} ({r['totalHits']} hit(s))")
+for m in r["matches"][:5]:
+    print(f"  - {m.get('list')}: {m.get('name')} (score {m.get('score')})")

infinihash_kyc.egg-info/PKG-INFO

@@ -0,0 +1,19 @@
+Metadata-Version: 2.4
+Name: infinihash-kyc
+Version: 0.1.0
+Summary: Python SDK for the Infinihash KYC API (KYC/KYB cases, sanctions screening).
+Author: Infinihash
+License: MIT
+Project-URL: Homepage, https://kyc.infinihash.com
+Project-URL: API reference, https://kyc-api.infinihash.com/api/v1/openapi.json
+Project-URL: Source, https://github.com/Infinihash/kyc-python
+Requires-Python: >=3.8
+Description-Content-Type: text/markdown
+License-File: LICENSE
+Requires-Dist: requests>=2.28
+Provides-Extra: test
+Requires-Dist: pytest>=7; extra == "test"
+Dynamic: license-file
+
+# kyc-python
+Python SDK for the Infinihash KYC API

infinihash_kyc.egg-info/SOURCES.txt

@@ -0,0 +1,11 @@
+LICENSE
+README.md
+pyproject.toml
+infinihash_kyc/__init__.py
+infinihash_kyc/client.py
+infinihash_kyc.egg-info/PKG-INFO
+infinihash_kyc.egg-info/SOURCES.txt
+infinihash_kyc.egg-info/dependency_links.txt
+infinihash_kyc.egg-info/requires.txt
+infinihash_kyc.egg-info/top_level.txt
+tests/test_client.py

infinihash_kyc.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

infinihash_kyc.egg-info/requires.txt

@@ -0,0 +1,4 @@
+requests>=2.28
+
+[test]
+pytest>=7

infinihash_kyc.egg-info/top_level.txt

@@ -0,0 +1 @@
+infinihash_kyc

infinihash_kyc/__init__.py

@@ -0,0 +1,5 @@
+"""Infinihash KYC Python SDK. See client.py for the full guide."""
+from .client import KYC, KYCError, ScreeningUnavailable, is_unscreened, ENTITY_TYPES, RISK_TIERS
+
+__all__ = ["KYC", "KYCError", "ScreeningUnavailable", "is_unscreened", "ENTITY_TYPES", "RISK_TIERS"]
+__version__ = "0.1.0"

infinihash_kyc/client.py

@@ -0,0 +1,174 @@
+"""Infinihash KYC Python SDK.
+
+Thin `requests` wrapper around the Infinihash KYC API (https://kyc-api.infinihash.com).
+Every call returns the JSON the server sent; errors surface as exceptions.
+
+Quick start
+-----------
+    from infinihash_kyc import KYC
+
+    kyc = KYC(api_key="...")                    # or set INFINIHASH_KYC_KEY
+
+    # Onboard a company: create the case, run the checks, wait for the decision
+    case = kyc.submissions.create(
+        entity_type="private_corp",
+        entity_name="Acme Widgets Ltd",
+        country="GB",
+        registration_number="01234567",
+    )
+    kyc.submissions.run(case["id"])
+    result = kyc.submissions.wait(case["id"])
+    print(result["status"], result["summary"]["action"])
+
+    # Screen one name against the sanctions lists
+    hit = kyc.screen("Vladimir Putin", country="RU")
+    print(hit["status"], hit["totalHits"])
+
+Fail-closed by design
+---------------------
+* `screen()` raises ScreeningUnavailable when the server says it did not screen
+  (HTTP 503 or `screened: false`). It never returns a result you could mistake
+  for "clear".
+* A case that could not be screened finishes in status `review` with
+  `summary.error` set. `wait()` returns it; `is_unscreened()` tells you.
+
+Errors
+------
+Non-2xx responses raise KYCError(status, message, body).
+"""
+from __future__ import annotations
+
+import os
+import time
+from typing import Any, Dict, List, Optional
+
+import requests
+
+DEFAULT_BASE_URL = "https://kyc-api.infinihash.com"
+DEFAULT_TIMEOUT = 30
+
+ENTITY_TYPES = [
+    "individual", "listed_public", "wholly_owned_sub", "pension_erisa", "governmental",
+    "nonprofit", "trust", "spv", "complex", "private_corp",
+]
+RISK_TIERS = ["low", "medium", "high"]
+TERMINAL_STATUSES = {"complete", "review", "failed"}
+
+# snake_case keyword -> API field. Anything not listed is rejected locally,
+# because the server silently drops unknown fields.
+_FIELDS = {
+    "lei": "lei", "crd": "crd", "jurisdiction": "jurisdiction", "ticker": "ticker",
+    "country": "country", "email": "email", "naics_code": "naicsCode",
+    "business_description": "businessDescription", "registration_number": "registrationNumber",
+    "incorporation_date": "incorporationDate", "principal_place_of_business": "principalPlaceOfBusiness",
+    "dob": "dob", "nationality": "nationality", "smart_import": "smartImport", "source_doc": "sourceDoc",
+    "address_line1": "addressLine1", "address_line2": "addressLine2", "city": "city",
+    "region": "region", "postal_code": "postalCode",
+    "officers": "officers", "owners": "owners", "signatory": "signatory",
+}
+
+
+class KYCError(Exception):
+    """Any non-2xx response. Attributes: status (int), body (dict | str | None)."""
+
+    def __init__(self, status: int, message: str, body: Any = None):
+        super().__init__(f"HTTP {status}: {message}")
+        self.status = status
+        self.body = body
+
+
+class ScreeningUnavailable(KYCError):
+    """The server did not screen the name. Never treat this as a clear result."""
+
+
+class _Submissions:
+    def __init__(self, client: "KYC"):
+        self._c = client
+
+    def create(self, entity_type: str, entity_name: str, risk_tier: str = "low", **fields: Any) -> Dict[str, Any]:
+        """POST /api/kyc. Returns {id, status, requiredChecks, _links, ...}."""
+        if entity_type not in ENTITY_TYPES:
+            raise ValueError(f"entity_type must be one of {ENTITY_TYPES}")
+        if risk_tier not in RISK_TIERS:
+            raise ValueError(f"risk_tier must be one of {RISK_TIERS}")
+        unknown = sorted(set(fields) - set(_FIELDS))
+        if unknown:
+            raise ValueError(f"unknown field(s) {unknown}; the API would silently drop them")
+        body = {"entityType": entity_type, "entityName": entity_name, "riskTier": risk_tier}
+        body.update({_FIELDS[k]: v for k, v in fields.items() if v is not None})
+        return self._c._request("POST", "/api/kyc", json=body)
+
+    def run(self, submission_id: str) -> Dict[str, Any]:
+        """POST /api/kyc/{id}/run. Starts the checks (asynchronous)."""
+        return self._c._request("POST", f"/api/kyc/{submission_id}/run")
+
+    def get(self, submission_id: str) -> Dict[str, Any]:
+        """GET /api/kyc/{id}. Status, checks and (once finished) the summary."""
+        return self._c._request("GET", f"/api/kyc/{submission_id}")
+
+    def list(self) -> List[Dict[str, Any]]:
+        """GET /api/kyc. Your submissions, newest first."""
+        return self._c._request("GET", "/api/kyc")
+
+    def wait(self, submission_id: str, timeout: float = 300, poll: float = 3) -> Dict[str, Any]:
+        """Poll until the case reaches complete / review / failed, or raise TimeoutError."""
+        deadline = time.monotonic() + timeout
+        while True:
+            sub = self.get(submission_id)
+            if sub.get("status") in TERMINAL_STATUSES:
+                return sub
+            if time.monotonic() >= deadline:
+                raise TimeoutError(f"submission {submission_id} still {sub.get('status')!r} after {timeout}s")
+            time.sleep(poll)
+
+    def export_pdf(self, submission_id: str) -> bytes:
+        """GET /api/kyc/{id}/export/pdf. The case report as PDF bytes."""
+        return self._c._request("GET", f"/api/kyc/{submission_id}/export/pdf", raw=True)
+
+
+def is_unscreened(submission: Dict[str, Any]) -> bool:
+    """True when the case finished without being screened (fail-closed review)."""
+    return bool(((submission or {}).get("summary") or {}).get("error"))
+
+
+class KYC:
+    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None,
+                 timeout: float = DEFAULT_TIMEOUT, session: Optional[requests.Session] = None):
+        self.api_key = api_key or os.environ.get("INFINIHASH_KYC_KEY")
+        if not self.api_key:
+            raise ValueError("api_key is required (or set INFINIHASH_KYC_KEY)")
+        self.base_url = (base_url or os.environ.get("INFINIHASH_KYC_URL") or DEFAULT_BASE_URL).rstrip("/")
+        self.timeout = timeout
+        self._s = session or requests.Session()
+        self._s.headers.update({"X-API-Key": self.api_key, "User-Agent": "infinihash-kyc-python/0.1.0"})
+        self.submissions = _Submissions(self)
+
+    def screen(self, name: str, country: Optional[str] = None, dob: Optional[str] = None) -> Dict[str, Any]:
+        """POST /api/kyc/screen. Raises ScreeningUnavailable if the name was not screened."""
+        body = {"name": name, **({"country": country} if country else {}), **({"dob": dob} if dob else {})}
+        try:
+            out = self._request("POST", "/api/kyc/screen", json=body)
+        except KYCError as e:
+            if e.status == 503:
+                raise ScreeningUnavailable(e.status, "screening unavailable -- NOT a clear result", e.body) from None
+            raise
+        if out.get("screened") is False or out.get("status") == "ERROR":
+            raise ScreeningUnavailable(200, "server reports the name was not screened", out)
+        return out
+
+    def health(self) -> Dict[str, Any]:
+        return self._request("GET", "/health", auth=False)
+
+    def _request(self, method: str, path: str, json: Any = None, raw: bool = False, auth: bool = True):
+        headers = None if auth else {"X-API-Key": None}
+        r = self._s.request(method, self.base_url + path, json=json, timeout=self.timeout, headers=headers)
+        if not 200 <= r.status_code < 300:
+            try:
+                body = r.json()
+                msg = body.get("error") if isinstance(body, dict) else str(body)
+            except ValueError:
+                body, msg = r.text, r.text[:200]
+            raise KYCError(r.status_code, msg or r.reason, body)
+        if raw:
+            return r.content
+        return r.json()