Python
Official Python SDK for Denser Retriever Platform
Installation
pip install denserai-retriever-sdkQuick Start
from denser_retriever import DenserRetriever
client = DenserRetriever(api_key="your-api-key")
def quick_example():
# Create a knowledge base
kb = client.create_knowledge_base("My First KB")
kb_id = kb["data"]["id"]
# Import text content
client.import_text_content_and_poll(
knowledge_base_id=kb_id,
title="Getting Started",
content="Denser Retriever enables semantic search across your documents."
)
# Search
results = client.query(query="semantic search")
print(results["data"])
# Cleanup
client.delete_knowledge_base(kb_id)
if __name__ == "__main__":
quick_example()Configuration
Initialize the client with your API credentials:
from denser_retriever import DenserRetriever
client = DenserRetriever(
api_key="YOUR_API_KEY", # Required: Your API key
timeout=30 # Optional: Request timeout in seconds (default: 30)
)You can find your API key in the Denser Retriever Dashboard.
API Reference
All methods return a dictionary with success (bool) and data fields.
Knowledge Base Management
create_knowledge_base
Create a new knowledge base.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
name | str | Yes | Knowledge base name |
description | str | No | Optional description |
kb = client.create_knowledge_base(
name="Technical Documentation",
description="Product docs and API references"
)
kb_id = kb["data"]["id"]Returns:
{
"success": True,
"data": {
"id": str,
"name": str,
"description": str | None,
"createdAt": str,
"updatedAt": str
}
}list_knowledge_bases
List all knowledge bases in your organization.
kbs = client.list_knowledge_bases()
for kb in kbs['data']:
print(f"{kb['name']} (ID: {kb['id']})")update_knowledge_base
Update knowledge base metadata.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
knowledge_base_id | str | Yes | Knowledge base ID |
name | str | No | New name |
description | str | No | New description |
client.update_knowledge_base(
knowledge_base_id=kb_id,
name="Updated Name",
description="Updated description"
)delete_knowledge_base
Permanently delete a knowledge base and all its documents.
client.delete_knowledge_base(kb_id)File Import Workflow
Uploading files requires a three-step process:
Step 1: Get Presigned URL
presign_upload_url
Generate a presigned URL for file upload.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
knowledge_base_id | str | Yes | Target knowledge base ID |
file_name | str | Yes | File name with extension |
size | int | Yes | File size in bytes (max: 52,428,800) |
presign = client.presign_upload_url(kb_id, "document.pdf", 1024000)
file_id = presign["data"]["fileId"]
upload_url = presign["data"]["uploadUrl"]
expires_at = presign["data"]["expiresAt"]Step 2: Upload File
Use the requests library to PUT the file to the presigned URL:
import requests
with open(file_path, "rb") as f:
requests.put(
upload_url,
data=f,
headers={"Content-Type": "application/octet-stream"}
)import urllib.request
with open(file_path, "rb") as f:
req = urllib.request.Request(
upload_url,
data=f,
headers={"Content-Type": "application/octet-stream"},
method="PUT"
)
with urllib.request.urlopen(req) as response:
passThe presigned URL expires after a certain time. Make sure to upload the file before expiration.
Step 3: Register and Process
import_file
Register the uploaded file for processing.
doc = client.import_file(file_id)
print(f"Document ID: {doc['data']['id']}, Status: {doc['data']['status']}")import_file_and_poll
Register file and automatically poll until processing completes.
Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
file_id | str | - | File ID from presign_upload_url |
options.intervalMs | int | 2000 | Polling interval in milliseconds |
options.timeoutMs | int | 600000 | Maximum wait time in milliseconds |
doc = client.import_file_and_poll(
file_id,
options={"intervalMs": 2000, "timeoutMs": 600000}
)
# Returns when status is "processed" or raises exception on "failed"/"timeout"Text Content Import
import_text_content
Import text content directly as a document.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
knowledge_base_id | str | Yes | Target knowledge base ID |
title | str | Yes | Document title (max: 256 chars) |
content | str | Yes | Text content (max: 1,000,000 chars) |
doc = client.import_text_content(
knowledge_base_id=kb_id,
title="API Documentation",
content="Complete API reference and usage examples..."
)import_text_content_and_poll
Import text content and poll until processing completes.
Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
knowledge_base_id | str | - | Target knowledge base ID |
title | str | - | Document title |
content | str | - | Text content |
options.intervalMs | int | 2000 | Polling interval in milliseconds |
options.timeoutMs | int | 600000 | Maximum wait time in milliseconds |
doc = client.import_text_content_and_poll(
knowledge_base_id=kb_id,
title="Release Notes",
content="Version 2.0 includes...",
options={"intervalMs": 1000, "timeoutMs": 300000}
)Document Management
list_documents
List all documents in a knowledge base.
docs = client.list_documents(kb_id)
for doc in docs['data']:
print(f"{doc['title']} - {doc['status']} ({doc['size']} bytes)")Returns:
{
"success": True,
"data": [
{
"id": str,
"title": str,
"type": str,
"size": int,
"status": str, # "pending" | "processing" | "processed" | "failed" | "timeout"
"createdAt": str
}
]
}get_document_status
Check document processing status.
status = client.get_document_status(doc_id)
print(f"Status: {status['data']['status']}")Status Values:
| Status | Description |
|---|---|
pending | Queued for processing |
processing | Currently being processed |
processed | Successfully processed and searchable |
failed | Processing failed |
timeout | Processing timed out |
delete_document
Permanently delete a document.
client.delete_document(doc_id)Query Method
query
Perform semantic search across knowledge bases.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
query | str | Yes | Search query (1-8192 chars) |
knowledge_base_ids | list[str] | No | Filter by specific knowledge bases |
limit | int | No | Maximum results (default: 10, max: 50) |
# Search across all knowledge bases
results = client.query("machine learning algorithms")
# Search within specific knowledge bases
results = client.query(
query="deployment guide",
knowledge_base_ids=[kb_id1, kb_id2],
limit=20
)
# Process results
for item in results['data']:
print(f"Score: {item['score']:.3f}")
print(f"Title: {item['title']}")
print(f"Content: {item['content']}")
print(f"Document: {item['document_id']}")
print(f"KB: {item['knowledge_base_id']}")
print(f"Metadata: {item.get('metadata')}")
print("---")Returns:
{
"success": True,
"data": [
{
"id": str,
"score": float,
"document_id": str,
"knowledge_base_id": str,
"title": str,
"type": str,
"content": str,
"metadata": {
"source": str | None,
"annotations": str
}
}
]
}Account Management
get_usage
Retrieve current usage statistics for your organization.
usage = client.get_usage()
print(f"Knowledge Bases: {usage['data']['knowledgeBaseCount']}")
print(f"Storage Used: {usage['data']['storageUsed']} bytes")Returns:
{
"success": True,
"data": {
"knowledgeBaseCount": int,
"storageUsed": int # bytes
}
}get_balance
Retrieve current credit balance for your account.
balance = client.get_balance()
print(f"Balance: {balance['data']['balance']} credits")Returns:
{
"success": True,
"data": {
"balance": float
}
}Error Handling
The SDK raises APIError for all API-related errors, providing structured error information for robust error handling.
APIError Class
class APIError(Exception):
def __init__(self, message: str, code: str | None = None,
http_status: int | None = None, data: dict | None = None):
self.message = message # Human-readable error message
self.code = code # Machine-readable error code
self.http_status = http_status # HTTP status code
self.data = data # Additional error contextError Handling Pattern
from denser_retriever import DenserRetriever, APIError
client = DenserRetriever(api_key="your-api-key")
try:
results = client.query(
query="search term",
knowledge_base_ids=["invalid-kb-id"],
limit=5
)
print(results["data"])
except APIError as error:
print(f"[{error.code}] {error.message}")
print(f"HTTP Status: {error.http_status}")
# Handle specific error codes
if error.code == "INSUFFICIENT_CREDITS":
print("Please top up your account to continue.")
elif error.code == "NOT_FOUND":
print("The requested resource does not exist.")
elif error.code == "STORAGE_LIMIT_EXCEEDED":
print(f"Storage quota exceeded: {error.data}")
elif error.code == "KNOWLEDGE_BASE_LIMIT_EXCEEDED":
print("Maximum knowledge bases reached.")
else:
print(f"An error occurred: {error.message}")
except Exception as error:
print(f"Unexpected error: {error}")Common Error Codes
| Error Code | HTTP Status | Description |
|---|---|---|
INPUT_VALIDATION_FAILED | 422 | Request parameters failed validation |
UNAUTHORIZED | 401 | Invalid or missing API key |
FORBIDDEN | 403 | Access to resource is denied |
NOT_FOUND | 404 | Requested resource does not exist |
INSUFFICIENT_CREDITS | 403 | Account has insufficient credits |
STORAGE_LIMIT_EXCEEDED | 403 | Storage quota exceeded |
KNOWLEDGE_BASE_LIMIT_EXCEEDED | 403 | Maximum knowledge bases reached |
INTERNAL_SERVER_ERROR | 500 | Server encountered an error |
Always implement proper error handling in your application to gracefully handle these error cases.
Next Steps
- Check the API Reference for detailed endpoint documentation