How to Set Up a CocoIndex Connector: Architecture and Implementation Guide

Setting up a CocoIndex connector requires implementing three integrated components—a Source API for data ingestion, a Target API for state declaration, and a root-state provider that registers handlers with the core engine to automate creation, updates, and deletion.

CocoIndex connectors act as the bridge between processing pipelines and external systems such as filesystems, PostgreSQL databases, and vector stores like Qdrant. This guide walks through the concrete implementation details required to set up a CocoIndex connector, using the local filesystem connector as the canonical reference implementation from the cocoindex-io/cocoindex repository.

The Three Components of a CocoIndex Connector

Every CocoIndex connector consists of three distinct parts that work together to provide bidirectional data flow and state management.

Source API

The Source API reads data from external systems and emits a walker that can be iterated synchronously or asynchronously. This handles ingestion from files, database rows, or vectors.

• localfs.walk_dir() – Walks directories recursively with optional live monitoring
• postgres._source – Handles row-level ingestion from PostgreSQL tables
• qdrant._source – Manages vector retrieval from Qdrant collections

Target API

The Target API declares the desired state of external resources, allowing the engine to automatically create, update, or delete them. Rather than imperatively writing files or dropping tables, you declare what should exist.

• localfs.declare_file() – Declares a file with specific content and fingerprint
• localfs.declare_dir_target() – Declares a directory structure target
• postgres._target and qdrant._target – Declare database tables and vector collections

Root-State Provider

The root-state provider registers the connector with the core engine, transforming concrete actions (write file, drop table) into TargetState objects that the engine can reconcile.

In python/cocoindex/connectors/localfs/_target.py lines 72-78, the registration occurs when the module is imported:

_root_provider = coco.register_root_target_states_provider(
    "cocoindex/localfs", _RootHandler()
)

This registration keys the connector by name (cocoindex/localfs) and provides a handler class that implements the reconciliation logic.

Architecture Walkthrough

The local filesystem connector illustrates the standard reconciliation pattern used across all CocoIndex connectors.

Root Provider Registration

When the connector module loads, it registers a provider keyed by the connector name. This registration happens immediately upon import:

_root_provider = coco.register_root_target_states_provider(
    "cocoindex/localfs", _RootHandler()
)

This call establishes the link between the string identifier "cocoindex/localfs" and the _RootHandler class that will process state changes.

Target State Creation

Public helper functions like declare_dir_target and declare_file build TargetState objects containing two critical elements:

1. A stable _RootKey built from an optional ContextKey for the base directory and a relative path
2. A _EntrySpec describing whether the target is a file, directory, or non-existence

In python/cocoindex/connectors/localfs/_target.py lines 13-21, the dir_target function constructs these objects:

key = _RootKey(base_dir_key=_get_base_dir_key(file_path),
               path=file_path.path.as_posix())
spec = _EntrySpec(entry_spec=_DirSpec(),
                  create_parent_dirs=create_parent_dirs)
return _root_provider.target_state(key, spec)

Similarly, declare_file (lines 13-20) uses _FileSpec instead of _DirSpec to represent file content rather than directory structure.

Reconciliation and Action Sink

The core engine calls the handler's reconcile method for each key to determine if the desired state differs from the stored state. In python/cocoindex/connectors/localfs/_target.py lines 54-60, the _reconcile_entry method compares fingerprints:

target_fp = fingerprint_bytes(entry_spec)               # file fingerprint

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

If a change is required, the method produces a _EntryAction (write file, delete file, create directory). These actions pass to _apply_actions_with_child, which executes filesystem operations at lines 75-80 of _execute_entry_action:

if action.entry_type == "file":
    if action.create_parents:
        path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(action.content)

For directories, this method returns a child handler so nested entries can be processed recursively.

Memo-Stable FilePath

When a ContextKey[pathlib.Path] serves as the base directory, the memo key uses the string stored in the key rather than the absolute path. This design makes pipelines robust to project directory moves, as the logical identifier remains stable regardless of where the code executes.

Complete Pipeline Implementation

A typical pipeline reads Markdown files from a source directory and writes transformed outputs declaratively. This example demonstrates the full setup pattern:

import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher

# Stable identifiers for the two directories

SOURCE_DIR = coco.ContextKey[pathlib.Path]("source_dir")
OUTPUT_DIR = coco.ContextKey[pathlib.Path]("output_dir")


@coco.fn
async def app_main() -> None:
    # Declare the target – a DirTarget that will manage the output directory

    out_target = await localfs.mount_dir_target(OUTPUT_DIR)

    # Walk the source directory (live mode optional)

    matcher = PatternFilePathMatcher(included_patterns=["**/*.md"])
    async for file in localfs.walk_dir(SOURCE_DIR, recursive=True,
                                      path_matcher=matcher):
        # Launch a processing component for each file

        await coco.mount(
            coco.component_subpath("file", str(file.file_path.path)),
            process_file,
            file,
            out_target,
        )


