Python SDK - Mixpeek

The Mixpeek Python SDK is auto-generated from our OpenAPI specification and always stays in sync with the latest API features.

Installation

Install via pip:

pip install mixpeek

Requirements: Python 3.9+

Quick Start

import mixpeek
from mixpeek.rest import ApiException

# Configure the client
configuration = mixpeek.Configuration(
    host="https://api.mixpeek.com"
)

# Create an API client
with mixpeek.ApiClient(configuration) as api_client:
    # Create an instance of the Collections API
    collections_api = mixpeek.CollectionsApi(api_client)

    try:
        # List collections
        collections = collections_api.list_collections(
            authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
            x_namespace="ns_xxxxxxxxxxxxx"
        )
        print("Collections:", collections)
    except ApiException as e:
        print(f"Exception: {e}")

Authentication

All API requests require authentication using a Bearer token:

authorization = "Bearer mxp_sk_xxxxxxxxxxxxx"  # Your API key
x_namespace = "ns_xxxxxxxxxxxxx"  # Your namespace ID or custom name

You can create API keys in the Mixpeek dashboard under Organization Settings.

Core Operations

Collections

import mixpeek
from mixpeek.rest import ApiException

configuration = mixpeek.Configuration(host="https://api.mixpeek.com")

with mixpeek.ApiClient(configuration) as api_client:
    collections_api = mixpeek.CollectionsApi(api_client)

    # Create a collection
    create_request = mixpeek.CreateCollectionRequest(
        alias="my-collection",
        description="Store multimodal documents"
    )

    collection = collections_api.create_collection(
        create_collection_request=create_request,
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )
    print(f"Created collection: {collection.collection_id}")

    # Get a collection
    retrieved = collections_api.get_collection(
        collection_identifier="my-collection",
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

    # List all collections
    all_collections = collections_api.list_collections(
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

    # Delete a collection
    collections_api.delete_collection(
        collection_identifier="my-collection",
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

Retrievers

import mixpeek

with mixpeek.ApiClient(configuration) as api_client:
    retrievers_api = mixpeek.RetrieversApi(api_client)

    # Create a retriever
    create_request = mixpeek.CreateRetrieverRequest(
        retriever_name="semantic-search",
        description="Search across all documents",
        collection_identifiers=["my-collection"],
        stages=[
            {
                "type": "embed",
                "model": "openai-text-embedding-3-small"
            },
            {
                "type": "vector_search",
                "top_k": 10
            }
        ]
    )

    retriever = retrievers_api.create_retriever(
        create_retriever_request=create_request,
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

    # Execute a retriever
    execute_request = mixpeek.ExecuteRetrieverRequest(
        query="find relevant documents about AI"
    )

    results = retrievers_api.execute_retriever(
        retriever_id=retriever.retriever_id,
        execute_retriever_request=execute_request,
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )
    print(f"Found {len(results.results)} results")

Adhoc Retrievers

Execute retrievers on-the-fly without creating them first:

import mixpeek

with mixpeek.ApiClient(configuration) as api_client:
    adhoc_api = mixpeek.AdhocRetrieversApi(api_client)

    # Execute adhoc retriever
    adhoc_request = mixpeek.AdhocExecuteRequest(
        query="search query",
        collection_identifiers=["my-collection"],
        stages=[
            {
                "type": "embed",
                "model": "openai-text-embedding-3-small"
            },
            {
                "type": "vector_search",
                "top_k": 10
            }
        ]
    )

    results = adhoc_api.execute_adhoc_retriever(
        adhoc_execute_request=adhoc_request,
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

Documents

import mixpeek

with mixpeek.ApiClient(configuration) as api_client:
    documents_api = mixpeek.DocumentsApi(api_client)

    # Upload documents
    upload_request = mixpeek.UploadDocumentsRequest(
        collection_id="col_abc123",
        documents=[
            {
                "url": "s3://bucket/video.mp4",
                "metadata": {"title": "Demo Video"}
            },
            {
                "url": "s3://bucket/image.jpg",
                "metadata": {"title": "Demo Image"}
            }
        ]
    )

    documents = documents_api.upload_documents(
        upload_documents_request=upload_request,
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

Standalone Namespaces (Bring Your Own Vectors)

Standalone namespaces let you store and query precomputed vectors directly — no buckets, collections, or extractors required. This is the fastest path for agents that already have embeddings. The high-level Mixpeek client exposes the MVS primitives ergonomically; every call sends your namespace as the X-Namespace header automatically.

from mixpeek import Mixpeek

mp = Mixpeek(api_key="mxp_sk_xxxxxxxxxxxxx", namespace="my-vectors")

# 1. Create a standalone namespace with one or more vector configs.
mp.namespaces.create(
    namespace_id="my-vectors",
    mode="standalone",
    vector_configs=[{"name": "text_8", "dimension": 8, "metric": "cosine"}],
)

# 2. Upsert documents with your own vectors (direct write, no pipeline).
mp.namespaces.documents.upsert(
    namespace_id="my-vectors",
    documents=[
        {
            "document_id": "doc-1",
            "vectors": {"text_8": [0.11, 0.32, -0.38, -0.41, 0.006, 0.34, -0.40, -0.52]},
            "payload": {"body": "hello world", "category": "docs"},
        }
    ],
    # Optional: request a write token for read-your-writes consistency.
    options={"write_token": True},
)

# 3. Query is unified on retrievers. Create a retriever with a feature_search
#    stage and execute it. The X-Namespace header is sent for you.
retriever = mp.retrievers.create(
    retriever_name="byov-search",
    stages=[
        {
            "stage_name": "feature_search",
            "stage_type": "filter",
            "config": {
                "stage_id": "feature_search",
                "parameters": {
                    "final_top_k": 5,
                    "searches": [
                        {
                            "feature_uri": "text_8",
                            "query": {"input_mode": "vector", "value": "{{INPUT.qv}}"},
                            "top_k": 5,
                        }
                    ],
                },
            },
        }
    ],
    input_schema={"qv": {"type": "array", "required": True}},
)

results = mp.retrievers.execute(
    retriever["retriever_id"],
    inputs={"qv": [0.11, 0.32, -0.38, -0.41, 0.006, 0.34, -0.40, -0.52]},
)

The removed POST /v1/search endpoint has been replaced by this retriever create + execute flow. Mixpeek.search(namespace_id=..., queries=[...]) is a shortcut that builds and runs an ephemeral feature_search retriever for you.

Read-your-writes after a direct upsert

A direct upsert is durable immediately, but retriever reads are eventually consistent by default. To read your own just-written document, pass the write_token from the upsert response back as the X-Write-Token header on execute — it routes the read to the primary shard and bypasses caches:

res = mp.namespaces.documents.upsert(
    namespace_id="my-vectors",
    documents=[{"document_id": "doc-2", "vectors": {"text_8": [...]}, "payload": {}}],
    options={"write_token": True},
)
# res["write_token"] -> send as X-Write-Token on the next retriever execute.

Promote to managed (auto-embedding)

Promote a standalone namespace to managed mode to map a vector index to an inference service. After promotion the same feature_search stage accepts input_mode: "text" and auto-embeds queries — no client-side embedding needed.

import urllib3, json

http = urllib3.PoolManager()
http.request(
    "POST",
    "https://api.mixpeek.com/v1/namespaces/my-vectors/promote",
    headers={"Authorization": "Bearer mxp_sk_xxxxxxxxxxxxx", "Content-Type": "application/json"},
    body=json.dumps({
        "vector_mappings": [
            {"existing_index": "text_8", "inference_service": "intfloat/multilingual-e5-large-instruct"}
        ]
    }).encode(),
)

Before promotion, a standalone vector index has no inference mapping, so input_mode: "text" / "content" queries return an actionable 400 telling you to promote the namespace or use input_mode=vector with a precomputed embedding.

Error Handling

The SDK provides comprehensive error handling through ApiException:

from mixpeek.rest import ApiException

try:
    collection = collections_api.get_collection(
        collection_identifier="non-existent",
        authorization="Bearer mxp_sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )
except ApiException as e:
    print(f"Status code: {e.status}")
    print(f"Reason: {e.reason}")
    print(f"Body: {e.body}")

    # Handle specific error codes
    if e.status == 404:
        print("Collection not found")
    elif e.status == 401:
        print("Authentication failed - check your API key")
    elif e.status == 403:
        print("Access forbidden - check your namespace")

Available APIs

The SDK includes these API classes:

AdhocRetrieversApi - Execute retrievers without saving them
AgentSessionsApi - Manage AI agent sessions
AlertsApi - Configure and manage alerts
AnalyticsApi - Access usage and performance analytics
BucketsApi - Manage object storage buckets
CollectionsApi - Manage document collections
DocumentsApi - Upload and manage documents
NamespacesApi - Manage multi-tenant namespaces
OrganizationsApi - Organization management
PluginsApi - Configure and manage plugins
RetrieversApi - Create and execute retrievers
TaxonomiesApi - Manage classification taxonomies
WebhooksApi - Configure webhook integrations

Configuration Options

Custom Host

configuration = mixpeek.Configuration(
    host="https://custom.api.endpoint.com"
)

Timeouts

configuration = mixpeek.Configuration(
    host="https://api.mixpeek.com"
)
configuration.timeout = 60  # Set timeout to 60 seconds

Resources

PyPI Package

View on PyPI

GitHub Repository

Source code and issues

API Reference

Complete API documentation

OpenAPI Spec

OpenAPI specification

Next Steps

Quickstart

Get started with Mixpeek

JavaScript SDK

Use the JavaScript/TypeScript SDK

MCP Server

Use Mixpeek with AI assistants

Examples

View example implementations

​Installation

​Quick Start

​Authentication

​Core Operations

​Collections

​Retrievers

​Adhoc Retrievers

​Documents

​Standalone Namespaces (Bring Your Own Vectors)

​Read-your-writes after a direct upsert

​Promote to managed (auto-embedding)

​Error Handling

​Available APIs

​Configuration Options

​Custom Host

​Timeouts

​Resources

PyPI Package

GitHub Repository

API Reference

OpenAPI Spec

​Next Steps

Quickstart

JavaScript SDK

MCP Server

Examples

Installation

Quick Start

Authentication

Core Operations

Collections

Retrievers

Adhoc Retrievers

Documents

Standalone Namespaces (Bring Your Own Vectors)

Read-your-writes after a direct upsert

Promote to managed (auto-embedding)

Error Handling

Available APIs

Configuration Options

Custom Host

Timeouts

Resources

Next Steps