MultiBase

SDK for working with your MultiBase instance in Python.

Proprietary software. All rights reserved. No license is granted for use, modification, or distribution.

pip install multibase-sdk
# or
uv add multibase-sdk

Setup

1. Create a WorkOS organization API key (WorkOS Dashboard → Organization → API Keys).
2. Set environment variables:

export MULTIBASE_API_KEY="your-workos-api-key"
# export MULTIBASE_WEBAPP_URL="https://app.eventual.ai" (optional, this is the default)

Usage

from multibase import Client

client = Client()
for project in client.list_projects():
    print(project.name, project.id)
    for collection in client.list_collections(project.name):
        print(" ", collection.name, f"({collection.item_count} items)")
for span in collection:
    print("   ", span.producer, span.start_ts, f"{span.duration_ms}ms")

Fetching sample data

Each collection span covers a time window on a producer. To load the underlying sample rows (the same data the query engine serves), call fetch_samples() on a bound span:

from multibase import Client

client = Client()
collection = client.get_collection("my-project", "Pedestrians")

# Discover available columns (project-scoped Arrow schema)
schema = collection.schema()  # or client.get_schema("my-project") / dataset.schema()
print(schema.names)

for span in collection:
    table = span.fetch_samples()
    print(table.num_rows, table.column_names)
    # Defaults include id, producer, ts, and every camera's
    # frames_<cam>_mp4_path / frames_<cam>_mp4_offset_sec columns.
    # Pass columns=[...] to project additional fields from the schema.
    # Camera stubs like "front_cam" are expanded to footage path/offset
    # columns automatically.

Returns a pyarrow.Table. Convert to pandas with table.to_pandas() if needed.

Concurrent fetches

Use fetch_samples_aio() to load sample data for many spans in parallel with the same Client:

import asyncio

from multibase import Client

async def load_all():
    client = Client()
    collection = client.get_collection("my-project", "Pedestrians")
    spans = list(collection)
    try:
        return await asyncio.gather(*(span.fetch_samples_aio() for span in spans))
    finally:
        await client.aclose()

tables = asyncio.run(load_all())

Datasets

Datasets are immutable snapshots of episodes, created in the webapp. Use them when you want a fixed training set that won't change if the source collection is edited:

from multibase import Client

client = Client()
dataset = client.get_dataset("my-project", "train-v1")
print(dataset.name, dataset.episode_count, dataset.sample_rate_hz)

for episode in dataset:
    table = episode.fetch_samples()
    print(table.num_rows, table.column_names)

PyTorch DataLoader

With the optional torch extra installed (pip install multibase-sdk[torch]), convert a dataset to a PyTorch DataLoader:

loader = dataset.to_torch_dataloader(batch_size=4, num_frames=8, seed=42)
for batch in loader:
    print(batch["front_cam"].shape)  # [batch, T, C, H, W]

To learn more about MultiBase, visit our website.