Hoe integreert u de KBO/BCE-API in Python met httpx en async/await?

Maak een klasse BCEClient die een httpx.AsyncClient instantieert met de headers X-API-Key en X-API-Secret van uw Company Belgium-account. De methoden get_company en search gebruiken await-aanroepen voor niet-blokkerende verzoeken. De volledige code past in circa 40 regels en integreert van nature in een FastAPI- of Django async-stack. API-sleutels worden geladen via os.getenv zodat ze nooit hardgecodeerd worden.

Waarom httpx gebruiken in plaats van requests voor de KBO/BCE-API in Python?

httpx ondersteunt natively async/await, wat essentieel is voor applicaties met hoge concurrentie zoals FastAPI-apps of massaverificatiescripts. Met requests (synchroon) blokkeert elke API-aanroep de thread; met httpx async verwerkt u honderden verzoeken parallel met een semafoor. httpx is ook compatibel met async testcontexten en genereert herbruikbare clients met HTTP keep-alive-verbindingen, wat de latency verlaagt.

Hoe implementeert u een exponentiële retry voor 429-fouten van de KBO/BCE-API in Python?

Maak een functie with_retry die uw API-aanroep inpakt in een lus. Bij status 429 of 5xx wacht u base_delay * 2^attempt seconden voor een nieuwe poging, waarbij u de header Retry-After respecteert als die aanwezig is. Voeg willekeurige jitter toe om thundering herd te vermijden als meerdere instanties parallel draaien. Na 4 mislukte pogingen gooit u een uitzondering. Dit patroon is identiek in Python en Node.js.

Kan FastAPI als proxy voor de KBO/BCE-API worden gebruikt in een Belgische applicatie?

Ja, dit is een gangbare architectuur. Maak een FastAPI-applicatie die uw BCEClient bij het opstarten instantieert via een lifespan contextmanager, en stel vervolgens routes bloot zoals GET /companies/{number} en GET /search die de aanroepen delegeren aan de BCEClient. Deze proxy laat u toe uw eigen authenticatie, business-logica, extra caching en aangepaste logs toe te voegen voor de officiele API te roepen. De toegevoegde latency is minder dan 5 ms lokaal.

KBO/BCE-API Python: volledige FastAPI async tutorial 2026

In het kort

Deze tutorial laat u zien hoe u de KBO/BCE-API van Company Belgium in Python integreert in minder dan een uur, met kopieerbare code. U bouwt een async httpx-client, een exponentiële retry voor 429 en 5xx-fouten, een TTL-cache in geheugen, en een FastAPI-wrapper die uw eigen business-routes blootstelt. Het resultaat past in 100 regels en is direct inzetbaar in productie.

Het resultaat aan het einde van het artikel

Aan het einde van deze tutorial heeft u:

Een async Python-client voor de KBO/BCE-API van Company Belgium

Een FastAPI-wrapper die een route /companies/{number} proxy blootstelt

Een exponentiële retry-strategie voor netwerkfouten

Een lokale in-memory cache om calls te beperken

Schone foutafhandeling (404, 429, 5xx)

Alle code is getest en direct kopieerbaar in uw project.

1. Vereisten

Python 3.11+ (voor native async)

Een API-key Company Belgium (gratis plan volstaat)

Een terminal en een editor

2. Setup

Terminal

1
2
3

mkdir bce-python-tuto && cd bce-python-tuto
python -m venv .venv && source .venv/bin/activate
pip install httpx fastapi uvicorn python-dotenv

Maak een .env-bestand:

Terminal

1
2
3

COMPANY_BELGIUM_API_KEY=pk_live_xxxxxxxxxxxx
COMPANY_BELGIUM_API_SECRET=sk_live_xxxxxxxxxxxx
COMPANY_BELGIUM_BASE_URL=https://api.companybelgium.be/v1

3. De minimale async client

Maak bce_client.py:

Python

import os
import asyncio
import httpx
from dotenv import load_dotenv

load_dotenv()

class BCEClient:
    def __init__(self, base_url=None, api_key=None, api_secret=None, timeout=10.0):
        self.base_url = base_url or os.getenv("COMPANY_BELGIUM_BASE_URL")
        self.api_key = api_key or os.getenv("COMPANY_BELGIUM_API_KEY")
        self.api_secret = api_secret or os.getenv("COMPANY_BELGIUM_API_SECRET")
        self.timeout = timeout
        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "X-API-Key": self.api_key,
                "X-API-Secret": self.api_secret,
            },
            timeout=timeout,
        )

    async def get_company(self, enterprise_number: str) -> dict:
        clean = enterprise_number.replace(".", "").replace(" ", "")
        r = await self._client.get(class="code-string">f"/companies/{clean}")
        r.raise_for_status()
        return r.json()["data"]

    async def search(self, query: str, limit: int = 20) -> list[dict]:
        r = await self._client.get("/companies/search", params={"q": query, "limit": limit})
        r.raise_for_status()
        return r.json()["data"]

    async def close(self):
        await self._client.aclose()


