Canonical Docs
Sign up

Python SDK

Type-safe client with sync and async support. Works with Python 3.11+.

Installation

pip install canonical-search

Quick Start

from canonical_search import CanonicalClient

client = CanonicalClient(api_key="YOUR_API_KEY")
results = client.search("AI healthcare startups")

for company in results.results:
    print(f"{company.name} — {company.domain}")
    print(f"  {company.description[:100]}")

Configuration

Direct

client = CanonicalClient(
    api_key="YOUR_API_KEY",
    base_url="https://trycanonical.ai",  # default
    timeout=30.0,                          # default
)

Environment Variables

export CANONICAL_API_KEY=YOUR_API_KEY
export CANONICAL_API_BASE_URL=https://trycanonical.ai  # optional
export CANONICAL_TIMEOUT=30.0                          # optional
from canonical_search.config import client_from_env

client = client_from_env()
results = client.search("fintech companies in Europe")

Search Methods

Sync

results = client.search(
    query="B2B SaaS companies",
    top_k=50,          # 1-1000, default 25
)

Async

results = await client.asearch(
    query="B2B SaaS companies",
    top_k=50,
)

Parameters

Parameter Type Default Description
query str required Natural language search query
top_k int 25 Number of results (1–1000)

Credits: search, search_structured, and find_similar_companies each cost 1 credit per strong (verified) result returned — partial matches (verdict == "partial") are never charged. The exact charge is on SearchResponse.credits_used, and the running balance on credits_remaining.

Structured Search

Search by typed filters instead of (or alongside) a free-text description — location, headcount, funding, founding year, founder pedigree, and exclusions.

from canonical_search import CanonicalClient, StructuredFilters, StructuredLocation

client = CanonicalClient(api_key="YOUR_API_KEY")

results = client.search_structured(
    description="AI diagnostic platforms for hospitals",  # optional semantic core
    filters=StructuredFilters(
        location=StructuredLocation(countries=["United States"]),
        employee_count_min=50,
        employee_count_max=100,
        funding_series=["series_a", "series_b"],
        founding_year_min=2018,
    ),
    intent="sales_prospecting",  # optional ranking profile
    top_k=25,
)

search_structured() parameters

Parameter Type Default Description
description str | None None Free-text semantic core (the only fuzzy field). Omit to search by filters only.
filters StructuredFilters | None None Typed filters (see below).
intent str | None None Ranking profile (see allowed values below).
top_k int 25 Number of results (1–1000).
include_partials bool false Also return LLM-evaluated partial matches (unbilled).

StructuredFilters fields

Field Type Description
location StructuredLocation | None HQ filter: cities / states / countries, each a list[str]. Most-granular wins.
employee_count_min / _max int | None Headcount range (inclusive).
funding_series list[str] | None Funding stages (see allowed values below).
funding_min_usd / _max_usd int | None Latest-round amount range in USD.
funding_post_money_min_usd / _max_usd int | None Post-money valuation range in USD.
funded_after str | None ISO date YYYY-MM-DD — funded on/after this date.
funding_investor list[str] | None Investor name variants; any match satisfies the filter.
people StructuredPeople | None Founder filters: has_repeat_founder, has_technical_cofounder (bool), founder_prior_categories / founder_prior_companies (list[str]).
founding_year_min / _max int | None Founding-year range (inclusive).
founding_year_exclude_min / _max int | None Contiguous year range to exclude (set both).
exclude_company_domains list[str] | None Domains to drop from results.
exclude StructuredExclude | None Semantic/structured negations: description, location, funding_series, funding_investor.

Allowed values

  • funding_series: pre_seed, seed, series_a, series_b, series_c, series_d, series_e, series_f, series_g_plus, growth, late, bridge, venture, angel, other
  • intent: sales_prospecting, sales_timing, sales_expansion, competitive_tracking, emerging_competitor_scan, talent_source, recruiter_employer_vet, jobseeker_stability, jobseeker_growth
  • founder_prior_categories: faang, big_tech, unicorn, top_startup, mbb

Unknown enum values raise InvalidRequestError (422) with the valid set. Also available as await client.asearch_structured(...); returns the same SearchResponse as search().

Find Similar

Given a seed company's domain, return canonically similar companies. Returns a SearchResponse.

results = client.find_similar_companies(
    company_domain="stripe.com",
    top_k=25,               # 1-1000, default 25
    intent="sales_timing",   # optional ranking profile
)

# async: await client.afind_similar_companies("stripe.com")
Parameter Type Default Description
company_domain str required Seed company domain (e.g. stripe.com). Case- and scheme-insensitive.
top_k int 25 Number of similar companies (1–1000).
intent str | None None Optional ranking profile (same slugs as structured search).

Credits: 1 credit per strong result returned (same as search).

Lookup

