Plugins & SDK

Extend CortexFlow-AI with typed, sandboxed plugins — without installing the full gateway.

Install the SDK

pip install cortexflow-sdk

cortexflow-sdk is a standalone, dependency-free package exposing Plugin, Tool, and ChannelAdapter — everything a plugin author needs, without FastAPI, channel SDKs, or a Qdrant client pulled in.

Plugin types

`plugin_type`	Implement	Contributes
`tool`	`get_tools()`	One or more `Tool` instances
`channel`	`get_channel_adapter()`	A `ChannelAdapter` instance
`tts` / `stt` / `memory`	—	Loaded by name; see gateway internals
`generic`	`on_load()` / `on_unload()`	Lifecycle hooks only

Writing a tool plugin

from cortexflow_sdk import Plugin, PluginMetadata, Tool, ToolResult


class WeatherTool(Tool):
    name = "get_weather"
    description = "Get the current weather for a city."
    parameters = {
        "city": {"type": "str", "description": "City name", "required": True},
    }
    permissions = ["network"]

    async def execute(self, city: str) -> ToolResult:
        return ToolResult(tool=self.name, output=f"Sunny in {city}")


class WeatherPlugin(Plugin):
    metadata = PluginMetadata(
        name="cortexflow-weather",
        version="1.0.0",
        plugin_type="tool",
        description="Adds a get_weather tool.",
        permissions=["network"],
    )

    def get_tools(self):
        return [WeatherTool()]

Registering your plugin

Declare a cortexflow.plugins entry point in your package's pyproject.toml:

[project.entry-points."cortexflow.plugins"]
cortexflow-weather = "my_plugin.plugin:WeatherPlugin"

Once your package is installed in the same environment as the gateway, PluginRegistry.discover() finds it automatically at startup.

Writing a channel adapter plugin

from cortexflow_sdk import ChannelAdapter, InboundMessage


class MyChannelAdapter(ChannelAdapter):
    channel_id = "my_channel"

    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def send(self, target, text, *, reply_to=None, attachments=None):
        ...

    async def _on_platform_event(self, raw_event: dict) -> None:
        await self._dispatch(InboundMessage(
            channel=self.channel_id,
            sender_id=raw_event["user_id"],
            sender_name=raw_event["user_name"],
            text=raw_event["text"],
        ))

Example plugins

Three complete, tested reference plugins — published on PyPI, source in examples/plugins/:

pip install cortexflow-github — source — a github_events tool listing recent repo events
pip install cortexflow-notion — source — a notion_search tool searching Notion pages/databases
pip install cortexflow-google-calendar — source — a calendar_list_events tool listing upcoming events

Each example reads its credential from an environment variable in Plugin.__init__ and hands it to the Tool it constructs — the Tool itself never raises, returning ToolResult.error for network/HTTP/missing-dependency failures.