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

> Learn to customize CocoIndex connector behavior by extending its declarative pipeline engine. Implement TargetHandler.reconcile() and register custom action sinks for greater control.

- Repository: [CocoIndex/cocoindex](https://github.com/cocoindex-io/cocoindex)
- Tags: how-to-guide
- Published: 2026-05-05

---

**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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`:

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```python

# 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`](https://github.com/cocoindex-io/cocoindex/blob/main/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.

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/_internal/context_keys.py).

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/connectors/localfs/_target.py) (lines 82-124 and 50-78).

### Reading from PostgreSQL Sources

Consume database rows using the PostgreSQL source connector:

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```python

# 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`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/_internal/context_keys.py) to inject dynamic configuration like database pools.
- **Reference implementation**: Study [`python/cocoindex/connectors/localfs/_target.py`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/connectors/localfs/_target.py) and [`python/cocoindex/connectors/postgres/_source.py`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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.