serpkite

The official Python SDK for SerpKite, the Google search API built for AI agents. You get clean JSON or Markdown, typed pydantic models, sync and asyncio clients, a native CrewAI tool, and credits that never expire.

pip install serpkite

It needs Python 3.9+ and depends only on httpx and pydantic v2.

Quickstart

Create a key at app.serpkite.com/keys and export it:

export SERPKITE_API_KEY=skt_live_...

from serpkite import SerpKite

sk = SerpKite()  # reads SERPKITE_API_KEY
res = sk.search("best espresso machine", country="us")
print(res.results[0].title, res.meta.credits_used)

md = sk.search("best espresso machine", format="markdown")  # str, token-lean for LLMs

Every response shares one envelope: request (the normalised request), results (the vertical's primary list), vertical-specific extras (answer_box, knowledge_graph, people_also_ask, related_searches, top_stories, places, ads on search) and meta (request_id, credits_used, cached, latency_ms, …). All keys are snake_case.

Models keep unknown fields (extra="allow"), so new API fields never break an older SDK. You can read them with res.model_extra or res.model_dump().

Client options

sk = SerpKite(
    api_key="skt_live_...",              # default: SERPKITE_API_KEY
    base_url="https://api.serpkite.com", # default: SERPKITE_BASE_URL or the public API
    timeout=60.0,                        # seconds or an httpx.Timeout
    max_retries=2,                       # 429 / 5xx / connection errors
    http_client=httpx.Client(proxy="http://..."),  # optional, bring your own
    default_headers={"X-Request-Id": "..."},
)

with SerpKite() as sk:  # closes the HTTP client on exit
    ...

Methods

Every query vertical takes q plus optional keyword parameters: country, language, location, uule, num, page, time (hour|day|week|month|year), tbs, device, safe, autocorrect, ads, max_age, engine (see Search engines & fallback) and include_content (search only).

The output format changes the return type, and the overloads make type checkers follow it:

sk.search("q")                          # SearchResponse
sk.search("q", format="markdown")       # str
sk.search("q", format="compact")        # dict: lean results + meta
sk.search("q", fields="results.title,results.link,answer_box")   # dict: projected
sk.webpage("https://example.com", format="markdown")             # str

Some more examples:

# Top 5 results with their pages as Markdown (+1 credit per fetched page)
res = sk.search("vector databases comparison", include_content=5)
for r in res.results:
    print(r.title, len(r.content or ""))

# Reviews, paged
page = sk.reviews(place_id="ChIJN1t_tDeuEmsRUsoyG83frY4", sort="newest")
nxt = sk.reviews(place_id="ChIJN1t_tDeuEmsRUsoyG83frY4", page_token=page.next_page_token)

# Where does example.com rank?
r = sk.rank("espresso machine", "example.com")
print(r.position, r.checked)

# Accept a cached result up to an hour old (half the credits on a hit)
sk.news("fed rate decision", max_age=3600)

Search controls

Domain filters and date ranges work on search, news, images and videos; boost_domains on search and news; highlights on search with include_content. None of them costs extra credits.

res = sk.search(
    "connection pooling",
    include_domains=["postgresql.org", "github.com/pgbouncer", ".edu"],  # host, path prefix or TLD (<= 20)
    exclude_domains=["pinterest.com"],
    boost_domains=["postgresql.org"],  # to the top, keeping the rest
    start_date="2026-01-01",  # YYYY-MM-DD or a date; end_date too
    include_content=3,
    highlights=True,  # 3 query-ranked passages per page instead of the whole page
)
for r in res.results:
    print(r.position, r.published_at, r.highlights[0].text if r.highlights else None)

Search engines & fallback

By default every request is answered by Google only (engine="google"); SerpKite already fails over across its own proxy pools. Opt in to other providers with engine:

# Fall back to other enabled providers when Google is blocked or times out
res = sk.search("best espresso machine", engine="auto")
print(res.meta.engine)  # "google", or e.g. "brave" if Google was unavailable
for step in res.meta.route or []:  # absent on cache hits
    print(step.provider, step.outcome, step.ms)  # google blocked 812 / brave ok 431

# Only these providers, in this order (any sequence of names works)
sk.news("espresso", engine=["google", "brave"])

• engine is "google" (default), "auto", "consensus", one provider ("brave", "bing", "yahoo", "duckduckgo", "mojeek", "wikipedia") or a list. "auto" and "consensus" can't be combined with other names; unknown names, or a provider that doesn't serve the endpoint, raise BadRequestError (invalid_request).
• engine="consensus" (search only) asks several independent indexes in parallel, merges the results by URL and ranks them by agreement: each result has sources (e.g. ["google", "brave"]), meta.engine is "consensus", and it costs the sum of one page per provider that returned results.
• meta.engine names the provider that answered; meta.route lists each attempt as a RouteStep (provider, outcome, ms). request.engine echoes what you asked for.
• Credits (meta.credits_used) follow the answering provider's price.

Async

AsyncSerpKite has the same surface, built on httpx.AsyncClient:

import asyncio
from serpkite import AsyncSerpKite

async def main() -> None:
    async with AsyncSerpKite() as sk:
        res, news = await asyncio.gather(
            sk.search("best espresso machine"),
            sk.news("espresso prices"),
        )
        print(res.results[0].link, news.results[0].source)

asyncio.run(main())

Errors

Every API error raises a serpkite.SerpKiteError with status, code, message and request_id. The class depends on the HTTP status:

import serpkite

try:
    sk.search("espresso")
except serpkite.InsufficientCreditsError:
    print("Top up at https://app.serpkite.com/billing")
except serpkite.SerpKiteError as e:
    print(e.status, e.code, e.message, e.request_id)

Failed, empty and blocked searches are not billed. The client retries 429s, 5xx responses and connection errors up to max_retries times. It uses exponential backoff with jitter and honours Retry-After. It never retries daily_limit_reached. Batch creation is retried only when the request provably never reached the server, unless you pass idempotency_key.

Map and extract

# The URLs of a site (1 credit): sitemaps + start page, canonicalised and deduplicated
site = sk.map("https://docs.example.com/", search="install", include_paths=["^/guides/"])

# Up to 20 URLs (HTML or PDF) as Markdown in one call: 1 credit per URL that came back
pages = sk.extract([u.url for u in site.results[:5]], query="install", highlights=3)
for f in pages.failed:  # not charged
    print("failed", f.url, f.error.code)

Crawl

task = sk.crawl(
    "https://docs.example.com/",
    limit=200,  # up to 1000 pages; max_depth up to 10
    include_paths=["^/guides/"],
    sitemap="include",  # include (default) | only | skip
    query="authentication",  # read the most relevant pages first
)
done = sk.wait_for_crawl(task.id)  # completed, failed or canceled
if done.result:
    for p in done.result.pages:
        print(p.url, len(p.markdown or ""))
    print(done.result.stats.stopped)  # done | limit | time_limit | size_limit | too_many_failures | canceled

• crawl reserves limit credits and charges 1 per page read (0.5 from cache); the rest is refunded. It is only retried on 429, so a lost response never starts a second crawl.
• wait_for_crawl polls with a growing interval (up to 15 s) and returns failed or canceled tasks; it raises TaskTimeoutError (a TimeoutError) when timeout expires.
• cancel_crawl(id) refunds a queued crawl; a running one stops at its next checkpoint (status="canceling") and is charged for the pages read.
• Pass webhook_url= to get a signed crawl.completed delivery instead of polling. A result over 4 MB arrives as result=None with result_omitted=True; fetch it with get_crawl.

Monitors

mon = sk.monitors.create(
    "ai agents",
    endpoint="news",
    interval="hourly",  # hourly | daily | weekly, or interval_seconds=3600…2592000
    webhook_url="https://example.com/hooks/serpkite",  # optional
)
sk.monitors.run(mon.id)  # due within ~30 s
sk.monitors.update(mon.id, active=False)
sk.monitors.update(mon.id, q="ai agent frameworks", num=20)  # change the saved search

# Run history, newest first (new results kept 24 h): one page, or every run
page = sk.monitors.runs(mon.id, limit=20)  # page.next_before → before=
for run in sk.monitors.iter_runs(mon.id):
    print(run.status, run.new_results)

# Watch a page for content changes (1 credit per check); metadata is echoed in webhooks
watch = sk.monitors.create(endpoint="webpage", url="https://example.com/pricing", metadata={"customer": "acme"})
sk.monitors.update(watch.id, metadata=None)  # clears it; leave it out to keep it
sk.monitors.delete(watch.id)

Each search run costs what its search costs (1 credit per 10 results, num 100 is 7; empty and failed runs are free) and reports only results it hasn't seen before (among the top num). Without a webhook_url, read new results from monitors.runs. A monitor pauses itself after 10 failed runs in a row; update(id, active=True) resumes it. webhook_url="" removes the webhook; a changed search reports every result as new once.

Batches

Batches run at half price. You queue 1-100 requests for one endpoint, then poll for the results or receive a signed webhook:

job = sk.batches.create(
    endpoint="search",
    requests=[{"q": "a"}, {"q": "b", "country": "de"}],
    webhook_url="https://example.com/hooks/serpkite",  # optional
)
for entry in job.batches:  # one entry per request, in order: Batch or ErrorResponse
    if isinstance(entry, serpkite.types.Batch):
        done = sk.batches.wait(str(entry.id), timeout=300, poll_interval=2)
        if done.status == "done":
            res = serpkite.types.SearchResponse.model_validate(done.result)

batches.wait raises BatchTimeoutError (a TimeoutError) when the timeout expires. Results are kept for 24 hours.

Pass idempotency_key= (e.g. a UUID you store with the batch) to make create safe to retry: for 24 hours the server replays the first response for the same key and body, so the client then also retries 5xx and connection errors.

Webhook deliveries (batch.completed, crawl.completed, monitor.results) are always signed. Check one with the raw body, then parse it into a typed event:

from serpkite import parse_webhook, verify_webhook
from serpkite.types import MonitorResultsEvent, TaskCompletedEvent

body = request.get_data()
if not verify_webhook(os.environ["SERPKITE_WEBHOOK_SECRET"], body, request.headers):
    abort(401)
event = parse_webhook(body, request.headers)
if isinstance(event, MonitorResultsEvent):  # dedupe retries on event.run_id
    print(event.name, len(event.new_results))
elif isinstance(event, TaskCompletedEvent):  # crawl.completed
    print(event.status, event.poll_url)

Or in one step: parse_webhook(body, request.headers, secret=...) verifies first and raises WebhookSignatureError on a bad signature or a stale timestamp.

CrewAI

pip install "serpkite[crewai]"

from crewai import Agent
from serpkite.crewai import SerpKiteSearchTool

tool = SerpKiteSearchTool()  # optional: endpoint="news", country="de", num=10, engine="auto"
agent = Agent(role="Researcher", goal="Find current facts", backstory="...", tools=[tool])

The tool returns Markdown, which uses far fewer tokens than JSON. The agent can set query, num, country, language and time. endpoint can be search, news, images, videos, maps, places, shopping, scholar, patents or autocomplete. engine (you set it, not the agent) opts in to fallback providers.

LangChain

See langchain-serpkite for LangChain tools, a retriever and a document loader.

Development

uv sync
uv run pytest
uv run ruff check . && uv run ruff format --check .
uv run mypy
./scripts/gen_models.sh   # regenerate src/serpkite/_models.py from backend/api/serp-api.yaml
uv run --with "crewai>=0.80" pytest tests/test_crewai.py   # optional CrewAI tests

License

MIT