leocrm/docs/plugin-development-guide.md

LeoCRM Plugin Development Guide

Version: 2.0
Date: 2026-07-23
Applies to: All plugin developers

---

1. Overview

Plugins are self-contained modules that extend LeoCRM's functionality. Each plugin lives in its own directory under app/plugins/builtins/ and declares its capabilities via a PluginManifest. The plugin system supports:

• API Routes — Register FastAPI route handlers
• Event Subscriptions — React to domain events (contact.created, etc.)
• Database Migrations — Versioned SQL migrations run on install
• RBAC Permissions — Declare and enforce permissions
• Frontend UI Contributions — Menu items, page routes, detail tabs, settings pages, dashboard widgets
• AI Agent Capabilities — Register tools for the AI assistant
• Lifecycle Hooks — on_install, on_activate, on_deactivate, on_uninstall

---

2. Plugin Structure

Every plugin lives under app/plugins/builtins/<plugin_name>/:

app/plugins/builtins/my_plugin/
├── __init__.py          # Package init (can be empty)
├── plugin.py            # Plugin class with manifest (required)
├── routes.py            # FastAPI route definitions
├── models.py            # SQLAlchemy models (optional)
├── schemas.py           # Pydantic schemas (optional)
├── services.py          # Business logic (optional)
├── migrations/          # SQL migration files
│   └── 0001_initial.sql
└── tests/               # Plugin tests (optional)
    ├── __init__.py
    └── test_plugin.py

---

3. Manifest Format

The PluginManifest is a Pydantic v2 model that declares all metadata and capabilities of a plugin. It is defined in app/plugins/manifest.py.

3.1 Core Fields

Field	Type	Required	Description
name	str	Yes	Unique plugin identifier (snake_case, max 80 chars, alphanumeric with underscores only)
version	str	Yes	Semantic version string (e.g. "1.0.0")
display_name	str	Yes	Human-readable plugin name (max 120 chars)
description	str	No	Plugin description (max 500 chars)
dependencies	list[str]	No	Other plugin names this plugin depends on
is_core	bool	No	Whether this is a core plugin that cannot be deactivated (default: false)

3.2 Route Definitions

from app.plugins.manifest import PluginRouteDef

routes=[
    PluginRouteDef(
        path="/api/v1/my-plugin",       # URL path prefix
        module="app.plugins.builtins.my_plugin.routes",  # Dotted module path
        router_attr="router",            # Attribute name of the APIRouter in the module
    ),
]

3.3 Event Subscriptions

events=[
    "contact.created",
    "contact.updated",
    "contact.deleted",
]

Event naming convention: <entity>.<action> (e.g. company.created, task.assigned).

3.4 Migrations

migrations=[
    "0001_initial.sql",
    "0002_add_indexes.sql",
]

Migration files are stored in the plugin's migrations/ directory and run in order on install.

3.5 Permissions

permissions=[
    "my_plugin:read",
    "my_plugin:write",
    "my_plugin:delete",
    "my_plugin:admin",
]

Permission naming convention: <plugin_name>:<action>.

3.6 Field Definitions (Field-Level Permissions)

from app.plugins.manifest import FieldDefinition

field_definitions=[
    FieldDefinition(
        module="companies",          # Module name (e.g. 'companies', 'contacts')
        field="annual_revenue",      # Field name
        label="Annual Revenue",      # Human-readable label
        sensitivity="sensitive",     # normal|sensitive|critical
    ),
]

3.7 Agent Capabilities

agent_capabilities=[
    "contact_search",       # Contact search capability
    "email_draft",          # Email draft generation
    "calendar_scheduling",  # Calendar scheduling
]

3.8 Frontend UI Fields (Phase 3)

FrontendMenuItem — Sidebar Navigation

from app.plugins.manifest import FrontendMenuItem

menu_items=[
    FrontendMenuItem(
        label_key="nav.myPlugin",     # i18n key for the menu label
        label="My Plugin",            # Fallback label if i18n key is missing
        path="/my-plugin",            # Frontend route path
        icon="Sparkles",              # lucide-react icon name
        group="",                     # Optional group label_key for tree-style nesting
        order=100,                    # Sort order within the sidebar
        badge_key="",                 # Optional store key for badge count
    ),
]

Field	Type	Required	Default	Description
label_key	str	Yes	—	i18n key for the menu label
label	str	No	""	Fallback label if i18n key is missing
path	str	Yes	—	Frontend route path, e.g. /mail
icon	str	No	"FileText"	lucide-react icon name
group	str	No	""	Optional group label_key for tree-style nesting
order	int	No	100	Sort order within the sidebar
badge_key	str	No	""	Optional store key for badge count

FrontendPageRoute — Page Routes

from app.plugins.manifest import FrontendPageRoute

page_routes=[
    FrontendPageRoute(
        path="/my-plugin",            # Frontend route path
        component="@/pages/MyPlugin", # Dotted path to the React component
        parent="",                    # Parent route path for nested routes
        protected=True,               # Whether the route requires authentication
        order=100,                    # Sort order
    ),
]

Field	Type	Required	Default	Description
path	str	Yes	—	Frontend route path, e.g. /mail or /mail/settings
component	str	Yes	—	Dotted path to the React component, e.g. @/pages/Mail
parent	str	No	""	Parent route path for nested routes (e.g. /settings for a settings sub-page)
protected	bool	No	True	Whether the route requires authentication
order	int	No	100	Sort order

FrontendDetailTab — Entity Detail Tabs

from app.plugins.manifest import FrontendDetailTab

detail_tabs=[
    FrontendDetailTab(
        entity_type="contact",        # Entity type this tab applies to
        label_key="tabs.myPlugin",    # i18n key for the tab label
        label="My Tab",               # Fallback label
        component="@/components/MyTab", # Dotted path to the React component
        icon="FileText",              # lucide-react icon name
        order=50,                     # Sort order within the detail view
        permission="my_plugin:read",  # Optional permission required to see this tab
    ),
]

Field	Type	Required	Default	Description
entity_type	str	Yes	—	Entity type this tab applies to, e.g. 'contact'
label_key	str	Yes	—	i18n key for the tab label
label	str	No	""	Fallback label
component	str	Yes	—	Dotted path to the React component
icon	str	No	"FileText"	lucide-react icon name
order	int	No	100	Sort order within the detail view
permission	str	No	""	Optional permission required to see this tab

FrontendSettingsPage — Settings Sub-Pages

from app.plugins.manifest import FrontendSettingsPage

settings_pages=[
    FrontendSettingsPage(
        path="my-plugin",             # Settings sub-route path
        label_key="settings.myPlugin", # i18n key for the settings nav label
        label="My Plugin",            # Fallback label
        component="@/pages/MyPluginSettings", # Dotted path to the React component
        icon="Sparkles",              # lucide-react icon name
        order=100,                    # Sort order within settings nav
        permission="my_plugin:admin", # Optional permission required
    ),
]

Field	Type	Required	Default	Description
path	str	Yes	—	Settings sub-route path, e.g. mail or notifications
label_key	str	Yes	—	i18n key for the settings nav label
label	str	No	""	Fallback label
component	str	Yes	—	Dotted path to the React component
icon	str	No	"Settings"	lucide-react icon name
order	int	No	100	Sort order within settings nav
permission	str	No	""	Optional permission required

FrontendDashboardWidget — Dashboard Widgets

from app.plugins.manifest import FrontendDashboardWidget

dashboard_widgets=[
    FrontendDashboardWidget(
        id="my_plugin_stats",         # Unique widget identifier
        label_key="widgets.myPlugin", # i18n key for the widget title
        label="My Plugin Stats",      # Fallback label
        component="@/components/MyWidget", # Dotted path to the React component
        icon="LayoutDashboard",       # lucide-react icon name
        order=100,                    # Sort order on the dashboard
        col_span=1,                   # Grid column span (1-4)
        row_span=1,                   # Grid row span
        permission="my_plugin:read",  # Optional permission required
    ),
]

Field	Type	Required	Default	Description
id	str	Yes	—	Unique widget identifier
label_key	str	Yes	—	i18n key for the widget title
label	str	No	""	Fallback label
component	str	Yes	—	Dotted path to the React component
icon	str	No	"LayoutDashboard"	lucide-react icon name
order	int	No	100	Sort order on the dashboard
col_span	int	No	1	Grid column span (1-4)
row_span	int	No	1	Grid row span
permission	str	No	""	Optional permission required

3.9 Complete Manifest Example

from app.plugins.base import BasePlugin
from app.plugins.manifest import (
    PluginManifest, PluginRouteDef, FieldDefinition,
    FrontendMenuItem, FrontendPageRoute, FrontendDetailTab,
    FrontendSettingsPage, FrontendDashboardWidget,
)

class MyPlugin(BasePlugin):
    manifest = PluginManifest(
        name="my_plugin",
        version="1.0.0",
        display_name="My Plugin",
        description="A comprehensive example plugin.",
        dependencies=["permissions"],
        is_core=False,
        routes=[
            PluginRouteDef(
                path="/api/v1/my-plugin",
                module="app.plugins.builtins.my_plugin.routes",
                router_attr="router",
            ),
        ],
        events=["contact.created", "contact.updated"],
        migrations=["0001_initial.sql"],
        permissions=["my_plugin:read", "my_plugin:write", "my_plugin:admin"],
        field_definitions=[
            FieldDefinition(
                module="contacts",
                field="custom_field",
                label="Custom Field",
                sensitivity="normal",
            ),
        ],
        agent_capabilities=["my_plugin:search"],
        menu_items=[
            FrontendMenuItem(
                label_key="nav.myPlugin",
                label="My Plugin",
                path="/my-plugin",
                icon="Sparkles",
                order=100,
            ),
        ],
        page_routes=[
            FrontendPageRoute(
                path="/my-plugin",
                component="@/pages/MyPlugin",
                protected=True,
            ),
        ],
        detail_tabs=[
            FrontendDetailTab(
                entity_type="contact",
                label_key="tabs.myPlugin",
                label="My Tab",
                component="@/components/MyTab",
                icon="FileText",
                order=50,
                permission="my_plugin:read",
            ),
        ],
        settings_pages=[
            FrontendSettingsPage(
                path="my-plugin",
                label_key="settings.myPlugin",
                label="My Plugin",
                component="@/pages/MyPluginSettings",
                icon="Sparkles",
                order=100,
                permission="my_plugin:admin",
            ),
        ],
        dashboard_widgets=[
            FrontendDashboardWidget(
                id="my_plugin_stats",
                label_key="widgets.myPlugin",
                label="My Plugin Stats",
                component="@/components/MyWidget",
                icon="LayoutDashboard",
                order=100,
                col_span=2,
                row_span=1,
                permission="my_plugin:read",
            ),
        ],
    )

---

3.1 Architecture Requirements (v3) — MUST READ

Every plugin must pass docs/plugin-checklist.md before merge. The checklist maps each rule to a failure class that actually occurred in this codebase. Summary of the mechanisms introduced by the architecture repair (Blocks A–C):

Contracts — the ONLY way to reach other plugins

from app.plugins.builtins.contracts import get_contract

contract = get_contract("kommunikation")
if contract is None or not hasattr(contract, "send_message"):
    logger.warning("kommunikation contract unavailable - skipping")
    return  # or degrade gracefully — NEVER import from the other plugin directly
result = await contract.send_message(...)

Rules:

• If a contract is missing, define it in the owning plugin's contracts.py (self-registers via get_contract_registry().register()).
• Contracts must be unregistered on deactivate and re-registered on activate (symmetry).
• CI gate: python scripts/check_cross_plugin_imports.py → 0 violations.

Lifecycle symmetry & ordering

