Building a custom target connector

Connect CocoIndex's declarative target-state system to external systems. Covers TargetHandler, tracking records, actions, sinks, root vs container targets, attachment providers, and idempotency best practices.

Version: v 1.0.14
Last reviewed: Jul 4, 2026

A custom target connector is the mechanism that connects CocoIndex’s declarative target state system to external systems. When you call methods like dir_target.declare_file() or table_target.declare_row(), a target connector handles the actual synchronization — determining what changed and applying those changes to the external system.

When to create a custom target connector

Most users will use built-in connectors (like localfs or postgres) and never need to create their own. Consider creating a custom target connector when:

You need to integrate with an external system not covered by existing connectors
You need custom change detection logic (e.g., content-based fingerprinting)
You need to manage hierarchical target states (containers with children)

Start Simple

For simple use cases where you just need to write data to an external system without sophisticated change tracking, consider using a regular function with memoization instead. Target states providers are most valuable when you need CocoIndex to track and clean up target states automatically.

Key data types

This section introduces the key data types. Each is marked as either you implement or CocoIndex provides to clarify responsibilities.

TargetHandler (you implement)

A TargetHandler implements the reconciliation logic. It’s a protocol with one required method and one optional method:

python

class TargetHandler(Protocol[ValueT, TrackingRecordT, OptChildHandlerT]):
    def reconcile(
        self,
        key: StableKey,
        desired_target_state: ValueT | NonExistenceType,
        prev_possible_records: Collection[TrackingRecordT],
        prev_may_be_missing: bool,
        /,
    ) -> TargetReconcileOutput[Any, TrackingRecordT, OptChildHandlerT] | None:
        ...

    # Optional: override to support attachment types (see "Implementing attachment providers")
    def attachments(self) -> dict[str, TargetHandler]:
        return {}  # Default: no attachments

Type Parameters:

ValueT: The specification for the target state (e.g., file content, row data)
TrackingRecordT: What’s stored to detect changes on future runs
OptChildHandlerT: The child handler type, or None for leaf targets. A container handler declares it as the third argument of its reconcile() return type, TargetReconcileOutput[Action, TrackingRecord, ChildHandler]; it types the provider you get from declare_target_state_with_child().

Parameters:

key: StableKey — a union of None | bool | int | str | bytes | uuid.UUID | Symbol | tuple[StableKey, ...]
desired_target_state: What the user declared, or NON_EXISTENCE if no longer declared
prev_possible_records: Tracking records from previous runs (may have multiple due to interrupted updates)
prev_may_be_missing: If True, the target state might not exist in the external system

Returns:

TargetReconcileOutput if an action is needed
None if no changes are required

Non-blocking and side-effect free

The reconcile() method must be non-blocking and free of side effects. It should only compare states and return an action — actual I/O operations happen later in the TargetActionSink. CocoIndex may call reconcile() more than once for the same target state when it retries a commit internally; only the output of the committed attempt is applied.

Type annotations

Annotate the prev_possible_records parameter with Collection[YourTrackingRecord] so CocoIndex can properly reconstruct stored tracking records during deserialization. See Serialization for details on supported types.

Values and actions are equal copies

CocoIndex holds every declared target state until its processing component commits, and every action until its sink has applied it. To keep that cheap, a value made of plain data — builtin containers and scalars, date and time types, Decimal, UUID, ipaddress and pathlib objects, numpy arrays and scalars, enum members, and NamedTuples — is held in a compact encoded form from the moment it’s declared. reconcile() then receives an equal copy of the value, with the same types (a tuple stays a tuple), rather than the declared object itself, and the sink receives equal copies of the actions reconcile() returned. A value holding anything else, such as a dataclass, a pydantic model, or a client handle, reaches reconcile() as the declared object.

So don’t rely on the identity of a declared value or an action, or pass mutable state through them; and since the copy is taken at declaration, changes made to a value after it was declared are not seen.

Tracking record (you define)

A tracking record captures the essential information needed to detect changes. Good tracking records:

Are minimal: Only include what’s needed for change detection
Are deterministic: Same input always produces the same record
Are serializable: Must be persistable (typically a NamedTuple or dataclass). Dataclasses and NamedTuples are serialized with msgspec automatically. For types requiring pickle, use @coco.serialize_by_pickle.

python

# Example: File tracking record
@dataclass(frozen=True, slots=True)
class _FileTrackingRecord:
    fingerprint: bytes  # Content hash for change detection

