How to Customize CocoIndex Connector Behavior: A Developer’s Guide to Extending the Declarative Pipeline Engine

CocoIndex connectors bridge the declarative pipeline engine to external systems through a four-layer architecture—target state provider, handler, action sink, and public API—that you can extend by implementing TargetHandler.reconcile() and registering a custom action sink with coco.register_root_target_states_provider().

Customizing connector behavior in cocoindex-io/cocoindex allows you to integrate proprietary storage systems, modify how existing connectors execute side-effects, or inject runtime configuration dynamically. All connectors follow a consistent pattern defined in the core library, making it straightforward to override specific layers without rewriting the entire integration.

Understanding the Connector Architecture

Every CocoIndex connector implements a four-layer stack that separates declaration from execution:

Layer	Responsibility	Core API
Target state provider	Registers a root handler that creates, updates, or deletes resources	coco.register_root_target_states_provider (see _root_provider in python/cocoindex/connectors/localfs/_target.py)
Target handler	Implements reconcile() to turn a desired state into an _EntryAction	Sub-class of coco.TargetHandler (e.g., _RootHandler, _EntryHandler)
Action sink	Executes low-level side-effects (write a file, run an INSERT)	coco.TargetActionSink.from_fn(_apply_actions_with_child)
Public API	Thin wrapper functions users call from pipelines (declare_dir_target, declare_file)	Functions annotated with @coco.fn

This architecture ensures that declarative specifications remain pure while the engine handles impure side-effects through the action sink.

The Connector Lifecycle

Understanding how data flows through these layers is essential for customization.

Root Key Definition

The root key provides a stable identifier to look up targets across runs. In the local filesystem connector, this is implemented as a NamedTuple:

class _RootKey(NamedTuple):
    base_dir_key: str | None   # context key for base directory

    path: str                  # relative path inside the base

Source: python/cocoindex/connectors/localfs/_target.py (lines 19-25).

Root Handler and Reconciliation

The _RootHandler receives the root key, resolves absolute paths, and delegates to _reconcile_entry. Its reconcile() method returns a TargetReconcileOutput containing the action and sink needed to achieve the desired state.

Source: python/cocoindex/connectors/localfs/_target.py (lines 40-66).

Action Creation and Execution

Entry actions store everything needed for side-effects. The _EntryAction tuple defines the operation type, target path, content, and parent directory creation flags. The action sink—created via coco.TargetActionSink.from_fn()—materializes these actions on the real system.

Source: python/cocoindex/connectors/localfs/_target.py (lines 11-15 and 31-40).

Building a Custom Connector from Scratch

To create a custom backend (e.g., a NoSQL store or REST service), implement four components:

1. Define a stable root key—typically a tuple of a context key and logical identifier.
2. Write a TargetHandler that translates desired state into actions.
3. Provide an action sink that performs side-effects.
4. Register the provider with coco.register_root_target_states_provider.

Minimal Skeleton for a Key-Value Connector

The following example implements a complete custom connector for a hypothetical key-value store:

# myconnector/_target.py

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import NamedTuple, Literal

import cocoindex as coco
from cocoindex._internal.context_keys import ContextKey, ContextProvider

# 1️⃣ Stable key -------------------------------------------------------------

class _RootKey(NamedTuple):
    client_key: str           # ContextKey that resolves to a KV-client

    bucket: str               # Logical bucket name


# 2️⃣ Desired state -----------------------------------------------------------

@dataclass(frozen=True, slots=True)
class _EntrySpec:
    """Value to store under a key."""
    value: bytes | str
    ttl: int | None = None   # optional time-to-live


# 3️⃣ Action definition -------------------------------------------------------

class _EntryAction(NamedTuple):
    client_key: str
    bucket: str
    key: str
    value: bytes
    ttl: int | None
    op: Literal["put", "delete"]


# 4️⃣ Action sink -------------------------------------------------------------

def _apply_actions(
    ctx: ContextProvider,
    actions: list[_EntryAction],
    /,
) -> list[None]:
    out = []
    for act in actions:
        client = ctx.get(act.client_key)          # Resolve the KV client

        if act.op == "put":
            client.put(act.bucket, act.key, act.value, ttl=act.ttl)
        else:  # delete

            client.delete(act.bucket, act.key)
        out.append(None)
    return out


_action_sink = coco.TargetActionSink["_EntryAction", None].from_fn(_apply_actions)


# 5️⃣ Handler ---------------------------------------------------------------