Everything registered in on_activate (contracts, services, event handlers, hooks, tools, search providers) MUST be deregistered in on_deactivate. Own cleanup runs FIRST, then super().on_deactivate(). On activate, call super().on_activate() FIRST, then own registrations.

Hook helpers live on the registry singleton, NOT as module functions:

from app.core.hooks import get_hook_registry
get_hook_registry().register_action("wiki.article.created", handler, owner_tag="wiki")
# deactivate:
get_hook_registry().unregister_all_for_plugin("wiki")

Dependencies (dependencies=[...])

Declare every plugin you depend on in the manifest. Activation order is topological (Kahn), activation fails with a clear error if a dependency is inactive, and deactivation of a dependency is blocked while your plugin is active.

Permission fields are MANDATORY on UI contributions

FrontendMenuItem.permission, FrontendPageRoute.permission, FrontendSettingsPage.permission, FrontendDashboardWidget.permission — empty string means "any authenticated user". The route renderer enforces them via ProtectedRoute; the sidebar/settings filter by them.

Frontend components: register for production builds

New page components must be added to STATIC_COMPONENT_MAP in frontend/src/components/plugins/PluginLoader.tsx. Vite cannot chunk runtime-built import paths — unregistered components work in dev but fail in production. Also verify the component file actually exists (ghost references show an error boundary in production).

Migrations touching plugin-owned tables

Alembic revisions must guard with to_regclass and skip when the table does not exist yet; the plugin-side SQL migration adds the same schema idempotently (dual-path convergence). See migrations 0119–0141 + plugin migration files for the pattern.

Tests run against real PostgreSQL

Use the ephemeral-database fixture pattern from app/plugins/builtins/automation/tests/test_automation.py: create/drop a DB per run, enable pgvector, import all models before create_all, and create real tenant/user rows instead of random UUIDs for FK columns.

---

4. Plugin Lifecycle

4.1 Installation (on_install)

Called when the plugin is installed, after migrations are run. Override to perform seed data or initial setup.

async def on_install(self, db: AsyncSession, service_container: ServiceContainer) -> None:
    """Perform initial setup after migrations."""
    # Create default settings
    settings_service = service_container.settings
    await settings_service.create_defaults(db, plugin_name=self.name)

4.2 Activation (on_activate)

Called when the plugin is activated. Override to register event listeners and prepare runtime state. The default implementation subscribes to events listed in the manifest.

async def on_activate(
    self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus
) -> None:
    """Register event listeners and prepare runtime state."""
    # Default: subscribes to manifest events
    for event_name in self.manifest.events:
        handler = self._make_event_handler(event_name)
        self._event_handlers[event_name] = handler
        event_bus.subscribe(event_name, handler)
    self._container = service_container

4.3 Deactivation (on_deactivate)

Called when the plugin is deactivated. Override to clean up runtime state. The default implementation unsubscribes all event listeners.

async def on_deactivate(
    self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus
) -> None:
    """Clean up runtime state."""
    for event_name, handler in self._event_handlers.items():
        event_bus.unsubscribe(event_name, handler)
    self._event_handlers.clear()

4.4 Uninstallation (on_uninstall)

Called when the plugin is uninstalled (before data tables are dropped). Override to clean up external resources.

async def on_uninstall(self, db: AsyncSession, service_container: ServiceContainer) -> None:
    """Clean up external resources before tables are dropped."""
    # Remove external API webhooks, etc.
    pass

---

5. UI Registration

Plugins contribute frontend UI elements through their manifest. The frontend PluginRegistry component fetches active manifests and populates the pluginStore (Zustand), which is then consumed by:

5.1 Sidebar Menu Items

Menu items from all active plugins are merged and sorted by order via getAllMenuItems(). The sidebar renders them alongside built-in navigation items.

5.2 Page Routes

Page routes are registered via PluginRouteRenderer, a catch-all route handler that checks the current URL against all plugin page_routes. When a match is found, it renders the plugin's page component using PluginPage (React.lazy + Suspense + ErrorBoundary).

5.3 Detail Tabs

Detail tabs are filtered by entity_type via getDetailTabsForEntity(entityType). Entity detail views (contacts, companies, etc.) render these tabs alongside built-in tabs.

5.4 Settings Pages

Settings pages are merged and sorted via getAllSettingsPages(). The settings navigation renders them alongside built-in settings pages.

5.5 Dashboard Widgets

Dashboard widgets are merged and sorted via getAllDashboardWidgets(). The dashboard grid renders them with their specified col_span and row_span.

5.6 Frontend Component Resolution

Component paths use the @/ alias (resolved to src/ by Vite). The PluginPage component converts @/pages/MyPlugin to ../pages/MyPlugin for dynamic import:

// PluginLoader.tsx
const importPath = componentPath.replace(/^@\//, '../');
const LazyComp = lazy(() =>
  import(/* @vite-ignore */ importPath).then((m) => ({
    default: m.default || m[Object.keys(m)[0]],
  }))
);

---

6. Event Bus

6.1 Subscribing to Events

Events are declared in the manifest and handled by methods named on_<event_name> with dots replaced by underscores:

# Manifest
events=["contact.created", "contact.updated"]

# Handler methods
async def on_contact_created(self, event_data: dict):
    contact_id = event_data.get("contact_id")
    # React to new contact

async def on_contact_updated(self, event_data: dict):
    contact_id = event_data.get("contact_id")
    # React to contact update

6.2 Publishing Events

Events are published via the EventBus service:

from app.core.event_bus import get_event_bus

event_bus = get_event_bus()
await event_bus.publish("contact.created", {"contact_id": str(contact.id)})

6.3 Event Naming Conventions

• Format: <entity>.<action>
• Standard actions: created, updated, deleted, assigned, completed
• Examples: company.created, task.assigned, email.sent
• Use past tense for actions

---

23. Migration Runner

7.1 Writing Migrations

SQL migration files are stored in the plugin's migrations/ directory and referenced in the manifest:

-- migrations/0001_initial.sql
CREATE TABLE my_plugin_items (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    name VARCHAR(200) NOT NULL,
    config JSONB DEFAULT '{}',
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE INDEX idx_my_plugin_items_tenant ON my_plugin_items(tenant_id);

7.2 Migration Naming

• Files are named with a zero-padded sequence number and a descriptive slug: 0001_initial.sql, 0002_add_indexes.sql
• Migrations run in alphanumeric order
• Each migration runs exactly once per plugin installation

7.3 Migration Guidelines

• Always include tenant_id for multi-tenant tables
• Use UUID primary keys with gen_random_uuid()
• Include created_at and updated_at timestamps
• Add appropriate indexes for foreign keys and frequently queried columns
• Use IF NOT EXISTS / IF EXISTS for idempotent operations

---

24. Service Container

After activation, plugins can access shared services via self.services:

class MyPlugin(BasePlugin):
    async def on_activate(self, db, service_container, event_bus):
        self._container = service_container

    async def do_something(self):
        # Access services after activation
        settings = self.services.settings
        audit = self.services.audit
        cache = self.services.cache

Available services (defined in app/core/service_container.py):

Service	Accessor	Description
Settings	self.services.settings	Global and per-tenant settings
Audit	self.services.audit	Audit logging
Cache	self.services.cache	Redis/In-memory cache
EventBus	self.services.event_bus	Event publishing/subscribing
Notifications	self.services.notifications	User notifications

---

25. RBAC / Permissions

9.1 Declaring Permissions

Permissions are declared in the manifest:

permissions=[
    "my_plugin:read",
    "my_plugin:write",
    "my_plugin:delete",
    "my_plugin:admin",
]

9.2 Securing Routes

Use the require_permission dependency:

from app.deps import get_current_user, require_permission
from fastapi import Depends

@router.get("", dependencies=[Depends(require_permission("my_plugin:read"))])
async def list_items(current_user: dict = Depends(get_current_user)):
    ...

@router.post("", status_code=201, dependencies=[Depends(require_permission("my_plugin:write"))])
async def create_item(data: ItemCreate, current_user: dict = Depends(get_current_user)):
    ...

9.3 Permission Naming Convention

• Format: <plugin_name>:<action>
• Standard actions: read, write, delete, share, admin
• Examples: calendar:read, dms:write, tags:delete

9.4 Field-Level Permissions

Field definitions in the manifest enable field-level access control:

field_definitions=[
    FieldDefinition(
        module="companies",
        field="annual_revenue",
        label="Annual Revenue",
        sensitivity="sensitive",  # normal|sensitive|critical
    ),
]

---

26. AI Agent Integration

10.1 Agent Capabilities

Declare AI capabilities in the manifest:

agent_capabilities=[
    "contact_search",
    "email_draft",
    "calendar_scheduling",
]

10.2 Tool Registry

Plugins can register tools for the AI assistant:

from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry

registry = get_tool_registry()

registry.register(
    name="search_contacts",
    description="Search contacts by name, email, or phone number",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "limit": {"type": "integer", "description": "Max results", "default": 10},
        },
        "required": ["query"],
    },
    handler=my_search_handler,
    plugin_name="my_plugin",
    required_permission="contacts:read",
    category="search",
)

10.3 Tool Handler

async def my_search_handler(arguments: dict, context: dict) -> str:
    query = arguments.get("query", "")
    limit = arguments.get("limit", 10)
    # Perform search...
    return json.dumps({"results": results})

10.4 Cleanup on Deactivation

async def on_deactivate(self, db, service_container, event_bus):
    registry = get_tool_registry()
    registry.unregister_plugin("my_plugin")

---

27. Search Integration

The Unified Search plugin provides hybrid cross-entity search (PostgreSQL FTS + pgvector) with KI query understanding, RRF rank fusion, visibility filtering, and field-level RBAC. Plugins can expose their entities to unified search by implementing a SearchProvider.

27.1 Creating a SearchProvider

Inherit from BaseSearchProvider and implement the required methods. The base class handles visibility filtering automatically (see 27.4).

# app/plugins/builtins/my_plugin/search_provider.py
from __future__ import annotations

import uuid
from typing import Any

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider


class MyEntitySearchProvider(BaseSearchProvider):
    """Search provider for MyEntity."""

    entity_type = "my_entity"
    supports_fts = True
    supports_vector = True
    supports_rag = False
    supports_graph = False

    async def _search_fts_filtered(
        self,
        db: AsyncSession,
        tsquery: str,
        tenant_id: uuid.UUID,
        limit: int,
        visible_ids: set[uuid.UUID] | None,
    ) -> list[dict[str, Any]]:
        """Full-text search filtered by visible_ids (None = no filter)."""
        if visible_ids is not None:
            sql = text(
                """
                SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
                FROM my_entities e
                WHERE e.tenant_id = :tid
                  AND e.deleted_at IS NULL
                  AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q)
                  AND e.id = ANY(:visible_ids)
                ORDER BY rank DESC
                LIMIT :lim
                """
            )
            result = await db.execute(
                sql, {"q": tsquery, "tid": tenant_id, "lim": limit, "visible_ids": list(visible_ids)}
            )
        else:
            sql = text(
                """
                SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
                FROM my_entities e
                WHERE e.tenant_id = :tid
                  AND e.deleted_at IS NULL
                  AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q)
                ORDER BY rank DESC
                LIMIT :lim
                """
            )
            result = await db.execute(sql, {"q": tsquery, "tid": tenant_id, "lim": limit})
        return [dict(r) for r in result.mappings().all()]

    async def _search_vector_filtered(
        self,
        db: AsyncSession,
        embedding: list[float],
        tenant_id: uuid.UUID,
        limit: int,
        visible_ids: set[uuid.UUID] | None,
    ) -> list[dict[str, Any]]:
        """Semantic vector search on the entity's embedding column."""
        # Same pattern as _search_fts_filtered but using `embedding <=> cast(:emb AS vector)`
        return []

    async def get_embedding_text(
        self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID
    ) -> str:
        """Return the text used for embedding generation (non-sensitive fields only)."""
        sql = text(
            "SELECT name, description FROM my_entities WHERE id = :eid AND tenant_id = :tid"
        )
        result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id})
        row = result.mappings().first()
        if not row:
            return ""
        return " ".join(str(v) for v in row.values() if v)

    def to_search_result(self, entity: object) -> dict[str, Any]:
        """Convert an ORM entity or dict to a search result dict."""
        if isinstance(entity, dict):
            entity_id = str(entity.get("id", ""))
            name = entity.get("name", "")
            description = entity.get("description", "")
        else:
            entity_id = str(getattr(entity, "id", ""))
            name = getattr(entity, "name", "")
            description = getattr(entity, "description", "")
        return {
            "entity_type": self.entity_type,
            "entity_id": entity_id,
            "title": name,
            "snippet": description or "",
            "score": 0.0,
            "data": {},
        }

