Semantik Plugin Development

This skill helps you create plugins for Semantik, a self-hosted semantic search engine. Plugins extend Semantik's capabilities for document ingestion, embedding, chunking, reranking, extraction, and AI agents.

Protocol Version

Current Version: 1.0.0

Breaking changes to protocols increment the major version. Your plugins continue to work as long as they satisfy the protocol interface.

Security Note

Plugins run in-process with the main Semantik application (no sandboxing). Only install plugins you trust. See Security Guide for details.

Quick Start

Create a minimal connector plugin in 5 minutes:

python

# my_connector.py
from typing import ClassVar, Any, AsyncIterator
import hashlib

class MyConnector:
    PLUGIN_ID: ClassVar[str] = "my-connector"
    PLUGIN_TYPE: ClassVar[str] = "connector"
    PLUGIN_VERSION: ClassVar[str] = "1.0.0"

    def __init__(self, config: dict[str, Any]) -> None:
        self._config = config

    async def authenticate(self) -> bool:
        return True

    async def load_documents(self, source_id: int | None = None) -> AsyncIterator[dict[str, Any]]:
        content = "Document content..."
        yield {
            "content": content,
            "unique_id": "doc-1",
            "source_type": self.PLUGIN_ID,
            "metadata": {},
            "content_hash": hashlib.sha256(content.encode()).hexdigest(),
        }

    @classmethod
    def get_config_fields(cls) -> list[dict[str, Any]]:
        return []

    @classmethod
    def get_secret_fields(cls) -> list[dict[str, Any]]:
        return []

    @classmethod
    def get_manifest(cls) -> dict[str, Any]:
        return {"id": cls.PLUGIN_ID, "type": cls.PLUGIN_TYPE, "version": cls.PLUGIN_VERSION,
                "display_name": "My Connector", "description": "Custom connector"}

Plugin Types

Type	Purpose	Key Method	Template
`connector`	Ingest documents from sources	`load_documents()`	connector.py
`embedding`	Convert text to vectors	`embed_texts()`	embedding.py
`chunking`	Split documents into chunks	`chunk()`	chunking.py
`reranker`	Reorder search results	`rerank()`	reranker.py
`extractor`	Extract entities/metadata	`extract()`	extractor.py
`agent`	LLM-powered capabilities	`execute()`	agent.py

Type-specific guides:

•Connector Guide - Document sources, async iterators
•Embedding Guide - Query/document modes, dimensions
•Chunking Guide - Text segmentation strategies
•Reranker Guide - Cross-encoder reranking
•Extractor Guide - Entity and metadata extraction
•Agent Guide - LLM agents, streaming, context

Cross-cutting guides:

•Testing Guide - Contract tests, mocks, fixtures
•Security Guide - Trust model, best practices
•Advanced Guide - Health checks, dependencies, migration

Development Approach

Protocol-Based (Recommended)

Use plain Python classes with no semantik imports. Plugins are validated by structural typing (duck typing):

python

class MyPlugin:
    PLUGIN_ID: ClassVar[str] = "my-plugin"
    PLUGIN_TYPE: ClassVar[str] = "connector"  # or embedding, chunking, etc.
    PLUGIN_VERSION: ClassVar[str] = "1.0.0"
    # ... implement required methods

Benefits:

•Zero dependencies on semantik
•Develop in separate repository
•Distribute via PyPI or git
•No version conflicts

ABC-Based (Advanced)

Inherit from semantik base classes when you need access to internal utilities:

python

from shared.connectors.base import BaseConnector

class MyConnector(BaseConnector):
    # ... inherit helper methods

Use when:

•Building embedding plugins with GPU management
•Need access to shared utilities
•Developing internal/builtin plugins

Required Class Variables

Every plugin must define:

python

from typing import ClassVar, Any

class MyPlugin:
    PLUGIN_ID: ClassVar[str] = "my-plugin"      # Unique ID (lowercase, hyphens)
    PLUGIN_TYPE: ClassVar[str] = "connector"    # One of 6 types
    PLUGIN_VERSION: ClassVar[str] = "1.0.0"     # Semantic version

Some plugin types require additional class variables:

Type	Additional Variables
`connector`	`METADATA` (dict with name, description, icon)
`embedding`	`INTERNAL_NAME`, `API_ID`, `PROVIDER_TYPE`, `METADATA`
`chunking`	(none)
`reranker`	(none)
`extractor`	(none)
`agent`	(none)

Manifest Method

All plugins must implement get_manifest():

python

@classmethod
def get_manifest(cls) -> dict[str, Any]:
    return {
        "id": cls.PLUGIN_ID,
        "type": cls.PLUGIN_TYPE,
        "version": cls.PLUGIN_VERSION,
        "display_name": "My Plugin",
        "description": "What the plugin does",
        # Optional fields:
        "author": "Your Name",
        "license": "MIT",
        "homepage": "https://github.com/...",
        "requires": ["other-plugin"],  # Dependencies
        "capabilities": {},  # Plugin-specific capabilities
    }

Configuration

Config Fields (UI)

Define configuration fields for the Semantik UI:

python

@classmethod
def get_config_fields(cls) -> list[dict[str, Any]]:
    return [
        {
            "name": "base_url",
            "type": "text",        # text, password, number, boolean, select
            "label": "Base URL",
            "description": "API endpoint",
            "required": True,
            "placeholder": "https://api.example.com",
        },
        {
            "name": "model",
            "type": "select",
            "label": "Model",
            "options": ["model-a", "model-b"],
            "default": "model-a",
        },
    ]

Secret Fields

