Write a Custom Target: Export Indexed Data Anywhere

CocoIndex supports custom targets: you can export data to any destination, whether it’s a local file, cloud storage, a REST API, or your own bespoke system, and still get incremental updates, change tracking, and automatic cleanup from the engine.

The code in this post uses the CocoIndex v1 API. We ran the complete example below on CocoIndex v1.1.1.

Why you would write one

Agents often read their context from a system CocoIndex has no connector for: an internal knowledge API, a vendor’s ingestion endpoint, a store another team owns. Sending data to it from a plain function is easy. Keeping it in sync with the sources is the hard part:

• Send only what changed. Re-posting every document on each run is slow and can cost money per call.
• Delete what is gone. When a source file is removed, its entry in the API should go too, or agents keep retrieving it.
• Recover from a run that stopped halfway. Some writes landed and some did not, and the next run has to converge anyway.

A custom target gives these jobs to the engine. You write one function that compares the desired state with what was written before, and one function that applies a batch of changes.

Swap targets without rewriting your pipeline

CocoIndex targets share one declarative pattern: you state what should exist, and the connector keeps the external system in sync. Writing HTML files to a folder and rows to Postgres look the same at the call site:

# Local files
localfs.declare_file(outdir / name, html, create_parent_dirs=True)

# Postgres: same declare pattern, different connector
table.declare_row(row=DocEmbedding(id=id, embedding=vec))

Because the transformation logic never touches write paths, swapping a target doesn’t ripple through the pipeline. Custom targets extend the same pattern to systems CocoIndex doesn’t ship a connector for: you implement the reconciliation once, and your pipeline code keeps declaring state the same way. They are the write-side counterpart of custom sources.

What a custom target looks like in v1

A custom target connector is the piece that connects CocoIndex’s declarative target states to an external system. You implement three small things:

1. A tracking record: the minimal information persisted per item to detect changes on future runs. For a file, a content fingerprint; for a row, a hash of the row data.
2. An action and a sink: the action describes one operation (write this file, delete this row); the TargetActionSink executes a batch of them against the external system.
3. A TargetHandler with one required method, reconcile(): given the desired state for a key and the tracking records from previous runs, return an action if something must change, or None to skip.

Registration ties it together: coco.register_root_target_states_provider() takes your handler and returns a provider your user-facing API uses to declare target states.

At runtime the engine drives the loop: it collects everything your pipeline declared, calls reconcile() per key (passing NON_EXISTENCE for keys that disappeared, which is how cleanup happens), batches the returned actions by sink, executes them, and persists the new tracking records for next time. reconcile() must be non-blocking and free of side effects; all I/O lives in the sink.

How change detection works

You never write an imperative sync loop. You describe the desired state, and the handler reconciles it against what existed before. To detect changes without re-reading the target, the tracking record stores just enough to compare, typically a content fingerprint from the connectorkits.fingerprint utilities, not the full content. When the fingerprint of the desired state matches the stored record, reconcile() returns None and the item is skipped.

An update can be interrupted after the sink has applied some actions but before the engine commits the new records. The engine then keeps every record the target may reflect until a later run confirms one. That is why reconcile() receives a collection of previous records, and why the skip condition checks that all of them match. A second flag, prev_may_be_missing, is True when the item may not exist in the target at all: a key with no stored record, or one whose interrupted action was a delete. Actions must be idempotent, so that re-applying a write that already landed has no ill effect.

The playground below runs the handler from this post on three output files. Change the sources and tick the interrupted run to see each call’s arguments and return value.

Example: export markdown files to local HTML

The example converts .md files to .html on disk and updates changed files incrementally. The pipeline side is the files_transform example. It uses the built-in localfs target; afterwards we build a simplified version of that file target ourselves.

The pipeline

One processing component per markdown file: read, render, declare the HTML file that should exist.

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    html = _markdown_it.render(await file.read_text())
    outname = "__".join(file.file_path.path.parts) + ".html"
    localfs.declare_file(outdir / outname, html, create_parent_dirs=True)


@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
        live=True,
    )
    await coco.mount_each(process_file, files.items(), outdir)

declare_file states what should exist. The connector writes the file, overwrites it on change, and deletes the .html when the source .md disappears, so the pipeline has no file I/O code.

Build the file target yourself

Here is a simplified version of that localfs connector, following the custom target connector guide. First the imports and types: the value a pipeline declares for one file, an action describing one file operation, and a tracking record holding a content fingerprint.

import pathlib
from collections.abc import Collection, Sequence
from dataclasses import dataclass
from typing import NamedTuple

