Python Income Tax API Integration Guide

Installation

The API works with the standard requests library. For async support (FastAPI, asyncio), use httpx. Neither library requires special configuration for the API.

# Synchronous (most use cases)
pip install requests

# Async support
pip install httpx

# Optional: load .env files for local development
pip install python-dotenv

No other dependencies are required. The API returns JSON, which Python's built-in json module handles — requests and httpx both expose a .json() method that handles parsing automatically.

Basic request

A minimal working example — calculate take-home pay for a UK gross salary of £60,000:

import os
import requests

API_KEY = os.environ.get("COUNTRYTAXCALC_API_KEY")
BASE_URL = "https://api.countrytaxcalc.com/api/v1"

response = requests.post(
    f"{BASE_URL}/calculate",
    headers={
        "Content-Type": "application/json",
        "X-API-Key": API_KEY
    },
    json={
        "income": 60000,
        "country": "GB"
    },
    timeout=10  # always set a timeout
)

response.raise_for_status()  # raises HTTPError for 4xx/5xx

data = response.json()

print(f"Country:        {data['country']}")
print(f"Gross income:   £{data['gross_income']:,}")
print(f"Net income:     £{data['totals']['net_income']:,}")
print(f"Monthly net:    £{data['totals']['monthly_net_income']:,.2f}")
print(f"Effective rate: {data['totals']['effective_tax_rate']}%")
print(f"Data source:    {data['metadata']['data_source']}")
print(f"Confidence:     {data['metadata']['accuracy_confidence']}")

Expected output:

Country:        United Kingdom
Gross income:   £60,000
Net income:     £45,357
Monthly net:    £3,779.75
Effective rate: 24.41%
Data source:    HMRC
Confidence:     0.95

Passing optional fields

# US federal + California state tax
response = requests.post(
    f"{BASE_URL}/calculate",
    headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
    json={
        "income": 120000,
        "country": "US",
        "jurisdiction": "CA",  # California
        "currency": "USD"
    },
    timeout=10
)

# Germany with income in USD (auto-converted to EUR for calculation)
response = requests.post(
    f"{BASE_URL}/calculate",
    headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
    json={
        "income": 100000,
        "country": "DE",
        "currency": "USD"  # API converts USD → EUR, then calculates
    },
    timeout=10
)

Response parsing

The response is nested JSON. Here's a helper that extracts the most commonly needed values:

def parse_tax_result(data: dict) -> dict:
    """Extract key tax figures from the API response."""
    totals = data["totals"]
    breakdown = data["breakdown"]
    metadata = data["metadata"]

    # Sum social contributions
    social_total = sum(
        c["amount"] for c in breakdown.get("social_contributions", [])
    )

    # Format confidence label
    confidence = metadata["accuracy_confidence"]
    confidence_label = "verified" if confidence >= 0.9 else "estimated"

    return {
        "country": data["country"],
        "country_code": data["country_code"],
        "currency": data["currency"],
        "currency_symbol": data["currency_symbol"],
        "gross_income": data["gross_income"],
        "net_income": totals["net_income"],
        "monthly_net": totals["monthly_net_income"],
        "total_tax": totals["total_tax"],
        "income_tax": breakdown["national_income_tax"]["amount"],
        "social_tax": social_total,
        "effective_rate": totals["effective_tax_rate"],
        "take_home_pct": totals["take_home_percentage"],
        "marginal_rate": breakdown["national_income_tax"]["marginal_rate"],
        "data_source": metadata["data_source"],
        "confidence": confidence,
        "confidence_label": confidence_label,
    }

# Usage
data = response.json()
result = parse_tax_result(data)
print(f"{result['country']}: {result['currency_symbol']}{result['net_income']:,} net "
      f"({result['effective_rate']}% effective, {result['confidence_label']})")

Accessing deductions and brackets

data = response.json()
breakdown = data["breakdown"]

# Named deductions (e.g. Personal Allowance)
for deduction in breakdown["deductions"]:
    print(f"  {deduction['name']}: {deduction['amount']:,}")

# Tax brackets applied
brackets = breakdown["national_income_tax"]["brackets_applied"]
for bracket in brackets:
    print(f"  {bracket['rate']*100:.0f}% on income above {bracket['threshold']:,}: "
          f"tax = {bracket['tax']:,}")

# Social contributions
for contrib in breakdown["social_contributions"]:
    print(f"  {contrib['name']} ({contrib['type']}): {contrib['amount']:,} "
          f"at {contrib['rate']*100:.1f}%")

Error handling

A production-grade client should handle all error cases explicitly. The API uses standard HTTP status codes with a consistent JSON error body:

import os
import time
import requests
from requests.exceptions import RequestException

