Python SDK

Official Python client for VectorAmp datasets, ingestion, search, and Intelligence. Typed and defaults to https://api.vectoramp.com.

Install

pip install vectoramp

From a checkout for local development:

git clone https://github.com/vectoramp/vectoramp-python.git
cd vectoramp-python
pip install -e '.[dev]'

Configure

The SDK sends API keys with the X-API-Key header. The API key is the only required input, and it can come from the environment.

from vectoramp import VectorAmp

# Reads VECTORAMP_API_KEY from the environment when api_key is omitted.
client = VectorAmp()

export VECTORAMP_API_KEY=vsk_...

Pass the key explicitly or override the host when needed:

client = VectorAmp(api_key="vsk_...")

Quickstart

A minimal init, a one-call create, then object → method calls. The only required argument to create is the name: the default embedding is VectorAmp-Embedding-4B (provider="vectoramp"), the dimension is inferred (2560), the metric defaults to cosine, and the index is always SABLE.

from vectoramp import VectorAmp

client = VectorAmp()

# One call creates a SABLE dataset with the managed VectorAmp embedding model.
dataset = client.datasets.create("product-docs")

# Object -> method: operate on the returned dataset directly.
dataset.add_texts([
    "VectorAmp stores and searches vectors at scale.",
    "SABLE is VectorAmp's high-performance vector index.",
])

results = dataset.search("How does VectorAmp search documents?", top_k=5)
answer = dataset.ask("What index does VectorAmp use?")

print(results["results"])
print(answer["answer"])

info

Dataset creation always requests SABLE. The SDK intentionally does not expose an index_type option.

Creating datasets

from vectoramp import openai

# Minimal: name only.
client.datasets.create("docs")

# Hybrid (dense + sparse) index.
client.datasets.create("docs", hybrid=True)

# OpenAI embedding (dim inferred: small -> 1536, large -> 3072).
client.datasets.create("docs", embedding=openai("small"))  # or openai("large")

# Custom / unknown model: pass dim explicitly.
client.datasets.create(
    "docs",
    embedding={"provider": "acme", "model": "acme-embed-v1"},
    dim=1024,
)

Built-in dimension inference: vectoramp/VectorAmp-Embedding-4B → 2560, openai/text-embedding-3-small → 1536, openai/text-embedding-3-large → 3072.

Object and service styles

Both access styles work everywhere the SDK allows it. Prefer the object → method style on a returned Dataset; the service style that passes dataset_id explicitly remains available.

# Object -> method style (preferred)
page = client.datasets.list(limit=50, offset=0)
dataset = page["datasets"][0]
dataset.search("release notes", top_k=10)
dataset.delete()

# Service style
dataset = client.datasets.get("dataset-uuid")
client.datasets.search("dataset-uuid", text="release notes", top_k=10)
client.datasets.delete("dataset-uuid")

List calls return pagination envelopes such as { "datasets": [...], "total": n, "limit": 50, "offset": 0 } with Dataset objects in the datasets field.

Insert, add text, and search

A record id may be a string or an integer; integer ids are sent as JSON numbers, not coerced to strings.

# Raw vectors
dataset.insert([
    {
        "id": "doc-001",
        "values": [0.1, 0.2, 0.3],
        "metadata": {"title": "Intro", "source": "manual"},
    },
    {
        "id": 42,  # preserved as a JSON number
        "values": [0.4, 0.5, 0.6],
        "metadata": {"title": "Appendix"},
    },
])

# Embed text with the dataset's model, then insert. Ids are auto-generated when
# omitted, and the source text is copied into metadata.text.
dataset.add_texts("VectorAmp uses SABLE for high-performance vector search.")
dataset.add_texts(
    ["VectorAmp uses SABLE for high-performance vector search."],
    ids=["sable-note"],
    metadatas=[{"source": "docs"}],
)

search accepts a bare string (text query) or a float vector; top_k defaults to 10.

# Text and vector search
dataset.search("wireless headphones", top_k=10, include_documents=True)
dataset.search([0.1, 0.2, 0.3], top_k=5)

# Hybrid + filtered search, with the optional model reranker.
dataset.search(
    text="wireless headphones",
    top_k=10,
    filters={"category": "electronics"},
    advanced_filters=[{"field": "price", "op": "lt", "value": 100}],
    hybrid=True,
    sparse_query="wireless headphones",
    alpha=0.7,
    rerank={"enabled": True},  # vectoramp / VectorAmp-Rerank-v1
)