27.2 Capability Flags

Each provider declares which search modes it supports via class attributes. The registry and API use these flags to decide which search paths to run and to report capabilities to clients.

Flag	Default	Meaning
supports_fts	True	Full-text search via PostgreSQL tsvector/tsquery.
supports_vector	True	Semantic vector search via pgvector embeddings.
supports_rag	False	Retrieval-augmented generation over document chunks.
supports_graph	False	Graph-based search (GraphRAG).

Set supports_vector = False (and implement _search_vector_filtered returning []) when an entity has no embedding column, e.g. chat messages or workflows.

27.3 Registering a Provider

Register providers during plugin activation. The Unified Search plugin calls auto_register_providers(db) on activation, which registers all built-in providers. For a custom plugin, register your provider directly in on_activate:

async def on_activate(self, db, service_container, event_bus) -> None:
    await super().on_activate(db, service_container, event_bus)
    from app.plugins.builtins.unified_search.provider_registry import get_search_registry
    from app.plugins.builtins.my_plugin.search_provider import MyEntitySearchProvider

    get_search_registry().register(MyEntitySearchProvider())

To add a provider to the built-in auto_register_providers list, import and append it to the provider_cls list in app/plugins/builtins/unified_search/provider_registry.py.

27.4 Permission Filtering (Automatic)

BaseSearchProvider.search_fts / search_vector automatically load the calling user's visible entity IDs via get_visible_ids and pass them to _search_fts_filtered / _search_vector_filtered. When is_system_admin is true or no user_id is provided, visible_ids is None and no filter is applied. Your _search_*_filtered implementation must honor visible_ids (add AND id = ANY(:visible_ids) when it is not None).

27.5 Auto-Indexing via Events / Outbox

Entities are indexed automatically when created or updated. The Unified Search plugin subscribes to domain events (e.g. contact.created, contact.updated, mail.synced, file.uploaded) and enqueues indexing jobs. For custom entities, publish the corresponding events or enqueue jobs directly:

from app.core.jobs import enqueue_job

# After creating/updating an entity
await enqueue_job("index_entity", "my_entity", str(entity_id))

Indexing jobs call index_entity(entity_type, entity_id, tenant_id, db), which uses the provider's get_embedding_text to generate an embedding and stores it in the entity's embedding column. The indexed_at column tracks dedup so unchanged entities are not re-embedded.

27.6 Lifecycle Hooks

The Unified Search plugin subscribes to lifecycle events to keep the index consistent:

Event	Handler	Effect
entity.deleted	handle_entity_delete	Removes embedding + TSV (and chunks for files).
entity.restored	handle_entity_restore	Rebuilds the search index.
entity.corrected	handle_entity_correction	Rebuilds the search index.

Publish these events (or call the lifecycle functions directly) when your plugin deletes, restores, or corrects entities so the search index stays in sync.

27.7 RAG Document Chunking

For RAG over long documents, use chunk_text from app.plugins.builtins.unified_search.chunking to split extracted text into overlapping chunks before embedding:

from app.plugins.builtins.unified_search.chunking import chunk_text

chunks = chunk_text(document_text, chunk_size=1000, overlap=200)
# Each chunk: {"chunk_index": 0, "chunk_text": "...", "chunk_hash": "sha256..."}

Chunks are stored in the document_chunks table and embedded at the chunk level for fine-grained vector retrieval. chunk_hash is a deterministic SHA-256 of the chunk text, used for dedup.

---

28. Testing Guide

11.1 Backend Tests

Tests live in the plugin's tests/ directory or in the central tests/ folder:

# tests/test_my_plugin.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_list_items_requires_permission(client: AsyncClient, auth_headers):
    response = await client.get("/api/v1/my-plugin", headers=auth_headers)
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_list_items_without_permission_returns_403(client: AsyncClient, no_perm_headers):
    response = await client.get("/api/v1/my-plugin", headers=no_perm_headers)
    assert response.status_code == 403

11.2 Frontend Tests

Frontend tests use vitest, @testing-library/react, and jsdom:

// frontend/src/components/plugins/__tests__/PluginRegistry.test.tsx
import React from 'react';
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render } from '@testing-library/react';
import { PluginRegistry } from '../PluginRegistry';
import { usePluginStore } from '@/store/pluginStore';

// Mock the API hook
vi.mock('@/api/pluginManifests', () => ({
  useActivePluginManifests: () => ({
    data: { plugins: [/* ... */], total: 1 },
    isLoading: false,
    error: null,
  }),
}));

describe('PluginRegistry', () => {
  beforeEach(() => {
    usePluginStore.getState().reset();
  });

  it('populates store with manifests on mount', () => {
    render(<PluginRegistry />);
    const manifests = usePluginStore.getState().manifests;
    expect(manifests).toHaveLength(1);
  });
});

---

28. Do's and Don'ts

Do's

• Do use snake_case for plugin names
• Do declare all permissions in the manifest
• Do secure every API route with require_permission
• Do include tenant_id in all multi-tenant tables
• Do use UUID primary keys
• Do prefix i18n keys with the plugin name
• Do clean up resources in on_deactivate and on_uninstall
• Do write tests for both backend and frontend
• Do follow the event naming convention <entity>.<action>
• Do use semantic versioning for plugin versions

Don'ts

• Don't hardcode tenant IDs or user IDs
• Don't use synchronous database operations
• Don't store secrets in the database without encryption
• Don't modify other plugins' data directly (use events instead)
• Don't create circular dependencies between plugins
• Don't use is_core=True unless the plugin is essential for system operation
• Don't skip error handling in event handlers (they run asynchronously)
• Don't register the same route path in multiple plugins

---

29. Examples

13.1 Minimal Plugin (verified by tests/test_gate_f_minimal_example.py)

⚠️ Three pitfalls verified by the Gate-F test suite — the original version of this example failed all three:

1. __init__.py is required with a re-export of the plugin class — discover_builtins() scans the package namespace and never finds classes that live only in plugin.py.
2. The route must carry the full path — main.py mounts plugin routers WITHOUT a prefix; an empty route path raises Prefix and path cannot be both empty.
3. Plugin routes are dispatched dynamically — they never appear in app.routes; verify them via HTTP request, not via route introspection.

# app/plugins/builtins/minimal_example/__init__.py
"""Minimal Example plugin package."""
from app.plugins.builtins.minimal_example.plugin import MinimalExamplePlugin

__all__ = ["MinimalExamplePlugin"]

# app/plugins/builtins/minimal_example/plugin.py
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest, PluginRouteDef

class MinimalExamplePlugin(BasePlugin):
    manifest = PluginManifest(
        name="minimal_example",
        version="1.0.0",
        display_name="Minimal Example",
        description="A minimal example plugin.",
        dependencies=[],
        routes=[
            PluginRouteDef(
                path="/api/v1/minimal-example",
                module="app.plugins.builtins.minimal_example.routes",
                router_attr="router",
            ),
        ],
        events=[],
        migrations=[],
        permissions=["minimal_example:read"],
    )

# app/plugins/builtins/minimal_example/routes.py
# NOTE: main.py mounts plugin routers WITHOUT a prefix — the manifest's
# route path is documentation only. The full path must live here.
from fastapi import APIRouter, Depends
from app.deps import get_current_user, require_permission

router = APIRouter()

@router.get(
    "/api/v1/minimal-example",
    dependencies=[Depends(require_permission("minimal_example:read"))],
)
async def list_items(current_user: dict = Depends(get_current_user)):
    return {"items": []}

13.2 Plugin with UI

See the complete manifest example in Section 3.9 for a plugin with full UI contributions (menu items, page routes, detail tabs, settings pages, dashboard widgets).

13.3 Plugin with Events

# app/plugins/builtins/event_example/plugin.py
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest

class EventExamplePlugin(BasePlugin):
    manifest = PluginManifest(
        name="event_example",
        version="1.0.0",
        display_name="Event Example",
        description="Demonstrates event handling.",
        dependencies=[],
        events=["contact.created", "contact.updated", "contact.deleted"],
        migrations=[],
        permissions=[],
    )

    async def on_contact_created(self, event_data: dict):
        contact_id = event_data.get("contact_id")
        print(f"Contact created: {contact_id}")

    async def on_contact_updated(self, event_data: dict):
        contact_id = event_data.get("contact_id")
        print(f"Contact updated: {contact_id}")

    async def on_contact_deleted(self, event_data: dict):
        contact_id = event_data.get("contact_id")
        print(f"Contact deleted: {contact_id}")

---

7. LLM Integration

LeoCRM stellt einen zentralen LLM-Client bereit über den alle LLM-Calls (Completion und Embedding) laufen. Keine direkten litellm.acompletion() oder litellm.aembedding() Aufrufe in Plugin-Code.

7.1 Completion

from app.ai.llm_client import llm_complete

result = await llm_complete(
    model="openai/gpt-4o",  # oder None für Default-Model
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Summarize this email."},
    ],
    temperature=0.3,
    max_tokens=1000,
    # Optional: API-Key/Base aus DB holen
    db=db,
    tenant_id=tenant_id,
    # Optional: JSON-Response erzwingen
    response_format={"type": "json_object"},
    # Optional: Tools für Function-Calling
    tools=[{"type": "function", "function": {...}}],
    # Optional: Retry-Konfiguration
    timeout=30,
    max_retries=2,
)

content = result["content"]       # str — LLM-Response-Text
usage = result["usage"]           # dict — {prompt_tokens, completion_tokens, total_tokens}
cost_usd = result["cost_usd"]     # float — geschätzte Kosten
model = result["model"]           # str — verwendetes Modell
raw_response = result["raw_response"]  # litellm-Response-Objekt für erweiterte Nutzung

7.2 Embedding

from app.ai.llm_client import llm_embed

# Einzelne Embedding
embeddings = await llm_embed(
    texts="Text to embed",
    model="openai/text-embedding-3-small",  # oder None für Default
    db=db,
    tenant_id=tenant_id,
    dimensions=768,  # Optional, für text-embedding-3 Modelle
)
# → [[0.01, 0.02, ...]]

# Batch-Embedding
embeddings = await llm_embed(
    texts=["Text 1", "Text 2", "Text 3"],
    db=db,
    tenant_id=tenant_id,
)
# → [[...], [...], [...]]

7.3 Provider-Auswahl und API-Key-Auflösung

Der zentrale Client löst API-Keys automatisch aus der Datenbank (AIProvider-Tabelle) oder Environment-Variablen. Priorität:

