Python SDK

pip install mixpeek

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 sk_xxxxxxxxxxxxx",
            x_namespace="ns_xxxxxxxxxxxxx"
        )
        print("Collections:", collections)
    except ApiException as e:
        print(f"Exception: {e}")

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

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 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 sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

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

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

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 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 sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )
    print(f"Found {len(results.results)} results")

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 sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

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 sk_xxxxxxxxxxxxx",
        x_namespace="ns_xxxxxxxxxxxxx"
    )

from mixpeek.rest import ApiException

try:
    collection = collections_api.get_collection(
        collection_identifier="non-existent",
        authorization="Bearer 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")

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

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