> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mixpeek.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Mixpeek CLI

> Command-line interface for building, testing, and deploying custom extractors

## Installation

```bash theme={null}
pip install mixpeek
mixpeek --version
```

## Quick Start

```bash theme={null}
# 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
```

See [Custom Extractors](/processing/custom-extractors) for full extractor development guide.

## Configuration

```bash theme={null}
export MIXPEEK_API_KEY="mxp_sk_your_api_key"
export MIXPEEK_NAMESPACE="ns_your_namespace"
```

| Option       | Environment Variable | Description                                                                |
| ------------ | -------------------- | -------------------------------------------------------------------------- |
| `--api-key`  | `MIXPEEK_API_KEY`    | Your Mixpeek API key                                                       |
| `--base-url` | `MIXPEEK_BASE_URL`   | API base URL (default: [https://api.mixpeek.com](https://api.mixpeek.com)) |

## Commands

### `mixpeek plugin init`

Create a new plugin from template.

```bash theme={null}
mixpeek plugin init <name> [options]
```

| Option          | Description                                                 |
| --------------- | ----------------------------------------------------------- |
| `--category`    | `text`, `image`, `video`, `audio`, `document`, `multimodal` |
| `--description` | Plugin description                                          |
| `--author`      | Author name                                                 |
| `--output`      | Output directory                                            |

```bash theme={null}
# Examples
mixpeek plugin init sentiment_analyzer --category text
mixpeek plugin init face_detector --category image --description "Detect faces"
```

### `mixpeek plugin test`

Validate and test plugin locally.

```bash theme={null}
mixpeek plugin test [options]
```

| Option          | Description                     |
| --------------- | ------------------------------- |
| `--path`        | Plugin directory (default: `.`) |
| `--sample-data` | JSON/CSV file with test data    |
| `--verbose`     | Detailed output                 |

**Validates:**

* Structure (manifest.py, pipeline.py exist)
* Schemas (valid Pydantic models)
* Pipeline (`build_steps()` callable)
* Tests (runs pytest if tests/ exists)

```bash theme={null}
# Examples
mixpeek plugin test
mixpeek plugin test --path ./my_extractor --verbose
mixpeek plugin test --sample-data samples.json
```

### `mixpeek plugin publish`

Upload and deploy plugin to Mixpeek.

```bash theme={null}
mixpeek plugin publish [options]
```

| Option        | Description                |
| ------------- | -------------------------- |
| `--path`      | Plugin directory           |
| `--namespace` | Target namespace ID        |
| `--dry-run`   | Validate without uploading |

**What happens:**

1. Validates structure and schemas
2. Runs security scan
3. Creates .tar.gz archive
4. Uploads to S3 via presigned URL
5. Confirms and triggers deployment

```bash theme={null}
# Examples
mixpeek plugin publish
mixpeek plugin publish --namespace ns_abc123 --dry-run
```

### `mixpeek plugin list`

List plugins in namespace.

```bash theme={null}
mixpeek plugin list [options]
```

| Option        | Description                             |
| ------------- | --------------------------------------- |
| `--namespace` | Namespace ID                            |
| `--source`    | `all`, `builtin`, `custom`, `community` |

```bash theme={null}
mixpeek plugin list --source custom
```

## Plugin Structure

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

### manifest.py

```python theme={null}
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",
    },
]
```

### processors/core.py

```python theme={null}
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
```

### pipeline.py

```python theme={null}
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}
```

### realtime.py (Enterprise)

```python theme={null}
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}
```

## Resource Types

| Type               | Use For                         |
| ------------------ | ------------------------------- |
| `ResourceType.CPU` | Text embeddings, classification |
| `ResourceType.GPU` | Local models (Whisper, CLIP)    |
| `ResourceType.API` | External APIs (OpenAI, Vertex)  |

## Row Conditions

```python theme={null}
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)
```

## Security Constraints

Plugins are scanned before deployment. **Forbidden:**

| Pattern                   | Reason          |
| ------------------------- | --------------- |
| `subprocess`, `os.system` | Shell execution |
| `eval`, `exec`            | Dynamic code    |
| `socket`                  | Direct network  |
| `ctypes`                  | Memory access   |
| `__import__`              | Dynamic imports |

## Using Your Plugin

After publishing:

```python theme={null}
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}
    }
)
```

## API Reference

| Endpoint                                           | Method | Description                      |
| -------------------------------------------------- | ------ | -------------------------------- |
| `/v1/namespaces/{id}/plugins/uploads`              | POST   | Get presigned upload URL         |
| `/v1/namespaces/{id}/plugins/uploads/{id}/confirm` | POST   | Confirm upload                   |
| `/v1/namespaces/{id}/plugins`                      | GET    | List plugins                     |
| `/v1/namespaces/{id}/plugins/{id}`                 | GET    | Get plugin details               |
| `/v1/namespaces/{id}/plugins/{id}`                 | DELETE | Delete plugin                    |
| `/v1/namespaces/{id}/plugins/{id}/deploy`          | POST   | Deploy for realtime (Enterprise) |
| `/v1/namespaces/{id}/plugins/{id}/status`          | GET    | Check deployment status          |

## Troubleshooting

| Issue               | Solution                                   |
| ------------------- | ------------------------------------------ |
| Plugin not found    | Check namespace, wait for deployment       |
| Import errors       | Ensure `__init__.py` files exist           |
| Security scan fails | Remove forbidden patterns                  |
| Validation errors   | Check manifest.py exports metadata/schemas |

Debug mode:

```bash theme={null}
mixpeek plugin test --verbose
mixpeek plugin publish --dry-run
```