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), include_content (search only) and include_answer (search and news, see AI answers).

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

# One place in full: structured hours, phones, time zone, photos
p = sk.place(place_id="ChIJN1t_tDeuEmsRUsoyG83frY4").results[0]
print(p.timezone, p.weekly_hours.monday if p.weekly_hours else None)  # [HoursInterval(open='09:00', ...)]

# Reviews, paged (the first 10 of a place)
page = sk.reviews(place_id="ChIJN1t_tDeuEmsRUsoyG83frY4", num=5)
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", "yandex") 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. extract, answer, crawl and research are billed even when the response is lost, so they are retried only on 429 and connection failures; answer and answer_stream wait at least 100 s for the response (the search runs before the model writes).

AI answers

from serpkite.types import AnswerStreamDelta, AnswerStreamDone

# A cited answer grounded on live Google results ("fast" answers from the result page;
# "standard", the default, also reads the top num_sources pages)
res = sk.answer("what is retrieval-augmented generation", depth="fast")
print(res.answer)  # Markdown with [1] markers
for c in res.citations:
    print(c.index, c.title, c.link)
print(res.meta.credits, res.meta.llm)  # search/answer split, model and tokens

# Stream it: the numbered sources first, then text deltas, then the final answer
for ev in sk.answer_stream("what is retrieval-augmented generation"):
    if isinstance(ev, AnswerStreamDelta):
        print(ev.text, end="", flush=True)
    elif isinstance(ev, AnswerStreamDone):
        print("\n", [c.link for c in ev.citations])

# Structured output: a JSON Schema dict or a pydantic model class (not with streaming)
from pydantic import BaseModel

class Building(BaseModel):
    name: str
    height_m: float

facts = sk.answer("tallest building in the world", output_schema=Building)
print(Building.model_validate(facts.output))

# A cited answer on top of a normal search or news response
news = sk.news("ECB rate decision", time="day", include_answer=True)
print(news.answer.text if news.answer else news.meta.warnings)  # refunded with a warning if it failed

# Summaries or JSON per page
pages = sk.extract(["https://example.com/pricing"], summary=True,
                   output_schema={"type": "object", "properties": {"price": {"type": "number"}}})
print(pages.results[0].summary, pages.results[0].output)

• A model failure raises APIError with code ai_unavailable (503); the model part is refunded. In a stream it arrives as an error event and is raised the same way. Closing the iterator early closes the connection, and the model part is not charged. AsyncSerpKite.answer_stream is an async iterator (async for ev in sk.answer_stream(q)).
• extract: summary=True and output_schema (+ optional prompt) each cost extra per page; a page whose model step failed has summary_error / output_error instead and isn't charged for it. Its default timeout becomes 75 seconds with either.
• AI prices (answers, include_answer, summaries, schema extraction, research) are listed at serpkite.com/pricing; meta.credits on an answer splits the search and the model part.

Research

task = sk.research(
    "Compare managed Postgres vector search options in 2026",
    effort="standard",  # lite | standard (default) | deep
)
done = sk.wait_research(task.id)  # completed, failed or canceled (15 min default timeout)
if done.status == "completed" and done.result:
    print(done.result.report)  # Markdown with [n] citations
    for src in done.result.sources:
        print(src.index, src.link, src.query)

• Each effort has a flat price (serpkite.com/pricing), reserved up front and charged only when the report is written; failed research is refunded. research is only retried on 429.
• get_research(id) shows progress (round, queries, sources, step) while it runs. cancel_research(id) refunds a queued task; a running one stops at its next round and is charged for the searches already run.
• output_schema (a dict or a pydantic model class) also returns the report as JSON in result.output; include_domains / exclude_domains, country and language steer the searches. Pass webhook_url= to get a signed research.completed delivery instead of polling.

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, research.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 or research.completed (event.kind)
    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