Fingerprinting

For content-based change detection, use the connectorkits.fingerprint utilities. This lets you detect changes without storing the full content:

python

from cocoindex.connectorkits.fingerprint import fingerprint_bytes, fingerprint_str, fingerprint_object

# For raw bytes
fp = fingerprint_bytes(content)

# For strings
fp = fingerprint_str(text)

# For arbitrary objects (uses memo key mechanism)
fp = fingerprint_object(obj)

Action and TargetActionSink (you implement)

An action (you define) describes what operation to perform on the external system:

python

# Example: File action
class _FileAction(NamedTuple):
    path: pathlib.Path
    content: bytes | None  # None means delete

A TargetActionSink batches and executes actions:

python

# Sync sink
sink = coco.TargetActionSink.from_fn(apply_actions)

# Async sink
sink = coco.TargetActionSink.from_async_fn(apply_actions_async)

The sink function receives context_provider as its first argument (for looking up connections from the environment), followed by a sequence of actions, and returns nothing:

python

import cocoindex as coco

def apply_actions(
    context_provider: coco.ContextProvider,
    actions: Sequence[_FileAction],
) -> None:
    for action in actions:
        if action.content is None:
            action.path.unlink(missing_ok=True)
        else:
            action.path.write_bytes(action.content)

A container target’s sink is built with from_fn_with_children() / from_async_fn_with_children() instead. Its callback takes a third argument, child_slots, holding a coco.ChildSlot for every action whose target state was declared with a child, and fulfills each slot with the child handler. See Implementing container targets.

TargetReconcileOutput (you return)

TargetReconcileOutput bundles what reconcile() returns when an action is needed:

python

class TargetReconcileOutput(Generic[ActionT, TrackingRecordT, OptChildHandlerT], NamedTuple):
    action: ActionT                                    # What to do
    sink: TargetActionSink[ActionT]                    # How to execute it
    tracking_record: TrackingRecordT | NonExistenceType  # What to remember
    child_invalidation: Literal["destructive", "lossy"] | None = None  # For container targets

OptChildHandlerT is the handler type of the child target states the action’s sink fulfills (None for leaf targets). Nothing in the tuple depends on it; it is how a container handler declares its child handler type.

The child_invalidation field is only relevant for container targets (those with children). See Child invalidation for details.

TargetStateProvider (CocoIndex provides)

A TargetStateProvider is a factory that creates TargetState objects. You don’t implement this class — CocoIndex gives you one when you register a handler or declare a target state with children.

python

# You get a provider from registration
provider = coco.register_root_target_states_provider("my.target", handler)

# Or from declaring a parent target state
child_provider = coco.declare_target_state_with_child(parent_target_state)

# Or from an attachment on a resolved child provider (see "Implementing attachment providers")
att_provider = child_provider.attachment("vector_index")

TargetState (CocoIndex provides)

A TargetState wraps a key and spec. You create these using the provider, then declare them:

python

# Create a target state
target_state = provider.target_state(key, spec)

# Declare it for reconciliation
coco.declare_target_state(target_state)

Implementing root target states

This section covers root target states — those not nested inside another target.

Life of a root target state

Understanding what happens at runtime:

Registration: You define a TargetHandler and call register_root_target_states_provider(). CocoIndex returns a TargetStateProvider — a factory for creating target states associated with your handler.
Declaration: During execution, user code calls provider.target_state(key, spec) to create TargetState objects, then declare_target_state() to declare them. CocoIndex collects all declared target states.
Reconciliation: When the processing unit finishes, CocoIndex calls your handler’s reconcile() method for each target state. For declared target states, desired_target_state contains the spec; for previously declared but now missing states, desired_target_state is NON_EXISTENCE (triggering cleanup). Your reconcile() compares the desired state with previous records and returns TargetReconcileOutput if an action is needed, or None if no changes are required.
Action Execution: CocoIndex batches actions by their TargetActionSink and executes them. The sink applies changes to the external system (database writes, file operations, API calls, etc.). A batch may combine the actions of several processing components that finish around the same time; if the sink fails, CocoIndex retries in smaller batches along component boundaries, so only the components whose actions actually fail are affected. The sink may therefore see an action from a failed batch again — keep actions idempotent.
Tracking Persistence: After successful execution, CocoIndex persists the new tracking records. On the next run, these become the prev_possible_records for change detection.

