Create a Source Plugin

Goal: Build a plugin that ingests documents from external sources (email, APIs, file systems)

Last updated: 2026-07-21

Goal: Build a plugin that ingests documents from external sources (email, APIs, file systems)

Use When: You need to automatically pull documents from Salesforce, Dropbox, custom APIs, or any external system.

What Source Plugins Do

Source plugins connect to external systems, fetch documents, and yield them for the Engine to create. Unlike other plugins that process existing documents, source plugins create new documents.

Key Method: Yield DocumentInput from fetch() - the Engine handles document creation, state persistence, and credential management.

Prerequisites

Before starting:

Install the SDK: pip install bizsupply-sdk
Read Plugin Interface Specification
Decide what credentials your source needs (API keys, OAuth tokens, etc.)

How Source Plugins Work

1. Engine loads credentials from secure storage -> injects as DynamicCredential
2. Engine loads your typed state model from database -> injects as state
3. Engine calls fetch(credentials, state, configs)
4. For each DocumentInput you yield, Engine creates the document
5. Engine auto-saves state after each document (crash-resilient)

Step 1: Define Your State Model

Source plugins use a typed state model to track sync progress between runs:

python

from bizsupply_sdk import BaseSourceState


class MySourceState(BaseSourceState):
    """State model for tracking sync progress."""
    last_item_id: str | None = None
    last_sync_time: str | None = None
    page: int = 0

The Engine loads this state automatically and saves it after each yielded document.

Step 2: Create the Plugin Code

Create my_source.py:

python

from datetime import datetime, timezone

from bizsupply_sdk import (
    SourcePlugin,
    BaseSourceState,
    DocumentInput,
    DynamicCredential,
)


class MySourceState(BaseSourceState):
    """State model for tracking sync progress."""
    last_item_id: str | None = None
    last_sync_time: str | None = None


class MySourcePlugin(SourcePlugin):
    """
    Ingests documents from a custom API.

    The Engine handles credentials, state, and document creation.
    Just yield DocumentInput objects from fetch()!
    """

    # REQUIRED: Unique identifier for this source type
    source_type = "my_custom_api"

    # REQUIRED: State model class for tracking sync progress
    source_state_model = MySourceState

    # Optional: Credentials the user must configure for this source
    credential_fields = ["api_url", "api_key", "timeout_seconds"]

    # Optional: Configurable parameters
    configurable_parameters = [
        {
            "parameter_name": "max_documents",
            "parameter_type": "int",
            "default_value": 100,
            "description": "Maximum documents to fetch per run",
        },
        {
            "parameter_name": "batch_size",
            "parameter_type": "int",
            "default_value": 50,
            "description": "Number of items per API request",
        },
    ]

    async def fetch(self, credentials, state, configs):
        """
        Fetch documents from external source.

        Args:
            credentials: DynamicCredential with configured fields
            state: MySourceState loaded from database
            configs: Runtime configuration (see Engine-Injected Configs below)

        Yields:
            DocumentInput for each document to create
        """
        self.logger.info("Starting document ingestion")

        # Access credentials by attribute or .get()
        api_url = credentials.api_url
        api_key = credentials.api_key
        timeout = int(credentials.get("timeout_seconds", 30))

        self.logger.info(f"Connecting to: {api_url}")
        self.logger.info(f"Last synced item: {state.last_item_id}")

        max_documents = int(configs.get("max_documents", 100))

        # =========================================================
        # YOUR FETCH LOGIC HERE
        # =========================================================
        # Replace with your actual API/service fetching code
        # Example:
        #
        # import httpx
        # async with httpx.AsyncClient() as client:
        #     response = await client.get(
        #         f"{api_url}/documents",
        #         headers={"Authorization": f"Bearer {api_key}"},
        #         params={"after": state.last_item_id},
        #         timeout=timeout,
        #     )
        #     items = response.json()["documents"]

        # Placeholder - replace with real fetch logic
        items = []

        count = 0
        for item in items:
            if count >= max_documents:
                break

            try:
                file_bytes = item.get("content")  # bytes
                filename = item.get("filename", "document.pdf")

                # Yield DocumentInput - Engine creates the document
                yield DocumentInput(
                    file_data=file_bytes,
                    filename=filename,
                    metadata={
                        "source": "my_api",
                        "external_id": item.get("id"),
                        "title": item.get("title"),
                        "created_at": item.get("created_at"),
                    },
                )

                # Update state - Engine auto-saves after each yield
                state.last_item_id = item.get("id")
                state.last_sync_time = datetime.now(timezone.utc).isoformat()
                count += 1

            except Exception as e:
                self.logger.error(f"Failed to process item: {e}")
                continue

        self.logger.info(f"Ingestion complete. Yielded {count} documents")

    async def has_new_data(self, credentials, state, configs):
        """
        Check if source has new data since last sync.

        Used by auto-sync to decide whether to trigger a full fetch.
        Should be lightweight (e.g., HEAD request or count query).
        """
        # Implement a lightweight check
        # Example: return await check_api_for_new_items(credentials, state)
        return True