class TaxAPIError(Exception):
    def __init__(self, status_code: int, error: str, message: str):
        self.status_code = status_code
        self.error = error
        self.message = message
        super().__init__(f"Tax API {status_code}: {message}")

def calculate_tax(income: float, country: str, currency: str = None,
                  jurisdiction: str = None, retries: int = 2) -> dict:
    """
    Call POST /calculate with retry logic for transient errors.
    Raises TaxAPIError for non-retryable errors.
    """
    API_KEY = os.environ.get("COUNTRYTAXCALC_API_KEY")
    url = "https://api.countrytaxcalc.com/api/v1/calculate"

    payload = {"income": income, "country": country}
    if currency:
        payload["currency"] = currency
    if jurisdiction:
        payload["jurisdiction"] = jurisdiction

    for attempt in range(retries + 1):
        try:
            resp = requests.post(
                url,
                headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
                json=payload,
                timeout=10
            )
        except RequestException as e:
            if attempt < retries:
                time.sleep(2 ** attempt)  # exponential backoff
                continue
            raise

        if resp.status_code == 200:
            return resp.json()

        # Try to parse error body
        try:
            err = resp.json()
            error_code = err.get("error", "unknown_error")
            message = err.get("message", resp.text)
        except Exception:
            error_code = "parse_error"
            message = resp.text

        if resp.status_code == 429:
            # Burst limit — retry after waiting
            retry_after = int(resp.headers.get("Retry-After", 60))
            if attempt < retries:
                time.sleep(retry_after)
                continue
            raise TaxAPIError(429, error_code, message)

        if resp.status_code >= 500 and attempt < retries:
            # Server error — retry with backoff
            time.sleep(2 ** attempt)
            continue

        # Non-retryable: 400, 401, 404, 422
        raise TaxAPIError(resp.status_code, error_code, message)

    raise TaxAPIError(0, "max_retries", "Maximum retries exceeded")


# Usage with error handling
try:
    result = calculate_tax(60000, "GB")
    print(f"Net income: £{result['totals']['net_income']:,}")
except TaxAPIError as e:
    if e.status_code == 401:
        print("Invalid API key — check COUNTRYTAXCALC_API_KEY environment variable")
    elif e.status_code == 404:
        print(f"Country not found: {e.message}")
    elif e.status_code == 429:
        print(f"Rate limit exceeded: {e.message}")
    else:
        print(f"API error {e.status_code}: {e.message}")

Async with httpx

For FastAPI applications or scripts that need to fetch multiple countries concurrently, use httpx with Python's asyncio:

import asyncio
import os
import httpx

API_KEY = os.environ.get("COUNTRYTAXCALC_API_KEY")
BASE_URL = "https://api.countrytaxcalc.com/api/v1"

async def calculate_tax_async(
    client: httpx.AsyncClient,
    income: float,
    country: str,
    currency: str = "USD",
    semaphore: asyncio.Semaphore = None
) -> dict:
    """Single async tax calculation."""
    payload = {"income": income, "country": country, "currency": currency}

    async with (semaphore or asyncio.Semaphore(1)):
        resp = await client.post(
            f"{BASE_URL}/calculate",
            headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
            json=payload,
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        return {
            "country": data["country"],
            "country_code": data["country_code"],
            "net_income": data["totals"]["net_income"],
            "effective_rate": data["totals"]["effective_tax_rate"],
            "currency": data["currency"],
            "confidence": data["metadata"]["accuracy_confidence"]
        }

async def compare_countries(income: float, countries: list[str], currency: str = "USD"):
    """Fetch tax results for multiple countries concurrently."""
    semaphore = asyncio.Semaphore(5)  # max 5 concurrent requests

    async with httpx.AsyncClient() as client:
        tasks = [
            calculate_tax_async(client, income, country, currency, semaphore)
            for country in countries
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter errors and sort by net income
    valid = [r for r in results if isinstance(r, dict)]
    errors = [r for r in results if isinstance(r, Exception)]

    if errors:
        for err in errors:
            print(f"Warning: {err}")

    return sorted(valid, key=lambda x: x["net_income"], reverse=True)

# Run it
countries = ["GB", "DE", "NL", "SE", "NO", "DK", "IE", "CH", "SG", "AU"]

results = asyncio.run(compare_countries(100000, countries, "USD"))

print(f"{'Country':<20} {'Net Income':>15} {'Effective Rate':>15}")
print("-" * 52)
for r in results:
    net = f"{r['currency']} {r['net_income']:,.0f}"
    rate = f"{r['effective_rate']}%"
    label = "" if r['confidence'] >= 0.9 else " (est.)"
    print(f"{r['country']:<20} {net:>15} {rate:>15}{label}")

Caching responses

For most products, the same country+income combination will be requested many times. Caching saves API quota and reduces latency. The Terms of Service permit caching for up to 24 hours.

In-memory cache (single process)

import time
from functools import lru_cache

# Simple time-aware cache wrapper
_cache: dict[str, tuple[dict, float]] = {}
CACHE_TTL = 86400  # 24 hours in seconds

def calculate_tax_cached(income: float, country: str, currency: str = None) -> dict:
    """Calculate tax with 24-hour in-memory caching."""
    key = f"{country}:{income}:{currency or 'local'}"
    now = time.time()

    if key in _cache:
        data, cached_at = _cache[key]
        if now - cached_at < CACHE_TTL:
            return data

    result = calculate_tax(income, country, currency)
    _cache[key] = (result, now)
    return result

Redis cache (multi-process / production)

import json
import os
import redis
import requests

redis_client = redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))
CACHE_TTL = 86400  # 24 hours

