Python SDK

Get started with TopK using Python SDK.

The TopK Python library provides convenient access to the TopK API from any Python 3.9+ application. The library includes type definitions for all request params and response fields, and features both synchronous and asynchronous clients.

Installation

pip install topk-sdk

# or

uv add topk-sdk

Building with a coding agent

Using Claude Code, Cursor, Codex, or another coding agent? Install the TopK skill so the agent writes correct TopK code. It covers schema design, chunking, hybrid queries, multi-tenancy, and the common mistakes.

npx skills add topk-io/topk --skill topk

Prerequisites

• API key — sign in to console.topk.io and generate an API key.
• Region — available regions are listed at docs.topk.io/regions.

Usage

Hybrid Search

import os
from topk_sdk import Client
from topk_sdk.schema import text, keyword_index, semantic_index
from topk_sdk.query import select, field, fn

client = Client(
    api_key=os.environ["TOPK_API_KEY"],
    region="aws-us-east-1-elastica",
)

# Create a collection
client.collections().create(
    "books",
    schema={
        "title": text().required().index(keyword_index()),
        "content": text().index(semantic_index()),
    },
)

# Upsert documents
client.collection("books").upsert([
    {
        "_id": "1",
        "title": "Catcher in the Rye",
        "content": "IF YOU REALLY WANT TO HEAR about it, the first thing you'll probably want to know is ...",
        "author": "J.D. Salinger",
        "rating": 3.8,
    },
    {
        "_id": "2",
        "title": "1984",
        "content": "It was a bright cold day in April, and the clocks were striking thirteen. Winston Smith, ...",
        "author": "George Orwell",
        "rating": 4.7,
    },
])

# Query with hybrid search
results = client.collection("books").query(
    select(
        # Select document fields to return
        "_id", "title", "author",
        # Compute semantic similarity of content field with the query
        similarity_score=fn.semantic_similarity(
            "content",
            "What is the meaning of life?",
        ),
    )
    # Filter documents by metadata
    .filter(field("rating") >= 3.0)
    # Rank using the computed similarity score and rating
    .sort(field("rating") * field("similarity_score"), asc=False)
    # Get top 10 highest ranked documents
    .limit(10)
)

Vector Search

import os
from topk_sdk import Client
from topk_sdk.schema import text, f32_vector, vector_index
from topk_sdk.query import select, field, fn

client = Client(
    api_key=os.environ["TOPK_API_KEY"],
    region="aws-us-east-1-elastica",
)

# Create a collection with a vector field (dimension must match your embedding model's output size)
client.collections().create(
    "books",
    schema={
        "title": text().required(),
        "embedding": f32_vector(dimension=1536).required().index(vector_index(metric="dot_product")),
    },
)

# Upsert documents with embeddings
client.collection("books").upsert([
    {"_id": "1", "title": "Catcher in the Rye", "embedding": [0.1, 0.2, ...]},
    {"_id": "2", "title": "1984",               "embedding": [0.9, 0.8, ...]},
])

# Query the nearest neighbors to a query vector
results = client.collection("books").query(
    select(
        "_id", "title",
        distance=fn.vector_distance("embedding", [0.8, 0.9, ...]),
    )
    # Return the 10 closest documents (ascending = closest first)
    .sort(field("distance"), asc=True)
    .limit(10)
)

Async usage

Simply import AsyncClient instead of Client and use async for / await with each API call:

import os
import asyncio
from topk_sdk import AsyncClient
from topk_sdk.schema import text, keyword_index, semantic_index
from topk_sdk.query import select, field, fn

client = AsyncClient(
    api_key=os.environ["TOPK_API_KEY"],
    region="aws-us-east-1-elastica",
)

async def main() -> None:
    # Hybrid search
    await client.collections().create(
        "books",
        schema={
            "title": text().required().index(keyword_index()),
            "content": text().index(semantic_index()),
        },
    )

    await client.collection("books").upsert([
        {"_id": "1", "title": "Catcher in the Rye", "content": "...", "rating": 3.8},
        {"_id": "2", "title": "1984", "content": "...", "rating": 4.7},
    ])

    results = await client.collection("books").query(
        select(
            # Select document fields to return
            "_id", "title",
            # Compute semantic similarity of content field with the query
            similarity_score=fn.semantic_similarity("content", "What is the meaning of life?"),
        )
        # Filter documents by metadata
        .filter(field("rating") >= 3.0)
        # Rank using the computed similarity score and rating
        .sort(field("rating") * field("similarity_score"), asc=False)
        # Get top 10 highest ranked documents
        .limit(10)
    )

asyncio.run(main())

Handling errors

from topk_sdk.error import (
    CollectionNotFoundError,
    PermissionDeniedError,
    QuotaExceededError,
    SlowDownError,
)

try:
    client.collections().get("books")
except CollectionNotFoundError:
    print("Collection does not exist")
except PermissionDeniedError:
    print("Check your API key")
except QuotaExceededError:
    print("Usage quota exceeded")
except SlowDownError:
    print("Rate limited — back off and retry")

Error	Description
CollectionNotFoundError	Collection does not exist
PartitionNotFoundError	Partition does not exist
CollectionAlreadyExistsError	Collection with this name already exists
CollectionValidationError	Invalid collection name or schema
DocumentValidationError	Invalid document
SchemaValidationError	Invalid schema
PermissionDeniedError	Invalid or missing API key
QuotaExceededError	Usage quota exceeded
RequestTooLargeError	Request payload too large
SlowDownError	Rate limited by the server
QueryLsnTimeoutError	Timed out waiting for write consistency

Retries

The client automatically retries transient errors (server unavailable, network failures) and LSN consistency timeouts. Retry behaviour can be configured via RetryConfig:

from topk_sdk import Client, RetryConfig, BackoffConfig

client = Client(
    api_key=os.environ.get("TOPK_API_KEY"),
    region="aws-us-east-1-elastica",
    retry_config=RetryConfig(
        max_retries=5,        # default: 3
        timeout=60_000,       # total retry chain timeout in ms, default: 30,000
        backoff=BackoffConfig(
            init_backoff=200, # default: 100 ms
            max_backoff=5_000, # default: 10,000 ms
        ),
    ),
)

Requirements

Python 3.9 or higher.