Step 3: Validate and Register

Validate your plugin before registering:

bash

bizsupply validate my_source.py

Then register:

bash

curl -X POST "https://api.bizsupply.ai/api/v1/plugins" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -F "name=My API Source" \
  -F "description=Ingests documents from My Custom API" \
  -F "code=@my_source.py"

When you register:

The plugin is AST-validated (no exec() at this point)
Plugin type, source_type, credential_fields, and other metadata are automatically extracted from your code
The submission is held in the review queue — no bypass for any plugin type

Review workflow

Plugins execute inside an isolated execution process on the platform, so every submission — including from platform administrators — is held in a review queue until approved by a platform administrator. The response tells you where your submission landed:

json

{"review_id": "...", "status": "pending_review"}

On approval (via POST /api/v1/reviews/{review_id}/approve by a platform administrator):

source_type is registered (if new) with the declared credential schema
The plugin becomes active under the same ID that was allocated at submit time
Users can then connect sources of that type through the bizsupply UI

Rejected submissions are never registered — the reviewer gives a rejection reason which is returned to the submitter. No bypass — this applies to all plugin types. See Create a Plugin for the full review flow.

Step 4: Configure Source Credentials

After registration, users configure credentials through the bizsupply UI:

Go to Settings > Sources > Add Source
Select your source type
Fill in the credential fields you defined
Save

The credentials are securely stored and injected into your plugin via the credentials parameter.

Key Methods

Method	Purpose
`await self.prompt_llm(prompt, file_data, mime_type)`	Call LLM (if needed)
`await self.get_prompt(prompt_id)`	Load prompt template (if needed)
`self.logger`	Plugin-specific logger

The Engine handles credential injection, state loading/saving, and document creation automatically.

Accessing Credentials

Credentials are injected as a DynamicCredential object. Access fields by attribute or .get():

python

async def fetch(self, credentials, state, configs):
    # Attribute access (raises AttributeError if missing)
    api_key = credentials.api_key
    api_url = credentials.api_url

    # .get() access (returns None or default if missing)
    timeout = credentials.get("timeout_seconds", 30)

    # Validate required fields early
    credentials.validate_required_fields(["api_key", "api_url"])

    # Check if a field exists
    if credentials.has_field("refresh_token"):
        # Handle OAuth refresh
        ...

All credential values are stored as strings. Convert to other types in your plugin code as needed:

python

timeout = int(credentials.get("timeout_seconds", "30"))
enabled = credentials.get("enabled") == "true"

State Management

State is injected as your typed model and auto-saved by the Engine after each yielded document:

python

class MySourceState(BaseSourceState):
    cursor: str | None = None
    page: int = 0

async def fetch(self, credentials, state, configs):
    # State is loaded from database with previous values
    self.logger.info(f"Resuming from cursor: {state.cursor}")

    for item in items:
        yield DocumentInput(...)

        # Mutate state directly - Engine auto-saves
        state.cursor = item.id
        state.page += 1

If the plugin crashes mid-run, the Engine resumes from the last saved state.

Engine-Injected Configs

The configs dict passed to fetch() and has_new_data() contains your plugin's configurable_parameters values, plus optional source-level settings injected by the Engine. These are user-configured per source and available to any source plugin automatically:

Key	Type	Injected When	Description
`oldest_document_date`	`str` (ISO 8601)	User set a date cutoff on the source	Skip documents older than this date. Parse with `datetime.fromisoformat()`.
`folder_filter`	`list[dict]` (`[{id, name}]`)	User selected specific folders/labels	Restrict to these folders. Each entry has `id` (provider-specific) and `name` (display).

Both are optional — if not set by the user, they won't be in configs. Always use .get():

python

async def fetch(self, credentials, state, configs):
    # Optional: date cutoff (skip old documents)
    oldest_document_date_str = configs.get("oldest_document_date")
    if oldest_document_date_str:
        cutoff = datetime.fromisoformat(oldest_document_date_str)
        # Apply cutoff to your API query...

    # Optional: folder/label filter (scan specific folders only)
    folder_filter = configs.get("folder_filter")
    if folder_filter:
        folder_ids = [f["id"] for f in folder_filter if f.get("id")]
        # Apply folder_ids to your API query...