async def demo():
    client = BCEClient()
    try:
        company = await client.get_company("0200.065.765")
        print(class="code-string">f"Naam: {company[class="code-string">'name']}")
        print(class="code-string">f"Vorm: {company[class="code-string">'legalForm']}")
        print(class="code-string">f"Adres: {company[class="code-string">'address'][class="code-string">'fullAddress']}")
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(demo())

Voer uit:

Terminal

python bce_client.py

U krijgt de fiche van een Belgische onderneming in JSON, in één verzoek.

4. Exponentiële retry voor transiente fouten

De API kan 429 (rate limit) of 5xx (tijdelijke storing) retourneren. In plaats van te falen, retry met backoff.

Voeg retry.py toe:

Python

import asyncio
import httpx
from typing import Callable, TypeVar, Awaitable

T = TypeVar("T")

async def with_retry(
    fn: Callable[[], Awaitable[T]],
    max_attempts: int = 4,
    base_delay: float = 0.5,
) -> T:
    last_exc = None
    for attempt in range(max_attempts):
        try:
            return await fn()
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status in (429, 500, 502, 503, 504) and attempt < max_attempts - 1:
                delay = base_delay * (2 ** attempt)
                retry_after = e.response.headers.get("Retry-After")
                if retry_after and retry_after.isdigit():
                    delay = max(delay, float(retry_after))
                await asyncio.sleep(delay)
                last_exc = e
                continue
            raise
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempt < max_attempts - 1:
                await asyncio.sleep(base_delay * (2 ** attempt))
                last_exc = e
                continue
            raise
    if last_exc:
        raise last_exc
    raise RuntimeError("retry exhausted")

Gebruik in de client:

Python

from retry import with_retry

async def get_company(self, enterprise_number: str) -> dict:
    clean = enterprise_number.replace(".", "").replace(" ", "")
    return await with_retry(lambda: self._get_company_once(clean))

async def _get_company_once(self, num: str) -> dict:
    r = await self._client.get(class="code-string">f"/companies/{num}")
    r.raise_for_status()
    return r.json()["data"]

5. Lokale in-memory cache

Om hetzelfde verzoek niet steeds opnieuw te betalen, voeg een TTL-cache toe:

Python

import time
from typing import Any

class TTLCache:
    def __init__(self, ttl_seconds: int = 300):
        self.ttl = ttl_seconds
        self._store: dict[str, tuple[float, Any]] = {}

    def get(self, key: str):
        item = self._store.get(key)
        if not item:
            return None
        expires_at, value = item
        if time.time() > expires_at:
            del self._store[key]
            return None
        return value

    def set(self, key: str, value: Any):
        self._store[key] = (time.time() + self.ttl, value)

Integreer in de client:

Python

class BCEClient:
    def __init__(self, ..., cache_ttl: int = 300):
        ...
        self.cache = TTLCache(ttl_seconds=cache_ttl)

    async def get_company(self, enterprise_number: str) -> dict:
        clean = enterprise_number.replace(".", "").replace(" ", "")
        cached = self.cache.get(class="code-string">f"company:{clean}")
        if cached:
            return cached
        data = await with_retry(lambda: self._get_company_once(clean))
        self.cache.set(class="code-string">f"company:{clean}", data)
        return data

Voor een gedeelde cache tussen processen (productie), schakel over op memcached, of gebruik de HTTP-cachelaag via Cache-Control-headers. Het principe blijft hetzelfde.

6. FastAPI-wrapper

Stel uw eigen API bloot die de BCE-API consumeert en uw business-logica toevoegt.

Maak main.py:

Python

from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
import httpx
from bce_client import BCEClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.bce = BCEClient()
    try:
        yield
    finally:
        await app.state.bce.close()

app = FastAPI(lifespan=lifespan)

@app.get("/companies/{number}")
async def get_company(number: str):
    try:
        return await app.state.bce.get_company(number)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            raise HTTPException(404, "Company not found")
        raise HTTPException(502, class="code-string">f"Upstream error: {e.response.status_code}")

@app.get("/search")
async def search(q: str, limit: int = 20):
    try:
        return await app.state.bce.search(q, limit)
    except httpx.HTTPStatusError as e:
        raise HTTPException(502, class="code-string">f"Upstream error: {e.response.status_code}")