Multiple Previous States

Due to interrupted updates, prev_possible_records may contain multiple records. CocoIndex tracks all possible states until a successful update confirms the current state. Your reconciliation logic should handle this by generating actions that work correctly regardless of which previous state is actual.

Step 1: Define your types

Start by defining the types for your provider:

python

from typing import NamedTuple, Collection
from dataclasses import dataclass
import cocoindex as coco

# Key: StableKey is used to identify target states — it's a union type:
#   None | bool | int | str | bytes | uuid.UUID | Symbol | tuple[StableKey, ...]

# Value: What the user declares
@dataclass
class _RowSpec:
    data: dict[str, Any]

# Tracking Record: What to persist for change detection
@dataclass(frozen=True, slots=True)
class _RowTrackingRecord:
    fingerprint: bytes

# Action: What operation to perform
class _RowAction(NamedTuple):
    key: coco.StableKey
    data: dict[str, Any] | None  # None = delete

Step 2: Implement the handler

python

class _RowHandler(coco.TargetHandler[_RowSpec, _RowTrackingRecord]):
    """Handler for database rows."""

    def __init__(self, connection: DatabaseConnection, table: str):
        self._conn = connection
        self._table = table
        self._sink = coco.TargetActionSink.from_async_fn(self._apply_actions)

    async def _apply_actions(
        self, context_provider: coco.ContextProvider, actions: Sequence[_RowAction]
    ) -> None:
        # Connection was passed in __init__ — context_provider not needed here,
        # but must be accepted per the sink protocol.
        for action in actions:
            if action.data is None:
                await self._conn.delete(self._table, action.key)
            else:
                await self._conn.upsert(self._table, action.key, action.data)

    def _compute_fingerprint(self, data: dict[str, Any]) -> bytes:
        from cocoindex.connectorkits.fingerprint import fingerprint_object
        return fingerprint_object(data)

    def reconcile(
        self,
        key: coco.StableKey,
        desired_target_state: _RowSpec | coco.NonExistenceType,
        prev_possible_records: Collection[_RowTrackingRecord],
        prev_may_be_missing: bool,
        /,
    ) -> coco.TargetReconcileOutput[_RowAction, _RowTrackingRecord] | None:
        # Handle deletion
        if coco.is_non_existence(desired_target_state):
            if not prev_possible_records and not prev_may_be_missing:
                return None  # Nothing to delete
            return coco.TargetReconcileOutput(
                action=_RowAction(key=key, data=None),
                sink=self._sink,
                tracking_record=coco.NON_EXISTENCE,
            )

        # Handle upsert
        target_fp = self._compute_fingerprint(desired_target_state.data)

        # Skip if unchanged
        if not prev_may_be_missing and all(
            prev.fingerprint == target_fp for prev in prev_possible_records
        ):
            return None

        return coco.TargetReconcileOutput(
            action=_RowAction(key=key, data=desired_target_state.data),
            sink=self._sink,
            tracking_record=_RowTrackingRecord(fingerprint=target_fp),
        )

Step 3: Register the provider

For root-level target states (not nested within another target), register a provider:

python

_row_provider = coco.register_root_target_states_provider(
    "mycompany.io/mydb/row",  # Unique provider name
    _RowHandler(connection, table),
)

Step 4: Create user-facing APIs

Wrap the provider in a user-friendly API:

python

class TableTarget:
    """User-facing API for declaring rows."""

    def __init__(self, provider: coco.TargetStateProvider[_RowSpec, None]):
        self._provider = provider

    def declare_row(self, *, row: dict[str, Any], key: tuple[str, ...]) -> None:
        spec = _RowSpec(data=row)
        target_state = self._provider.target_state(key, spec)
        coco.declare_target_state(target_state)

Implementing container targets

Container targets (directories, tables) have children (files, rows). This section covers how non-root target states work and how to implement them.

Non-root target states

For targets nested inside another target (e.g., files inside a directory), the lifecycle is similar to root targets but how you get the provider is different.

For root targets, you call register_root_target_states_provider() and immediately get a provider with your handler. For non-root targets, the handler comes from the parent’s sink execution:

Declaration: Call declare_target_state_with_child(parent_ts) — returns an unresolved child provider immediately
Resolution: When the parent reconciles and its sink executes, CocoIndex hands the sink a ChildSlot for that action. The sink fulfills it with the child handler, and CocoIndex resolves the child provider with that handler.
Usage: The child provider can now create child target states, which follow the same reconciliation → execution → tracking flow as root targets.