These configs are also injected into has_new_data() — apply the same filters there so the sync check matches the fetch scope.

Document Metadata

When yielding documents, include relevant metadata:

python

yield DocumentInput(
    file_data=file_bytes,
    filename="invoice_123.pdf",
    mime_type="application/pdf",  # Optional: auto-detected if not provided
    metadata={
        "source": "my_api",
        "external_id": "doc_123",
        "title": "Invoice #456",
        "author": "vendor@acme.com",
        "created_at": "2025-01-15",
    },
)

Common Mistakes

Using Old execute() Method

python

# WRONG - old v1.0 API
async def execute(self, context: PluginContext):
    credentials = self.get_source_credentials()
    state = await self.get_source_state()
    for item in items:
        doc = await self.create_document(file_data, filename, metadata)
    await self.update_source_state(state)
    return created_documents

# CORRECT - yield DocumentInput from fetch()
async def fetch(self, credentials, state, configs):
    for item in items:
        yield DocumentInput(file_data=..., filename=...)
        state.cursor = item.id  # Engine auto-saves

Missing sourcestatemodel

python

# WRONG - no state model defined
class MySource(SourcePlugin):
    source_type = "my_api"
    # Missing source_state_model!

# CORRECT
class MySource(SourcePlugin):
    source_type = "my_api"
    source_state_model = MySourceState

Hardcoding Credentials

python

# WRONG - security risk
api_key = "sk-1234567890"

# CORRECT - use injected credentials
api_key = credentials.api_key

Missing SDK Import

python

# WRONG
class MyPlugin(SourcePlugin):  # NameError!
    ...

# CORRECT
from bizsupply_sdk import SourcePlugin, BaseSourceState, DocumentInput, DynamicCredential

class MyPlugin(SourcePlugin):
    ...

Example: OAuth Mailbox Source (Gmail / Outlook)

Mailbox sources frequently let users select multiple folders or labels. Each folder must keep its own resume cursor — a single global cursor would always follow whichever folder had the newest activity, causing quieter folders' new messages to be excluded by the next sync's after filter.

The pattern: model state as a dict of per-folder cursors, keyed by the provider's folder/label id.

python

from datetime import datetime

from pydantic import BaseModel, Field

from bizsupply_sdk import SourcePlugin, BaseSourceState, DocumentInput, DynamicCredential


class MailFolderCursor(BaseModel):
    """Per-folder sync cursor."""
    last_message_id: str | None = None
    last_received_at: datetime | None = None
    last_email_index: int = 0


class GmailState(BaseSourceState):
    """One cursor per scanned Gmail label.

    Use a sentinel key (e.g. ``"__all__"``) when no folder filter is set,
    so single-scope and multi-folder configurations share the same shape.
    """
    cursors: dict[str, MailFolderCursor] = Field(default_factory=dict)


class GmailSourcePlugin(SourcePlugin):
    source_type = "gmail"
    source_state_model = GmailState
    credential_fields = ["access_token", "refresh_token", "token_uri"]

    async def list_items(self, credentials, state, configs):
        # For each folder, fetch only messages newer than that folder's cursor.
        for folder_id in self._folders_to_scan(configs):
            cursor = state.cursors.get(folder_id) or MailFolderCursor()
            async for msg_ref in self._list_for_folder(folder_id, cursor, credentials):
                cursor.last_message_id = msg_ref["id"]
                cursor.last_received_at = msg_ref["received_at"]
                cursor.last_email_index += 1
                state.cursors[folder_id] = cursor
                yield msg_ref

When advancing the watermark, advance only that folder's cursor — never a single global field.

Example: Microsoft 365 Source (OAuth, Microsoft Graph API)

The built-in Microsoft365Plugin follows the same per-folder-cursor pattern. Outlook folders are addressed by Graph folder id (resolve display names to ids once at the top of list_items):

python

from datetime import datetime

from pydantic import BaseModel, Field

from bizsupply_sdk import SourcePlugin, BaseSourceState, DocumentInput, DynamicCredential


class MicrosoftFolderCursor(BaseModel):
    last_message_id: str | None = None
    last_received_at: datetime | None = None
    last_email_index: int = 0


class Microsoft365State(BaseSourceState):
    cursors: dict[str, MicrosoftFolderCursor] = Field(default_factory=dict)


