Python SDK
The generated aacsearch client — installation, constructor, keyword-only methods, AACSearchError and the retry caveat.
The Python client lives in packages/sdk-python, distribution name aacsearch, package aacsearch. It requires Python 3.9+ and requests>=2.31, and is generated from the same OpenAPI document as the TypeScript SDK.
pip install -e packages/sdk-python # inside the monorepofrom aacsearch import AACSearchClient, AACSearchError
client = AACSearchClient("aac_live_9f3c2a1b…", base_url="https://search.example.com/api/v1")
result = client.search(index="documentation", query={"q": "pricing", "per_page": 5})
print(result["found"], result["hits"][0]["document"]["title"])Constructor
AACSearchClient(
api_key: str,
base_url: str = "http://localhost:3000/api/v1",
max_retries: int = 2,
)It keeps a requests.Session and sends Authorization: Bearer <key> plus Content-Type: application/json on every call. An empty api_key raises before any request:
AACSearchClient("")
# AACSearchError: api_key is required (status=401, code=MISSING_API_KEY)Errors
class AACSearchError(Exception):
status: int # HTTP status
code: str # UNAUTHORIZED, QUOTA_EXCEEDED, VALIDATION_ERROR…
details: Any # details from the error envelope
request_id: str | Nonetry:
client.upsert_documents(
index="documentation",
body={"documents": [{"external_id": "pricing", "content": "Pro is 29 USD per month."}]},
)
except AACSearchError as error:
if error.code == "RATE_LIMITED":
time.sleep(2)
elif error.code == "QUOTA_EXCEEDED":
raise SystemExit(f"Quota exhausted until {(error.details or {}).get('resetsAt')}")
else:
raise RuntimeError(f"{error.code} {error} (request {error.request_id})")__str__ returns the message from the envelope, which is why str(error) reads like a sentence rather than a status code.
Retries
The client defines a _request helper that retries 429 and 5xx up to max_retries with 250 ms × 2ⁿ backoff. The generated methods call requests directly and do not go through it, so in the current build you should assume no automatic retries: wrap calls that matter in your own retry with jitter, and treat 429 as Retry-After-less — read error.details["reset_at"] when present.
Methods
All methods are keyword-only after self, use snake_case, and return the parsed JSON as a dict. query arguments become query-string parameters; body arguments become the JSON body.
Search and discovery
| Method | HTTP |
|---|---|
search(index=..., query={...}) | GET /indexes/{index}/search |
search_post(index=..., body={...}) | POST /indexes/{index}/search |
suggest(index=..., query={...}) | GET /indexes/{index}/suggest |
multi_search(body={...}) | POST /indexes/multi_search |
nl_search(index=..., body={...}) | POST /indexes/{index}/nl-search |
index_stats(index=..., query={...}) | GET /indexes/{index}/stats |
health(), capabilities(), openapi() | unauthenticated discovery endpoints |
metrics() | GET /metrics (needs the operations token) |
hits = client.search(
index="documentation",
query={"q": "pricing", "query_by": "title,content", "filter_by": "tags:=[billing]", "per_page": 5},
)
for hit in hits["hits"]:
print(hit["document"]["title"], hit["score"])Documents
| Method | HTTP |
|---|---|
upsert_documents(index=..., body={...}) | POST /indexes/{index}/documents |
get_document(index=..., externalId=...) | GET /indexes/{index}/documents/{externalId} |
delete_document(index=..., externalId=...) | DELETE /indexes/{index}/documents/{externalId} |
delete_documents(index=..., query={...}) | DELETE /indexes/{index}/documents?filter_by=… |
bulk_update_documents(index=..., body={...}) | POST /indexes/{index}/documents/bulk |
export_documents(index=...) | GET /indexes/{index}/documents/export |
Note the camelCase externalId parameter — it is the one place the generated signature kept the wire name.
client.upsert_documents(
index="documentation",
body={
"documents": [
{
"external_id": "pricing",
"title": "Pricing",
"content": "Pro is 29 USD per month with 500,000 searches.",
"tags": ["billing"],
}
],
"action": "upsert",
},
)
document = client.get_document(index="documentation", externalId="pricing")
client.delete_documents(index="documentation", query={"filter_by": "source_type:=connector"})AI
| Method | HTTP |
|---|---|
answer(index=..., body={...}) | POST /indexes/{index}/answer |
assistant_ask(index=..., body={...}) | POST /indexes/{index}/assistant |
assistant_stream(index=..., body={...}) | POST /indexes/{index}/assistant/stream — yields events |
assistant_native(index=..., body={...}) | POST /indexes/{index}/assistant/native |
assistant_threads(index=...) | GET /indexes/{index}/assistant |
assistant_thread(index=..., conversationId=...) | GET /indexes/{index}/assistant/{conversationId} |
conversation_models(index=...) | GET /indexes/{index}/conversation-models |
voice_search(index=..., body={...}) | POST /indexes/{index}/voice |
image_search(index=..., body={...}) | POST /indexes/{index}/image |
answer = client.answer(index="documentation", body={"question": "How much does pro cost?"})
print(answer["answer"])
for citation in answer["citations"]:
print(f"[{citation['index']}] {citation['title']} {citation['url'] or ''}")
for event in client.assistant_stream(index="documentation", body={"question": "And the limits?"}):
if event["type"] == "delta":
print(event["payload"].get("text", ""), end="")assistant_stream is a generator over parsed SSE frames: {"type": "citations" | "delta" | "done" | "error", "payload": …}.
Relevance, operations, analytics
| Method | HTTP |
|---|---|
list_synonyms / put_synonym / delete_synonym | /indexes/{index}/synonyms |
list_curations / put_curation / delete_curation | /indexes/{index}/curations |
track_event(index=..., body={...}) | POST /indexes/{index}/events |
reindex(index=...) | POST /indexes/{index}/reindex |
scoped_key(index=..., body={...}) | POST /indexes/{index}/scoped-key |
usage(), wallet() | /billing/usage, /billing/wallet |
analytics_overview(query={...}) | GET /analytics/overview |
analytics_overview_csv(query={...}) | GET /analytics/overview.csv |
slow_queries(query={...}), trending_queries(query={...}) | /analytics/slow-queries, /analytics/trending |
export_organization(), purge_organization(query={...}) | /organization/export, /organization/data |
trigger_snapshot(), run_jobs(query={...}) | /admin/snapshot, /admin/jobs/run |
Caveats
The generated Python client mirrors the TypeScript one, including its mismatches with the server:
multi_search(body={...})posts to the federated route/indexes/multi_searchand returns one result per search.export_documentscallsresponse.json()on an NDJSON stream and will raise. Readresponse.textinstead.webhook_ingestposts no body and no signature headers; sign the payload yourself and userequests.multi_suggestsends no body while the endpoint requiresindexesandq.
For anything outside the generated surface, the raw endpoints in the API reference are the contract — the client adds no behaviour of its own beyond base URL handling and JSON encoding.