1. Explizit übergebener api_key Parameter
2. DB-Lookup über get_api_credentials(db, tenant_id)
3. Environment-Variablen (AI_API_KEY, AI_API_BASE, AI_PROVIDER)
4. Mock-Mode (kein API-Key → Keyword-basierte Fallback-Antworten)

from app.ai.llm_client import get_api_credentials, build_model

# API-Credentials aus DB holen
api_key, api_base, provider_type = await get_api_credentials(db, tenant_id)

# Model-String bauen (provider/model)
model = build_model("gpt-4o", provider_type)  # → "openai/gpt-4o"

7.4 Error-Handling

Der zentrale Client klassifiziert Errors automatisch:

• Transient (Timeout, Rate-Limit 429, Service-Unavailable 503) → Retry mit Exponential-Backoff
• Permanent (Auth 401/403, Validation, Model-Not-Found) → Sofortiger Fehler, kein Retry

try:
    result = await llm_complete(model="openai/gpt-4o", messages=[...])
except Exception as e:
    # Transient errors wurden bereits retried
    # Permanent errors kommen hier an
    logger.error(f"LLM call failed permanently: {e}")

7.5 Cost-Tracking

llm_complete() gibt cost_usd zurück — automatisch berechnet aus Token-Usage. Plugins sollen diesen Wert in ihren Cost-Tracking-Mechanismus übernehmen.

result = await llm_complete(...)
total_cost += result["cost_usd"]

7.6 Was NICHT zu tun ist

• ❌ import litellm und direkte litellm.acompletion() / litellm.aembedding() Aufrufe
• ❌ Eigene API-Key-Verwaltung — immer über get_api_credentials() oder llm_complete(db=db, tenant_id=tenant_id)
• ❌ Eigene Retry-Logik — llm_complete() hat bereits Retry mit Backoff
• ❌ Eigene Cost-Tracking-Logik — llm_complete() gibt cost_usd zurück
• ❌ Eigene Provider-Auswahl — build_model() und get_api_credentials() zentralisieren das

---

8. Event-System Rollen

LeoCRM hat 4 Event-Systeme mit klar getrennten Rollen. Nicht dieselbe Funktion über Hook UND EventBus triggern.

Übersicht

System	Rolle	Persistenz	Use Case
HookRegistry	Lifecycle-Erweiterungspunkte	In-Memory	contact.before_create, mail.after_send, dms.after_delete — Plugins können Daten anpassen oder reagieren
EventBus	Flüchtige interne Events	In-Memory	notification.created, ui.contact_selected — asynchrone Notifikationen, UI-Events, Proactive Suggestions
Outbox	Dauerhafte Domain Events	DB (transactional)	contact.created, mail.received, task.completed — reliable Delivery, Retry, DLQ, Worker-Polling
WebhookDispatcher	Externe HTTP-Zustellung	DB + HTTP	Externe Webhooks an registrierte URLs — Retry, Auth, Payload-Signatur

8.1 HookRegistry (app/core/hooks.py)

Wann verwenden: Wenn ein Plugin bei einem Lifecycle-Punkt Daten anpassen oder reagieren will.

from app.core.hooks import get_hook_registry

reg = get_hook_registry()

# Action — kein Return, nur Seiteneffekte
reg.register_action("contact.before_create", self._on_contact_create, priority=10)

# Filter — Return modifizierten Wert
def _format_name(self, name: str) -> str:
    return name.title()
reg.register_filter("contact.format_display_name", self._format_name, priority=10)

Aufruf im Core/Plugin-Service:

from app.core.hooks import do_action, apply_filters

await do_action("contact.before_create", contact_data, db=db)
display_name = await apply_filters("contact.format_display_name", contact.name)

Verfügbare Hooks

Actions (fire-and-forget, kein Return-Wert):

Hook Name	Modul	Parameter
contact.before_create	app/services/contact_service.py	data, db, tenant_id, user_id
contact.after_create	app/services/contact_service.py	serialized, db, tenant_id, user_id
contact.before_update	app/services/contact_service.py	data, db, tenant_id, user_id, contact_id
contact.after_update	app/services/contact_service.py	snapshot, db, tenant_id, user_id, contact_id
contact.before_delete	app/services/contact_service.py	db, tenant_id, contact_id, user_id
contact.after_delete	app/services/contact_service.py	db, tenant_id, contact_id, user_id
company.before_create	app/routes/companies.py	body, db, tenant_id, user_id
company.after_create	app/routes/companies.py	serialized, db, tenant_id, user_id
company.before_update	app/routes/companies.py	body, db, tenant_id, user_id, company_id
company.after_update	app/routes/companies.py	serialized, db, tenant_id, user_id, company_id
company.before_delete	app/routes/companies.py	db, tenant_id, user_id, company_id
company.after_delete	app/routes/companies.py	snapshot, db, tenant_id, user_id, company_id
mail.before_send	app/plugins/builtins/mail/services.py	mail_data (Filter)
mail.after_send	app/plugins/builtins/mail/services.py	mail_data, db, account, msg_id
mail.after_receive	app/plugins/builtins/mail/services.py	payload, db, tenant_id
mail.before_delete	app/plugins/builtins/mail/services.py	mail_id, tenant_id, permanent, db
mail.after_delete	app/plugins/builtins/mail/services.py	mail_id, tenant_id, permanent, db
mail.before_move	app/plugins/builtins/mail/services.py	mail_id, tenant_id, source_folder_id, target_folder_id, db
mail.after_move	app/plugins/builtins/mail/services.py	mail_id, tenant_id, source_folder_id, target_folder_id, db
dms.before_upload	app/plugins/builtins/dms/routes.py	upload_data (Filter)
dms.after_upload	app/plugins/builtins/dms/routes.py	payload, db, tenant_id, user_id
dms.before_update	app/plugins/builtins/dms/routes.py	data, db, tenant_id, user_id, file_id
dms.after_update	app/plugins/builtins/dms/routes.py	payload, db, tenant_id, user_id, file_id
dms.before_delete	app/plugins/builtins/dms/routes.py	db, tenant_id, user_id, file_id
dms.after_delete	app/plugins/builtins/dms/routes.py	db, tenant_id, user_id, file_id
dms.before_restore	app/plugins/builtins/dms/routes.py	db, tenant_id, user_id, file_id
dms.after_restore	app/plugins/builtins/dms/routes.py	payload, db, tenant_id, user_id, file_id
dms.folder.before_create	app/plugins/builtins/dms/routes.py	body, db, tenant_id, user_id
dms.folder.after_create	app/plugins/builtins/dms/routes.py	payload, db, tenant_id, user_id
dms.folder.before_delete	app/plugins/builtins/dms/routes.py	db, tenant_id, user_id, folder_id
dms.folder.after_delete	app/plugins/builtins/dms/routes.py	db, tenant_id, user_id, folder_id
calendar.before_appointment	app/plugins/builtins/calendar/routes.py	body, tenant_id, user_id, cal_id
calendar.after_appointment	app/plugins/builtins/calendar/routes.py	entry_id, tenant_id, user_id
calendar.before_update	app/plugins/builtins/calendar/routes.py	body, tenant_id, user_id, entry_id
calendar.after_update	app/plugins/builtins/calendar/routes.py	entry_id, tenant_id, user_id
calendar.before_delete	app/plugins/builtins/calendar/routes.py	tenant_id, user_id, entry_id
calendar.after_delete	app/plugins/builtins/calendar/routes.py	tenant_id, user_id, entry_id
task.before_create	app/plugins/builtins/tasks/services.py	data, db, tenant_id, user_id
task.after_create	app/plugins/builtins/tasks/services.py	serialized, db, tenant_id, user_id
task.before_update	app/plugins/builtins/tasks/services.py	data, db, tenant_id, task_id
task.after_update	app/plugins/builtins/tasks/services.py	serialized, db, tenant_id, task_id
task.before_delete	app/plugins/builtins/tasks/services.py	db, tenant_id, task_id
task.after_delete	app/plugins/builtins/tasks/services.py	db, tenant_id, task_id
comm.conversation.before_create	app/plugins/builtins/kommunikation/services.py	tenant_id, user_id
comm.conversation.after_create	app/plugins/builtins/kommunikation/services.py	conversation_id, tenant_id, user_id
comm.before_message	app/plugins/builtins/kommunikation/services.py	conversation_id, tenant_id, sender_id
comm.after_message	app/plugins/builtins/kommunikation/services.py	message_id, conversation_id, tenant_id, sender_id
comm.before_edit	app/plugins/builtins/kommunikation/services.py	message_id, tenant_id, user_id
comm.after_edit	app/plugins/builtins/kommunikation/services.py	message_id, tenant_id, user_id
comm.before_delete	app/plugins/builtins/kommunikation/services.py	message_id
comm.after_delete	app/plugins/builtins/kommunikation/services.py	message_id
agent.before_run	app/plugins/builtins/automation/agent_runner.py	agent_id, tenant_id, trigger_type
agent.after_run	app/plugins/builtins/automation/agent_runner.py	agent_id, tenant_id, status, result
workflow.before_start	app/services/workflow_service.py	instance_id, workflow_id, tenant_id, user_id
workflow.after_start	app/services/workflow_service.py	instance_id, workflow_id, tenant_id, user_id
workflow.after_complete	app/services/workflow_service.py	instance_id, workflow_id, tenant_id, user_id
workflow.after_cancel	app/services/workflow_service.py	instance_id, workflow_id, tenant_id, user_id
tag.before_create	app/plugins/builtins/tags/routes.py	body, db, tenant_id, user_id
tag.after_create	app/plugins/builtins/tags/routes.py	payload, db, tenant_id, user_id
tag.before_assign	app/plugins/builtins/tags/routes.py	body, db, tenant_id, user_id
tag.after_assign	app/plugins/builtins/tags/routes.py	payload, db, tenant_id, user_id
tag.before_unassign	app/plugins/builtins/tags/routes.py	body, db, tenant_id, user_id
tag.after_unassign	app/plugins/builtins/tags/routes.py	payload, db, tenant_id, user_id
tag.before_delete	app/plugins/builtins/tags/routes.py	db, tenant_id, user_id, tag_id
tag.after_delete	app/plugins/builtins/tags/routes.py	db, tenant_id, user_id, tag_id

Filters (modifizieren Wert, Return erforderlich):

Hook Name	Modul	Parameter
contact.format_display_name	app/services/contact_service.py	name, data, db, tenant_id, user_id
dms.before_upload	app/plugins/builtins/dms/routes.py	upload_data (filename, mime_type)
mail.before_send	app/plugins/builtins/mail/services.py	mail_data
search.before_search	app/plugins/builtins/unified_search/search_engine.py	query_analysis
search.after_search	app/plugins/builtins/unified_search/search_engine.py	all_results

8.2 EventBus (app/core/event_bus.py)

Wann verwenden: Für flüchtige interne Notifikationen, UI-Events, Proactive Suggestions. Nicht für Events die reliable Delivery brauchen.

from app.core.event_bus import get_event_bus

event_bus = get_event_bus()

# Subscribe
event_bus.subscribe("notification.created", self._on_notification)

# Publish (ephemeral — geht verloren bei Crash)
await event_bus.publish("notification.created", {"user_id": "...", "message": "..."})

8.3 Outbox (app/core/outbox.py)

Wann verwenden: Für dauerhafte Domain Events die reliable Delivery, Retry und Worker-Verarbeitung brauchen.

from app.core.outbox import enqueue_outbox_event

# In derselben Transaktion wie die Business-Operation
await enqueue_outbox_event(db, tenant_id, "contact.created", {
    "contact_id": str(contact.id),
    "tenant_id": str(tenant_id),
})
# Transaction commit → Event ist durable → Worker pollt und published an EventBus