import cocoindex as coco
from cocoindex.connectorkits.fingerprint import fingerprint_bytes

_FileContent = bytes  # the value declared for one file


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


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

The sink applies a batch of actions. Every operation is idempotent (missing_ok, exist_ok), so a batch that runs twice leaves the same files:

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)

The handler holds the reconciliation logic: delete when the state is gone, skip when every possible previous record matches the new fingerprint, write otherwise.

class _FileHandler(coco.TargetHandler[_FileContent, _FileTrackingRecord]):
    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 / str(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),
        )

Registering the handler returns a provider. The handler gets its output directory as an argument, and a thin user-facing wrapper turns the provider into the declare_file call the pipeline uses. The pipeline renders HTML as a str, so the wrapper encodes it to the bytes the handler fingerprints:

OUTDIR = pathlib.Path("./output_html")

_provider = coco.register_root_target_states_provider(
    "myproject.io/file", _FileHandler(base_path=OUTDIR)
)


def declare_file(filename: str, content: str | bytes) -> None:
    if isinstance(content, str):
        content = content.encode()
    coco.declare_target_state(_provider.target_state(filename, content))

The provider name ("myproject.io/file") must be unique in your app, and the key you pass to target_state() identifies the file across runs.

Use it in the pipeline

The pipeline changes in one line: it calls your declare_file with a file name instead of localfs.declare_file with a path.

from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from markdown_it import MarkdownIt

_markdown_it = MarkdownIt("gfm-like")


@coco.fn(memo=True)
async def process_file(file: FileLike) -> None:
    html = _markdown_it.render(await file.read_text())
    outname = "__".join(file.file_path.path.parts) + ".html"
    declare_file(outname, html)


@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
        live=True,
    )
    await coco.mount_each(process_file, files.items())


app = coco.App(
    coco.AppConfig(name="CustomFileTarget"),
    app_main,
    sourcedir=pathlib.Path("./data"),
)

Containers (a directory whose files are children, a table whose rows are children) follow the same pattern one level deeper: the parent’s sink is built with a child_handler factory that returns the handler for the children. The connector guide covers that, along with attachments for auxiliary state like vector indexes.

Run it

Put the code blocks from Build the file target yourself and Use it in the pipeline in one main.py, add a few .md files under data/, and run:

pip install "cocoindex>=1.1.1" "markdown-it-py[linkify,plugins]"
echo "COCOINDEX_DB=./cocoindex.db" > .env
cocoindex update main        # catch-up: scan, sync, exit
cocoindex update -L main     # live: keep watching for file changes

We ran it with three files, then edited guide.md and deleted old.md. The second run’s summary reports what changed:

✅ process_file: 3 total | 1 reprocessed, 1 deleted, 1 unchanged

The sink rewrote output_html/data__guide.md.html and deleted output_html/data__old.md.html. notes.md was unchanged, so its component was skipped. Live mode keeps the output synchronized as files change, which suits content that changes often, such as an internal wiki.

From the v0 connector API

If you wrote a v0 custom target, here is where each piece went:

The v0 API split targets into setup (DDL-like) and data (DML-like) methods. The v1 model unifies both as target states reconciled the same way, which is what makes containers, children, and attachments compose: a table and its rows are the same concept at two levels.

Connector design rules

• Idempotent actions: actions should be safe to run more than once (ON CONFLICT DO UPDATE, missing_ok=True, exist_ok=True), because interrupted runs re-apply them, and a failed batch is retried in smaller batches.
• Keep tracking records minimal: a fingerprint beats storing content; the engine persists them per item across runs.
• Handle multiple previous states: skip only when all prev_possible_records match and prev_may_be_missing is false.
• Use ContextKey for external resource identity: put the stable logical name in the target state key, never hosts or credentials, so rotating a password doesn’t orphan tracked state.

When to build a custom target

Most pipelines use built-in connectors and never need one. Reach for a custom target connector when you integrate with a system CocoIndex doesn’t cover, need custom change detection (such as content-based fingerprinting), or manage hierarchical state (containers with children). For a simple “write this somewhere” step with no cleanup requirement, a regular memoized function is often enough. A connector is worth writing when you want the engine to track and clean up what it wrote.

If you build a connector for a system others use too, tell us about it: CocoIndex on GitHub.

About the author

Linghua JinCofounder & CEO of CocoIndex · @linghuaj

CEO and cofounder of CocoIndex. Writes about incremental data infrastructure, agents, and the engineering decisions behind the engine.