vectorless-sdk

Official Python SDK for Vectorless — structure-preserving document retrieval.

Install

pip install vectorless-sdk

Quick Start

from vectorless import VectorlessClient

# Deployed instance with API key
client = VectorlessClient(
    base_url="https://api.vectorless.dev",
    api_key="vl_live_...",
)

# Self-hosted, no auth needed
# client = VectorlessClient(base_url="http://localhost:8080")

# 1. Ingest a document
result = client.ingest_document(
    "./research-paper.pdf",
    filename="research-paper.pdf",
    metadata={"department": "engineering"},
)

# 2. Wait for processing (parsing → summarizing → ready)
doc = client.wait_for_ready(
    result.document_id,
    on_progress=lambda s: print(f"Status: {s}"),
)

# 3. Explore the document tree
tree = client.get_document_tree(doc.id)
for section in tree.sections:
    print("  " * section.depth + f"{section.title} ({section.tokens} tokens)")

# 4. Query — LLM navigates the tree to find relevant sections
response = client.query(doc.id, "What methodology was used?")
for section in response.sections:
    print(f"\n## {section.title}\n{section.content}")
print(f"Strategy: {response.strategy} | {response.elapsed_ms}ms")

Async Support

Every method has an async counterpart:

from vectorless import AsyncVectorlessClient

async with AsyncVectorlessClient(
    base_url="https://api.vectorless.dev",
    api_key="vl_live_...",
) as client:
    result = await client.ingest_document(b"# Hello\n\nWorld", filename="hello.md")
    doc = await client.wait_for_ready(result.document_id)

    # Concurrent section fetch
    sections = await client.get_sections(["sec_1", "sec_2", "sec_3"])

Transport Protocols

Choose the wire protocol at init time:

# HTTP/REST — default, uses httpx
client = VectorlessClient(transport="http")

# ConnectRPC — protobuf JSON encoding, native streaming
client = VectorlessClient(transport="connect")

┌────────────────────────────────────────────────┐
│   VectorlessClient / AsyncVectorlessClient     │
├────────────────────────────────────────────────┤
│            Transport Abstraction                │
│  ┌──────────────────┐ ┌─────────────────────┐ │
│  │  HttpTransport    │ │  ConnectTransport   │ │
│  │  AsyncHttpTransport│ │AsyncConnectTransport│ │
│  │  REST/JSON        │ │  ConnectRPC JSON    │ │
│  │  SSE streaming    │ │  Native streaming   │ │
│  │  httpx            │ │  httpx              │ │
│  └────────┬─────────┘ └──────────┬──────────┘ │
│           │                       │            │
│     vectorless-server       vectorless-server   │
└────────────────────────────────────────────────┘

Streaming Queries

Watch retrieval progress in real-time:

# Sync
for event in client.query_stream(doc_id, "Explain the results"):
    if event.type == "section_selected" and event.section:
        print(f"Found: {event.section.title}")
    if event.type == "completed":
        print(f"Done in {event.elapsed_ms}ms")

# Async
async for event in client.query_stream(doc_id, "Explain the results"):
    ...

API Reference

Client Configuration

Methods

Both VectorlessClient (sync) and AsyncVectorlessClient (async) expose:

Error Handling

from vectorless import (
    VectorlessError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
)

try:
    doc = client.get_document("doc_123")
except NotFoundError:
    print("Document not found")
except AuthenticationError:
    print("Check your API key")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except VectorlessError as e:
    print(f"Error {e.status}: {e.message}")

Context Manager

# Sync
with VectorlessClient(api_key="vl_...") as client:
    doc = client.get_document("doc_123")

# Async
async with AsyncVectorlessClient(api_key="vl_...") as client:
    doc = await client.get_document("doc_123")

Environment Variables

Requirements

• Python 3.9+
• Dependencies: httpx, pydantic

License

MIT