Features: DLQ (error_message, failed_at), Replay (replay_failed_event), Consumer Registry, Stats.

Verfügbare Outbox Events

Event Name	Modul	Payload	Aggregate
contact.created	app/services/contact_service.py	contact_id, tenant_id, displayname, type	contact
contact.updated	app/services/contact_service.py	contact_id, tenant_id, changes	contact
lead.created	app/services/contact_service.py	contact_id, tenant_id	contact
mail.received	app/plugins/builtins/mail/services.py	mail_id, tenant_id, account_id, folder_id, subject, from_address	mail
mail.send	app/plugins/builtins/mail/services.py	mail_id, tenant_id, account_id	mail
file.created	app/plugins/builtins/dms/routes.py	file_id, tenant_id, name, mime_type, size_bytes	dms_file
file.deleted	app/plugins/builtins/dms/routes.py	file_id, tenant_id	dms_file
file.restored	app/plugins/builtins/dms/routes.py	file_id, tenant_id	dms_file
dms.file.uploaded	app/plugins/builtins/dms/routes.py	(legacy alias for file.created)	dms_file
task.completed	app/plugins/builtins/tasks/services.py	task_id, tenant_id, title, assigned_to	task
workflow.started	app/services/workflow_service.py	instance_id, workflow_id, tenant_id	workflow_instance
workflow.completed	app/services/workflow_service.py	instance_id, workflow_id, tenant_id, status	workflow_instance
workflow.cancelled	app/services/workflow_service.py	instance_id, workflow_id, tenant_id	workflow_instance
agent.run_started	app/plugins/builtins/automation/agent_runner.py	agent_id, tenant_id, trigger_type	agent
agent.run_completed	app/plugins/builtins/automation/agent_runner.py	agent_id, tenant_id, status, cost_usd	agent

8.4 WebhookDispatcher (app/core/webhook_dispatcher.py)

Wann verwenden: Für externe HTTP-Zustellung an registrierte Webhook-URLs.

from app.core.webhook_dispatcher import register_webhook_event_handlers

# Wird automatisch im Worker registriert — Plugins müssen nur Webhook-Configs erstellen
# und Events über die Outbox publishen

8.5 Entscheidungsregel

Braucht das Event reliable Delivery + Retry?
  → JA → Outbox
  → NEIN → Braucht es Daten-Anpassung (Filter)?
           → JA → HookRegistry (register_filter)
           → NEIN → Braucht es nur Reaktion (Action)?
                    → JA → HookRegistry (register_action)
                    → NEIN → Ist es eine flüchtige Notifikation / UI-Event?
                             → JA → EventBus
                             → NEIN → Geht es an externe Systeme?
                                      → JA → WebhookDispatcher (über Outbox)
                                      → NEIN → Braucht kein Event

Verboten: Dieselbe Funktion über Hook UND EventBus triggern — das führt zu Doppel-Ausführung und Race-Conditions.

---

9. Schema Authority

LeoCRM hat eine klare Schema-Authority-Hierarchie. Kein neuer Schema-Mechanismus.

Authority-Regeln

Schema-Typ	Authority	Wie
Core-Tabellen	Alembic-Migrationen	alembic revision --autogenerate -m "..." → alembic upgrade head
Plugin-Tabellen	Plugin-Migrationsweg	plugin/migrations/ → sync_plugin_schema.py bei Aktivierung
Runtime Auto-Sync	Nicht authoritative	Base.metadata.create_all in Tests/dev — nie in Produktion

Verbindliche Regeln

1. Core-Schema-Änderungen immer über Alembic-Migrationen — nie manuelle SQL-Statements in Produktion
2. Plugin-Schema-Änderungen über Plugin-Migrationen — nie Core-Migrationen für Plugin-Tabellen
3. Runtime Auto-Sync (create_all, sync_plugin_schema.py) ist Convenience für Dev/Tests — nicht für Produktion authoritative
4. Migration-Staffelung beachten: neu → migrieren → umstellen → testen → release → alt entfernen
5. Keine Schema-Drift — wenn Core und Plugin dasselbe Modell nutzen, ist Core authoritative

Plugin-Migrationen

# plugin/migrations/001_initial.py
from alembic import op

def upgrade():
    op.create_table("my_plugin_table", ...)

def downgrade():
    op.drop_table("my_plugin_table")

Plugin-Migrationen werden bei Plugin-Aktivierung automatisch ausgeführt (sync_plugin_schema.py). Bei Deaktivierung bleiben die Tabellen erhalten (Soft-Deactivate). Bei Uninstall werden sie gedroppt.

---

10. Trigger

Plugins können durch vier Trigger-Typen aktiviert werden: Domain-Events, UI-Events, Cron-Jobs und manuelle Trigger. Alle vier Typen konvergieren im selben Execution-Kern (run_automation). Webhook-Trigger folgt in Phase G.

10.0 Trigger-Typen-Übersicht

Trigger-Typ	trigger_type	Event-Quelle	Durability	Dispatch-Pfad
Domain-Event	event	Outbox → Worker → EventBus	durable (at-least-once)	EventBus * → TriggerDispatcher → run_automation
UI-Event	ui	WebSocket → EventBus (ephemeral)	ephemeral (keine Persistenz)	EventBus * → TriggerDispatcher → run_automation
Cron/Schedule	schedule	ARQ Cron → scheduler_tick	durable (ARQ-Queue)	scheduler_tick → enqueue_job("run_automation")
Manual	manual	API-Route /execute	durable (HTTP-Request)	Route → run_automation(trigger_type="manual")

Wichtig: UI-Events (ui.*) dürfen niemals in die Outbox geschrieben werden. Sie sind ephemeral und fließen direkt über den EventBus.

10.1 Domain-Event-Trigger (durable)

Domain-Events werden über die Transaction Outbox gepublished: ein Service schreibt das Event in die event_outbox Tabelle innerhalb derselben DB-Transaktion. Der ARQ-Worker pollt die Outbox alle 5 Sekunden und published die Events auf den EventBus.

Der TriggerDispatcher (app/core/trigger_dispatcher.py) abonniert den * Wildcard-Handler auf dem EventBus und evaluiert jedes eingehende Event. Bei einer Übereinstimmung mit einer AutomationDefinition (trigger_type="event", trigger_config.event_name matcht) wird run_automation aufgerufen.

Flow:

Service → enqueue_outbox_event() → event_outbox Tabelle
                                         ↓ (commit)
ARQ Worker (process_outbox_job, alle 5s)
                                         ↓
EventBus.publish_with_results(event_name, envelope)
                                         ↓
TriggerDispatcher._on_event(payload)
                                         ↓
DB-Query: AutomationDefinition WHERE trigger_type='event' AND trigger_config->>'event_name' = event_name
                                         ↓
run_automation(trigger_type="event", trigger_data=payload)

Beispiel — Domain-Event in Automation umwandeln:

# AutomationDefinition erstellen
POST /api/v1/automation/
{
    "name": "notify-on-contact-create",
    "trigger_type": "event",
    "trigger_config": {"event_name": "contact.created"},
    "conditions": [],
    "actions": [
        {"type": "notification", "config": {"user_id": "...", "title": "Neuer Kontakt"}}
    ]
}

Beispiel — Plugin veröffentlicht Domain-Event:

from app.core.outbox import enqueue_outbox_event

await enqueue_outbox_event(db, tenant_id, "contact.created", {
    "contact_id": str(contact.id),
    "tenant_id": str(tenant_id),
})
# Event wird beim Commit der Transaktion persistent.
# Der Worker published es an den EventBus.
# Der TriggerDispatcher findet passende Automations und führt sie aus.

10.2 UI-Event-Trigger (ephemeral)

UI-Events (ui.*) sind ephemeral — sie werden nicht in die Outbox geschrieben. Sie fließen direkt vom Frontend über WebSocket → EventBus → TriggerDispatcher.

Der TriggerDispatcher erkennt UI-Events am ui. Prefix und setzt trigger_type="ui" beim Dispatch. AutomationDefinition-Einträge mit trigger_type="ui" werden automatisch gematcht.

Flow:

Frontend (WebSocket) → ws_helpers → EventBus.publish("ui.contact_selected", payload)
                                         ↓
TriggerDispatcher._on_event(payload)
    → is_ui_event("ui.contact_selected") = True
    → trigger_type = "ui"
                                         ↓
DB-Query: AutomationDefinition WHERE trigger_type='ui' AND trigger_config->>'event_name' = 'ui.contact_selected'
                                         ↓
run_automation(trigger_type="ui", trigger_data=payload)

⚠️ Kritische Regel: UI-Events dürfen niemals über enqueue_outbox_event() gepublished werden. Verwende ausschließlich event_bus.publish():

from app.core.event_bus import get_event_bus

# RICHTIG — ephemeral, nur EventBus
event_bus = get_event_bus()
await event_bus.publish("ui.contact_selected", {
    "event_name": "ui.contact_selected",
    "tenant_id": str(tenant_id),
    "data": {"contact_id": str(contact_id)},
})

# FALSCH — würde UI-Event in Outbox persistieren
# await enqueue_outbox_event(db, tenant_id, "ui.contact_selected", {...})  # ❌

Beispiel — UI-Event-Automation erstellen:

POST /api/v1/automation/
{
    "name": "track-contact-selection",
    "trigger_type": "ui",
    "trigger_config": {"event_name": "ui.contact_selected"},
    "conditions": [],
    "actions": [
        {"type": "api_call", "config": {"url": "https://analytics.example.com/track", "method": "POST"}}
    ]
}

10.3 Cron-Trigger (schedule)

Cron-Jobs werden über AutomationCronJob-Einträge in der Datenbank verwaltet. Der ARQ-Worker führt scheduler_tick alle 5 Minuten aus, liest fällige Cron-Jobs und enqueued run_automation mit trigger_type="scheduled".

Flow:

ARQ Cron (alle 5min) → scheduler_tick(ctx)
                                         ↓
DB-Query: AutomationCronJob WHERE is_active=True AND next_run_at <= now()
                                         ↓
enqueue_job("run_automation", job.target_id, trigger_type="scheduled")
                                         ↓
run_automation(trigger_type="scheduled", trigger_data={})
                                         ↓
Update: last_run_at = now, next_run_at = calculate_next_run(cron_expression)

Beispiel — Cron-Job im Plugin-Manifest deklarieren:

from app.plugins.manifest import CronJobContribution

cron_jobs=[
    CronJobContribution(
        name="my_plugin_daily_cleanup",
        cron_expression="0 3 * * *",
        job_type="custom",
        plugin_name="my_plugin",
    ),
],

Beispiel — Cron-Job zur Automation verknüpfen:

POST /api/v1/automation/cron-jobs/
{
    "name": "daily-report-auto",
    "cron_expression": "0 9 * * *",
    "job_type": "automation",
    "target_id": "<automation-definition-uuid>",
    "is_active": true
}

10.4 Manuelle Trigger (manual)

Manuelle Trigger werden über die API-Route POST /api/v1/automation/{id}/execute ausgelöst. Die Route ruft run_automation direkt mit trigger_type="manual" auf.

Flow:

HTTP POST /api/v1/automation/{id}/execute
                                         ↓
AutomationRun (status="running") wird in DB erstellt
                                         ↓
run_automation(trigger_type="manual", trigger_data={"triggered_by": user_id})
                                         ↓
AutomationRun wird mit Ergebnis aktualisiert (status="completed"/"partial")

Beispiel — Manuelle Automation auslösen:

curl -X POST https://crm.media-on.de/api/v1/automation/{id}/execute \
  -H "Cookie: session=..."

Beispiel — Eigene Manual-Trigger-Route im Plugin:

