VectorBridge Python SDK

A modern Python SDK for the VectorBridge.ai API with first‑class support for both synchronous and asynchronous usage. Access user/admin operations, AI knowledge management, vector queries, history tracking, and more.

Installation

pip install vector-bridge

Choose Sync or Async

• Sync client: VectorBridgeClient
• Async client: AsyncVectorBridgeClient
• Factory: create_client(async_client: bool = False, **kwargs)

from vector_bridge import VectorBridgeClient, AsyncVectorBridgeClient, create_client

Quick Start (Sync)

from vector_bridge import VectorBridgeClient

# API key auth (application access)
client = VectorBridgeClient(integration_name="default", api_key="your_api_key")
print(client.ping())  # "OK"

# Or JWT token auth
client = VectorBridgeClient(integration_name="default", access_token="your_jwt_token")
print(client.ping())

Quick Start (Async)

import asyncio
from vector_bridge import AsyncVectorBridgeClient

async def main():
    async with AsyncVectorBridgeClient(integration_name="default", api_key="your_api_key") as client:
        print(await client.ping())  # "OK"

        # Or use JWT token
        # async with AsyncVectorBridgeClient(integration_name="default", access_token="your_jwt_token") as client:
        #     print(await client.ping())

asyncio.run(main())

Using the Factory

from vector_bridge import create_client

# Sync
client = create_client(integration_name="default", api_key="your_api_key")

# Async
async_client = create_client(integration_name="default", api_key="your_api_key", async_client=True)

History Tracking

Track changes to items, users, and integrations with comprehensive history endpoints:

from vector_bridge.schema.helpers.enums import ItemType, HistoryChangeType

# Sync - Get history for a specific item
history = client.history.get_item_history(
    item_id="item123",
    item_type=ItemType.FILE,
    limit=50
)

# Get all history for current integration
history = client.history.get_integration_history(
    limit=50,
    item_type=ItemType.FOLDER,
    change_type=HistoryChangeType.CREATED
)

# Get history by user
history = client.history.get_user_history(
    user_id="user123",
    limit=25
)

# Get my activity
my_activity = client.history.get_my_activity(limit=25)

# Delete history entries
client.history.delete_history_entry(
    history_id="hist123",
    item_id="item123",
    item_type=ItemType.FILE
)

# Async equivalents
# await client.history.get_item_history(...)
# await client.history.get_integration_history(...)
# await client.history.get_user_history(...)
# await client.history.get_my_activity(...)
# await client.history.delete_history_entry(...)

OTP Authentication

Generate and validate one-time passwords for authentication:

# Sync - Generate OTP for login
response = client.otp.generate_otp(email="user@example.com")

# Validate OTP and get token
token = client.otp.validate_otp(email="user@example.com", code="123456")

# Generate sign-up code
response = client.otp.generate_sign_up_code(email="newuser@example.com")

# Validate sign-up code
client.otp.validate_sign_up_code(email="newuser@example.com", code="123456")

# Reset password
client.otp.reset_password(
    email="user@example.com",
    code="123456",
    password="new_secure_password"
)

# Async equivalents
# await client.otp.generate_otp(...)
# await client.otp.validate_otp(...)
# await client.otp.generate_sign_up_code(...)
# await client.otp.validate_sign_up_code(...)
# await client.otp.reset_password(...)

Vector Queries

from weaviate.collections.classes.filters import Filter

# Sync search
search = client.queries.run_search_query(
    vector_schema="Documents",
    target_vector="default",
    near_text="attention mechanism",
    limit=5,
)

# Sync find similar
# from uuid import UUID
# similar = client.queries.run_find_similar_query(
#     vector_schema="Documents",
#     target_vector="default",
#     near_id=UUID("8c03ff2f-36f9-45f7-9918-48766c968f45"),
#     limit=5,
# )

# Async equivalents
# await client.queries.run_search_query(...)
# await client.queries.run_find_similar_query(...)

Client Highlights (Sync/Async)

Most day‑to‑day endpoints are under client.<module> (not client.admin). Available client modules:

User Modules

• client.users - User management (get me, list users, update profile, change password)
• client.otp - One-time password authentication (generate, validate, sign-up, password reset)
• client.history - Track changes to items, users, and integrations
• client.ai_knowledge - AI knowledge management (file storage, vector DB)
• client.queries - Vector search and similarity queries
• client.logs - Application logs
• client.api_keys - API key management
• client.usage - Usage statistics and metrics

Admin Modules

• client.admin.organization - Organization management
• client.admin.security_groups - Security group management
• client.admin.integrations - Integration configuration
• client.admin.database.state - VectorDB schema state (get schema, check readiness)
• client.admin.settings - System settings and DB connection checks

Example (sync):

me = client.users.get_me()
logs = client.logs.list_logs(integration_name=client.integration_name, limit=25)
history = client.history.get_my_activity(limit=10)
users = client.users.get_users_in_my_organization(limit=50)

Example (async):

# me = await client.users.get_me()
# logs = await client.logs.list_logs(integration_name=client.integration_name, limit=25)
# history = await client.history.get_my_activity(limit=10)
# users = await client.users.get_users_in_my_organization(limit=50)

AI Knowledge: File Storage (Sync and Async)

The AI Knowledge file storage API has been updated to align with the latest VectorBridge API (v2.2.1). All endpoints now use RESTful resource paths.

Sync file workflow

from vector_bridge import VectorBridgeClient
from vector_bridge.schema.ai_knowledge.filesystem import AIKnowledgeFileSystemItemCreate, AIKnowledgeFileSystemItemUpdate
from vector_bridge.schema.helpers.enums import FileAccessType

