Plugins

Since v0.2.0, intake uses an extensible architecture based on plugins. Parsers, exporters and connectors are discovered automatically via Python entry_points (PEP 621).

How It Works

Automatic Discovery

intake uses importlib.metadata.entry_points() to discover installed plugins. Each plugin is registered under an entry_points group:

Group	Type	Example
`intake.parsers`	Input format parsers	`markdown = "intake.ingest.markdown:MarkdownParser"`
`intake.exporters`	Output format exporters	`architect = "intake.export.architect:ArchitectExporter"`
`intake.connectors`	Direct API connectors	(empty — preparation for Phase 2)

When intake loads a registry (ParserRegistry, ExporterRegistry), it attempts to discover plugins first. If discovery fails, it falls back to manual (hardcoded) registration.

Verify Installed Plugins

# List all discovered plugins
intake plugins list

# With details (module, load errors)
intake plugins list -v

# Verify compatibility
intake plugins check

Built-in Plugins

intake includes 13 built-in plugins registered as entry_points in pyproject.toml:

Parsers (11)

Name	Module	Format
`markdown`	`intake.ingest.markdown:MarkdownParser`	`.md`, `.markdown`
`plaintext`	`intake.ingest.plaintext:PlaintextParser`	`.txt`, stdin
`yaml`	`intake.ingest.yaml_input:YamlInputParser`	`.yaml`, `.yml`, `.json`
`pdf`	`intake.ingest.pdf:PdfParser`	`.pdf`
`docx`	`intake.ingest.docx:DocxParser`	`.docx`
`jira`	`intake.ingest.jira:JiraParser`	`.json` (auto-detected)
`confluence`	`intake.ingest.confluence:ConfluenceParser`	`.html` (auto-detected)
`image`	`intake.ingest.image:ImageParser`	`.png`, `.jpg`, `.webp`, `.gif`
`url`	`intake.ingest.url:UrlParser`	`http://`, `https://`
`slack`	`intake.ingest.slack:SlackParser`	`.json` (auto-detected)
`github_issues`	`intake.ingest.github_issues:GithubIssuesParser`	`.json` (auto-detected)

Exporters (2)

Name	Module	Format
`architect`	`intake.export.architect:ArchitectExporter`	`pipeline.yaml` + spec
`generic`	`intake.export.generic:GenericExporter`	`SPEC.md` + `verify.sh` + spec

Protocols

intake defines two generations of protocols:

V1 — Core Protocols

V1 protocols are the original system protocols. They are minimal and sufficient for simple parsers and exporters:

# Parser V1
@runtime_checkable
class Parser(Protocol):
    def can_parse(self, source: str) -> bool: ...
    def parse(self, source: str) -> ParsedContent: ...

# Exporter V1
@runtime_checkable
class Exporter(Protocol):
    def export(self, spec_dir: str, output_dir: str) -> list[str]: ...

All built-in parsers and exporters use V1. The registries accept both V1 and V2.

V2 — Plugin Protocols

V2 protocols extend V1 with metadata, discovery capabilities, and additional functionality. They are designed for external plugins:

@runtime_checkable
class ParserPlugin(Protocol):
    @property
    def meta(self) -> PluginMeta: ...

    @property
    def supported_extensions(self) -> set[str]: ...

    def confidence(self, source: str) -> float: ...
    def can_parse(self, source: str) -> bool: ...
    def parse(self, source: str) -> ParsedContent: ...

@runtime_checkable
class ExporterPlugin(Protocol):
    @property
    def meta(self) -> PluginMeta: ...

    @property
    def supported_agents(self) -> list[str]: ...

    def export(self, spec_dir: str, output_dir: str) -> ExportResult: ...

@runtime_checkable
class ConnectorPlugin(Protocol):
    @property
    def meta(self) -> PluginMeta: ...

    @property
    def uri_schemes(self) -> list[str]: ...

    def can_handle(self, uri: str) -> bool: ...
    async def fetch(self, uri: str) -> list[FetchedSource]: ...
    def validate_config(self) -> list[str]: ...

Support Dataclasses

@dataclass
class PluginMeta:
    name: str
    version: str
    description: str
    author: str

@dataclass
class ExportResult:
    files_created: list[str]
    primary_file: str
    instructions: str

@dataclass
class FetchedSource:
    local_path: str
    original_uri: str
    format_hint: str
    metadata: dict[str, str]

Creating an External Plugin

Step 1: Create the Package

mi-plugin-intake/
├── pyproject.toml
└── src/
    └── mi_plugin/
        ├── __init__.py
        └── parser.py

Step 2: Implement the Protocol

# src/mi_plugin/parser.py
from __future__ import annotations

from intake.ingest.base import ParsedContent


class AsanaParser:
    """Parser for Asana JSON exports."""

    def can_parse(self, source: str) -> bool:
        # Detect if the file is an Asana export
        ...

    def parse(self, source: str) -> ParsedContent:
        # Parse the file and return normalized ParsedContent
        ...

You can implement either the V1 (Parser) or V2 (ParserPlugin) protocol. The registry accepts both.

Step 3: Register as an entry_point

# pyproject.toml
[project.entry-points."intake.parsers"]
asana = "mi_plugin.parser:AsanaParser"

Step 4: Install

pip install mi-plugin-intake

The plugin is discovered automatically:

intake plugins list    # "asana" should appear
intake plugins check   # should report OK

Now you can use Asana files as a source:

intake init "Sprint review" -s asana-export.json

Hooks

The HookManager allows registering callbacks that run in response to pipeline events:

from intake.plugins.hooks import HookManager

manager = HookManager()

# Register a callback
def on_parse_complete(data: dict) -> None:
    print(f"Parsed: {data['source']}")

manager.register("parse_complete", on_parse_complete)

# Emit an event
manager.emit("parse_complete", {"source": "reqs.md", "format": "markdown"})

Features

Callbacks run in registration order
Exceptions are caught and logged without blocking other callbacks
registered_events returns the names of events with registered callbacks
Ready for wiring in future pipeline phases

PluginRegistry API

from intake.plugins.discovery import PluginRegistry

registry = PluginRegistry()

# Discover all plugins
registry.discover_all()

# Discover by group
registry.discover_group("intake.parsers")

# Get plugins by type
parsers = registry.get_parsers()       # dict[str, object]
exporters = registry.get_exporters()   # dict[str, object]
connectors = registry.get_connectors() # dict[str, object]

# List plugin information
for info in registry.list_plugins():
    print(f"{info.name} ({info.group}) v{info.version} - V2: {info.is_v2}")

# Verify compatibility
issues = registry.check_compatibility(info)

PluginInfo

Field	Type	Description
`name`	str	Plugin name
`group`	str	entry_point group
`module`	str	Python module
`distribution`	str	Package that provides it
`version`	str	Package version
`is_builtin`	bool	Whether it is a built-in intake plugin
`is_v2`	bool	Whether it implements the V2 protocol
`load_error`	str \| None	Load error (if it failed)

Exceptions

Exception	Description
`PluginError`	Base plugin error
`PluginLoadError`	Failed to load a plugin (import error, module not found)