@router.post("/api/v1/my-plugin/run-sync")
async def run_manual_sync(request: Request, db: AsyncSession = Depends(get_db)):
    """Manueller Trigger — Benutzer startet Sync von der UI."""
    # Business logic oder: run_automation direkt aufrufen
    from app.plugins.builtins.automation.execution_engine import run_automation
    result = await run_automation(
        ctx={},
        automation_id=str(automation_id),
        trigger_type="manual",
        trigger_data={"triggered_by": str(user_id)},
    )
    return {"status": "started", "result": result}

10.5 TriggerDispatcher — Der generische Event→Automation Dispatcher

Der TriggerDispatcher (app/core/trigger_dispatcher.py) ist das zentrale Bindeglied zwischen EventBus und Automation-Engine. Er ersetzt frühere hardcodierte Event-Handler-Stubs (on_contact_created etc.) durch eine generische Wildcard-Subscription.

Registrierung:

• In app/main.py lifespan (API-Container)
• In app/core/worker.py on_startup (Worker-Container)

from app.core.trigger_dispatcher import register_trigger_dispatcher
from app.core.event_bus import get_event_bus

event_bus = get_event_bus()
register_trigger_dispatcher(event_bus)
# Dispatcher abonniert '*' auf dem EventBus

Matching-Logik:

1. Jedes Event auf dem EventBus erreicht _on_event(payload)
2. event_name wird aus payload["event_name"] extrahiert
3. ui.* Prefix → trigger_type="ui", sonst trigger_type="event"
4. DB-Query: aktive AutomationDefinition mit passendem trigger_type und trigger_config.event_name
5. Jede Match-Definition wird über run_automation ausgeführt

Wichtig: Der Dispatcher ist generisch — es gibt keine hardcodierte Event-Liste. Jedes registrierte Outbox-Event kann eine Automation triggern, sobald eine AutomationDefinition mit passendem trigger_config.event_name existiert.

---

11. Message-System

Plugins können System-Nachrichten in Chat-Räume posten, eigene Chat-Räume erstellen und Mini-Apps registrieren. Das Communication-Plugin (kommunikation) stellt die Infrastruktur bereit.

11.1 System-Nachrichten posten

from app.plugins.builtins.kommunikation.contracts import create_plugin_room, send_message

# Plugin-Room für einen Benutzer erstellen
await create_plugin_room(
    db, tenant_id, user_id,
    plugin_name="my_plugin",
    title="My Plugin",
    participant_type="system",
    user_role="reader",
)

# Nachricht in den Room senden
await send_message(
    db, tenant_id, conversation_id,
    sender_id=None,
    sender_type="system",
    content="**Sync abgeschlossen**\n120 Kontakte aktualisiert",
    content_format="markdown",
    blocks=None,
    metadata={"event_type": "sync.completed"},
)

11.2 Rich Content Blocks

Nachrichten können strukturierte Blocks enthalten (app/plugins/builtins/kommunikation/content_types.py):

blocks = [
    {
        "block_type": "action_card",
        "block_data": {
            "title": "Backup fehlgeschlagen",
            "body": "Letztes Backup um 03:00 Uhr ist gescheitert.",
            "actions": [
                {"label": "Öffnen", "action": "/settings/backup", "type": "primary"},
                {"label": "Archivieren", "action": "dismiss", "type": "secondary"},
            ],
        },
    },
    {
        "block_type": "contact_card",
        "block_data": {"contact_id": str(contact_id), "name": "Max Mustermann"},
    },
    {
        "block_type": "miniapp",
        "block_data": {"app_id": "my_miniapp", "config": {"contact_id": str(contact_id)}},
    },
]

await send_message(db, tenant_id, conv_id, sender_id=None, sender_type="system",
                   content="Neuer Kontakt", content_format="markdown", blocks=blocks)

Unterstützte Block-Typen: text, markdown, html, image, audio, video, file, action_card, contact_card, miniapp.

11.3 Mini-Apps registrieren

Mini-Apps werden im Manifest deklariert:

from app.plugins.manifest import MiniAppContribution

miniapps=[
    MiniAppContribution(
        app_id="my_miniapp",
        name="My Mini App",
        icon="AppWindow",
        description="Interactive mini-app embedded in chat",
        render_schema={"type": "object", "properties": {"contact_id": {"type": "string"}}},
    ),
],

Siehe system_notif Plugin als Referenz-Implementierung.

---

12. Search

Plugins können Search-Provider registrieren, um ihre Entitäten in der Unified Search bereitzustellen.

12.1 SearchProvider implementieren

Implementiere das SearchProvider-Protokoll oder erbe von BaseSearchProvider für automatische Visibility-Filterung:

from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider

class MyEntitySearchProvider(BaseSearchProvider):
    entity_type = "my_entity"

    async def _search_fts_filtered(self, db, tsquery, tenant_id, limit, visible_ids):
        # FTS-Query mit visible_ids Filter
        ...

    async def _search_vector_filtered(self, db, embedding, tenant_id, limit, visible_ids):
        # Vector-Search mit pgvector + visible_ids Filter
        ...

    async def get_embedding_text(self, db, entity_id, tenant_id) -> str:
        # Text-Repräsentation für Embedding-Generierung
        ...

    def to_search_result(self, entity) -> dict:
        return {"id": str(entity.id), "type": "my_entity", "title": entity.name}

12.2 Provider registrieren

from app.plugins.builtins.unified_search.provider_registry import get_search_registry

async def on_activate(self, db, service_container, event_bus):
    await super().on_activate(db, service_container, event_bus)
    registry = get_search_registry()
    registry.register(MyEntitySearchProvider())

async def on_deactivate(self, db, service_container, event_bus):
    get_search_registry().unregister("my_entity")
    await super().on_deactivate(db, service_container, event_bus)

12.3 Unterstützte Modi

Modus	Beschreibung
FTS	PostgreSQL Full-Text Search (tsquery)
Vector	Semantische Suche via pgvector (embedding)
RAG	Retrieval-Augmented Generation (Vector + LLM)
Graph	Beziehungs-Suche (zukünftig)

12.4 Auto-Indexierung

Wenn eine Entität erstellt/aktualisiert wird, wird ein Outbox-Event gepublished. Der Worker generiert das Embedding und aktualisiert den Such-Index automatisch. Plugins müssen nur get_embedding_text() korrekt implementieren.

---

13. File Storage

Plugins speichern Files über das zentrale Storage-Backend (app/core/storage.py). Keine eigenen Storage-Backends implementieren.

13.1 save_with_metadata()

from app.core.storage import save_with_metadata, get_storage_backend

# Speichert mit MIME-Prüfung, Size-Limit und Hash-Berechnung
metadata = await save_with_metadata(
    path=f"my_plugin/{tenant_id}/{file_name}",
    data=file_bytes,
    allowed_mimes=["application/pdf", "image/png", "image/jpeg"],  # None = Default-Allowlist
    max_size_mb=10,  # None = aus Config
)
# Returns: {path, mime_type, size, hash, storage_path}

13.2 MIME-Prüfung und Path-Traversal-Schutz

from app.core.storage import validate_mime, validate_size, compute_hash

# MIME wird content-based erkannt (python-magic) mit Extension-Fallback
mime = validate_mime(path, data, allowed_mimes=["application/pdf"])
# ValueError bei nicht erlaubtem MIME-Typ

# Path-Traversal wird im LocalStorage automatisch blockiert:
# _full_path() normt den Pfad und prüft, ob er innerhalb base_path bleibt

13.3 Storage-Backend lesen

backend = get_storage_backend()
data = await backend.read(path)
url = await backend.get_url(path, expires=3600)
exists = await backend.exists(path)
await backend.delete(path)

Für File-Embedding siehe Kapitel 7.2 (LLM Integration — Embedding).

---

14. Redis

Plugins nutzen den globalen Redis-Singleton. Keine eigenen Connections, kein aioredis.from_url().

14.1 Redis-Client

from app.core.auth import get_redis

redis = get_redis()  # Globaler Singleton
await redis.setex(f"my_plugin:lock:{resource_id}", 30, "locked")
value = await redis.get(f"my_plugin:lock:{resource_id}")
await redis.delete(f"my_plugin:lock:{resource_id}")

14.2 Cache-Wrapper

from app.core.cache import cache_get, cache_set, cache_delete, cache_flush_pattern

# Set mit TTL (Default: 300s)
await cache_set(f"my_plugin:summary:{tenant_id}", {"count": 42}, ttl=60)

# Get
data = await cache_get(f"my_plugin:summary:{tenant_id}")

# Delete
await cache_delete(f"my_plugin:summary:{tenant_id}")

# Pattern-Flush (alle Keys mit Prefix)
await cache_flush_pattern("my_plugin:*")

Verboten: aioredis.from_url() in Plugin-Code — immer get_redis() oder get_cache() verwenden.

---

15. Permissions

Plugins deklarieren Permissions im Manifest und nutzen das bestehende RBAC/ABAC-System.

15.1 Permissions deklarieren

manifest = PluginManifest(
    name="my_plugin",
    permissions=[
        "my_plugin:read",
        "my_plugin:write",
        "my_plugin:admin",
    ],
    ...
)

15.2 Permissions in Routes erzwingen

from app.deps import require_permission

@router.get("/api/v1/my-plugin/items",
            dependencies=[Depends(require_permission("my_plugin:read"))])
async def list_items(...):
    ...

@router.post("/api/v1/my-plugin/items",
             dependencies=[Depends(require_permission("my_plugin:write"))])
async def create_item(...):
    ...

15.3 Wildcard-Support

Das Permission-System unterstützt Wildcards: my_plugin:*, *:read, *:* (Superadmin). Bare * ist verboten.

15.4 Field-Level Permissions

from app.plugins.manifest import FieldDefinition

field_definitions=[
    FieldDefinition(module="my_plugin", field="secret_code",
                    label="Secret Code", sensitivity="sensitive"),
],

Wichtig: Tools, Skills und MCP dürfen keine Rechte verleihen. Siehe docs/permissions.md und docs/permissions_plugin_dev.md.

---

16. AI Tools

Plugins können Tools in der globalen ToolRegistry registrieren, die von AI-Agenten aufgerufen werden können.

16.1 Tool registrieren

from app.plugins.builtins.ai_assistant.contracts import get_tool_registry

async def on_activate(self, db, service_container, event_bus):
    await super().on_activate(db, service_container, event_bus)
    registry = get_tool_registry()
    registry.register(
        name="my_plugin_lookup_contact",
        description="Look up a contact by name in My Plugin",
        parameters={
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Contact name"},
            },
            "required": ["name"],
        },
        handler=self._lookup_contact_handler,
        plugin_name="my_plugin",
        required_permission="my_plugin:read",
        category="contacts",
    )

async def _lookup_contact_handler(self, arguments: dict, context: dict) -> str:
    name = arguments.get("name", "")
    # Business logic — return JSON string
    return f'{{"found": true, "name": "{name}"}}'

16.2 Tool bei Deaktivierung entfernen

async def on_deactivate(self, db, service_container, event_bus):
    get_tool_registry().unregister_plugin("my_plugin")
    await super().on_deactivate(db, service_container, event_bus)

16.3 Permission-Prüfung

Jedes Tool hat ein required_permission-Feld. Der AI-Service prüft dies vor der Ausführung:

# Wird automatisch im Service geprüft:
if tool.required_permission:
    if not check_permission(user_context, tool.required_permission):
        return f"Error: Permission '{tool.required_permission}' required"

16.4 Tool-Schema (OpenAI-kompatibel)

Das AITool-Dataclass konvertiert automatisch ins OpenAI Function-Calling-Format via to_openai_schema().

