The typed Python SDK for the Blitz API — B2B data, search, and enrichment.
- Fully typed — Pydantic v2 response models with attribute access and IDE
autocomplete,
TypedDictrequest filters, and a shippedpy.typedmarker so mypy/pyright see the types in your own code. - Sync & async —
BlitzAPIandAsyncBlitzAPIoverhttpx. - Resilient — built-in client-side rate limiting, retries with backoff on
429/5xx, and a typed exception hierarchy. - Forward-compatible — new fields the API adds never break deserialization.
- 1:1 with the API — request filters and response fields are snake_case, matching docs.blitz-api.ai.
Create and manage API keys at app.blitz-api.ai.
Billing. Blitz bills per result. A bare
for person in client.search.people(...)loop streams every match up to the server-side limit (people: 50k results), which can be a lot of records. Bound spend withmax_items(a client-side total cap on.collect()/.auto_paging_iter(), never sent on the wire) — details in Pagination.
- Installation
- Quickstart
- Example: find, enrich, collect
- Authentication
- Endpoints
- Pagination
- Usage & rate limit (
fair_usage) - Configuration
- Error handling
- Forward compatibility
- Development
pip install blitz-api-py
# or: poetry add blitz-api-py / uv add blitz-api-pyRequires Python 3.10+.
from blitz_api import BlitzAPI
from blitz_api.types import Industry, JobLevel
# api_key defaults to the BLITZ_API_KEY environment variable.
with BlitzAPI() as client:
# Health-check the key before a batch job.
info = client.account.key_info()
print(info.valid, info.records_remaining, info.max_requests_per_seconds)
# LinkedIn profile URL -> verified work email.
email = client.enrichment.email(
person_linkedin_url="https://www.linkedin.com/in/example-person",
)
if email.found:
print(email.email)
# Search people with typed, autocompleted filters.
people = client.search.people(
company={"industry": {"include": [Industry.SOFTWARE_DEVELOPMENT]}},
people={"job_level": [JobLevel.VP]},
max_results=10,
)
for person in people.results:
print(person.full_name, person.headline)import asyncio
from blitz_api import AsyncBlitzAPI
async def main() -> None:
async with AsyncBlitzAPI() as client:
result = await client.enrichment.company(
company_linkedin_url="https://www.linkedin.com/company/placeholder-co",
)
print(result.company.name if result.company else None)
asyncio.run(main())A complete flow — find people, enrich each one's verified work email, collect the
contacts. max_items caps the total fetched so the run can't surprise you with records.
from blitz_api import BlitzAPI
from blitz_api.types import Industry, JobLevel
client = BlitzAPI() # reads BLITZ_API_KEY
# 1. Find up to 25 VPs at software companies (typed filters, 1:1 with the API).
leads = client.search.people(
company={"industry": {"include": [Industry.SOFTWARE_DEVELOPMENT]}},
people={"job_level": [JobLevel.VP]},
max_results=25,
).collect(max_items=25) # client-side total cap — bounds record spend
# 2. Enrich each lead's verified work email from their LinkedIn profile URL.
contacts: list[dict[str, str | None]] = []
for person in leads:
if not person.linkedin_url:
continue
result = client.enrichment.email(person_linkedin_url=person.linkedin_url)
if result.found:
contacts.append({"name": person.full_name, "email": result.email})
print(f"Collected {len(contacts)} contacts")What comes back is typed and snake_case. A Person from the search above (fields are a
superset — only what the profile has is populated, and unknown fields the API adds
later are preserved):
Person(
full_name="Jordan Lee",
# Derived from the first position as "<job title> | @<employer>" — not the
# free-text headline written on the LinkedIn profile.
headline="VP of Engineering | @Acme",
linkedin_url="https://www.linkedin.com/in/example-person",
location=Location(city="San Francisco", state_code="CA", country_code="US", continent="North America"),
# Positions from the profile, in profile order — read `job_is_current` for the
# current role rather than assuming index 0.
experiences=[Experience(job_title="VP of Engineering", company_name="Acme", job_is_current=True)],
# first_name, last_name, skills, education, certifications, … also present
)How many positions land in experiences[] depends on the endpoint — and upstream
currently contradicts itself for search.people: the changelog entry of 2026-09-21 says
it returns only the position that matched your filters, while the API reference still
says it returns the full position history. Don't rely on either from a search result. If
you need the person's whole career in profile order, enrichment.person returns it
unambiguously (1 record on success, free on a miss):
if person.linkedin_url: # optional on Person; the call requires a str
result = client.enrichment.person(person_linkedin_url=person.linkedin_url)
if result.found and result.person:
for exp in result.person.experiences: # every position held, in profile order
print(exp.job_title, exp.company_name, exp.job_is_current)And enrichment.email(...) returns:
EmailEnrichmentResponse(
found=True,
email="[email protected]",
all_emails=[EmailMatch(email="[email protected]", email_domain="acme.com")],
)Pass the key explicitly or via the BLITZ_API_KEY environment variable:
client = BlitzAPI(api_key="sk_...") # explicit
client = BlitzAPI() # reads BLITZ_API_KEYThe key is sent in the x-api-key header. Never expose it in client-side code —
always call the API from your backend.
All methods are grouped into seven namespaces:
| Namespace | Methods |
|---|---|
client.account |
key_info() |
client.search |
people(), companies(), employee_finder(), waterfall_icp() |
client.jobs |
search(), company() |
client.company |
tam_by_jobs(), tam_by_people() |
client.enrichment |
person(), email(), phone(), email_to_person(), phone_to_person(), company(), domain_to_linkedin(), linkedin_to_domain(), company_distribution_by_country(), company_distribution_by_department() |
client.utils |
current_date() |
client.changelog |
list() |
Every method returns a typed Pydantic model (see blitz_api.types). Enum-backed
filter fields (e.g. Industry, JobLevel, Continent) accept either an enum
member or a raw string.
from blitz_api import BlitzAPI
from blitz_api.types import JobLevel, MxProvider
client = BlitzAPI()
# VPs who started a job in the last 90 days, at companies on Google Workspace, who
# graduated in 2015 or later. `is_recent_hire` is a key of `people`.
page = client.search.people(
company={"mx": {"provider": {"include": [MxProvider.GOOGLE]}}},
people={
"job_level": [JobLevel.VP],
"education": {"end_year": {"min": 2015}},
"is_recent_hire": True,
},
)
# Also search jobs a person held before: your job and company filters must hold in the
# same job, which comes back in `experiences` with `job_is_current` set. Past jobs only:
client.search.people(people={"include_active_jobs": False, "include_past_jobs": True})
# The waterfall takes the same three flags per cascade tier: try recent hires first, then
# fall back to anyone in the role.
client.search.waterfall_icp(
company_linkedin_url="https://www.linkedin.com/company/openai",
cascade=[
{"include_title": ["CTO"], "is_recent_hire": True},
{"include_title": ["CTO"]},
],
)Where the flags go depends on the route, and the API rejects a flag in the wrong place with a
422:
| Route | is_recent_hire |
include_active_jobs / include_past_jobs |
|---|---|---|
search.people |
key of people |
key of people |
company.tam_by_people |
key of people |
— |
search.waterfall_icp |
key of each cascade tier |
key of each cascade tier |
search.employee_finder |
top-level argument | — |
enrichment.company_distribution_by_country / _by_department |
top-level argument | — |
Setting both include_*_jobs flags to False is a 422. MxProvider.ANY stands for
every provider: in include it keeps companies with mail data, in exclude it keeps
companies with none.
The search, jobs and TAM methods return an auto-paginating page: iterate it and the
SDK fetches each subsequent page for you. search.people/search.companies,
jobs.search/jobs.company and company.tam_by_jobs/company.tam_by_people are
cursor-based; search.employee_finder is page-based — all behave identically here.
max_resultsis the page size, not a total. It's results per page, and the API bills 1 record per result returned. A barefor person in client.search.people(...)loop streams every match up to the server-side limit (people: 50k results / 1k pages; employee finder: 10k; jobs: 5k), which can be a lot of records. Bound it withmax_itemson.collect()/.auto_paging_iter()(a client-side total cap — never sent on the wire),breakout of the loop, or drive pages manually.
# Iterate every matching person across all pages — no cursor handling needed.
for person in client.search.people(people={"job_level": ["VP"]}):
print(person.full_name)
# Async works the same way.
async for person in await async_client.search.people(people={"job_level": ["VP"]}):
...
# Bound how much you pull.
for person in client.search.people(...).auto_paging_iter(max_items=200):
...
# Per-page control: inspect totals / cursors as you go.
for page in client.search.companies(company={...}).iter_pages(max_pages=5):
print(page.total_results, len(page.results), page.cursor)
# Jobs are cursor-paginated too — surface hiring signals matching your ICP.
for job in client.jobs.search(
job={"title": {"include": ["Head of Sales"]}, "date_posted": {"last_days": 30}},
company={"industry": {"include": ["Software Development"]}, "size": {"include": ["51-200"]}},
).auto_paging_iter(max_items=200):
print(job.company_name, job.title, job.location.city if job.location else None)
# The TAM builders page the same way — distinct companies, each with its match count.
for match in client.company.tam_by_people(
company={"industry": {"include": ["Software Development"]}},
people={"job_level": ["VP"], "min_per_company": 3},
).auto_paging_iter(max_items=200):
print(match.company.name if match.company else None, match.matched_people)
# Or page manually.
page = client.search.people(people={...}, max_results=50)
print(page.results, page.cursor)
nxt = page.get_next_page() # None once exhaustedThe page types (CursorPage, PageNumberPage, and their Async* variants) are
exported from blitz_api.
Every /v2 response carries a fair_usage block: what the call cost, what's left on
the plan, your rate-limit headroom, and a request id to quote to support. It's on every
response model and on every page, so you can meter a long run without extra key_info()
calls.
result = client.enrichment.email(person_linkedin_url="https://www.linkedin.com/in/example")
if result.fair_usage is not None:
print(result.fair_usage.records_used) # what this call cost
print(result.fair_usage.records_remaining) # a number, or "unlimited"
print(result.fair_usage.next_reset_at) # None on an unlimited plan
print(result.fair_usage.request_id) # quote this to support
if result.fair_usage.rate_limit is not None: # absent on account.key_info()
print(result.fair_usage.rate_limit.remaining_this_second)
# Pages carry the block of the request that fetched them.
for page in client.search.people(people={"job_level": ["VP"]}).iter_pages(max_pages=5):
print(page.fair_usage.records_used if page.fair_usage else None)InsufficientRecordsError (402) carries it too, so you can see when the balance
resets without a second call:
except InsufficientRecordsError as err:
if err.fair_usage is not None:
print("resets at", err.fair_usage.next_reset_at)client = BlitzAPI(
api_key=None, # falls back to BLITZ_API_KEY
base_url="https://api.blitz-api.ai",
timeout=30.0, # seconds, or an httpx.Timeout
max_retries=3, # retries on 429 / 5xx / network errors
rate_limit_rps=5.0, # client-side throttle; None to disable
)The client-side rate limiter is a sliding window — at most rate_limit_rps requests
in any rolling second — applied per endpoint: each endpoint (e.g. .email vs
.phone) is throttled independently, mirroring the API's own limit, which is also per
endpoint (10 req/s on every plan — 50 on plans created before 2026-09-30; check yours
via client.account.key_info().max_requests_per_seconds). The default 5.0
deliberately sits at half the lowest cap — raise it to your key's limit to use the
full budget. A single
client instance therefore stays under the limit on every endpoint, so a burst on one
never blocks another. Across multiple processes — which share an endpoint's budget — you
may still hit 429; the retry path handles that.
Every method also accepts a per-call timeout (seconds or an httpx.Timeout) when one
endpoint needs longer than the client default:
client.search.people(people={"job_level": ["VP"]}, timeout=10.0)from blitz_api import (
BlitzError, AuthenticationError, InsufficientRecordsError,
NotFoundError, RateLimitError, APIStatusError, APIConnectionError,
APITimeoutError, APIResponseValidationError,
)
try:
client.enrichment.email(person_linkedin_url="...")
except InsufficientRecordsError as err:
... # 402 — out of records; err.fair_usage.next_reset_at
except AuthenticationError:
... # 401 — bad key
except APIStatusError as err:
print(err.status_code, err.message, err.body)
except APIResponseValidationError:
... # 2xx whose body wasn't valid / didn't match the model
except BlitzError:
... # base class for everything this SDK raises429 and 5xx are retried automatically (with backoff) up to max_retries;
401/402/404 raise immediately. Connect timeouts and connection errors are retried,
but a read timeout is not — the server may already have processed (and billed) the
request, so it surfaces as APITimeoutError rather than risking a double charge.
Response models subclass a base configured with extra="allow", so a field the API adds
before this SDK models it is still present on the parsed object (via attribute access or
.model_extra). Known fields stay precisely typed.
See CONTRIBUTING.md for local setup, the test/type/lint commands, the enum code generator, and the automated release process.