The child handler often needs context from the parent’s action execution. For example, a file handler needs to know the directory path that was created. By creating the handler inside the parent’s sink, the handler has access to this runtime context.

Child invalidation

When a container target undergoes certain changes, the child target states may be affected. The child_invalidation field in TargetReconcileOutput lets you signal this to CocoIndex:

"destructive" — The container change destroys all existing children (e.g., a primary key change that requires dropping and recreating a table). CocoIndex will ignore all previous tracking records for children under this container and treat them as new.
"lossy" — The container change may cause data loss for existing children (e.g., a schema change that removes columns). CocoIndex will force an upsert for all children by setting prev_may_be_missing=True, even if their data appears unchanged.
None (default) — No impact on children. Normal change detection applies.

Set child_invalidation in the parent handler’s reconcile() method when you detect that the container itself has changed in a way that affects its children:

python

class _DirHandler(coco.TargetHandler[_DirSpec, _DirTrackingRecord, _EntryHandler]):
    def reconcile(self, key, desired_target_state, prev_possible_records, prev_may_be_missing, /):
        # Detect if the container change is destructive or lossy
        invalidation = None
        if self._is_destructive_change(desired_target_state, prev_possible_records):
            invalidation = "destructive"
        elif self._is_lossy_change(desired_target_state, prev_possible_records):
            invalidation = "lossy"

        return coco.TargetReconcileOutput(
            action=_DirAction(...),
            sink=self._sink,
            tracking_record=_DirTrackingRecord(...),
            child_invalidation=invalidation,
        )

Deleting a container

When a container target state is no longer declared, CocoIndex runs the container handler’s non-existence action and does not reconcile the children individually. The container action is therefore responsible for its children, in one of two shapes:

Destroy — for a container CocoIndex owns, remove it together with its contents (DROP TABLE, shutil.rmtree, drop_table, …). The _apply_dir_actions sink in Step 2 below does this.
Abandon — for a user-managed container (a message topic, a table the user created), leave the contents untouched. Once the container is no longer declared, CocoIndex has released its claim on it and must not write to it again: another writer may now own those keys, writing to a topic the user has since deleted would recreate it on brokers that auto-create topics, and the deletion policy captured from the old declaration (such as a deletion_value_fn closure) no longer exists.

The zvec connector (python/cocoindex/connectors/zvec/_target.py, _apply_collection_actions) implements both: it destroys system-managed collections and leaves user-managed ones alone. App.drop() applies the same shapes to every root container — it drops tables and abandons topics without visiting rows or messages — so document on your connector page which shape it uses. The user-facing contract is in How CocoIndex syncs target states.

Step 1: Define parent and child handlers

The parent handler reconciles the container itself. The child handler reconciles entries within it:

python

# Parent handler for directory
class _DirHandler(coco.TargetHandler[_DirSpec, _DirTrackingRecord, _EntryHandler]):
    def reconcile(self, key, desired_target_state, prev_possible_records, prev_may_be_missing, /):
        # Reconcile the directory itself
        ...

# Child handler for entries within a directory
class _EntryHandler(coco.TargetHandler[_EntrySpec, _EntryTrackingRecord]):
    def __init__(self, base_path: pathlib.Path):
        self._base_path = base_path

    def reconcile(self, key, desired_target_state, prev_possible_records, prev_may_be_missing, /):
        # Reconcile files/subdirs within the directory
        path = self._base_path / key
        ...

Step 2: Fulfill child slots from the sink

The parent’s sink is built with from_fn_with_children() (or from_async_fn_with_children()). Besides context_provider and actions, its callback receives child_slots: a mapping from the index of each action whose target state was declared with a child to the coco.ChildSlot for that child provider. The sink creates the container and fulfills the slot with the child handler:

python

def _apply_dir_actions(
    context_provider: coco.ContextProvider,
    actions: Sequence[_DirAction],
    child_slots: Mapping[int, coco.ChildSlot[_EntryHandler]],
) -> None:
    for i, action in enumerate(actions):
        if action.should_delete:
            # An orphan delete has no child provider, hence no slot.
            shutil.rmtree(action.path, ignore_errors=True)
        else:
            action.path.mkdir(parents=True, exist_ok=True)
            # Hand the created path to the child handler.
            child_slots[i].fulfill(_EntryHandler(action.path))