---

17. MCP

MCP (Model Context Protocol) dient als dünne Exposure-Schicht auf bestehende Tools/Services. MCP erhält keine eigenen Rechte.

17.1 Architektur

External MCP Server → MCP Client Plugin → ToolRegistry → Existing Tools/Services

MCP-Tools werden mit dem Naming-Schema mcp__{server}__{tool} in der ToolRegistry registriert.

17.2 Tool-Registrierung

# MCP Client Plugin registriert externe Tools automatisch:
registry.register(
    name=f"mcp__{server_name}__{tool_name}",
    description=f"[MCP:{server_name}] {tool.description}",
    parameters=tool.parameters,
    handler=_make_handler(server_cfg, tool.name),
    plugin_name="mcp_client",
    required_permission="mcp-client:read",  # Bestehende Permission
    category="mcp-external",
)

17.3 Auth- und Run-as-Kontext

Der vorhandene Auth-/Run-as-Kontext und normale Permission-Prüfungen bleiben maßgeblich. MCP-Tools erben die Permissions des aufrufenden Benutzers — MCP kann keine Rechte verleihen, die der Benutzer nicht hat.

Verboten: Eigene Auth-Bypass-Logik in MCP-Handlern. Immer den Standard-Permission-Check verwenden.

---

18. UI-Events

Plugins können auf UI-Events reagieren und eigene UI-Events publishen. UI-Events sind ephemeral — über EventBus, nicht über Outbox.

18.1 UI-Events abonnieren

from app.core.event_bus import get_event_bus

async def on_activate(self, db, service_container, event_bus):
    await super().on_activate(db, service_container, event_bus)
    event_bus.subscribe("ui.contact_selected", self._on_contact_selected)
    event_bus.subscribe("ui.page_navigated", self._on_page_navigated)
    event_bus.subscribe("ui.mail_opened", self._on_mail_opened)

async def _on_contact_selected(self, payload: dict) -> None:
    contact_id = payload.get("contact_id")
    # React to contact selection — e.g. preload data

18.2 UI-Events publishen

event_bus = get_event_bus()
await event_bus.publish("ui.my_plugin_widget_ready", {
    "widget_id": "summary",
    "tenant_id": str(tenant_id),
    "user_id": str(user_id),
})

18.3 Bekannte UI-Events

Event	Payload	Beschreibung
ui.contact_selected	{contact_id, user_id}	Benutzer hat Kontakt ausgewählt
ui.page_navigated	{path, user_id}	Benutzer hat Seite navigiert
ui.mail_opened	{mail_id, user_id}	Benutzer hat E-Mail geöffnet

Wichtig: UI-Events gehen bei Crash verloren. Für reliable Delivery Outbox verwenden (Kapitel 8).

---

19. AI UI Control

Das ai_ui_control Plugin ermöglicht AI-Agenten, die Frontend-UI zu steuern: Navigation, Filter, Kontakte öffnen, Modals, Tabs und Settings.

19.1 Command-Typen

from app.plugins.builtins.ai_ui_control.schemas import UICommandType

# Unterstützte Actions:
# navigate      → {action: 'navigate', path: '/contacts/123'}
# filter        → {action: 'filter', entity: 'contacts', filter: {type: 'company'}}
# open_contact  → {action: 'open_contact', contact_id: '...'}
# modal         → {action: 'modal', modal: 'edit', contact_id: '...'}
# tab           → {action: 'tab', tab: 'emails', contact_id: '...'}
# settings      → {action: 'settings', section: 'ai', key: 'model', value: 'gpt-4'}

19.2 Command senden (REST)

# AI-Agent sendet Command via REST:
POST /api/v1/ai-ui-control/command
{
    "action": "navigate",
    "path": "/contacts/abc-123",
    "description": "Opening contact detail page"
}
# Response: {command_id, status: "pending", action: "navigate"}

19.3 Command empfangen (WebSocket)

Das Frontend verbindet sich via WebSocket /ws/ai-ui-control und empfängt Commands in Echtzeit. Nach Ausführung sendet das Frontend Feedback zurück:

UICommandFeedback(
    command_id="...",
    status=UICommandStatus.success,
    action=UICommandType.navigate,
    current_path="/contacts/abc-123",
)

19.4 Permissions

AI UI Control benötigt ai_ui_control:write für Commands und ai_ui_control:read für Status-Abfragen.

Wichtig: Persistente Mutationen (Daten ändern, erstellen, löschen) dürfen nicht über AI UI Control laufen. Diese müssen über reguläre Tools/Services mit Permission-Prüfungen gehen. AI UI Control ist nur für UI-Navigation und Anzeige.

---

20. Sensitive Data

Plugins müssen sensible Daten explizit deklarieren und sicherstellen, dass diese nicht in Snapshots, Such-Index, Embeddings, Exporten oder Logs landen.

20.1 SENSITIVE_FIELDS deklarieren

Im Manifest über field_definitions mit sensitivity="sensitive" oder sensitivity="critical":

from app.plugins.manifest import FieldDefinition

field_definitions=[
    FieldDefinition(module="my_plugin", field="api_key",
                    label="API Key", sensitivity="critical"),
    FieldDefinition(module="my_plugin", field="internal_notes",
                    label="Internal Notes", sensitivity="sensitive"),
],

20.2 Was NICHT in Snapshots/Index/Embeddings/Export/Logs darf

• Passwords, Tokens, API Keys, Session-IDs — niemals loggen, indexieren oder embedden
• Personenbezogene Daten (DSGVO-relevant) — nicht in Such-Embeddings ohne explizite Freigabe
• Interne Notizen mit sensitivity="sensitive" — nicht in Exporten ohne Berechtigung

20.3 Error-Logging Sanitization

Das Error-Logging-Endpoint (/api/v1/errors) sanitized automatisch sensible Keys:

# app/routes/errors.py — _SENSITIVE_PATTERNS
# Erkennt: token, password, secret, authorization, cookie, session,
#           api_key, access_token, refresh_token, csrf, bearer, private_key
# Diese Felder werden durch "[redacted]" ersetzt.

20.4 AI/Data Exposure Policy

• AI-Tools dürfen keine sensiblen Felder an LLM-Provider senden, ohne dass der Benutzer die entsprechende Permission hat
• Embeddings dürfen nur aus nicht-sensiblen Texten generiert werden
• Export-Service respektiert Field-Level Permissions (hidden, readonly, read)

---

21. Migration-Staffelung

Schema-Änderungen in Plugins müssen gestaffelt durchgeführt werden, um Downtime und Datenverlust zu vermeiden.

21.1 Sechs-Schritt-Prozess

1. Neue Struktur erstellen (neue Tabelle/Spalte/Index)
   → Migration 0002_add_new_column.sql

2. Daten migrieren (Backfill)
   → Migration 0003_backfill_data.sql (oder Worker-Job)

3. Reads/Writes umstellen
   → Code schreibt in neue UND alte Struktur (Dual-Write)
   → Code liest aus neuer Struktur (mit Fallback auf alte)

4. Tests
   → Unit-Tests mit neuer Struktur
   → Integration-Tests mit Dual-Write
   → Migration-Tests (upgrade + downgrade)

5. Stabiler Release
   → Deploy mit neuer Struktur + Dual-Write
   → Verify: alle Daten korrekt migriert

6. Alte Struktur entfernen
   → Migration 0004_drop_old_column.sql
   → Code: Dual-Write entfernen, nur neue Struktur

21.2 Plugin-Migration-Beispiel

-- migrations/0002_add_status_v2.sql
ALTER TABLE my_plugin_items ADD COLUMN status_v2 VARCHAR(20) DEFAULT 'active';

-- migrations/0003_backfill_status.sql
UPDATE my_plugin_items SET status_v2 = CASE
    WHEN status = 'pending' THEN 'pending'
    WHEN status = 'done' THEN 'completed'
    ELSE 'active'
END;

# Schritt 3: Dual-Write in Service
item.status = old_status  # alte Spalte
item.status_v2 = map_to_new_status(old_status)  # neue Spalte

Wichtig: Jeder Schritt ist ein separater Release. Nie Struktur ändern und Daten migrieren in einer Migration.

---

22. Error-Handling

Plugins werfen strukturierte Errors über ApiError mit code, detail, field und status. Tracebacks werden niemals an den User gesendet.

22.1 ApiError werfen

from app.core.error_codes import ApiError

@router.post("/api/v1/my-plugin/items")
async def create_item(body: ItemCreate, db: AsyncSession = Depends(get_db)):
    if not body.name:
        raise ApiError(code="validation_error", detail="Name is required", field="name")

    existing = await check_duplicate(db, body.name)
    if existing:
        raise ApiError(code="not_found", detail="Item already exists", status=409)

    # Service unavailable
    raise ApiError(code="service_unavailable", detail="External API timeout")

22.2 Standardisierte Error-Codes

Code	HTTP Status	Beschreibung
not_found	404	Resource nicht gefunden
permission_denied	403	Keine Berechtigung
validation_error	422	Validierung fehlgeschlagen
rate_limited	429	Zu viele Requests
internal_error	500	Interner Fehler
service_unavailable	503	Service temporär nicht verfügbar

22.3 Error-Propagation-Kette

Plugin → ApiError(code, detail)
  → Core Exception Handler → JSON Response {error: {code, detail, field}}
    → Frontend API Client → TanStack Query onError
      → ErrorBoundary / Toast Notification → User

22.4 trace_id-Korrelation

Jeder Error bekommt eine trace_id für End-to-End-Tracing:

# WebSocket-Errors via ws_helpers:
from app.core.ws_helpers import send_ws_error
await send_ws_error(websocket, code="validation_error",
                    detail="Invalid input", trace_id=trace_id)

22.5 Frontend ErrorBoundary

Plugin-Seiten und MiniApps müssen eine React ErrorBoundary haben. Bei unhandled Errors wird eine freundliche Fehlermeldung angezeigt — kein Stacktrace.

22.6 Partial-Failure bei Batch-Operationen

Bei Batch-Operationen (z.B. Bulk-Import) wird nicht die gesamte Operation abgebrochen. Erfolgreiche Items werden committed, fehlgeschlagene Items werden mit Fehlergrund gesammelt zurückgegeben:

results = {
    "success": [item_id_1, item_id_2],
    "failed": [{"item": item_3, "error": "validation_error: Name required"}],
}

Verboten: Unbehandelte Tracebacks an den User senden. Alle Plugin-Errors müssen als ApiError geworfen werden.

---

30. API Versioning Strategie

LeoCRM verwendet URL-basiertes API-Versioning (/api/v1/). Diese Strategie ist verbindlich für alle zukünftigen API-Änderungen.

Regeln

Änderungstyp	Versionierung	Beispiele
Non-breaking	Innerhalb v1	Neue Endpoints, neue optionale Felder, neue Query-Parameter
Breaking	Neue v2-Router parallel	Feld entfernt, Feld-Typ geändert, Endpoint entfernt, Semantik geändert

Breaking Change Prozess

1. Neuen Router erstellen — APIRouter(prefix="/api/v2/...") parallel zu v1
2. v1 Routes deprecated markieren — @router.get("/api/v1/...", deprecated=True) + Deprecation Header
3. Übergangszeit — 1 Release-Zyklus beide Versionen parallel
4. v1 Routes entfernen — nach Übergangszeit + Verifikation dass keine Clients mehr v1 nutzen

Plugin API Versioning

Plugins deklarieren ihre API-Prefixe im Manifest (routes.prefix). Plugin-API-Änderungen folgen derselben Strategie — Breaking Changes erfordern neue Prefix-Version.

---

