Plugin protocol

Plugins are Python classes implementing BasePlugin. The framework discovers them at startup from a registry list.

Class#

python
from app.plugins.base import BasePlugin
from app.plugins.protocol import (
    PluginContext,
    PluginManifest,
    PluginPermission,
    ToolDefinition,
)

class MyPlugin(BasePlugin):
    @property
    def manifest(self) -> PluginManifest: ...

    async def get_system_context(self, ctx: PluginContext) -> str | None: ...

    async def execute_tool(
        self, tool_name: str, args: str, ctx: PluginContext,
    ) -> str: ...

    # Optional:
    async def get_dynamic_tools(self, ctx: PluginContext) -> list[dict] | None:
        """Return additional function-calling tool defs at runtime
        (e.g. user-installed skills)."""

PluginManifest#

python
@dataclass(frozen=True)
class PluginManifest:
    id: str                               # "namespace.name", unique
    name: str                             # display
    version: str                          # semver
    description: str
    tools: list[ToolDefinition] = ...
    permissions: list[PluginPermission] = field(default_factory=list)
    injects_context: bool = False         # if True, AI sees get_system_context output
    scope: str | None = None              # "private" = caller's data only, etc.

ToolDefinition#

python
@dataclass(frozen=True)
class ToolDefinition:
    name: str                             # function-call name
    description: str                      # what + when to use
    parameters: dict                      # OpenAI function-calling schema
    usage: str | None = None              # "[/foo <bar>]" hint for text-mode UIs
    permissions_required: list[PluginPermission] = field(default_factory=list)
    side_effects: bool = False            # True → routed through approval flow

PluginPermission#

python
class PluginPermission(str, Enum):
    READ_MESSAGES = "read_messages"
    WRITE_MESSAGES = "write_messages"
    READ_NOTES = "read_notes"
    WRITE_NOTES = "write_notes"
    READ_USERS = "read_users"
    EXTERNAL_HTTP = "external_http"
    SEARCH = "search"
    MANAGE_TASKS = "manage_tasks"

PluginContext#

python
@dataclass
class PluginContext:
    db: AsyncSession
    tenant: TenantContext
    settings: Settings
    redis: aioredis.Redis | None
    room_id: str
    sender_id: str                        # AI participant making the call
    user_token: str | None = None         # human user's JWT (for token exchange)
    scaigrid: ScaiGridClient | None = None
    arq: ArqRedis | None = None
    approval_service: ApprovalService | None = None
    event_service_factory: Callable | None = None

Tool result conventions#

Return a string. Markdown is fine. Empty string == "no result".
For errors, return a friendly message ending with a "Fallback suggestions" line (the AI's tool-etiquette rule expects this shape).

Example:

python
return (
    "Drive search failed: HTTP 401.\n"
    "Fallback suggestions: `find_note` for tenant-private notes; "
    "`web_search` for public info."
)

Registering#

Add to app/plugins/builtin/__init__.py:

python
from app.plugins.builtin.my_plugin import MyPlugin

BUILTIN_PLUGINS = [
    # … existing entries …
    MyPlugin(),
]

Restart the API; the plugin is discoverable in GET /v1/admin/plugins. Enable it with:

bash
curl -X POST "$BASE/v1/admin/plugins/<plugin-id>/enable" \
  -H "Authorization: Bearer $ADMIN_TOKEN"

Lifecycle hooks (optional)#

python
class MyPlugin(BasePlugin):
    async def on_install(self, ctx: PluginContext) -> None:
        """Called once when an admin enables the plugin for a tenant."""

    async def on_uninstall(self, ctx: PluginContext) -> None:
        """Called once when an admin disables."""

Testing#

The db and tenant_context fixtures in tests/conftest.py give you everything you need:

python
async def test_my_plugin(db, tenant_context, settings):
    plugin = MyPlugin()
    ctx = PluginContext(
        db=db, tenant=tenant_context, settings=settings,
        redis=None, room_id="r", sender_id="ai",
    )
    out = await plugin.execute_tool("my_tool", '{"x": 1}', ctx)
    assert "expected" in out