@coco.fn(memo=True)
async def process_file(file: coco.resources.file.FileLike,
                       target: localfs.DirTarget) -> None:
    # Read, transform, write back to the same relative path

    text = await file.read_text()
    transformed = text.upper()           # example transformation

    target.declare_file(
        filename=file.file_path.path,    # keep relative path

        content=transformed,
        create_parent_dirs=True,
    )

Step 1 – Context provisioning (usually in lifespan):

async def lifespan(builder: coco.EnvBuilder) -> None:
    builder.provide(SOURCE_DIR, pathlib.Path("./docs"))
    builder.provide(OUTPUT_DIR, pathlib.Path("./out"))

Step 2 – Mount the app via CLI or programmatically:

coco run path/to/app_main.py

The same pattern applies to Postgres, Qdrant, Neo4j, and other connectors. Only the source function (<connector>._source) and target API (<connector>._target) differ between implementations.

Common Target State Patterns

Declaring a Single File Target

For simple file outputs without a full directory target wrapper:

@coco.fn
def write_readme():
    coco.mount(
        localfs.declare_file,
        localfs.FilePath("README.md", base_dir=OUTPUT_DIR),
        content="# Project\nGenerated by CocoIndex",

        create_parent_dirs=True,
    )

This uses declare_file implemented in python/cocoindex/connectors/localfs/_target.py lines 82-95.

Declaring Nested Directory Targets

Create hierarchical directory structures by declaring sub-targets:

parent = await localfs.mount_dir_target(OUTPUT_DIR)
sub = parent.declare_dir_target("subfolder", create_parent_dirs=True)
sub.declare_file("inner.txt", b"inner content")

The DirTarget.declare_dir_target implementation resides in python/cocoindex/connectors/localfs/_target.py lines 27-33.

Using ContextKey Directly in Source Walks

Source walkers accept ContextKey objects directly for memo-stable path resolution:

matcher = PatternFilePathMatcher(included_patterns=["**/*.txt"])
async for file in localfs.walk_dir(SOURCE_DIR, recursive=True,
                                  path_matcher=matcher, live=True):
    await process(file)

The walk_dir signature and parameters are documented in docs/src/content/docs/connectors/localfs.mdx lines 70-78.

Key Implementation Files

• python/cocoindex/connectors/localfs/_target.py – Core target-state handler for the local filesystem connector, including registration, reconciliation, and action sink logic
• python/cocoindex/_internal/target_state.py – Generic engine abstractions including TargetHandler, TargetReconcileOutput, and ChildTargetDef
• python/cocoindex/_internal/context_keys.py – Definition of ContextKey used for memo-stable base directories
• python/cocoindex/connectors/postgres/_target.py – SQL database connector implementation example
• python/cocoindex/connectors/qdrant/_target.py – Vector store connector implementation example
• docs/src/content/docs/connectors/localfs.mdx – User-facing documentation for source and target APIs

Summary

• Three-part architecture: Every connector implements a Source API for reading, a Target API for declaring state, and a root-state provider for registration
• Declarative targets: Use declare_file() and mount_dir_target() to specify desired state rather than imperatively writing files
• Automatic reconciliation: The engine compares SHA-256 fingerprints stored in TargetState objects to determine whether to write, skip, or delete entries
• Memo-stable keys: Use ContextKey[pathlib.Path] to ensure pipeline consistency across different execution environments
• Consistent pattern: All connectors follow the registration/reconciliation contract defined in coco._internal.target_state, making it straightforward to switch between filesystem, database, and vector store backends

Frequently Asked Questions

What are the three main components of a CocoIndex connector?

The three components are the Source API (for reading data and emitting walkers), the Target API (for declaring desired resource states), and the root-state provider (which registers the connector with the engine via coco.register_root_target_states_provider()). These work together to enable the engine to automatically reconcile external system states with your pipeline's declared intent.

How does the CocoIndex engine determine whether to update a file?

The engine calculates a SHA-256 fingerprint of the desired file content in python/cocoindex/connectors/localfs/_target.py lines 54-60. It compares this fingerprint against stored records from previous runs. If the fingerprints match, the engine returns None indicating no change is needed. If they differ, the engine generates a _EntryAction to write the new content.

What is the purpose of ContextKey in CocoIndex connectors?

ContextKey objects provide memo-stable identifiers for resources like base directories. When you use coco.ContextKey[pathlib.Path]("source_dir"), the engine uses the string key "source_dir" rather than the absolute filesystem path for memoization. This ensures that pipelines remain stable when the project directory moves, as the logical identifier persists across different execution environments.

How do I register a custom connector with the CocoIndex engine?

Register your connector by calling coco.register_root_target_states_provider() with a unique string key (such as "cocoindex/localfs") and a handler instance that implements the reconciliation interface. This registration typically occurs at module import time, as shown in python/cocoindex/connectors/localfs/_target.py lines 72-78. Once registered, users can reference your connector by its key string in their pipelines.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}