31. Import/Export Handler auf Shared Helpers

LeoCRM stellt wiederverwendbare Bausteine für Import/Export-Funktionalität bereit. Plugins können eigene Import/Export-Handler auf dieser Basis aufsetzen.

31.1 Shared Helpers (app/services/import_export_helpers.py)

Die folgenden Funktionen sind verfügbar und können von jedem Plugin importiert werden:

from app.services.import_export_helpers import (
    parse_csv,       # CSV → list[dict] mit Encoding-Detection
    parse_json,      # JSON → list[dict]
    parse_xlsx,      # XLSX → list[dict] (openpyxl)
    write_csv,       # list[dict] → CSV bytes
    write_json,      # list[dict] → JSON bytes
    write_xlsx,      # list[dict] → XLSX bytes
    map_fields,      # Source-Column → Target-Field Mapping
    suggest_mapping, # Auto-Mapping-Vorschlag
    validate_row,    # Zeilen-Validierung mit required + validators
    build_error_report,  # Strukturierter Fehler-Report
    build_import_result, # Standardisiertes Import-Ergebnis
    detect_format,   # Format-Erkennung (csv/json/xlsx)
    parse_file,      # Auto-Detect + Parse
)

31.2 Eigener Import-Handler

Ein Plugin kann einen eigenen Import-Handler erstellen:

from app.services.import_export_helpers import parse_file, map_fields, validate_row, build_import_result

async def import_my_entity(db, tenant_id, user_id, content: bytes, filename: str, field_mapping: dict | None = None):
    rows = parse_file(filename, content)
    total = len(rows)
    errors = []
    valid_rows = []

    for idx, row in enumerate(rows, start=1):
        if field_mapping:
            row = map_fields(row, field_mapping)
        row_errors = validate_row(row, required=["name"], validators={"email": {"type": "email"}})
        if row_errors:
            for e in row_errors:
                errors.append({"row": idx, "field": "", "message": e})
        else:
            valid_rows.append(row)

    # ... DB-Insert mit Partial-Failure ...
    for idx, row in enumerate(valid_rows, start=1):
        try:
            # Insert entity
            pass
        except Exception as exc:
            errors.append({"row": idx, "field": "", "message": str(exc)})

    failed_row_count = len({e["row"] for e in errors})
    return build_import_result(
        total=total,
        succeeded=len(valid_rows),
        failed=failed_row_count,
        errors=errors,
    )

31.3 Partial-Failure-Semantik

Import-Handler müssen Partial-Failure implementieren:

1. Validierung pro Zeile: validate_row() prüft required fields und validators
2. Fehlerhafte Zeilen sammeln: Fehler werden in errors-Liste mit {row, field, message} gesammelt
3. Erfolgreiche Zeilen committen: Gültige Zeilen werden in DB geschrieben, pro Zeile try/except
4. Status-Klassifizierung: build_import_result() setzt Status auf success, partial_success, oder failed
5. Fehler-Report: build_error_report() erstellt strukturierten Report mit total_errors und errors-Liste

31.4 Background Processing für große Imports

Für Dateien > 1000 Zeilen soll der Import als ARQ-Job laufen:

from app.services.import_export_jobs import create_import_job, get_import_job_status

# In Route:
if len(rows) > 1000:
    job_id = await create_import_job(
        entity_type="my_entity",
        csv_content=content.decode("utf-8"),
        tenant_id=tenant_id,
        user_id=user_id,
        field_mapping=mapping,
    )
    return {"status": "pending", "job_id": job_id}

Job-Status wird in Redis gespeichert (leocrm:import_job:{job_id}) mit TTL 1h.

31.5 Frontend-Integration

Das Frontend nutzt die API-Client-Funktionen aus importExport.ts:

• previewImport(file, entityType) — Vorschau mit Mapping-Vorschlag
• validateImport(file, entityType, fieldMapping) — Validierung ohne Import
• importCsv(file, entityType, dryRun, fieldMapping) — Import mit optionalem Mapping
• getImportJobStatus(jobId) — Polling für Background-Jobs
• exportData(entityType, format) — Export als CSV/XLSX/JSON

---

32. Agents (Phase F)

Phase F introduces a full agent system: agent definitions, a ReAct loop, a tool registry, a skill registry, a permission model, an approval workflow, and workstream integration. This chapter explains how plugins can contribute agents, tools, and skills.

32.1 Agent Definition

An agent is defined by an AgentDefinition record (automation plugin). Key fields:

• name, description — display and purpose.
• llm_model, provider, api_key, api_base — LLM configuration (secrets never exposed via API).
• system_prompt — the agent's base instructions.
• max_steps, max_tokens, max_duration_seconds — execution limits.
• budget_limit_usd — cumulative cost cap per agent.
• tool_ids, skill_ids — which tools and skills the agent may use.
• mode — reactive (manual/proactive trigger) or proactive.
• trace_mode — standard or extended (extended posts ReAct steps to the workstream).
• ai_use_case_metadata — allowed data categories for the data policy.

Create an agent via POST /api/v1/agents or directly in code:

from app.plugins.builtins.automation.models import AgentDefinition

agent = AgentDefinition(
    tenant_id=tenant_id,
    name="Support Bot",
    description="Answers support questions",
    llm_model="gpt-4o",
    system_prompt="You are a helpful support assistant.",
    tool_ids=["mail_read", "contact_search"],
    skill_ids=["support_skill"],
    max_steps=10,
    budget_limit_usd=5.0,
)

32.2 Registering Tools in the ToolRegistry

Tools are registered in the central ToolRegistry (ai_assistant plugin). A tool exposes an OpenAI-style function schema and a handler.

from app.plugins.builtins.ai_assistant.contracts import get_tool_registry
from app.ai.agent_tools import AITool

async def _search_contacts_handler(args: dict, ctx: dict) -> dict:
    # ... business logic ...
    return {"results": [...]}

registry = get_tool_registry()
registry.register(AITool(
    name="contact_search",
    description="Search contacts by name or email",
    parameters={
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
    handler=_search_contacts_handler,
    required_permission="contacts:read",
))

• required_permission gates the tool: a user only gets the tool if they hold that permission.
• The handler receives (args, ctx) where ctx contains tenant_id, user_id, db, and agent_run_id.

32.3 Registering Skills in the SkillRegistry

Skills bundle instructions and allowed tools. They are registered in the singleton SkillRegistry.

from app.ai.skill_registry import SkillDefinition, get_skill_registry

skill = SkillDefinition(
    name="support_skill",
    description="Support workflow instructions",
    instructions="Use contact_search then mail_read to answer support tickets.",
    allowed_tool_ids=["contact_search", "mail_read"],
    category="support",
)
get_skill_registry().register(skill)

• get_by_names([...]) resolves skills and skips unknown names.
• A skill never grants a tool the user does not already have permission for — the effective tool set is the intersection of user, agent, skill, and tool permissions.

32.4 ReAct Loop

The ReAct loop (app.ai.agent_loop.run_react_loop) drives the agent:

1. Build context (system prompt + user message).
2. Call the LLM with the available tool schemas.
3. If the LLM returns a tool call, execute the tool handler and append the observation.
4. Repeat until a final answer, max_steps, timeout, or budget is reached.

from app.ai.agent_loop import run_react_loop

result = await run_react_loop(
    agent_definition=agent,
    messages=[{"role": "user", "content": "Find the latest invoice"}],
    tools=tool_schemas,
    tool_registry=registry,
    db=db,
    tenant_id=tenant_id,
    user_id=user_id,
    agent_run_id=run_id,
    max_steps=20,
    timeout_seconds=300,
)

result is a ReActResult with status, steps, final_content, total_cost_usd, and error. Statuses: completed, stopped_max_steps, stopped_timeout, stopped_error, budget_exceeded.

32.5 Permission Model

Effective agent permissions are the intersection of four layers:

User permissions ∩ Agent tool_ids ∩ Skill allowed_tool_ids ∩ Tool required_permission

• resolve_agent_permissions(db, tenant_id, user_id, agent) returns an AgentPermissionContext with effective_tool_ids and can_use_tool(name).
• System admins bypass the permission check and get all tools configured on the agent.
• filter_visible_agents respects agents:read; check_agent_execute_permission respects agents:execute.
• Optimistic locking: PATCH/DELETE on agents require a matching version; a mismatch returns 409 conflict.

32.6 Approval Workflow

Tools that require human approval pause the loop and create an ApprovalRequest (app.core.approval).

• Status lifecycle: pending → approved | rejected | expired.
• Create: create_approval_request(db, tenant_id, entity_type=..., entity_id=..., action=..., requested_by=..., metadata=...).
• Resolve: resolve_approval_request(db, tenant_id, request_id, decision="approved"|"rejected", approver_id=..., comment=...).
• Expire: expire_approval_request(db, tenant_id, request_id).
• API: POST /api/v1/approvals, POST /api/v1/approvals/{id}/approve|reject.

32.7 Workstream Integration

Agents post messages, steps, and results to the communication system (app.ai.agent_workstream):

• post_agent_message — text or block message, marked AI-generated.
• post_agent_step — ReAct step as an action_card (only in extended trace mode).
• post_agent_result — final result with status, steps_taken, total_cost_usd, run_id.
• post_approval_request — approval card.

All messages are marked with AI-generated transparency metadata.

32.8 Data Policy & Transparency

• enforce_data_policy(db, tenant_id, messages, agent) strips sensitive fields, enforces allowed data categories from ai_use_case_metadata, and checks provider compliance before content reaches the LLM.
• mark_as_ai_generated(content, metadata) adds ai_generated: true and ai_metadata to any outbound message.

32.9 Pre-Built Agents

LeoCRM ships pre-built agents in the automation plugin. Plugins can register additional agents at activation time by creating AgentDefinition records and registering their tools/skills in the registries.

---

This document is authoritative for all plugin development at LeoCRM.

Dokumente-Generator-Beitrag (Phase L)

Module koennen dem Dokumente-Generator Bloecke, Platzhalter und Daten beisteuern — Contract-Muster wie Import/Export (#359-Philosophie). Alle drei Hooks sind optional:

# contracts.py des Plugins
class MyContract:
    @staticmethod
    def document_entity_types() -> list[str]:
        return ["myentity"]

    @staticmethod
    def document_placeholders(entity_type: str) -> list[dict]:
        """[{key, label, example}] — speist Editor-Palette + Preview-Defaults."""
        if entity_type != "myentity":
            return []
        return [{"key": "title", "label": "Titel", "example": "Beispiel AG"}]

    @staticmethod
    async def document_data(db, tenant_id, entity_id, entity_type: str) -> dict:
        """Eine Entitaet als Template-Daten laden ({} wenn nicht gefunden)."""
        obj = await db.get(MyModel, entity_id)
        if obj is None or obj.tenant_id != tenant_id:
            return {}
        return {"title": obj.title}

    @staticmethod
    def document_blocks() -> list[dict]:
        """Zusaetzliche Palette-Bloecke (generisch als Key-Value-Tabelle gerendert)."""
        return [{"type": "myentity_summary", "label": "Entitaets-Uebersicht", "category": "modul", "fields": ["title", "status"]}]

Regeln:

• Rendering uebernimmt report_generator (document_renderer.py) — Module liefern NIE Markup (XSS/SSRF-Sandbox bleibt intakt)
• document_placeholders-Beispiele dienen als Preview-Fallbacks (StrictUndefined-vermeidend)
• Built-in-Bloecke duerfen nicht ueberschrieben werden (Contributions mit existierendem Typ werden ignoriert)
• Bilder laufen ausschliesslich als DocumentAsset/data:-URI (WeasyPrint-URL-Fetcher blockt extern)