_dir_sink = coco.TargetActionSink.from_fn_with_children(_apply_dir_actions)

Every slot must be fulfilled exactly once before the sink returns; an unfulfilled slot fails the commit. Because each slot is tied to its action by CocoIndex, a handler can never be attached to the wrong target state. Actions without a child provider — orphan deletes, or leaf actions when one sink is shared between a container and its children — have no entry in child_slots, so a batch of leaf actions costs nothing extra.

Step 3: Create user-facing API

The user-facing API uses declare_target_state_with_child() and exposes methods for declaring children:

python

class DirTarget:
    """User-facing API for declaring files in a directory."""

    def __init__(self, provider: coco.TargetStateProvider[_EntrySpec, None]):
        self._provider = provider

    def declare_file(self, filename: str, content: bytes) -> None:
        spec = _EntrySpec(content=content)
        target_state = cast(
            coco.TargetState[None],
            self._provider.target_state(filename, spec),
        )
        coco.declare_target_state(target_state)


@coco.fn
def declare_dir_target(path: pathlib.Path) -> DirTarget:
    """Declare a directory target and return an API for declaring files."""
    parent_ts = _root_provider.target_state(
        key=str(path),
        value=_DirSpec(),
    )
    # Child provider is pending until parent sink runs
    child_provider = coco.declare_target_state_with_child(parent_ts)
    return DirTarget(child_provider)

Migrating from `ChildTargetDef`

Before child slots, a container target’s sink was built with from_fn() / from_async_fn() and returned a list of coco.ChildTargetDef(handler=...) (or None) index-aligned with actions. That contract is deprecated and will be removed in a future release. Sinks written against it still run and emit a DeprecationWarning, and the pre-slot annotations still type-check:

coco.TargetActionSink[Action, ChildHandler] is accepted; the second type argument is deprecated and ignored (evaluating it at runtime warns). Write coco.TargetActionSink[Action].
coco.TargetReconcileOutput[Action, TrackingRecord, ChildHandler] is unchanged: the third argument is how a container handler declares its child handler type.

To migrate a container sink:

Build it with from_fn_with_children() / from_async_fn_with_children() instead of from_fn() / from_async_fn().
Add the third positional parameter, child_slots: Mapping[int, coco.ChildSlot[ChildHandler]], and return None.
Replace each outputs[i] = coco.ChildTargetDef(handler=h) with child_slots[i].fulfill(h); drop the None entries and the return outputs.

python

# Before
def _apply_dir_actions(
    context_provider: coco.ContextProvider,
    actions: Sequence[_DirAction],
) -> list[coco.ChildTargetDef[_EntryHandler] | None]:
    outputs: list[coco.ChildTargetDef[_EntryHandler] | None] = []
    for action in actions:
        if action.should_delete:
            shutil.rmtree(action.path, ignore_errors=True)
            outputs.append(None)
        else:
            action.path.mkdir(parents=True, exist_ok=True)
            outputs.append(coco.ChildTargetDef(handler=_EntryHandler(action.path)))
    return outputs


_dir_sink = coco.TargetActionSink[_DirAction, _EntryHandler].from_fn(_apply_dir_actions)


# After
def _apply_dir_actions(
    context_provider: coco.ContextProvider,
    actions: Sequence[_DirAction],
    child_slots: Mapping[int, coco.ChildSlot[_EntryHandler]],
) -> None:
    for i, action in enumerate(actions):
        if action.should_delete:
            shutil.rmtree(action.path, ignore_errors=True)
        else:
            action.path.mkdir(parents=True, exist_ok=True)
            child_slots[i].fulfill(_EntryHandler(action.path))


_dir_sink = coco.TargetActionSink.from_fn_with_children(_apply_dir_actions)

Leaf sinks need no change. If a leaf sink shares its callback with a container level (one sink per database serving table DDL and row writes), migrate it the same way: leaf actions simply have no entry in child_slots.

Implementing attachment providers

Attachment providers let a child handler expose auxiliary target states alongside its regular children. For example, a PostgreSQL table handler manages rows as regular children, but can also manage vector indexes and SQL command attachments as separate attachment types — each tracked independently.

When to use attachments