tip

rerank={"enabled": True} (or rerank=True) enables the optional model reranker after initial retrieval; provider/model default to vectoramp / VectorAmp-Rerank-v1. The separate rerank_depth_override only tunes SABLE's vector-index candidate depth and is not the model reranker.

Organization secrets and vector deletion

Store external provider credentials as organization secrets, then reference the secret from datasets. The default OpenAI BYOM reference is emb:openai:api_key.

# One-step convenience: saves/updates the OpenAI org secret, then creates
# the dataset with embedding.secret_ref = "emb:openai:api_key".
dataset = client.datasets.create(
    "openai-docs",
    openai_api_key=os.environ["OPENAI_API_KEY"],
)

# Or manage the secret explicitly.
client.secrets.put_openai_api_key(os.environ["OPENAI_API_KEY"])

dataset = client.datasets.create_openai("openai-docs", model="small")

# Delete individual vector ids from search.
dataset.delete_vectors(["doc-001", 42], write_concern="quorum")
client.datasets.delete_vectors(dataset.id, ["doc-002"])

Ingestion

Start from an existing source id, or pass a typed source builder as a one-liner — the SDK creates the source, extracts its id, and starts the job.

from vectoramp import WebSource, ConfluenceSource

# Existing source id
dataset.ingest_source("source-uuid")

# One-liner with a web source builder
dataset.ingest_source(WebSource(start_urls=["https://docs.example.com/"], max_depth=1))

# The same one-liner works for any source type, e.g. Confluence
dataset.ingest_source(
    ConfluenceSource(
        base_url="https://acme.atlassian.net",
        username="user@example.com",
        api_token="…",
        spaces=["ENG"],
    )
)

Create reusable sources through client.sources (an alias of client.ingestion):

web = client.sources.create_web(start_urls=["https://docs.example.com/"], max_depth=1)
s3 = client.sources.create_s3(bucket="my-bucket", prefix="documents/")
gcs = client.sources.create_gcs(bucket="my-gcs-bucket", prefix="documents/")
jira = client.sources.create_jira(cloud_id="atlassian-cloud-id", project_keys=["ENG"])
confluence = client.sources.create_confluence(
    cloud_id="atlassian-cloud-id",  # or base_url="https://acme.atlassian.net"
    username="user@example.com",
    api_token="…",                  # auth_mode defaults to "basic"
    spaces=["ENG", "DOCS"],         # empty/omitted = all accessible spaces
    include_attachments=True,       # default False
)
gdrive = client.sources.create_google_drive(folder_ids=["drive-folder-id"])
upload = client.sources.create_file_upload()

sync_mode is omitted unless you set it, so the server applies its default of "incremental" for the connectors that support it. Pass sync_mode="full" to force a full re-sync.

Use GenericSource (or the low-level client.ingestion.create_source(...)) as an escape hatch for source types the SDK has not modeled yet:

from vectoramp import GenericSource

client.sources.create(
    GenericSource(name="custom-source", source_type="custom", config={"any_api_field": "value"})
)

Manage and clean up sources

Validate a config before creating, delete a source, or find and remove sources nothing is using.

# Validate without persisting.
client.sources.validate("web", {"start_urls": ["https://docs.example.com"]})

# See what's using a source (active schedules + in-flight jobs).
refs = client.sources.get_references("source-uuid")  # -> {schedules, schedule_count, active_job_count, in_use}

# Delete. Raises 409 if in use unless force=True (force also disables its schedules).
client.sources.delete("source-uuid")
client.sources.delete("source-uuid", force=True)

# Find unused sources, or remove all of them in one call.
unused = client.sources.list_unused()
result = client.sources.cleanup_unused()  # -> {"deleted": [...], "count": n}

Authenticated sources & connections

Two ways to authenticate a gdrive/gcs/confluence/jira source. Use a service account for headless servers/CI, or a reusable connection for an interactive one-time grant.

# Option 1 — service account (headless, no browser).
client.sources.create_google_drive(
    folder_ids=["drive-folder-id"],
    service_account_json={"type": "service_account", "client_email": "ingest@proj.iam.gserviceaccount.com", "private_key": "…"},
)

# Option 2 — connection (interactive once, reused server-side).
# connect() creates the grant, prints/opens authorization_url, and polls until connected.
conn = client.connections.connect("google", source_type="gdrive")
client.sources.create_google_drive(folder_ids=["drive-folder-id"], connection_id=conn["id"])