def calculate_tax_redis(income: float, country: str, currency: str = None) -> dict:
    """Calculate tax with Redis caching."""
    cache_key = f"tax:{country}:{income}:{currency or 'local'}"

    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Cache miss — call API
    result = calculate_tax(income, country, currency)

    # Store in Redis with TTL
    redis_client.setex(cache_key, CACHE_TTL, json.dumps(result))
    return result

Complete multi-country comparison script

A self-contained Python script that fetches take-home pay for a list of countries and prints a ranked comparison. Uses the synchronous requests library with the caching and error-handling patterns from above.

"""
multi_country_comparison.py
Compares take-home pay across countries using the CountryTaxCalc API.

Usage:
  export COUNTRYTAXCALC_API_KEY=your_key_here
  python multi_country_comparison.py
"""

import os
import time
import requests

API_KEY = os.environ.get("COUNTRYTAXCALC_API_KEY")
if not API_KEY:
    raise ValueError("COUNTRYTAXCALC_API_KEY environment variable not set")

BASE_URL = "https://api.countrytaxcalc.com/api/v1"

COUNTRIES = ["GB", "DE", "NL", "SE", "NO", "DK", "IE", "CH", "FR", "AU",
             "CA", "SG", "NZ", "US", "AE"]
GROSS_INCOME = 100_000
CURRENCY = "USD"


def fetch_tax(country: str) -> dict | None:
    """Fetch tax calculation for a single country. Returns None on error."""
    try:
        resp = requests.post(
            f"{BASE_URL}/calculate",
            headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
            json={"income": GROSS_INCOME, "country": country, "currency": CURRENCY},
            timeout=10
        )
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", 60))
            print(f"  Rate limited — waiting {retry_after}s")
            time.sleep(retry_after)
            resp = requests.post(
                f"{BASE_URL}/calculate",
                headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
                json={"income": GROSS_INCOME, "country": country, "currency": CURRENCY},
                timeout=10
            )
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as e:
        print(f"  Error fetching {country}: {e}")
        return None


def main():
    print(f"Comparing take-home pay for ${GROSS_INCOME:,} USD gross income\n")
    print(f"{'#':<3} {'Country':<22} {'Net Income (local)':>20} {'Eff. Rate':>12} {'Source':<8}")
    print("-" * 72)

    results = []
    for country in COUNTRIES:
        data = fetch_tax(country)
        if data:
            results.append(data)
        time.sleep(0.1)  # small delay to avoid burst limit

    # Sort by effective tax rate (ascending = lowest tax first)
    results.sort(key=lambda d: d["totals"]["effective_tax_rate"])

    for i, d in enumerate(results, 1):
        symbol = d["currency_symbol"]
        net = d["totals"]["net_income"]
        rate = d["totals"]["effective_tax_rate"]
        source = d["metadata"]["data_source"]
        conf = d["metadata"]["accuracy_confidence"]
        approx = " ~" if conf < 0.9 else "  "

        net_str = f"{symbol}{net:>12,.0f}{approx}"
        print(f"{i:<3} {d['country']:<22} {net_str:>20} {rate:>11.1f}%  {source}")

    print(f"\n~ = estimated (accuracy_confidence < 0.9)")
    print(f"Disclaimer: estimates only. Consult a tax professional for advice.")


if __name__ == "__main__":
    main()

Environment variables and local development

The recommended way to manage your API key in Python:

# .env (add this file to .gitignore — never commit it)
COUNTRYTAXCALC_API_KEY=your_api_key_here

# In your Python code
from dotenv import load_dotenv
import os

load_dotenv()  # loads .env file in local dev; does nothing in production

API_KEY = os.environ.get("COUNTRYTAXCALC_API_KEY")
if not API_KEY:
    raise ValueError("COUNTRYTAXCALC_API_KEY is not set")

# requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0

# For async
httpx>=0.25.0

In production, set COUNTRYTAXCALC_API_KEY via your hosting platform's environment variable / secrets management system rather than a .env file.