Mark fields that contain secrets (encrypted at rest):

python

@classmethod
def get_secret_fields(cls) -> list[dict[str, Any]]:
    return [
        {"name": "api_key", "label": "API Key", "required": True},
    ]

Environment Variables

Use the _env suffix pattern for secrets:

python

# In config schema - user enters env var name
"api_key_env": "OPENAI_API_KEY"

# At runtime, semantik resolves it
config = {"api_key": "sk-actual-key-value"}  # Resolved

Testing

Manual Verification

bash

pip install -e .
python -c "
from my_plugin import MyConnector
print(f'ID: {MyConnector.PLUGIN_ID}')
print(f'Type: {MyConnector.PLUGIN_TYPE}')
print(f'Manifest: {MyConnector.get_manifest()}')
"

Protocol Validation

python

import pytest

class TestMyPlugin:
    def test_has_required_attributes(self):
        assert hasattr(MyPlugin, "PLUGIN_ID")
        assert hasattr(MyPlugin, "PLUGIN_TYPE")
        assert hasattr(MyPlugin, "PLUGIN_VERSION")
        assert MyPlugin.PLUGIN_TYPE == "connector"

    def test_manifest_format(self):
        manifest = MyPlugin.get_manifest()
        assert "id" in manifest
        assert "type" in manifest
        assert "display_name" in manifest

    @pytest.mark.asyncio
    async def test_core_functionality(self):
        plugin = MyPlugin(config={})
        # Test plugin-specific methods

With Semantik Test Mixins

If semantik is installed:

python

from shared.plugins.testing.contracts import ConnectorProtocolTestMixin

class TestMyConnector(ConnectorProtocolTestMixin):
    plugin_class = MyConnector

Packaging

pyproject.toml

toml

[project]
name = "semantik-plugin-myconnector"
version = "1.0.0"
requires-python = ">=3.10"
dependencies = []  # Your dependencies only

[project.entry-points."semantik.plugins"]
my-connector = "my_plugin.connector:MyConnector"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

See templates/pyproject.toml for a complete template.

Entry Point Format

code

plugin-id = "module.path:ClassName"

•plugin-id: Should match PLUGIN_ID
•module.path: Python import path
•ClassName: Your plugin class

Installation

bash

# Development
pip install -e .

# From git
pip install git+https://github.com/you/semantik-plugin-myconnector.git

# Via Semantik API
POST /api/v2/plugins/install
{"install_command": "git+https://github.com/..."}

Common Issues

Plugin Not Loading

•
Check entry point is registered:
bash
```
pip show semantik-plugin-myconnector
```

•

Verify PLUGIN_TYPE is valid:

python

assert PLUGIN_TYPE in ["connector", "embedding", "chunking", "reranker", "extractor", "agent"]

•

Check for import errors:

python

try:
    from my_plugin import MyConnector
except ImportError as e:
    print(f"Error: {e}")

Validation Errors

Error	Fix
`missing required keys: {'content'}`	Add all required fields to returned dict
`Invalid role: 'xyz'`	Use valid string from MESSAGE_ROLES
`content_hash must be 64 characters`	Use `hashlib.sha256(text.encode()).hexdigest()`

Async Issues

All I/O methods must be async:

python

# Wrong
def load_documents(self):
    yield {"content": "..."}

# Right
async def load_documents(self) -> AsyncIterator[dict]:
    yield {"content": "..."}

Templates

Ready-to-use templates in templates/:

File	Description
`connector.py`	Document source connector
`embedding.py`	Embedding model provider
`chunking.py`	Text chunking strategy
`reranker.py`	Search result reranker
`extractor.py`	Entity/metadata extractor
`agent.py`	LLM-powered agent
`pyproject.toml`	Package configuration

Copy a template and modify:

bash

cp templates/connector.py my_connector.py
# Edit PLUGIN_ID, PLUGIN_VERSION, and implement methods

Data Format Reference

Connector Documents (IngestedDocumentDict)

python

{
    "content": str,              # Full text (required)
    "unique_id": str,            # Unique identifier (required)
    "source_type": str,          # Your PLUGIN_ID (required)
    "metadata": dict,            # Source metadata (required)
    "content_hash": str,         # SHA-256, 64 hex chars (required)
    "file_path": str | None,     # Local path (optional)
}

Chunk Format (ChunkDict)

python

{
    "content": str,              # Chunk text (required)
    "metadata": {                # Chunk metadata (required)
        "chunk_index": int,
        "start_offset": int,
        "end_offset": int,
    },
    "chunk_id": str | None,      # Unique ID (optional)
    "embedding": list[float] | None,  # Pre-computed (optional)
}

Rerank Result (RerankResultDict)

python

{
    "index": int,                # Original document index (required)
    "score": float,              # Relevance score (required)
    "text": str | None,          # Document text (optional)
    "metadata": dict | None,     # Metadata (optional)
}

Agent Message (AgentMessageDict)

python

{
    "id": str,                   # Unique ID (required)
    "role": str,                 # user, assistant, system, tool_call, tool_result, error
    "type": str,                 # text, thinking, tool_use, tool_output, partial, final, error
    "content": str,              # Message content (required)
    "timestamp": str,            # ISO 8601 (required)
    "is_partial": bool,          # Streaming partial (optional)
    "sequence_number": int,      # Message order (optional)
}

Getting Help

•Semantik docs: See semantik/docs/external-plugins.md for protocol details
•Protocol reference: See semantik/docs/plugin-protocols.md for full specifications
•Examples: Check semantik/packages/shared/plugins/builtins/ for built-in plugins