Use attachments when a target has auxiliary state beyond its primary children — indexes, triggers, materialized views, or any side-resource that should be managed alongside the main data. Attachments use symbol keys as namespace separators so they never conflict with regular child keys.

Target state path hierarchy

Attachments create additional levels in the target state path using symbol keys (denoted with @ prefix in documentation):

@my_connector/table                    [level 1 — root provider]
  (db_key, schema, table_name)         [level 2 — table state]
    @vector_index                      [level 3 — attachment namespace (symbol key)]
      index_name_1                     [level 4 — attachment instance]
      index_name_2                     [level 4]
    @sql_command_attachment             [level 3 — another attachment namespace]
      cmd_name_1                       [level 4]
    row_pk_1                           [level 3 — regular child (row)]
    row_pk_2                           [level 3]

The symbol keys (@vector_index, @sql_command_attachment) are path namespaces — not target states themselves. They separate attachment instances from regular children at the same level.

How it works

Handler implements attachments(): The child handler (e.g., row handler) returns a dict of all supported attachment types and their handlers.
Engine eagerly registers attachment providers: When the child handler is fulfilled, the engine calls attachments() and registers all returned types. This ensures orphaned attachments can be cleaned up even when not declared in the current run.
User code calls provider.attachment(): On a resolved child provider, this retrieves the cached attachment sub-provider.
Target states declared under the attachment provider are tracked independently from regular children.

Step 1: Implement the attachment handler

An attachment handler is just a regular TargetHandler — it implements reconcile() and has an action sink:

python

class _VectorIndexSpec(NamedTuple):
    column: str
    metric: str
    method: str

class _VectorIndexAction(NamedTuple):
    name: str
    spec: _VectorIndexSpec | None  # None means delete

class _VectorIndexHandler:
    def __init__(self, pool, table_name):
        self._pool = pool
        self._table_name = table_name
        self._sink = coco.TargetActionSink.from_async_fn(self._apply_actions)

    async def _apply_actions(
        self, context_provider: coco.ContextProvider, actions: Sequence[_VectorIndexAction]
    ) -> None:
        async with self._pool.acquire() as conn:
            for action in actions:
                if action.spec is None:
                    await conn.execute(f'DROP INDEX IF EXISTS "{action.name}"')
                else:
                    await conn.execute(f'CREATE INDEX "{action.name}" ...')

    def reconcile(self, key, desired_target_state, prev_possible_records, prev_may_be_missing, /):
        # Standard reconcile pattern — compare fingerprints, return action or None
        ...

Step 2: Add `attachments()` to the parent handler

The parent handler (which manages regular children) returns a dict of all supported attachment types:

python

class _RowHandler(coco.TargetHandler[_RowValue, _RowFingerprint]):
    def __init__(self, pool, table_name, schema_name, table_schema):
        self._pool = pool
        self._table_name = table_name
        self._schema_name = schema_name
        # ...

    def attachments(self) -> dict[str, _VectorIndexHandler | _SqlCommandHandler]:
        return {
            "vector_index": _VectorIndexHandler(self._pool, self._table_name, self._schema_name),
            "sql_command_attachment": _SqlCommandHandler(self._pool, self._table_name, self._schema_name),
        }

    def reconcile(self, ...):
        # Regular row reconciliation
        ...

Step 3: Expose attachment APIs on the user-facing target

The user-facing target class calls provider.attachment() to get the attachment sub-provider, then declares target states on it:

python

class TableTarget:
    def __init__(self, provider, table_schema):
        self._provider = provider
        self._table_schema = table_schema

    def declare_row(self, *, row):
        # Regular child target state
        coco.declare_target_state(self._provider.target_state(pk_values, row_dict))

    def declare_vector_index(self, *, name, column, metric="cosine", method="ivfflat"):
        spec = _VectorIndexSpec(column=column, metric=metric, method=method)
        att_provider = self._provider.attachment("vector_index")
        coco.declare_target_state(att_provider.target_state(name, spec))

Tracking records for teardown

When an attachment has a teardown step (like DROP INDEX), store the full spec as the tracking record instead of a fingerprint. This lets you recover the teardown information from prev_possible_records when the attachment is deleted or changed. See the _SqlCommandHandler in the PostgreSQL connector for an example.

Best practices

Use `ContextKey` for external resource identity

When a target connector manages state in an external resource (a database, a search index, an object store, etc.), use a ContextKey string as part of the target state key — not connection parameters like host, port, or credentials.