client = VectorBridgeClient(integration_name="default", api_key="your_api_key")

# Create folder
folder_data = AIKnowledgeFileSystemItemCreate(
    name="Project Documents",
    private=True,
    created_by="user-123",
)
folder = client.ai_knowledge.file_storage.create_folder(
    file_data=folder_data,
    folder_description="Docs for Q4",
)

# Upload a file with progress tracking
file_data = AIKnowledgeFileSystemItemCreate(
    name="spec.pdf",
    parent_id=folder.uuid,
    tags=["spec", "q4"],
    created_by="user-123",
)
upload = client.ai_knowledge.file_storage.upload_file(
    file_data=file_data,
    file_path="./docs/spec.pdf",
)

for p in upload.progress_updates:
    print("progress:", p)
item = upload.item

# Update metadata
client.ai_knowledge.file_storage.update_file_or_folder(
    item_id=item.uuid,
    updated_properties=AIKnowledgeFileSystemItemUpdate(starred=True, tags=["spec", "approved"]) 
)

# Share read-only with a user
client.ai_knowledge.file_storage.grant_or_revoke_user_access(
    item_id=item.uuid, user_id="user-123", has_access=True, access_type=FileAccessType.READ
)

# Share with a security group
client.ai_knowledge.file_storage.grant_or_revoke_security_group_access(
    item_id=item.uuid, group_id="group-456", has_access=True, access_type=FileAccessType.WRITE
)

# List items under folder
from weaviate.collections.classes.filters import _Filters
items = client.ai_knowledge.file_storage.execute_files_and_folders_list_query(
    filters=_Filters.by_property("parent_id").equal(folder.uuid)
)

# Create file reference (link between files)
client.ai_knowledge.file_storage.create_files_reference(
    from_uuid=item.uuid,
    to_uuid="other-file-uuid",
    metadata={"relation": "attachment"}
)

# Delete file reference
client.ai_knowledge.file_storage.delete_files_reference(
    from_uuid=item.uuid,
    reference_id="reference-id-123"
)

# Get item path
path = client.ai_knowledge.file_storage.get_file_or_folder_path(item_id=item.uuid)

# Delete item
client.ai_knowledge.file_storage.delete_folder_or_file(item_id=item.uuid)

Async file workflow

import asyncio
from vector_bridge import AsyncVectorBridgeClient
from vector_bridge.schema.ai_knowledge.filesystem import AIKnowledgeFileSystemItemCreate, AIKnowledgeFileSystemItemUpdate
from vector_bridge.schema.helpers.enums import FileAccessType

async def main():
    async with AsyncVectorBridgeClient(integration_name="default", api_key="your_api_key") as client:
        folder_data = AIKnowledgeFileSystemItemCreate(
            name="Research",
            private=False,
            created_by="user-123",
        )
        folder = await client.ai_knowledge.file_storage.create_folder(
            file_data=folder_data,
            folder_description="ML papers",
        )

        file_data = AIKnowledgeFileSystemItemCreate(
            name="attention.pdf",
            parent_id=folder.uuid,
            tags=["nlp", "transformers"],
            created_by="user-123",
        )
        upload = await client.ai_knowledge.file_storage.upload_file(
            file_data=file_data,
            file_path="./papers/attention.pdf",
        )

        async for p in upload.progress_updates:
            print("progress:", p)
        item = await upload.item

        await client.ai_knowledge.file_storage.update_file_or_folder(
            item_id=item.uuid,
            updated_properties=AIKnowledgeFileSystemItemUpdate(starred=True)
        )

        await client.ai_knowledge.file_storage.grant_or_revoke_user_access(
            item_id=item.uuid, user_id="user-123", has_access=True, access_type=FileAccessType.READ
        )

        await client.ai_knowledge.file_storage.grant_or_revoke_security_group_access(
            item_id=item.uuid, group_id="group-456", has_access=True, access_type=FileAccessType.WRITE
        )

        from weaviate.collections.classes.filters import _Filters
        items = await client.ai_knowledge.file_storage.execute_files_and_folders_list_query(
            filters=_Filters.by_property("parent_id").equal(folder.uuid)
        )

        await client.ai_knowledge.file_storage.create_files_reference(
            from_uuid=item.uuid,
            to_uuid="other-file-uuid",
            metadata={"relation": "citation"}
        )

        path = await client.ai_knowledge.file_storage.get_file_or_folder_path(item_id=item.uuid)

asyncio.run(main())

System Settings (Admin)

The admin settings module provides system configuration and database connection testing:

# Get system settings
settings = client.admin.settings.get_settings()
print(settings.distribution_type)  # DistributionType.SELF_HOSTED
print(settings.files.max_size_bytes)  # File upload limits
print(settings.files.types)  # Supported file types

# Check VectorDB connection
client.admin.settings.check_db_connection(
    url="https://your-weaviate-instance.com",
    api_key="your-weaviate-api-key"
)

# Async
# settings = await client.admin.settings.get_settings()
# await client.admin.settings.check_db_connection(url="...", api_key="...")

Error Handling

Errors are raised as domain‑specific exceptions. For generic HTTP errors you may see HTTPException with status_code and detail.

from vector_bridge import VectorBridgeClient
from vector_bridge.schema.errors.base import BaseError

client = VectorBridgeClient(integration_name="default", access_token="your_jwt_token")
try:
    client.ping()
except BaseError as e:
    print(f"Error: {e}")

License

MIT License. See LICENSE for details.