class _RootHandler(coco.TargetHandler[_EntrySpec, None, None]):
    def reconcile(
        self,
        key: coco.StableKey,
        desired: _EntrySpec | coco.NonExistenceType,
        prev: list[None],
        may_be_missing: bool,
        /,
    ) -> coco.TargetReconcileOutput[_EntryAction, None, None] | None:
        bucket, entry_key = key  # type: ignore[assignment]

        if coco.is_non_existence(desired):
            return coco.TargetReconcileOutput(
                action=_EntryAction(
                    client_key="kv_client",
                    bucket=bucket,
                    key=entry_key,
                    value=b"", op="delete", ttl=None
                ),
                sink=_action_sink,
                tracking_record=coco.NON_EXISTENCE,
            )
        assert isinstance(desired, _EntrySpec)
        payload = (
            desired.value.encode()
            if isinstance(desired.value, str)
            else desired.value
        )
        return coco.TargetReconcileOutput(
            action=_EntryAction(
                client_key="kv_client",
                bucket=bucket,
                key=entry_key,
                value=payload,
                ttl=desired.ttl,
                op="put",
            ),
            sink=_action_sink,
            tracking_record=None,
        )


# 6️⃣ Provider registration ---------------------------------------------------

_root_provider = coco.register_root_target_states_provider(
    "myconnector/kv", _RootHandler()
)


# 7️⃣ Public API -------------------------------------------------------------

class BucketTarget(coco.ResolvesTo["BucketTarget"]):
    """High-level wrapper returned from `declare_bucket`."""
    _provider: coco.TargetStateProvider[_EntrySpec, None, coco.MaybePendingS]

    def __init__(self, provider):
        self._provider = provider

    def put(self, key: str, value: bytes | str, *, ttl: int | None = None):
        spec = _EntrySpec(value=value, ttl=ttl)
        target = self._provider.target_state((self._bucket, key), spec)
        coco.declare_target_state(target)


def declare_bucket(
    client: ContextKey[object],
    bucket: str,
) -> BucketTarget:
    """Create a bucket target that writes key-value pairs."""
    key = _RootKey(client.key, bucket)
    provider = coco.declare_target_state_with_child(_root_provider.target_state(key, _EntrySpec(b"", None)))
    return BucketTarget(provider)

Key implementation details from the reference codebase:

• RootKey definition mirrors the pattern in localfs/_target.py lines 19-25.
• TargetHandler.reconcile returns a TargetReconcileOutput as seen in lines 92-124 of the local filesystem implementation.
• TargetActionSink.from_fn converts pure functions into sinks (lines 11-15).
• register_root_target_states_provider makes the connector discoverable (lines 71-74).

Customizing Existing Connectors

When you only need to modify how actions execute—adding logging, retries, or payload transformation—you can wrap the built-in action sink without modifying core logic.

import logging
from cocoindex.connectors.localfs import _action_sink_with_child as base_sink

def logging_sink(ctx, actions, /):
    for act in actions:
        logging.info("FS action %s on %s", act.entry_type, act.path)
    return base_sink(ctx, actions)

# Re-register a provider that uses the wrapped sink

coco.register_root_target_states_provider(
    "cocoindex/localfs-logged",
    _RootHandler(),
    action_sink=logging_sink,        # <-- custom sink

)

Now every declare_file or declare_dir_target targeting cocoindex/localfs-logged emits structured logs while preserving the original connector's reconciliation logic.

Using Context Keys for Dynamic Configuration

Connectors often require runtime resources like database pools or credential objects. These are injected via context keys defined in python/cocoindex/_internal/context_keys.py.

from cocoindex._internal.context_keys import ContextKey
from cocoindex.connectors.postgres import PgTableSource

# In your application bootstrap:

coco.context_provider.provide(
    ContextKey["asyncpg.Pool"]("pg_pool"),
    await asyncpg.create_pool(dsn="postgresql://...")
)

# In a pipeline component:

@coco.fn
async def sync_users(pg_pool: ContextKey[asyncpg.Pool]):
    source = PgTableSource(pg_pool, table_name="users")
    async for row in source.fetch_rows():
        # process row …

The ContextKey machinery is used by built-in connectors like PgTableSource (see python/cocoindex/connectors/postgres/_source.py lines 62-78) and the local filesystem target to resolve dynamic configuration at runtime.

Practical Code Examples

Declaring Local File Targets

Use the built-in local filesystem connector to declaratively manage directories and files:

import pathlib, cocoindex as coco
from cocoindex import localfs