Target state keys must be stable across runs for correct reconciliation. CocoIndex uses keys to match current declarations with previously tracked states. If the key is stable, previously tracked states are associated with the current target, so CocoIndex can correctly reconcile — e.g., deleting rows that are no longer declared. If the key changes (because a connection parameter changed), CocoIndex cannot associate previous tracked states with the current target, and treats the target as being in a cleared state — losing the ability to clean up old data.

ContextKey solves this by providing a user-defined stable logical name (e.g., "my_pg") that is decoupled from transient connection details:

python

# User creates a stable logical name for the resource
db = coco.ContextKey[asyncpg.Pool]("my_pg")

# Target connector uses db.key (the string "my_pg") in the target state key
class _TableKey(NamedTuple):
    db_key: str            # Stable — from ContextKey.key
    schema_name: str | None
    table_name: str

key = _TableKey(db_key=db.key, schema_name=schema_name, table_name=table_name)

# At action time, resolve the live connection from context_provider
pool = context_provider.get(key.db_key, asyncpg.Pool)

This way, changing a password, switching replicas, or rotating credentials won’t invalidate tracked states or break reconciliation.

Resolve connections at action time, never at declare time. Target states outlive the code that declared them: when a declaration disappears or an app is dropped, CocoIndex generates delete actions in a run where your mount/declare code never executes. At that point the only inputs available are the persisted target state key and the current environment — which is why the sink receives a ContextProvider and why the key must carry ContextKey.key. Don’t capture a live connection when the target is declared; a connection captured at declare time won’t exist for cleanup, and resolving in the sink keeps create, update, and delete on one code path. Once resolved in the root sink, pass the typed connection down to child handlers directly (e.g., store _pool: asyncpg.Pool, not the key string).

See _TableKey in cocoindex/connectors/postgres/_target.py and cocoindex/connectors/surrealdb/_target.py for reference implementations.

Idempotent actions

Actions should be idempotent — applying the same action multiple times should have the same effect as applying it once:

python

# Good: Idempotent
path.mkdir(parents=True, exist_ok=True)
path.unlink(missing_ok=True)
await conn.execute("INSERT ... ON CONFLICT DO UPDATE ...")

# Bad: Not idempotent
path.mkdir()  # Fails if exists
await conn.execute("INSERT ...")  # Fails on duplicate key

Handle multiple previous states

Due to interrupted updates, prev_possible_records may contain multiple records. Design your reconciliation logic to handle this:

python

# Check if ALL previous states match (conservative approach)
if not prev_may_be_missing and all(
    prev.fingerprint == target_fp for prev in prev_possible_records
):
    return None  # Safe to skip

Efficient change detection

Choose tracking records that enable efficient change detection without storing full content:

Scenario	Tracking Record
File content	Content hash (fingerprint)
Database row	Row data hash
Schema/structure	Schema definition
Directory existence	`None` (presence is enough)

Skip reconciling unchanged values

When a handler’s tracking record is exactly the fingerprint of the declared value, set tracks_value_fingerprint = True on the handler class:

python

from cocoindex.connectorkits.fingerprint import fingerprint_object

class RowHandler(coco.TargetHandler[dict[str, Any], bytes]):
    tracks_value_fingerprint = True

    def reconcile(self, key, desired_target_state, prev_possible_records, prev_may_be_missing, /):
        if coco.is_non_existence(desired_target_state):
            ...
        target_fp = fingerprint_object(desired_target_state)
        if not prev_may_be_missing and all(
            prev == target_fp for prev in prev_possible_records
        ):
            return None
        return coco.TargetReconcileOutput(
            action=..., sink=self._sink, tracking_record=target_fp
        )

CocoIndex then doesn’t call reconcile() for a declared value that is unchanged — one that reconcile() above would return None for — so an update where most values are unchanged costs almost nothing per value. reconcile() is still called for new, changed, and removed target states, and whenever prev_may_be_missing is True.

Only set it when both hold:

The tracking record is fingerprint_object(desired_target_state) itself — not a fingerprint of a part of the value or of a derived value, and not wrapped in another type.
An unchanged fingerprint means there is nothing to do.

The attribute has no effect on a container handler, whose reconcile() always runs so that its sink can fulfill the child slot.

reconcile() must keep its own comparison: it is still called for values that aren’t plain data (anything other than None, bool, int, float, str, bytes, and list / tuple / dict-with-str-keys of those), unchanged or not.