Manage connections directly when you want to drive the grant flow yourself:

conn = client.connections.create("google", source_type="gdrive")  # -> {id, status: "pending", authorization_url}
print("Open in a browser:", conn["authorization_url"])
client.connections.get(conn["id"])                                 # poll until status == "connected"
client.connections.list(provider="google")
client.connections.delete(conn["id"])

Minimal file upload ingestion

For local files you do not need to create a source first. When source_name is omitted, the SDK creates a file_upload source with a generated name, initializes presigned uploads, uploads bytes to the returned URLs, and completes the upload job.

job = dataset.ingest_files(["./docs/whitepaper.pdf", "./docs/overview.txt"])

# Optional only when you want a specific source name
job = dataset.ingest_files(["./docs/whitepaper.pdf"], source_name="product-docs-upload")

List jobs with pagination:

jobs = client.ingestion.list_jobs(dataset_id=dataset.id, limit=50, offset=0)
job = client.ingestion.get_job("job-uuid")
client.ingestion.retry_job("job-uuid")

Dataset documents

List retained source documents and download originals when download_available is true. Pagination is cursor-based; keep passing next_cursor until it is empty.

from pathlib import Path

cursor = None
while True:
    page = client.datasets.list_documents(
        "dataset-uuid", limit=50, cursor=cursor, status="ready"
    )
    for document in page.get("documents", []):
        if document.get("download_available"):
            content = client.datasets.download_document("dataset-uuid", document["id"])
            Path(document.get("file_name") or document["id"]).write_bytes(content)
    cursor = page.get("next_cursor")
    if not cursor:
        break

# Object style
dataset.list_documents(limit=50)
dataset.download_document("document-id")

The SDK follows download redirects and returns raw bytes. Storage buckets, object keys, and payload references are never exposed.

Intelligence

ask defaults top_k=5, include_sources=True, and dataset "all" when unscoped.

answer = dataset.ask("What are the key product features?", top_k=5)
print(answer["answer"])

for event in client.ask_stream("Summarize the docs", dataset_id=dataset.id):
    if event["chunk_type"] == "text":
        print(event["content"], end="")

Multi-turn conversations

The Intelligence API is stateless. To ask a follow-up, send the prior turns as conversation history; you decide how many previous messages to include.

history = [
    {"role": "user", "content": "What are the key features?"},
    {"role": "assistant", "content": "Hybrid search, reranking, and managed ingestion."},
]
answer = client.ask("Which of those help with relevance?", conversation_history=history)
print(answer["answer"])

Persistent Intelligence sessions

Persist multi-turn conversations server-side instead of resending history.

session = client.intelligence.create_session(
    title="Q4 planning", dataset_id=dataset.id, metadata={"team": "product"}
)
client.intelligence.append_message(
    session["id"], role="user", content="Summarize the latest planning docs"
)
messages = client.intelligence.list_messages(session["id"], limit=100)
sessions = client.intelligence.list_sessions(limit=50)

Intelligence answers return sources[] and chunks[]. Inline [1] citations refer to sources[0]; preview_ref is an opaque preview token, never a raw storage key.

Method reference

Both access styles work everywhere the SDK allows it: client.datasets.search(id, …) and dataset.search(…). Required arguments are listed first; optional arguments show their default.

Client (VectorAmp)

• VectorAmp(api_key=None, *, base_url="https://api.vectoramp.com", timeout=30.0) — api_key falls back to VECTORAMP_API_KEY.
• client.ask(query, *, dataset_id=None, top_k=5, conversation_history=None, include_sources=True) → answer dict.
• client.ask_stream(query, *, dataset_id=None, top_k=5, conversation_history=None, include_sources=True) → iterator of SSE chunks.
• client.close() (also a context manager).

Datasets (client.datasets / Dataset)

