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 |