How to Integrate Address APIs in Python: Verification, Autocomplete & Geocoding
A complete Python guide to address verification, autocomplete, parsing, and geocoding with Sthan.io APIs. Includes synchronous requests, async aiohttp examples, error handling, and production best practices.
Prerequisites & Setup
Before you begin, make sure you have the following:
- Python 3.8+ installed (download here)
- pip package manager (included with Python 3.4+)
- A Sthan.io account (sign up for free -- no credit card required)
- Your Profile Name and Profile Password from the Sthan.io Dashboard
Install Dependencies
Install the requests library for making HTTP calls:
pip install requests
For async support (covered in the aiohttp section), also install:
pip install aiohttp
Sthan.io uses a 2-step JWT authentication flow. First, you send your profileName and
profilePassword as headers to GET /Auth/Token to receive a JWT access token.
Then, you include that token as a Bearer token in the Authorization header for all subsequent API calls.
Tokens expire after a set period, so your code should handle token refresh.
Authentication
The first step is obtaining a JWT token from the Sthan.io Auth endpoint. This token is then used as a Bearer token
for all subsequent API requests. Below is a complete, reusable SthanClient class that handles
authentication and token management.
Step 1: Get a JWT Token
Send a GET request to /Auth/Token with your credentials in the headers:
import requests
from urllib.parse import quote
# Step 1: Authenticate and get a JWT token
AUTH_URL = "https://api.sthan.io/Auth/Token"
headers = {
"profileName": "YOUR_PROFILE_NAME",
"profilePassword": "YOUR_PROFILE_PASSWORD"
}
response = requests.get(AUTH_URL, headers=headers)
response.raise_for_status()
data = response.json()
access_token = data["result"]["access_token"]
expiration = data["result"]["expiration"]
print(f"Token obtained, expires: {expiration}")
The response follows the standard Sthan.io envelope format:
{
"id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"result": {
"access_token": "eyJhbGciOiJIUzI1NiIs...",
"expiration": "2026-03-08T09:00:00Z"
},
"statusCode": 200,
"isError": false,
"errors": []
}
Step 2: Build a Reusable SthanClient Class
Wrap authentication and API calls in a reusable class. This client handles token acquisition, caching, and provides methods for each address API endpoint.
import requests
from urllib.parse import quote
from datetime import datetime, timezone
class SthanClient:
"""Reusable client for Sthan.io address APIs."""
BASE_URL = "https://api.sthan.io"
def __init__(self, profile_name: str, profile_password: str):
self.profile_name = profile_name
self.profile_password = profile_password
self._token: str | None = None
self._token_expiry: datetime | None = None
self.session = requests.Session()
self.session.headers.update({"Content-Type": "application/json"})
def _get_token(self) -> str:
"""Get a valid JWT token, refreshing if expired."""
now = datetime.now(timezone.utc)
if self._token and self._token_expiry and now < self._token_expiry:
return self._token
response = self.session.get(
f"{self.BASE_URL}/Auth/Token",
headers={
"profileName": self.profile_name,
"profilePassword": self.profile_password,
},
timeout=10,
)
response.raise_for_status()
data = response.json()
self._token = data["result"]["access_token"]
# Parse expiration and subtract 60s buffer
self._token_expiry = datetime.fromisoformat(
data["result"]["expiration"].replace("Z", "+00:00")
)
return self._token
def _auth_headers(self) -> dict:
"""Return headers with a valid Bearer token."""
token = self._get_token()
return {"Authorization": f"Bearer {token}"}
def autocomplete(self, text: str) -> dict:
"""Search for address suggestions as the user types."""
url = f"{self.BASE_URL}/AutoComplete/USA/Address/{quote(text)}"
resp = self.session.get(url, headers=self._auth_headers(), timeout=10)
resp.raise_for_status()
return resp.json()
def verify_address(self, address: str) -> dict:
"""Verify and standardize a US address."""
url = f"{self.BASE_URL}/AddressVerification/Usa/Single/{quote(address)}"
resp = self.session.get(url, headers=self._auth_headers(), timeout=15)
resp.raise_for_status()
return resp.json()
def parse_address(self, address: str) -> dict:
"""Parse an address string into structured components."""
url = f"{self.BASE_URL}/AddressParser/USA/Single/{quote(address)}"
resp = self.session.get(url, headers=self._auth_headers(), timeout=15)
resp.raise_for_status()
return resp.json()
def geocode(self, address: str) -> dict:
"""Forward geocode an address to latitude/longitude."""
url = f"{self.BASE_URL}/Geocoding/USA/Forward/{quote(address)}"
resp = self.session.get(url, headers=self._auth_headers(), timeout=15)
resp.raise_for_status()
return resp.json()
# Usage
client = SthanClient("YOUR_PROFILE_NAME", "YOUR_PROFILE_PASSWORD")
SthanClient uses requests.Session() for
connection pooling. This reuses TCP connections across requests, significantly
reducing latency when making multiple API calls. Never create a new requests.get()
call for each request in production.
Address Autocomplete
The Address Autocomplete API returns address suggestions as the user types, providing real-time results with sub-100ms response times. This is ideal for building type-ahead search fields in forms.
Endpoint
GET /AutoComplete/USA/Address/{text}
Python Example
# Address Autocomplete -- get suggestions as the user types
result = client.autocomplete("1600 pennsylvania ave")
# The result contains matching address suggestions
if not result["isError"]:
suggestions = result["result"]
for suggestion in suggestions:
print(suggestion)
else:
print("Error:", result["errors"])
Sample Response
{
"id": "b2c3d4e5-f6a7-8901-bcde-f12345678901",
"result": [
"1600 Pennsylvania Ave NW, Washington, DC 20500",
"1600 Pennsylvania Ave, Lincoln, NE 68508",
"1600 Pennsylvania Ave SE, Washington, DC 20003"
],
"statusCode": 200,
"isError": false,
"errors": []
}
Address Verification
The Address Verification API validates and standardizes US addresses against authoritative postal data. It corrects formatting errors, fills in missing components (ZIP+4, county), and confirms deliverability. This is essential for e-commerce checkout, shipping, and mailing workflows.
Endpoint
GET /AddressVerification/Usa/Single/{address}
Python Example
# Verify and standardize a US address
result = client.verify_address("1600 pennsylvania ave nw washington dc")
if not result["isError"]:
verified = result["result"]
print(f"Street: {verified['addressLine1']}")
print(f"City: {verified['city']}")
print(f"State: {verified['state']}")
print(f"ZIP: {verified['zipCode']}")
print(f"ZIP+4: {verified['plus4']}")
print(f"County: {verified['county']}")
else:
print("Verification failed:", result["errors"])
Sample Response
{
"id": "c3d4e5f6-a7b8-9012-cdef-123456789012",
"result": {
"addressLine1": "1600 Pennsylvania Ave NW",
"city": "Washington",
"state": "DC",
"zipCode": "20500",
"plus4": "0005",
"county": "District of Columbia",
"deliveryPoint": "99",
"carrierRoute": "C000",
"congressionalDistrict": "98",
"isValid": true
},
"statusCode": 200,
"isError": false,
"errors": []
}
Address Parsing
The Address Parser API breaks a freeform address string into structured components -- street number, street name, direction, suffix, unit, city, state, and ZIP code. This is useful when you receive unstructured address data from user input, CSVs, or third-party systems and need to normalize it into discrete fields.
Endpoint
GET /AddressParser/USA/Single/{address}
Python Example
# Parse a freeform address into structured components
result = client.parse_address("1600 Pennsylvania Ave NW Apt 2B Washington DC 20500")
if not result["isError"]:
parsed = result["result"]
print(f"Number: {parsed.get('number', '')}")
print(f"Street: {parsed.get('street', '')}")
print(f"Suffix: {parsed.get('suffix', '')}")
print(f"Direction: {parsed.get('directional', '')}")
print(f"Unit Type: {parsed.get('secondaryDesignator', '')}")
print(f"Unit Number: {parsed.get('secondaryNumber', '')}")
print(f"City: {parsed.get('city', '')}")
print(f"State: {parsed.get('state', '')}")
print(f"ZIP: {parsed.get('zip', '')}")
else:
print("Parsing failed:", result["errors"])
Sample Response
{
"id": "d4e5f6a7-b8c9-0123-defa-234567890123",
"result": {
"number": "1600",
"street": "Pennsylvania",
"suffix": "Ave",
"directional": "NW",
"secondaryDesignator": "Apt",
"secondaryNumber": "2B",
"city": "Washington",
"state": "DC",
"zip": "20500"
},
"statusCode": 200,
"isError": false,
"errors": []
}
Forward Geocoding
The Forward Geocoding API converts a US address string into geographic coordinates (latitude and longitude). Use this for mapping, distance calculations, delivery zone assignment, and location-based services.
Endpoint
GET /Geocoding/USA/Forward/{address}
Python Example
# Forward geocode an address to lat/lon coordinates
result = client.geocode("1600 Pennsylvania Ave NW, Washington, DC 20500")
if not result["isError"]:
geo = result["result"]
print(f"Latitude: {geo['latitude']}")
print(f"Longitude: {geo['longitude']}")
print(f"Accuracy: {geo.get('accuracy', 'N/A')}")
else:
print("Geocoding failed:", result["errors"])
Sample Response
{
"id": "e5f6a7b8-c9d0-1234-efab-345678901234",
"result": {
"latitude": 38.8977,
"longitude": -77.0365,
"accuracy": "Rooftop"
},
"statusCode": 200,
"isError": false,
"errors": []
}
GET /Geocoding/USA/Reverse/{lat}/{lon}.
Async with aiohttp
For high-throughput applications -- batch processing, web servers, or verifying large address
lists -- use Python's asyncio with aiohttp to make concurrent, non-blocking
API calls. This approach can process hundreds of addresses in the time sequential code handles a dozen.
Install aiohttp
pip install aiohttp
Async SthanClient
import aiohttp
import asyncio
from urllib.parse import quote
from datetime import datetime, timezone
class AsyncSthanClient:
"""Async client for Sthan.io address APIs using aiohttp."""
BASE_URL = "https://api.sthan.io"
def __init__(self, profile_name: str, profile_password: str):
self.profile_name = profile_name
self.profile_password = profile_password
self._token: str | None = None
self._token_expiry: datetime | None = None
self._session: aiohttp.ClientSession | None = None
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=15)
)
return self._session
async def _get_token(self) -> str:
"""Get a valid JWT token, refreshing if expired."""
now = datetime.now(timezone.utc)
if self._token and self._token_expiry and now < self._token_expiry:
return self._token
session = await self._get_session()
async with session.get(
f"{self.BASE_URL}/Auth/Token",
headers={
"profileName": self.profile_name,
"profilePassword": self.profile_password,
},
) as resp:
resp.raise_for_status()
data = await resp.json()
self._token = data["result"]["access_token"]
self._token_expiry = datetime.fromisoformat(
data["result"]["expiration"].replace("Z", "+00:00")
)
return self._token
async def _auth_headers(self) -> dict:
token = await self._get_token()
return {"Authorization": f"Bearer {token}"}
async def verify_address(self, address: str) -> dict:
"""Verify and standardize a US address (async)."""
session = await self._get_session()
headers = await self._auth_headers()
url = f"{self.BASE_URL}/AddressVerification/Usa/Single/{quote(address)}"
async with session.get(url, headers=headers) as resp:
resp.raise_for_status()
return await resp.json()
async def close(self):
"""Close the underlying HTTP session."""
if self._session and not self._session.closed:
await self._session.close()
Batch Processing Example
Verify multiple addresses concurrently using asyncio.gather(). Use a
Semaphore to control the level of concurrency and avoid overwhelming the API.
async def verify_batch(addresses: list[str], max_concurrent: int = 10):
"""Verify a list of addresses concurrently."""
client = AsyncSthanClient("YOUR_PROFILE_NAME", "YOUR_PROFILE_PASSWORD")
semaphore = asyncio.Semaphore(max_concurrent)
async def verify_one(address: str) -> dict:
async with semaphore:
try:
result = await client.verify_address(address)
return {"address": address, "result": result, "success": True}
except Exception as e:
return {"address": address, "error": str(e), "success": False}
tasks = [verify_one(addr) for addr in addresses]
results = await asyncio.gather(*tasks)
await client.close()
return results
# Run it
addresses = [
"1600 Pennsylvania Ave NW, Washington, DC",
"350 Fifth Avenue, New York, NY",
"1 Infinite Loop, Cupertino, CA",
"233 S Wacker Dr, Chicago, IL",
]
results = asyncio.run(verify_batch(addresses))
for r in results:
if r["success"]:
verified = r["result"]["result"]
print(f"OK: {verified['addressLine1']}, {verified['city']}, {verified['state']}")
else:
print(f"FAIL: {r['address']} -- {r['error']}")
max_concurrent semaphore based on
your plan's rate limits.
Error Handling & Best Practices
1. Retry Logic with Exponential Backoff
Handle transient failures (network errors, 429 rate limit, 500 server errors) with automatic retries and exponential backoff:
import time
import requests
from urllib.parse import quote
def request_with_retry(
session: requests.Session,
url: str,
headers: dict,
max_retries: int = 3,
base_delay: float = 1.0,
) -> requests.Response:
"""Make an HTTP GET request with exponential backoff retry."""
for attempt in range(max_retries + 1):
try:
response = session.get(url, headers=headers, timeout=15)
# Success or client error (not retryable)
if response.status_code < 500 and response.status_code != 429:
response.raise_for_status()
return response
# Rate limited -- wait and retry
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", base_delay))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
continue
# Server error -- retry with backoff
if attempt < max_retries:
delay = base_delay * (2 ** attempt)
print(f"Server error {response.status_code}. Retrying in {delay}s...")
time.sleep(delay)
except requests.exceptions.ConnectionError:
if attempt < max_retries:
delay = base_delay * (2 ** attempt)
print(f"Connection error. Retrying in {delay}s...")
time.sleep(delay)
else:
raise
raise Exception(f"Failed after {max_retries} retries")
2. URL Encoding Addresses
Addresses with special characters (#, &, spaces) must be
URL-encoded before including them in the API path. Use urllib.parse.quote():
from urllib.parse import quote
# Correctly encode addresses with special characters
address = "123 Main St #4B, New York, NY"
encoded = quote(address)
# Result: "123%20Main%20St%20%234B%2C%20New%20York%2C%20NY"
url = f"https://api.sthan.io/AddressVerification/Usa/Single/{encoded}"
3. Token Refresh Pattern
JWT tokens expire after a set period. The SthanClient class shown above handles
this automatically by checking the expiration field before each request.
If your token has expired, simply re-authenticate:
# The SthanClient._get_token() method handles this automatically.
# If you're managing tokens manually:
from datetime import datetime, timezone, timedelta
def is_token_valid(expiry_str: str) -> bool:
"""Check if a JWT token is still valid (with 60s buffer)."""
expiry = datetime.fromisoformat(expiry_str.replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
# Refresh 60 seconds before actual expiry
return now < expiry - timedelta(seconds=60)
4. Secure Credential Storage
Never hardcode API credentials in your source code. Use environment variables:
import os
# Load credentials from environment variables
profile_name = os.environ.get("STHAN_PROFILE_NAME")
profile_password = os.environ.get("STHAN_PROFILE_PASSWORD")
if not profile_name or not profile_password:
raise ValueError("Missing STHAN_PROFILE_NAME or STHAN_PROFILE_PASSWORD env vars")
client = SthanClient(profile_name, profile_password)
# Set environment variables before running your script
export STHAN_PROFILE_NAME="your-profile-name"
export STHAN_PROFILE_PASSWORD="your-profile-password"
python your_script.py
5. Production Tips
- Connection pooling: Always use
requests.Session()to reuse TCP connections. Creating a new connection per request adds 50-100ms of overhead. - Timeouts: Set explicit timeouts (10-15s) on every request to avoid hanging indefinitely. The
SthanClientabove usestimeout=10for auth andtimeout=15for API calls. - Logging: Log API errors with the request URL (without credentials) and status code for debugging. Use Python's built-in
loggingmodule. - Rate limiting: Check your plan's rate limits and implement client-side throttling to stay within bounds.
- Caching: Cache verification results for addresses you've already validated. Use
functools.lru_cacheor Redis for production caching.
Key Takeaways
- Use a reusable client class with
requests.Session()for connection pooling and automatic token management - URL-encode addresses with
urllib.parse.quote()before including them in API paths - For bulk processing, use aiohttp + asyncio with a semaphore to control concurrency -- 10x faster than sequential requests
- Implement exponential backoff retry logic for transient errors and rate limiting
- Never hardcode credentials -- use environment variables or a secrets manager
- Set explicit timeouts (10-15s) on every request to prevent hanging
Share This Article
Frequently Asked Questions
requests for scripts, CLIs, and simple integrations. Use aiohttp for web applications, bulk processing, or when you need to verify hundreds of addresses concurrently.
Retry-After header value if present in the response.
See the Error Handling section above for a complete implementation.
Additionally, consider batching your requests and using asyncio with semaphores to control concurrency.
asyncio with aiohttp to verify multiple addresses concurrently.
Create an AsyncSthanClient class, then use asyncio.gather() to process batches of addresses in parallel.
Control concurrency with asyncio.Semaphore to avoid overwhelming the API.
This approach can process hundreds of addresses per second while respecting rate limits.
See the Async with aiohttp section for a complete batch processing example.
asyncio library with aiohttp to make non-blocking API calls.
Create an async client class with async def methods, use aiohttp.ClientSession for connection pooling, and process multiple addresses concurrently with asyncio.gather().
This is significantly faster than sequential requests -- you can verify 100 addresses in the time it takes to verify 5-10 sequentially.
Python 3.8+ has mature async support that works well for API integrations.
The aiohttp section above includes a complete async client implementation.
asyncio.run()),
f-string formatting, and type hints that make API integration code cleaner and more maintainable.
The requests library supports Python 3.7+, and aiohttp requires Python 3.8+.
If you're starting a new project, use the latest stable Python version (3.12+) for best performance and security.
urllib.parse.quote() function to URL-encode addresses before including them in API URLs.
This properly encodes spaces, hash signs (#), ampersands (&), and other special characters.
For example, "123 Main St #4" becomes "123%20Main%20St%20%234".
The SthanClient class in this guide handles URL encoding automatically in every method call.
See the Error Handling section for more details.
os.environ.get("STHAN_PROFILE_NAME")) for local development,
and a secrets manager (AWS Secrets Manager, Azure Key Vault, HashiCorp Vault) for production.
You can also use a .env file with the python-dotenv library for local development --
just make sure to add .env to your .gitignore.
For CI/CD pipelines, use your platform's built-in secrets management (GitHub Secrets, GitLab CI Variables, etc.).
Start Building with Sthan.io
Get 100,000 free API calls per month. Verify, parse, geocode, and autocomplete addresses in Python -- no credit card required.