Mixpeek CLI

pip install mixpeek
mixpeek --version

# 1. Create plugin
mixpeek plugin init my_extractor --category text

# 2. Edit processors/core.py with your logic

# 3. Test locally
cd my_extractor && mixpeek plugin test

# 4. Publish
mixpeek plugin publish --namespace ns_xxx

export MIXPEEK_API_KEY="mxp_sk_your_api_key"
export MIXPEEK_NAMESPACE="ns_your_namespace"

mixpeek plugin init <name> [options]

# Examples
mixpeek plugin init sentiment_analyzer --category text
mixpeek plugin init face_detector --category image --description "Detect faces"

mixpeek plugin test [options]

# Examples
mixpeek plugin test
mixpeek plugin test --path ./my_extractor --verbose
mixpeek plugin test --sample-data samples.json

mixpeek plugin publish [options]

# Examples
mixpeek plugin publish
mixpeek plugin publish --namespace ns_abc123 --dry-run

mixpeek plugin list [options]

mixpeek plugin list --source custom

my_extractor/
├── manifest.py      # Metadata + schemas
├── pipeline.py      # Batch processing
├── realtime.py      # HTTP endpoint (optional, Enterprise)
└── processors/
    └── core.py      # Your logic

from pydantic import BaseModel, Field
from typing import List

class MyInput(BaseModel):
    text: str

class MyOutput(BaseModel):
    embedding: List[float]

class MyParams(BaseModel):
    threshold: float = Field(default=0.5)

metadata = {
    "feature_extractor_name": "my_extractor",
    "version": "1.0.0",
    "description": "My extractor",
    "category": "text",
}

input_schema = MyInput
output_schema = MyOutput
parameter_schema = MyParams
supported_input_types = ["text"]

features = [
    {
        "feature_name": "my_embedding",
        "feature_type": "embedding",
        "embedding_dim": 384,
        "distance_metric": "cosine",
    },
]

from dataclasses import dataclass
import pandas as pd

@dataclass
class MyConfig:
    threshold: float = 0.5

class MyProcessor:
    def __init__(self, config: MyConfig, progress_actor=None):
        self.config = config
        self._model = None

    def _load_model(self):
        if self._model is None:
            from sentence_transformers import SentenceTransformer
            self._model = SentenceTransformer("all-MiniLM-L6-v2")

    def __call__(self, batch: pd.DataFrame) -> pd.DataFrame:
        self._load_model()
        texts = batch["text"].fillna("").tolist()
        batch["my_embedding"] = self._model.encode(texts).tolist()
        return batch

from typing import Any, Dict, Optional
from engine.plugins.extractors.pipeline import (
    PipelineDefinition, ResourceType, RowCondition, StepDefinition, build_pipeline_steps
)
from .manifest import MyParams, metadata
from .processors.core import MyConfig, MyProcessor

def build_steps(
    extractor_request: Any,
    container: Optional[Any] = None,
    base_steps: Optional[list] = None,
    **kwargs
) -> Dict[str, Any]:
    params = MyParams(**(extractor_request.extractor_config.parameters or {}))

    steps = [
        StepDefinition(
            service_class=MyProcessor,
            resource_type=ResourceType.CPU,
            config=MyConfig(threshold=params.threshold),
            condition=RowCondition.IS_TEXT,
        ),
    ]

    pipeline = PipelineDefinition(name=metadata["feature_extractor_name"], version=metadata["version"], steps=steps)
    return {"steps": (base_steps or []) + build_pipeline_steps(pipeline), "prepare": lambda ds: ds}

from typing import Any, Dict

class RealtimeHandler:
    def __init__(self):
        self._model = None

    def predict(self, request: Dict[str, Any]) -> Dict[str, Any]:
        if self._model is None:
            from sentence_transformers import SentenceTransformer
            self._model = SentenceTransformer("all-MiniLM-L6-v2")

        text = request.get("text", "")
        embedding = self._model.encode([text])[0].tolist()
        return {"embedding": embedding}

RowCondition.IS_TEXT       # text/* MIME types
RowCondition.IS_IMAGE      # image/* MIME types
RowCondition.IS_VIDEO      # video/* MIME types
RowCondition.IS_AUDIO      # audio/* MIME types
RowCondition.IS_PDF        # application/pdf
RowCondition.ALWAYS        # All rows (default)

client.collections.create(
    collection_name="my_collection",
    source={"type": "bucket", "bucket_ids": ["bkt_..."]},
    feature_extractor={
        "feature_extractor_name": "my_extractor",
        "version": "1.0.0",
        "parameters": {"threshold": 0.7}
    }
)

mixpeek plugin test --verbose
mixpeek plugin publish --dry-run