Shared action sinks

If all instances of a handler use the same action logic, create a shared sink. The action function must accept context_provider as its first argument:

python

import cocoindex as coco

def _apply_actions(
    context_provider: coco.ContextProvider, actions: Sequence[_MyAction]
) -> None:
    ...

# Module-level shared sink
_shared_sink = coco.TargetActionSink.from_fn(_apply_actions)

class _MyHandler(coco.TargetHandler[...]):
    def reconcile(self, ...):
        return coco.TargetReconcileOutput(
            action=...,
            sink=_shared_sink,  # Reuse the same sink
            tracking_record=...,
        )

Complete example: local file system

Here’s a simplified version of the localfs connector showing the complete pattern:

python

from __future__ import annotations
import pathlib
from dataclasses import dataclass
from typing import Collection, NamedTuple, Sequence
import cocoindex as coco
from cocoindex.connectorkits.fingerprint import fingerprint_bytes


# Types
_FileContent = bytes
_FileFingerprint = bytes


class _FileAction(NamedTuple):
    path: pathlib.Path
    content: _FileContent | None  # None = delete


@dataclass(frozen=True, slots=True)
class _FileTrackingRecord:
    fingerprint: _FileFingerprint


# Action execution
def _apply_actions(context_provider: coco.ContextProvider, actions: Sequence[_FileAction]) -> None:
    for action in actions:
        if action.content is None:
            action.path.unlink(missing_ok=True)
        else:
            action.path.parent.mkdir(parents=True, exist_ok=True)
            action.path.write_bytes(action.content)


_file_sink = coco.TargetActionSink[_FileAction].from_fn(_apply_actions)


# Handler
class _FileHandler(coco.TargetHandler[_FileContent, _FileTrackingRecord]):
    __slots__ = ("_base_path",)
    _base_path: pathlib.Path

    def __init__(self, base_path: pathlib.Path):
        self._base_path = base_path

    def reconcile(
        self,
        key: coco.StableKey,
        desired_target_state: _FileContent | coco.NonExistenceType,
        prev_possible_records: Collection[_FileTrackingRecord],
        prev_may_be_missing: bool,
        /,
    ) -> coco.TargetReconcileOutput[_FileAction, _FileTrackingRecord] | None:
        path = self._base_path / key

        if coco.is_non_existence(desired_target_state):
            if not prev_possible_records and not prev_may_be_missing:
                return None
            return coco.TargetReconcileOutput(
                action=_FileAction(path=path, content=None),
                sink=_file_sink,
                tracking_record=coco.NON_EXISTENCE,
            )

        target_fp = fingerprint_bytes(desired_target_state)

        if not prev_may_be_missing and all(
            prev.fingerprint == target_fp for prev in prev_possible_records
        ):
            return None

        return coco.TargetReconcileOutput(
            action=_FileAction(path=path, content=desired_target_state),
            sink=_file_sink,
            tracking_record=_FileTrackingRecord(fingerprint=target_fp),
        )

See the full implementations in:

cocoindex/connectors/localfs/target.py — File system targets
cocoindex/connectors/postgres/target.py — PostgreSQL tables and rows

Building a custom target connector

When to create a custom target connector

Key data types

TargetHandler (you implement)

Tracking record (you define)

Action and TargetActionSink (you implement)

TargetReconcileOutput (you return)

TargetStateProvider (CocoIndex provides)

TargetState (CocoIndex provides)

Implementing root target states

Life of a root target state

Step 1: Define your types

Step 2: Implement the handler

Step 3: Register the provider

Step 4: Create user-facing APIs

Implementing container targets

Non-root target states

Child invalidation

Deleting a container

Step 1: Define parent and child handlers

Step 2: Fulfill child slots from the sink

Step 3: Create user-facing API

Migrating from ChildTargetDef

Implementing attachment providers

When to use attachments

Target state path hierarchy

How it works

Step 1: Implement the attachment handler

Step 2: Add attachments() to the parent handler

Step 3: Expose attachment APIs on the user-facing target

Best practices

Use ContextKey for external resource identity

Idempotent actions

Handle multiple previous states

Efficient change detection

Skip reconciling unchanged values

Shared action sinks

Complete example: local file system

Migrating from `ChildTargetDef`

Step 2: Add `attachments()` to the parent handler

Use `ContextKey` for external resource identity