Method	Required	Optional (default)	Returns
create(name, …)	name	dim (inferred), metric ("cosine"), embedding, embedding_provider ("vectoramp"), embedding_model ("VectorAmp-Embedding-4B"), hybrid (False), filters, metadata_schema, tuning	Dataset
list(…)	—	limit (50), offset (0)	page with Dataset objects
get(dataset_id)	dataset_id	—	Dataset
delete(dataset_id) / dataset.delete()	dataset_id	—	JSON
stats(dataset_id) / dataset.stats()	dataset_id	—	JSON
search(dataset_id, query=None, …) / dataset.search(query, …)	dataset_id	query (str text or float vector), vector, text, top_k (10), filters, advanced_filters, hybrid, sparse_query, alpha, rerank, include_documents, include_metadata, include_embeddings, embedding_provider, embedding_model, nprobe_override, rerank_depth_override	search response
insert(dataset_id, vectors) (+ insert_vectors) / dataset.insert(vectors)	dataset_id, vectors	— (record id may be str or int)	JSON
embed(dataset_id, …) / dataset.embed(…)	dataset_id	text or texts (one required)	JSON
add_texts(dataset_id, texts, …) / dataset.add_texts(texts, …)	dataset_id, texts (str or list)	ids (auto-generated), metadatas	JSON
list_documents(dataset_id, …) / dataset.list_documents(…)	dataset_id	limit (50), cursor, status	page
download_document(dataset_id, document_id) / dataset.download_document(document_id)	dataset_id, document_id	—	bytes
ensure_engine(dataset_id)	dataset_id	—	JSON
dataset.ask(query, …)	query	top_k (5), conversation_history, include_sources (True)	answer dict
dataset.ingest_source(source, …)	source (id or builder)	pipeline_id	job
dataset.ingest_files(paths, …)	paths	source_name, description	job

Embedding helper: openai("small"|"large"). Builder classes: WebSource, S3Source, GCSSource, GoogleDriveSource (source_type="gdrive"), JiraSource, ConfluenceSource, FileUploadSource, GenericSource.

Sources / ingestion (client.sources is an alias of client.ingestion)

Method	Required	Optional (default)
create(source) / create_source(…)	a builder, or source_type + config	name, description, metadata
create_web(start_urls=…)	start_urls	name, max_depth, max_pages, allowed_domains, include_assets, max_assets_per_page, selectors, headers
create_s3(bucket=…)	bucket	name, region ("us-east-1"), prefix, access_key_id, secret_access_key, role_arn, file_patterns, max_file_size_mb, sync_mode
create_gcs(bucket=…)	bucket	name, prefix, project_id, credentials_json, file_patterns, sync_mode
create_jira(cloud_id=…)	cloud_id	name, access_token, project_keys, jql, include_comments (True), sync_mode
create_confluence(cloud_id=… or base_url=…)	cloud_id or base_url	name, auth_mode ("basic"), username, api_token, oauth_credentials, spaces, include_attachments (False), sync_mode
create_google_drive(folder_ids=… or file_ids=…)	folder_ids or file_ids	name, auth_mode, oauth_credentials, service_account_json, connection_id, include_shared_drives, sync_mode
create_file_upload(…)	—	name, storage_provider ("s3"), sync_mode ("full")
list_sources(…) / get_source(source_id)	— / source_id	limit (50), offset (0)
validate(source_type, config)	source_type, config	—
get_references(source_id)	source_id	— (→ schedules, schedule_count, active_job_count, in_use)
delete(source_id)	source_id	force (False)
list_unused()	—	—
cleanup_unused()	—	— (→ deleted, count)
start_job(source_id=…, dataset_id=…)	source_id, dataset_id	pipeline_id
list_jobs(…) / get_job(job_id) / retry_job(job_id)	— / job_id / job_id	dataset_id, limit (50), offset (0)
ingest_files(dataset_id=…, paths=…)	dataset_id, paths	source_name, description

Schedules (client.schedules)

• list(*, limit=50, offset=0), get(schedule_id).
• create(*, source_id, dataset_id, cron, timezone=None, pipeline_id=None, enabled=None, name=None, metadata=None).
• update(schedule_id, …) — only passed fields change.
• delete(schedule_id), trigger(schedule_id).

Connections (client.connections)

• list(*, provider=None) — provider filters to "google" or "atlassian".
• create(provider, *, source_type=None) → {id, provider, status: "pending", authorization_url}.
• get(connection_id) → {id, provider, account_email, status}; poll until status == "connected".
• delete(connection_id).
• connect(provider, *, source_type=None) — helper that creates the grant, opens/prints authorization_url, polls until connected, and returns the connection.

Intelligence (client.intelligence)

• query(query, *, dataset_id=None, top_k=5, conversation_history=None, include_sources=True).
• stream(query, *, …) — iterator of SSE chunks.
• create_session(*, title=None, workspace_id=None, dataset_id=None, metadata=None).
• list_sessions(*, limit=50), get_session(session_id), delete_session(session_id).
• append_message(session_id, *, role, content, metadata=None), list_messages(session_id, *, limit=100).