class Microsoft365Plugin(SourcePlugin):
    source_type = "microsoft_365"
    source_state_model = Microsoft365State
    credential_fields = ["access_token", "refresh_token"]

    async def list_items(self, credentials, state, configs):
        async with httpx.AsyncClient() as client:
            for folder_id in await self._resolve_folders(client, configs):
                cursor = state.cursors.get(folder_id) or MicrosoftFolderCursor()
                # Use cursor.last_received_at in the OData $filter so the
                # quieter folder isn't masked by a noisier one's progress.
                async for msg_ref in self._list_for_folder(client, folder_id, cursor):
                    cursor.last_message_id = msg_ref["id"]
                    cursor.last_received_at = msg_ref["received_at"]
                    state.cursors[folder_id] = cursor
                    yield msg_ref

    async def has_new_data(self, credentials, state, configs):
        # Return True as soon as any folder has a message past its own cursor.
        for folder_id in self._configured_folders(configs):
            cursor = state.cursors.get(folder_id) or MicrosoftFolderCursor()
            if await self._folder_has_new(folder_id, cursor):
                return True
        return False

Key patterns:

Use httpx.AsyncClient for async HTTP (avoids blocking the event loop).
Resolve folder display names to opaque Graph ids once per run; key cursors by id.
Cross-folder dedup if the same message can appear in two folders (Microsoft: by internetMessageId).
/me/ Graph endpoints for delegated auth — the Engine injects tokens for the connected user.
The platform handles OAuth consent and token refresh automatically.

Example: Google Drive Source (Incremental Sync with Page Tokens)

The built-in GoogleDriveSourcePlugin demonstrates two advanced patterns: incremental sync using the Drive Changes API page token, and LLM-based relevance filtering before yielding documents.

python

from bizsupply_sdk import SourcePlugin, BaseSourceState, DocumentInput, DynamicCredential


class GoogleDriveSourceState(BaseSourceState):
    """State model for tracking incremental sync progress.

    changes_page_token is the key to efficient incremental sync — on first run
    all files are indexed and the token is saved. Subsequent runs use the token
    to retrieve only files added, modified, or deleted since the last sync.
    """
    changes_page_token: str | None = None  # Drive Changes API page token
    last_synced_at: str | None = None
    total_files_processed: int = 0


class GoogleDriveSourcePlugin(SourcePlugin):
    source_type = "google_drive"
    source_state_model = GoogleDriveSourceState
    credential_fields = ["access_token", "refresh_token"]

    async def fetch(self, credentials, state, configs):
        access_token = credentials.access_token

        if state.changes_page_token is None:
            # First run — full index: list all files, save start page token
            files = await self._list_all_files(access_token)
            state.changes_page_token = await self._get_start_page_token(access_token)
        else:
            # Subsequent runs — only changed files since last token
            files, state.changes_page_token = await self._get_changes(
                access_token, state.changes_page_token
            )

        for file_meta in files:
            file_data = await self._download_file(access_token, file_meta)
            if file_data is None:
                continue
            yield DocumentInput(
                file_data=file_data,
                filename=file_meta["name"],
                metadata={"source": "google_drive", "file_id": file_meta["id"]},
            )
            state.last_synced_at = ...  # Engine auto-saves state after each yield
            state.total_files_processed += 1

    async def has_new_data(self, credentials, state, configs):
        # Lightweight check: compare current Changes API token to saved token
        if state.changes_page_token is None:
            return True  # Never synced
        current_token = await self._get_start_page_token(credentials.access_token)
        return current_token != state.changes_page_token

Key patterns from this implementation:

Page token = sync cursor — save it in state after each run; the Changes API only returns what changed
First-run full index — when changes_page_token is None, fetch everything and store the starting token
has_new_data() is lightweight — compare the current token against the saved one (no file listing needed)
LLM relevance filter — optional filter_prompt_id config lets users filter out irrelevant files before they enter the pipeline

Example: Salesforce Source

python

from bizsupply_sdk import SourcePlugin, BaseSourceState, DocumentInput, DynamicCredential


class SalesforceState(BaseSourceState):
    last_sync_id: str | None = None


class SalesforceSourcePlugin(SourcePlugin):
    """Ingests documents from Salesforce."""

    source_type = "salesforce"
    source_state_model = SalesforceState
    credential_fields = ["instance_url", "access_token", "api_version"]

    configurable_parameters = [
        {
            "parameter_name": "object_type",
            "parameter_type": "str",
            "default_value": "ContentDocument",
            "description": "Salesforce object to query",
        },
    ]

    async def fetch(self, credentials, state, configs):
        instance_url = credentials.instance_url
        access_token = credentials.access_token
        # ... connect to Salesforce, yield DocumentInput objects

    async def has_new_data(self, credentials, state, configs):
        # Check Salesforce for new items since last sync
        return True

Next Steps

Use Plugins - Execute your plugin in a pipeline
Create a Classification Plugin - Classify ingested documents
Plugin Service API - All available service methods

PLUGINSCreate a Plugin PLUGINSCreate a Classification Plugin