pip install bizsupply-sdk

result = await self.prompt_llm(
    prompt,
    file_data=None,
    mime_type=None,
    schema=None,
)

result = await self.prompt_llm(
    prompt="Is this an invoice? Return JSON: {\"is_invoice\": true/false}"
)
if result:
    is_invoice = result.get("is_invoice")

result = await self.prompt_llm(
    prompt="Extract invoice data from this document.",
    file_data=file_data,
    mime_type=mime_type,
)

from pydantic import BaseModel

class InvoiceData(BaseModel):
    vendor_name: str
    total: float
    date: str

result = await self.prompt_llm(
    prompt="Extract vendor, total, and date from this invoice.",
    file_data=file_data,
    mime_type=mime_type,
    schema=InvoiceData,
)
# result is a dict matching the schema: {"vendor_name": "...", "total": ..., "date": "..."}

template = await self.get_prompt(prompt_id)

prompt_id = configs.get("classification_prompt_id")
template = await self.get_prompt(prompt_id)

prompt = template.format(
    labels=available_labels,
    document_content="[See attached document file]",
)

result = await self.prompt_llm(prompt=prompt, file_data=file_data, mime_type=mime_type)

fields_json = self.format_fields_for_prompt(fields)

[
  {"name": "invoice_total", "type": "number", "description": "Total amount due"},
  {"name": "vendor_name", "type": "string", "description": "Name of the vendor"}
]

async def extract(self, document, file_data, mime_type, fields, configs):
    fields_json = self.format_fields_for_prompt(fields)

    result = await self.prompt_llm(
        prompt=f"Extract these fields:\n{fields_json}",
        file_data=file_data,
        mime_type=mime_type,
    )

    return ExtractionResult(data=result or {})

from bizsupply_sdk import Document

# In classify() or extract()
self.logger.info(f"Processing {document.document_id}: {document.original_filename}")
self.logger.info(f"Labels: {document.labels}")
self.logger.info(f"Existing data: {document.data}")
self.logger.info(f"Metadata: {document.metadata}")

from bizsupply_sdk import ExtractionResult

return ExtractionResult(
    data={
        "invoice_total": 1500.00,
        "vendor_name": "ACME Corp",
        "invoice_date": "2025-01-15",
    },
    llm_fields=["invoice_total", "vendor_name", "invoice_date"],
    document_type="invoice",
)

from bizsupply_sdk import DocumentInput

yield DocumentInput(
    file_data=pdf_bytes,
    filename="invoice_123.pdf",
    mime_type="application/pdf",
    metadata={
        "source": "gmail",
        "sender": "[email protected]",
        "subject": "Invoice #123",
        "received_at": "2025-01-15T10:30:00Z",
    },
)

from bizsupply_sdk import BaseSourceState

class MySourceState(BaseSourceState):
    cursor: str | None = None
    last_sync_time: str | None = None
    page: int = 0
    processed_ids: list[str] = []

async def fetch(self, credentials, state, configs):
    self.logger.info(f"Resuming from cursor: {state.cursor}")

    for item in items:
        yield DocumentInput(...)
        state.cursor = item.id      # Engine saves after this yield
        state.page += 1

from bizsupply_sdk import DynamicCredential

async def fetch(self, credentials, state, configs):
    # Validate required fields early (raises ValueError if missing)
    credentials.validate_required_fields(["api_key", "api_url"])

    # Attribute access (raises AttributeError if missing)
    api_key = credentials.api_key
    api_url = credentials.api_url

    # .get() with default (returns default if missing)
    timeout = int(credentials.get("timeout_seconds", "30"))

    # Check if optional field exists and is non-empty
    if credentials.has_field("refresh_token"):
        # Handle OAuth refresh
        ...

    # List all available fields
    self.logger.info(f"Available credential fields: {credentials.keys()}")

from bizsupply_sdk import OntologyField

# In extract(), fields are injected by the Engine
for field in fields:
    self.logger.info(f"Field: {field.name} ({field.dtype}): {field.description}")

# Use format_fields_for_prompt() to convert to JSON for prompts
fields_json = self.format_fields_for_prompt(fields)

from bizsupply_sdk import OntologyNode

from bizsupply_sdk import OntologyManifest

from bizsupply_sdk import ExtendedDocument

def score(self, document):
    # Access fields by semantic name
    contract_value = document.get("contract_value")

    # Access related documents (linked by MATCH_RULES)
    for invoice in document.aggregations:
        price = invoice.get("price_per_kwh")
        supplier = invoice.get("supplier_name")

    # Escape hatch for raw dict access
    raw_data = document.raw
    doc_id = raw_data.get("document_id")

from bizsupply_sdk import ScoredDocument

def compute(self, results):
    # results is list[ScoredDocument]
    # Simple: best (lowest) score
    return min(r.score for r in results)

    # Weighted by volume:
    total_weight = sum(r.document.get("volume") or 1 for r in results)
    return sum(
        r.score * (r.document.get("volume") or 1) for r in results
    ) / total_weight

from bizsupply_sdk import MatchRule

from bizsupply_sdk import MatchRule, MatchCondition

MATCH_RULES = [
    MatchRule(
        name="contract_invoice_match",
        left_group=["contract", "energy"],
        right_group=["invoice", "energy"],
        description="Link energy contracts to their invoices",
        conditions=[
            MatchCondition(
                left_field="client_tax_id",
                right_field="client_tax_id",
                match_type="==",
            ),
            MatchCondition(
                left_field="cpe_point_of_delivery",
                right_field="cpe_point_of_delivery",
                match_type="==",
            ),
        ],
    ),
]

from bizsupply_sdk import MatchCondition

# Plugin base classes
from bizsupply_sdk import ClassificationPlugin, ExtractionPlugin, SourcePlugin

# Benchmark base class
from bizsupply_sdk import BaseBenchmark

# Core models (plugins)
from bizsupply_sdk import Document, ExtractionResult, DocumentInput
from bizsupply_sdk import BaseSourceState, DynamicCredential
from bizsupply_sdk import OntologyField, OntologyNode, OntologyManifest

# Benchmark models
from bizsupply_sdk import ExtendedDocument, ScoredDocument
from bizsupply_sdk import MatchRule, MatchCondition

# Protocol (for type hints)
from bizsupply_sdk import PluginServicesProtocol