CompanyBelgium
BCE/KBO API in Python: complete tutorial with FastAPI, async/await and copy-paste code
Tutorials
APIBCE/KBOIntegrationTutorialPerformance

BCE/KBO API in Python: complete tutorial with FastAPI, async/await and copy-paste code

Integrate the BCE/KBO API in Python in under an hour. Full tutorial: setup, httpx async requests, FastAPI wrapper, error handling, exponential retry, local cache. All code is directly copy-paste-able.

May 6, 20269 min read

In brief

This tutorial shows you how to integrate the Company Belgium BCE/KBO API in Python in under an hour, with copy-paste-ready code. You build an async httpx client, an exponential retry for 429s and 5xx errors, a TTL cache in memory, and a FastAPI wrapper that exposes your own business routes. The result fits in 100 lines and is directly deployable to production.

The result at the end of the article

By the end of this tutorial you will have:

  • An async Python client for the Company Belgium BCE/KBO API
  • A FastAPI wrapper exposing a /companies/{number} proxy route
  • An exponential retry strategy for network errors
  • A local in-memory cache to reduce calls
  • Clean error handling (404, 429, 5xx)

All code is tested and copy-paste-able into your project.

1. Prerequisites

  • Python 3.11+ (for native async support)
  • A Company Belgium API key (free plan is enough to follow along)
  • A terminal and an 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

Create a .env file:

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. The minimal async client

Create bce_client.py:

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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"Name: {company[class="code-string">'name']}")
        print(class="code-string">f"Form: {company[class="code-string">'legalForm']}")
        print(class="code-string">f"Address: {company[class="code-string">'address'][class="code-string">'fullAddress']}")
    finally:
        await client.close()


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

Run:

Terminal 
1
python bce_client.py

You get a Belgian company's record in JSON, in one request.

4. Exponential retry for transient errors

The API may return 429 (rate limit) or 5xx (transient outage). Instead of failing, retry with backoff.

Add retry.py:

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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")

Use it in the client:

Python 
1
2
3
4
5
6
7
8
9
10
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. Local in-memory cache

To avoid paying for the same request repeatedly, add a TTL cache:

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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)

Integrate into the client:

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
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

For a shared cache across processes (production), switch to memcached, or use the HTTP cache layer via Cache-Control headers. The principle is the same.

6. FastAPI wrapper

Expose your own API consuming the BCE API and adding business logic.

Create main.py:

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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}")

Run:

Terminal 
1
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. Typical use cases

A. Verify a supplier before payment

Python 
1
2
3
4
5
6
7
8
9
10
11
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 = Active
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return False
        raise
    finally:
        await client.close()

B. Pre-fill a client form

Python 
1
2
3
4
5
6
7
8
9
10
11
12
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. Bulk verification (with semaphore for rate limit)

Python 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import asyncio

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

    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. Mistakes to avoid

  • Opening one HTTP client per request — costly, breaks keep-alive. Reuse a client.
  • Not handling 429 — you will get throttled fast. Implement retry.
  • Not caching — you pay 10× the same call for the same SME.
  • No timeout — your workers will block. Always set a timeout.
  • Logging the API key — use os.getenv, never hardcode. See security best practices.

    • How Company Belgium fits into a Python stack

    The Company Belgium API is REST/JSON and compatible with any Python HTTP client (requests, httpx, aiohttp). No SDK required to start, but:

    • OpenAPI doc available — you can generate a typed client with openapi-python-client
    • Webhooks: receive changes on your FastAPI endpoint, rather than polling
    • Rate limit headers: X-RateLimit-Remaining, Retry-After — your retry code uses them directly
    • Official Python SDK available on pro plans (cache, retry and types built in)

    Further reading: see the BCE API comparison, the endpoints reference and the rate limits / caching architecture.

    Bottom line

    Integrating the BCE API in Python fits in 100 lines:

    • Async httpx client with auth header
    • Exponential retry on transient codes
    • TTL cache in memory
    • FastAPI wrapper to expose your own business routes
    • Semaphore for bulk operations

    The code in this tutorial is production-ready. Tune TTLs and concurrency per your plan, and you have a robust integration.

    Frequently asked questions

    How do you integrate the BCE/KBO API in Python with httpx and async/await?

    Create a BCEClient class that instantiates an httpx.AsyncClient with the X-API-Key and X-API-Secret headers from your Company Belgium account. The get_company and search methods use await calls for non-blocking requests. The full code fits in about 40 lines and integrates naturally into a FastAPI or Django async stack. API keys are loaded via os.getenv so they are never hardcoded.

    Why use httpx instead of requests for the BCE/KBO API in Python?

    httpx natively supports async/await, which is essential for high-concurrency applications like FastAPI apps or bulk verification scripts. With requests (synchronous), every API call blocks the thread; with async httpx, you can process hundreds of requests in parallel with a semaphore. httpx is also compatible with async test contexts and produces reusable clients with HTTP keep-alive connections, which reduces latency.

    How do you implement exponential retry for 429 errors from the BCE/KBO API in Python?

    Create a with_retry function that wraps your API call in a loop. On status 429 or 5xx, wait base_delay * 2^attempt seconds before retrying, honouring the Retry-After header if present. Add random jitter to avoid the thundering herd if multiple instances run in parallel. After 4 failed attempts, raise an exception. This pattern is identical in Python and Node.js.

    Can FastAPI be used as a proxy for the BCE/KBO API in a Belgian application?

    Yes, this is a common architecture. Create a FastAPI application that instantiates your BCEClient at startup via a lifespan context manager, then expose routes such as GET /companies/{number} and GET /search that delegate calls to the BCEClient. This proxy lets you add your own authentication, business logic, additional caching and custom logs before calling the official API. The added latency is under 5 ms locally.

    Ready to get started?

    Create your free account and get your API keys in minutes.

    Comments

    Loading comments…

    Leave a comment

    Not published — used only to notify you.

    Comments are moderated before publication.

    Related articles

    Accountant: automating the financial analysis of your customer portfolio

    Accountant: automating the financial analysis of your customer portfolio

    An accountant manages 50 to 300 client files. Manually analysing annual accounts, computing ratios, spotting red flags and preparing a report for each yearly meeting takes hundreds of hours a year. Here is how to automate 80% of that work while raising service quality.

    Calculating the credit limit of a B2B customer: 4 proven methods

    Calculating the credit limit of a B2B customer: 4 proven methods

    How much credit exposure can you safely grant a B2B customer? Discover the four methods used by credit insurers and finance departments of Belgian SMEs: equity cap, turnover cap, asset cap and financial scoring — with worked examples.

    Searching Belgian companies in natural language: the complete /recherche guide

    Searching Belgian companies in natural language: the complete /recherche guide

    Type "accountants in Liège", "restaurants 1000 Brussels" or a BCE number: the /recherche bar turns your sentence into NACE, postcode and name filters, then returns a page enriched with sector statistics. Tour of use cases, URL shortcuts and linguistic subtleties.

    Back to blog