Skip to main content
Agents & LLMs: a Markdown version of this page is available by appending .md to the URL, and the full documentation index is at llms.txt.
blitz-api-py is the official, typed Python SDK for the Blitz API. It ships sync and async clients over httpx, Pydantic v2 response models, TypedDict request filters, and a py.typed marker so mypy/pyright see the types in your own code.

PyPI

blitz-api-py on the Python Package Index.

GitHub

Source, changelog, and issues.

Install

pip install blitz-api-py
# or: uv add blitz-api-py
Requires Python 3.10+.

Quickstart

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.remaining_credits, 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)

Authentication

Pass the key explicitly or via the BLITZ_API_KEY environment variable. It is sent in the x-api-key header on every request.
client = BlitzAPI(api_key="sk_...")   # explicit
client = BlitzAPI()                    # reads BLITZ_API_KEY
Never expose your API key in client-side code (browsers, mobile apps). Always call Blitz from your backend.

Async

Use AsyncBlitzAPI for async/await. Every method mirrors the sync client.
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/openai",
        )
        print(result.company.name if result.company else None)

asyncio.run(main())

Endpoints

All methods are grouped into four namespaces. Enum-backed filter fields (e.g. Industry, JobLevel, Continent) accept either an enum member or a raw string.

client.account

info = client.account.key_info()
print(info.valid, info.remaining_credits, info.max_requests_per_seconds)
print(info.allowed_apis)
# People across many companies (cursor-paginated — see Pagination below).
people = client.search.people(
    company={
        "industry": {"include": ["IT Services and IT Consulting"]},
        "employee_range": ["51-200", "201-500"],
        "hq": {"sales_region": ["EMEA"]},
    },
    people={
        "job_level": ["VP", "Director"],
        "job_function": ["Sales & Business Development"],
        "min_connections": 200,
    },
    max_results=25,
)
for person in people.results:
    print(person.full_name, person.headline, person.linkedin_url)

# Companies by ICP (cursor-paginated).
companies = client.search.companies(
    company={
        "keywords": {"include": ["SaaS"]},
        "industry": {"include": ["Software Development"]},
        "hq": {"country_code": ["FR", "DE"]},
        "employee_range": ["51-200", "201-500"],
    },
    max_results=25,
)
for company in companies.results:
    print(company.name, company.industry, company.employees_on_linkedin)

# All employees at one company (page-paginated).
employees = client.search.employee_finder(
    company_linkedin_url="https://www.linkedin.com/company/openai",
    job_level=["C-Team", "VP", "Director"],
    job_function=["Sales & Business Development"],
    sales_region=["NORAM"],
    max_results=50,
)
print(f"Page {employees.page} of {employees.total_pages}")
for person in employees.results:
    print(person.full_name, person.headline)

# Single best decision-maker via a priority cascade.
match = client.search.waterfall_icp(
    company_linkedin_url="https://www.linkedin.com/company/openai",
    cascade=[
        {"include_title": ["CTO", "VP Engineering"], "location": ["WORLD"], "include_headline_search": False},
        {"include_title": ["Engineering Director", "Engineering Manager"], "location": ["WORLD"], "include_headline_search": False},
    ],
    max_results=5,
)
for item in match.results:
    print(f"[Tier {item.icp} | Rank #{item.ranking}]", item.person.full_name)

client.enrichment

# 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)

# LinkedIn profile URL -> direct phone (US only).
phone = client.enrichment.phone(person_linkedin_url="https://www.linkedin.com/in/example-person")
if phone.found:
    print(phone.phone)

# Work email -> full person profile.
by_email = client.enrichment.email_to_person(email="jane.doe@acme.com")
if by_email.found:
    print(by_email.person.full_name)

# Phone number -> full person profile (US only).
by_phone = client.enrichment.phone_to_person(phone="+14155551234")
if by_phone.found:
    print(by_phone.person.full_name)

# Company LinkedIn URL -> full company profile.
company = client.enrichment.company(company_linkedin_url="https://www.linkedin.com/company/openai")
print(company.company.name, company.company.industry, company.company.employees_on_linkedin)

# Website domain -> Company LinkedIn URL.
to_li = client.enrichment.domain_to_linkedin(domain="openai.com")
if to_li.found:
    print(to_li.company_linkedin_url)

# Company LinkedIn URL -> email domain.
to_domain = client.enrichment.linkedin_to_domain(company_linkedin_url="https://www.linkedin.com/company/openai")
if to_domain.found:
    print(to_domain.email_domain)

client.utils

# Current server date/time.
now = client.utils.current_date()

# Company employees grouped by country.
distribution = client.utils.company_employment_distribution(
    company_linkedin_url="https://www.linkedin.com/company/openai",
)

Pagination

The search methods (people, companies, employee_finder) return an auto-paginating page — iterate it and the SDK fetches each subsequent page for you.
# Stream every match across all pages — no cursor handling needed.
for person in client.search.people(people={"job_level": ["VP"]}):
    print(person.full_name)
See Pagination for max_items, manual paging, page metadata, and the per-result billing caveat.

Configuration

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
)
A single client instance stays under your key’s request-per-second limit (rate_limit_rps, default 5) and retries automatically on 429 — see Rate limits & retries. Each method also accepts a per-call timeout:
client.search.people(people={"job_level": ["VP"]}, timeout=10.0)

Error handling

from blitz_api import (
    BlitzError, AuthenticationError, InsufficientCreditsError,
    NotFoundError, RateLimitError, APIStatusError, APIConnectionError,
    APITimeoutError, APIResponseValidationError,
)

try:
    client.enrichment.email(person_linkedin_url="...")
except InsufficientCreditsError:
    ...                            # 402 — out of credits
except AuthenticationError:
    ...                            # 401 — bad key
except APIStatusError as err:
    print(err.status_code, err.message, err.body)
except APIResponseValidationError:
    ...                            # 2xx whose body didn't match the model
except BlitzError:
    ...                            # base class for everything this SDK raises
401 / 402 / 404 raise immediately; 429 and 5xx are retried automatically; read timeouts surface as APITimeoutError (not retried). See Rate limits & retries for the full retry and timeout behavior.

Types & enums

Response models are Pydantic v2 objects with attribute access and IDE autocomplete. Enum helpers live in blitz_api.types:
from blitz_api.types import Industry, JobLevel, Continent

client.search.people(
    company={"industry": {"include": [Industry.SOFTWARE_DEVELOPMENT]}},
    people={"job_level": [JobLevel.VP]},
)
Enum-backed fields accept an enum member or a raw string, so a value missing from the vendored taxonomy never blocks you. New fields the API adds are preserved on the parsed model rather than breaking deserialization.

Next steps

TypeScript / JavaScript SDK

The same API surface for Node and the browser-adjacent runtimes.

Field Normalization

Accepted values for industry, job level, job function, and geography filters.

API reference

Full request/response schemas and an interactive try-it console.

Recipes

End-to-end workflows: build an ICP list, enrich it, and sync to your CRM.