Start:

Terminal

uvicorn main:app --reload --port 8000

Test:

Terminal

1
2

curl http://localhost:8000/companies/0200.065.765
curl "http://localhost:8000/search?q=espero"

7. Typische gebruikscases

A. Een leverancier verifiëren vóór betaling

Python

async def is_supplier_valid(vat_number: str) -> bool:
    client = BCEClient()
    try:
        company = await client.get_company(vat_number)
        return company["status"] == "AC"  class="code-comment"># AC = Actief
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return False
        raise
    finally:
        await client.close()

B. Een cliëntformulier voorinvullen

Python

async def prefill_client_form(vat_number: str) -> dict:
    client = BCEClient()
    try:
        c = await client.get_company(vat_number)
        return {
            "name": c["name"],
            "legalForm": c["legalForm"],
            "address": c["address"]["fullAddress"],
            "nace_main": c["mainActivity"]["naceCode"],
        }
    finally:
        await client.close()

C. Massaverificatie (met semafoor voor rate limit)

Python

import asyncio

async def verify_many(vat_numbers: list[str]) -> dict[str, bool]:
    client = BCEClient()
    sem = asyncio.Semaphore(5)  class="code-comment"># max 5 gelijktijdige verzoeken

    async def check(n: str):
        async with sem:
            try:
                company = await client.get_company(n)
                return n, company["status"] == "AC"
            except Exception:
                return n, False

    try:
        results = await asyncio.gather(*[check(n) for n in vat_numbers])
        return dict(results)
    finally:
        await client.close()

8. Te vermijden fouten

Een HTTP-client per verzoek openen — duur, breekt keep-alive. Hergebruik een client.

429 niet afhandelen — u wordt snel gethrottled. Implementeer retry.

Niet cachen — u betaalt 10× dezelfde call voor dezelfde kmo.

Geen timeout — uw workers blokkeren. Stel systematisch een timeout in.

API-key loggen — gebruik os.getenv, hardcode nooit. Zie de beveiligingsprincipes.

Hoe Company Belgium aansluit in een Python-stack

De Company Belgium-API is REST/JSON en compatibel met elke Python HTTP-client (requests, httpx, aiohttp). Geen SDK vereist om te starten, maar:

OpenAPI-doc beschikbaar — u kunt een getypeerde client genereren met openapi-python-client

Webhooks: ontvang wijzigingen op uw FastAPI-endpoint in plaats van te pollen

Rate limit headers: X-RateLimit-Remaining, Retry-After — uw retry-code gebruikt ze direct

Officiële Python-SDK beschikbaar op pro-plannen (met cache, retry en geïntegreerde types)

Verder lezen: zie de BCE-API-vergelijking, de endpoints-referentie en de rate limits / caching-architectuur.

Samenvatting

De integratie van de BCE-API in Python past in 100 regels:

Async httpx-client met auth header

Exponentiële retry op transiente codes

TTL-cache in geheugen

FastAPI-wrapper om eigen business-routes bloot te stellen

Semafoor voor massa-operaties

De code in deze tutorial is direct productieklaar. Pas de TTL's en concurrenties aan op basis van uw plan, en u hebt een robuuste integratie.

KBO/BCE-API in Python: volledige tutorial met FastAPI, async/await en kopieerbare code

In het kort

Het resultaat aan het einde van het artikel

1. Vereisten

2. Setup

3. De minimale async client

4. Exponentiële retry voor transiente fouten

5. Lokale in-memory cache

6. FastAPI-wrapper

7. Typische gebruikscases

A. Een leverancier verifiëren vóór betaling

B. Een cliëntformulier voorinvullen

C. Massaverificatie (met semafoor voor rate limit)

8. Te vermijden fouten

Hoe Company Belgium aansluit in een Python-stack

Samenvatting

Veelgestelde vragen

Klaar om te beginnen?

Reacties

Laat een reactie achter

Gerelateerde artikelen

Accountant: de financiële analyse van uw klantenportefeuille automatiseren

De kredietlimiet van een B2B-klant berekenen: 4 beproefde methodes

Belgische bedrijven zoeken in natuurlijke taal: de volledige gids voor /recherche

Altman Z-Score voor Belgische kmo's: praktische gids met cijfervoorbeelden

MAR 2026: de essentiële gids voor het Belgische rekeningenstelsel

Ondernemingsnummer vs vestigingseenheid: het verschil in de KBO

Notarissen: uiteindelijke begunstigden verifiëren vóór een authentieke akte