> ## 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](https://docs.mixpeek.com/docs/processing/custom-extractors.md) 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 [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 ``` This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.