Resolve free-text company names to ranked candidates before passing a domain to another call. Returns a LookupResponse.

resp = client.lookup_companies(["stripe", "adyen"], k=5)

for name, result in resp.results.items():
    if result.auto_resolve_recommended:
        print(name, "→", result.primary_candidate_domain)
    else:
        # genuine toss-up — let the user pick from result.candidates
        print(name, "is ambiguous:", [c.domain for c in result.candidates])

# async: await client.alookup_companies([...])
Parameter Type Default Description
names list[str] required Free-text company names. Max 20 per call.
k int 5 Candidates to return per name (1–25).
disambiguation_mode str "auto_when_confident" One of auto_when_confident, always_auto, always_ask.

Credits: 1 credit per call when at least one candidate is returned — a no-match lookup is free.

Company Details

Drill into one company by domain — full profile, leadership people, and corporate relationships. Returns a CompanyDetails. Costs 1 credit when the company is found.

details = client.get_company_details("stripe.com")

print(details.company.name, details.company.employee_count)
for person in details.people:
    print(person.name, "—", person.role)

# async: await client.aget_company_details("stripe.com")
Parameter Type Default Description
company_domain str required Company domain (the public handle from search / lookup). Raises NotFoundError if unknown.

Account Status

Check your credit balance, plan, and rate limits. Read-only and free — charges no credits. Returns an AccountStatus.

status = client.get_account_status()
print(status.plan, status.credits.total, "credits left")
print(status.rate_limits.per_minute, "req/min")

# async: await client.aget_account_status()

Response Types

SearchResponse

class SearchResponse:
    results: list[Company]    # Matching companies
    count: int                # Number of results
    query: str                # Original query
    credits_used: int         # Credits consumed
    credits_remaining: int | None  # Remaining credits

Company

class Company:
    name: str
    website: str
    domain: str                        # public handle — use for follow-up calls
    description: str
    headquarters: str | None
    employee_count: int | None
    founding_year: int | None
    dimensions: dict | None            # e.g. {"industry": ["Healthcare"]}
    funding: FundingSummary | None     # latest round, totals, investors

FundingSummary

class FundingSummary:
    total_usd: float | None                  # equity rounds only
    round_count: int
    latest_date: str | None                  # ISO "YYYY-MM-DD"
    latest_amount_usd: float | None
    latest_series: str | None                # e.g. "series_b"
    latest_post_money_valuation: float | None
    latest_lead_investor: str | None
    all_investors: list[str]

CompanyDetails

class CompanyDetails:            # get_company_details(...)
    company: CompanyProfile | None    # full profile (name, hq, funding, dimensions, defunct)
    people: list[Person]              # founders + execs (may be empty)
    leadership_data: str | None       # "available" | "not_available" | "error"
    relationships: Relationships | None  # corporate family (present when edges exist)
    credits_used: int | None
    credits_remaining: int | None

class Person:
    name: str
    role: str                         # e.g. "ceo", "cofounder"
    role_secondary: str | None
    headline: str | None
    linkedin_url: str | None
    prior_companies: list[PriorCompany]

LookupResponse

class LookupResponse:             # lookup_companies(...)
    results: dict[str, LookupResult]  # keyed by each input name
    credits_used: int
    credits_remaining: int | None

class LookupResult:
    confidence: str                   # "high" | "medium" | "low"
    auto_resolve_recommended: bool    # True = safe to use primary; False = ask the user
    primary_candidate_domain: str | None
    candidates: list[LookupCandidate] # each has name, domain, headquarters, description

AccountStatus

class AccountStatus:              # get_account_status()
    credits: Credits                  # .total / .subscription / .extra / .subscription_resets_at
    plan: str | None                  # "free" | "starter" | "pro"
    rate_limits: RateLimits           # .per_minute / .per_day
    billing_url: str

Error Handling

from canonical_search import (
    CanonicalClient,
    AuthenticationError,
    InsufficientCreditsError,
    NotFoundError,
    InvalidRequestError,
    RateLimitError,
    APIError,
)

client = CanonicalClient(api_key="YOUR_API_KEY")

try:
    results = client.search("fintech")
except AuthenticationError:
    print("Invalid or missing API key")
except InsufficientCreditsError:
    print("Out of credits — top up at /billing")
except RateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after}s")
except APIError as e:
    print(f"API error: {e}")

All exceptions subclass CanonicalError and carry .status_code and .detail (the parsed error body); RateLimitError adds .retry_after (seconds).

Exception HTTP Status Meaning
AuthenticationError 401 Invalid or missing API key
InsufficientCreditsError 402 Not enough credits for the request
NotFoundError 404 Company / resource doesn't exist
InvalidRequestError 400 / 422 Malformed request or failed validation
RateLimitError 429 Rate limit exceeded (see .retry_after)
APIError 5xx Server error