@coco.fn
def generate_report(out_dir: pathlib.Path) -> None:
    # Mount a DirTarget that automatically creates the base directory

    target = coco.use_mount(
        coco.component_subpath("report"),
        localfs.declare_dir_target,
        out_dir,
        create_parent_dirs=True,
    )
    # Write two files inside the target directory

    target.declare_file("summary.txt", "Report generated ✅")
    sub = target.declare_dir_target("details")
    sub.declare_file("details.json", b'{"steps": 42}')

Key API calls reference declare_dir_target and declare_file in python/cocoindex/connectors/localfs/_target.py (lines 82-124 and 50-78).

Reading from PostgreSQL Sources

Consume database rows using the PostgreSQL source connector:

import cocoindex as coco
from cocoindex.connectors.postgres import PgTableSource

@coco.fn
async def sync_items(pg_pool: coco.ContextKey[asyncpg.Pool]) -> None:
    # Create a source that yields rows as a dataclass

    source = PgTableSource(
        pg_pool,
        table_name="items",
        row_type=ItemRecord,   # a @dataclass with fields matching columns

    )
    async for row in source.fetch_rows():
        # Do something with each row, e.g. upsert into another system

        await process(row)

PgTableSource implements a dual-mode iterator (RowFetcher) supporting both sync and async operations (source: python/cocoindex/connectors/postgres/_source.py lines 62-78).

Writing to a Custom Cloud Store

Implement a connector for proprietary cloud storage by defining actions and a sink:

# mycloud/_target.py (excerpt)

from cocoindex import TargetActionSink, TargetReconcileOutput, register_root_target_states_provider

class _UploadAction(NamedTuple):
    bucket: str
    key: str
    payload: bytes

def _cloud_sink(ctx, actions, /):
    client = ctx.get("cloud_client")
    for act in actions:
        client.upload(act.bucket, act.key, act.payload)
    return [None] * len(actions)

_cloud_sink = TargetActionSink["_UploadAction", None].from_fn(_cloud_sink)

class _RootHandler(coco.TargetHandler[_EntrySpec, None, None]):
    # Returns TargetReconcileOutput containing _UploadAction objects

    ...

_cloud_provider = register_root_target_states_provider(
    "mycloud/store", _RootHandler()
)

# Public API

def declare_bucket(bucket: str) -> BucketTarget:
    return BucketTarget(_cloud_provider.target_state((_ROOT_KEY, bucket), _EntrySpec(b"", None)))

Now pipelines can call declare_bucket("images").put("logo.png", img_bytes) to invoke your custom cloud SDK through the CocoIndex reconciliation engine.

Summary

• Four-layer architecture: All connectors implement a target state provider, handler, action sink, and public API.
• Customization points: Extend connectors by implementing TargetHandler.reconcile() or wrap existing action sinks for cross-cutting concerns like logging.
• Registration: Use coco.register_root_target_states_provider() to make custom connectors available to the engine.
• Runtime resources: Leverage ContextKey objects from python/cocoindex/_internal/context_keys.py to inject dynamic configuration like database pools.
• Reference implementation: Study python/cocoindex/connectors/localfs/_target.py and python/cocoindex/connectors/postgres/_source.py for production patterns.

Frequently Asked Questions

What is the minimum code needed to implement a custom CocoIndex connector?

You need three components: a NamedTuple root key for stable identification, a class inheriting from coco.TargetHandler that implements reconcile(), and an action sink created via coco.TargetActionSink.from_fn(). Finally, register these with coco.register_root_target_states_provider(). This pattern is illustrated in the local filesystem connector at python/cocoindex/connectors/localfs/_target.py.

How do I modify an existing connector without forking the repository?

Wrap the connector's action sink function. Import the original sink (e.g., _action_sink_with_child from cocoindex.connectors.localfs), create a wrapper function that adds your logic (logging, retries, metrics), then re-register the provider using coco.register_root_target_states_provider() with your custom sink as the action_sink parameter.

Can I use async operations in my custom connector's action sink?

Yes. The action sink receives a ContextProvider as its first argument and a list of actions as the second. You can define async logic within the sink function or use asyncio utilities to handle asynchronous side-effects. The PostgreSQL source connector demonstrates async patterns in python/cocoindex/connectors/postgres/_source.py.

Where should I store connection credentials when customizing connectors?

Never hardcode credentials. Instead, define a ContextKey for your connection object (pool, client, or credentials dict) and use coco.context_provider.provide() during application startup to inject the live resource. Your connector's action sink then retrieves this resource via ctx.get(context_key) at execution time, keeping sensitive data out of pipeline definitions.

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