Add a source

Create a direct package under backend/app/sync_v2/sources. This minimal contract-complete example emits one movie. Replace the fixture data with batched provider I/O through context.http.

backend/app/sync_v2/sources/example_source/
├── __init__.py
└── plugin.py

sources/example_source/plugin.py

from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Mapping, Sequence

from app.sync_v2.sdk import (
    ConfigField,
    ConfigFieldType,
    ConfigSchema,
    SourceContext,
    SourceItem,
    SourceManifest,
    SourceRuntime,
    SourceScanPage,
    SourceScanRequest,
    SourceScopeDefinition,
    SourceScopeOption,
)


class ExampleSourceRuntime(SourceRuntime):
    def __init__(self, context: SourceContext) -> None:
        self.context = context

    async def list_scope_options(
        self, scope_key: str
    ) -> Sequence[SourceScopeOption]:
        if scope_key != "libraries":
            return ()
        return (SourceScopeOption("main", "Main library"),)

    async def scan_items(self, request: SourceScanRequest) -> SourceScanPage:
        if "main" not in request.scope.get("libraries", ()):
            return SourceScanPage(items=())
        item = SourceItem(
            item_key=f"{self.context.source_id}:movie-42",
            raw_item_key="movie-42",
            canonical_media_key="movie:imdb:tt0133093",
            title="Example movie",
            media_type="movie",
            identifiers={"imdb": "tt0133093"},
            fields={
                "watchlist": True,
                "playback_status": "unplayed",
                "rating": str(Decimal("0.8")),
            },
            diagnostics={"library": "Main library"},
        )
        return SourceScanPage(items=(item,))


@dataclass(frozen=True, slots=True)
class ExampleSourcePlugin:
    manifest = SourceManifest(
        source_type="example_source",
        label="Example Source",
        plugin_version="1.0.0",
        supported_media_types=("movie",),
        auth_modes=("none",),
        capabilities=frozenset({
            "watchlist.observe",
            "playback_status.observe",
            "rating.observe",
        }),
        resources={"website": "https://example.com/"},
        scopes=(SourceScopeDefinition("libraries", "Libraries"),),
    )
    instance_schema = ConfigSchema(
        title="Example Source connection",
        fields=(ConfigField(
            "base_url",
            ConfigFieldType.STRING,
            "Server URL",
            required=True,
            pattern=r"https?://.+",
        ),),
    )
    features = ()
    auth = None
    actions = ()

    def create_runtime(self, context: SourceContext) -> SourceRuntime:
        return ExampleSourceRuntime(context)

    async def run_action(
        self, context: SourceContext, key: str, params: Mapping[str, Any]
    ) -> Mapping[str, Any]:
        raise ValueError(f"Unknown Example Source action: {key}")


def create_plugin() -> ExampleSourcePlugin:
    return ExampleSourcePlugin()

Canonical observation checklist

• item_key is stable inside the source instance and prefixes the instance when needed.
• raw_item_key is the provider’s stable media key.
• canonical_media_key identifies the same logical media across duplicate copies.
• identifiers contains provider IDs already supplied by the source.
• fields uses provider-neutral observations required by declared capabilities.
• extensions is reserved for JSON-safe provider data and every key is namespaced.
• mapping_descriptors supplies opaque catalog lookup material when direct IDs are insufficient.
• ratings are canonical 0..1 values represented exactly.

Do not derive target decisions in the source. For example, emit playback status and watchlist observations; let the profile’s target feature decide whether they mean Planning, Watched, diary, or no operation.

Optional contracts

Add a capability and override the matching runtime method together:

• incremental cursors with changes.read;
• provider events with events.parse and optional coalescing/preflight hooks;
• exact playback history with view_history.events;
• cached source artwork with artwork.read;
• source collections with collections.read and collections.members.read.

When event parsing is supported, return changed canonical items and changed_capabilities; do not name a concrete target feature in the source.

Verification

Add tests for pagination, scope filtering, canonical fields, duplicate copies, history completeness, event parsing, cancellation, provider errors, and secret-safe diagnostics. Refresh discovery and confirm the source appears in Sources without changing an API route or frontend registry.