Metadata-Version: 2.4
Name: deltacat-client
Version: 0.1.27
Summary: Lightweight REST client and thin MCP scaffolding for the DeltaCAT API server.
Project-URL: Homepage, https://github.com/ray-project/deltacat
Author: Ray Team
License-Expression: Apache-2.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: <3.13,>=3.10
Requires-Dist: attrs>=22.2.0
Requires-Dist: deltacat-io-core<0.2,>=0.1.27
Requires-Dist: httpx<0.29.0,>=0.23.0
Requires-Dist: python-dateutil>=2.8.0
Requires-Dist: requests>=2.31
Provides-Extra: all
Requires-Dist: deltacat-io-core[all]<0.2,>=0.1.27; extra == 'all'
Provides-Extra: daft
Requires-Dist: deltacat-io-core[daft]<0.2,>=0.1.27; extra == 'daft'
Provides-Extra: io
Requires-Dist: deltacat-io-core[io]<0.2,>=0.1.27; extra == 'io'
Provides-Extra: lance
Requires-Dist: deltacat-io-core[lance]<0.2,>=0.1.27; extra == 'lance'
Provides-Extra: lerobot
Requires-Dist: deltacat-io-core[lerobot]<0.2,>=0.1.27; extra == 'lerobot'
Provides-Extra: mcp
Requires-Dist: mcp<2,>=1.10; extra == 'mcp'
Provides-Extra: packds
Requires-Dist: deltacat-io-core[packds]<0.2,>=0.1.27; extra == 'packds'
Provides-Extra: pandas
Requires-Dist: deltacat-io-core[pandas]<0.2,>=0.1.27; extra == 'pandas'
Provides-Extra: polars
Requires-Dist: deltacat-io-core[polars]<0.2,>=0.1.27; extra == 'polars'
Description-Content-Type: text/markdown

<p align="center">
  <img src="https://github.com/ray-project/deltacat/raw/2.0/media/deltacat-logo-alpha-750.png" alt="deltacat logo" style="width:55%; height:auto; text-align: center;">
</p>

`deltacat-client` is the primary Python client package for
[DeltaCAT](https://github.com/ray-project/deltacat). It lets you read and
write tables, run jobs, and build data pipelines against a DeltaCAT API
server without installing the full storage, compute, or server runtime
stack.

The client talks to a DeltaCAT server over HTTP. Metadata operations
(schema validation, transaction management, compaction) stay server-side,
while large data reads and writes go directly between the client and cloud
storage using short-lived credentials the server vends on demand.


## Overview

The client is organized around a root `Client` object with resource-oriented subclients:

| Subclient | Purpose |
|-----------|---------|
| `client.catalog` | Namespaces, tables, read/write, transactions |
| `client.jobs` | Job submission, claiming, lifecycle, progress |
| `client.publications` | Incremental producers into the DeltaCAT Lakehouse (pipeline root nodes) |
| `client.subscriptions` | Incremental consumers from the DeltaCAT Lakehouse (pipeline leaf nodes) |
| `client.transforms` | Incrementally transform data within the DeltaCAT Lakehouse (pipeline intermediate nodes) |
| `client.pipelines` | Wire publications, transforms, and subscriptions into a connected DAG |


## Installation

```bash
pip install deltacat-client
```

Optional extras for local data materialization:

```bash
pip install "deltacat-client[all]"    # Full client-side read/write stack
pip install "deltacat-client[pandas]" # Pandas DataFrame support
pip install "deltacat-client[polars]" # Polars DataFrame support
pip install "deltacat-client[daft]"   # Daft DataFrame support
pip install "deltacat-client[lance]"  # Lance dataset support
pip install "deltacat-client[mcp]"    # Typed async MCP HTTP client
```


## Getting Started

Before using the client, you need a running DeltaCAT API server. See
[Server Setup](docs/server-setup.md) for instructions.

DeltaCAT lets you manage **Tables** across one or more **Catalogs**. A
**Table** is a named collection of data files. A **Catalog** is a named
data lake that contains tables. For the full data model, see the
[DeltaCAT README](../README.md#getting-started).

### Quick Start

```python
from deltacat_client import Client
import pyarrow as pa

# Connect to a DeltaCAT server
client = Client("http://localhost:8080")

# Write data to a table. The table is created automatically; by default
# the client uses mode="auto" and lets the server infer the file format.
data = pa.table({
    "id": [1, 2, 3],
    "name": ["Cheshire", "Dinah", "Felix"],
    "age": [3, 7, 5],
})
client.catalog.write(data, table="cool_cats")

# Read the data back
df = client.catalog.read(table="cool_cats", read_as="pandas")
print(df)
```

### Core Concepts

Expand the sections below to see examples of core client operations.

<details>

<summary><b>Authentication</b></summary>

When connecting to a production server with auth enabled, provide a bearer token:

```python
from deltacat_client import Client

client = Client(
    "https://deltacat-api.example.com",
    bearer_token="your-api-token",
)
```

Admins can also onboard a new user, grant their initial access, and receive a
one-time token to share over a secure out-of-band channel:

```python
from deltacat_client import Client

admin = Client(
    "https://deltacat-api.example.com",
    bearer_token="admin-api-token",
)

created = admin.auth.create_user(
    user_id="newadmin@example.com",
    display_name="New Admin",
    email="newadmin@example.com",
    initial_role="ADMIN",
    resource_type="catalog",
    resource_name="*",
    issue_token_label="bootstrap",
    idempotency_key="create-newadmin-001",
)

bootstrap_token = created.token.token
print("Share this token once via a secure secret channel:", bootstrap_token)
```

The server validates your token and maps it to a user identity for permission checks (read, write, admin). The client automatically includes the token on every request. For data file access, the server vends short-lived STS credentials that the client uses to read from and write to cloud storage directly.

See [Configuration](docs/configuration.md) for the detailed auth model, bootstrap tokens, and role-based access control.

</details>

<details>

<summary><b>Reading Data</b></summary>

```python
# Read a table as a PyArrow table (the default format)
arrow_table = client.catalog.read(namespace="robotics", table="episodes")

# Read as Pandas, Polars, or Daft
df = client.catalog.read(namespace="robotics", table="episodes", read_as="pandas")
polars_result = client.catalog.read(namespace="robotics", table="episodes", read_as="polars")
daft_df = client.catalog.read(namespace="robotics", table="episodes", read_as="daft")

# Filter rows and limit results
df = client.catalog.read(
    namespace="robotics",
    table="episodes",
    read_as="pandas",
    filter_predicate={"eq": ["task", "pick_screwdriver"]},
    limit=5000,
)

# Time travel: read the table as it existed at a prior point in time.
# The as_of value is a nanosecond-precision Unix epoch timestamp.
df = client.catalog.read(
    namespace="robotics",
    table="episodes",
    read_as="pandas",
    as_of=1712697600000000000,
)
```

See [Reading and Writing](docs/reading-and-writing.md) for scalable reads,
lazy vs. eager materialization, and format-specific behavior.
For PackDS-backed training tables, see [Training Data](docs/training-data.md).

</details>

<details>

<summary><b>Computed Fields</b></summary>

Computed fields derive values while a PyArrow read is materialized. They may be
supplied for one request or persisted through schema evolution so later reads
resolve them automatically. Physical inputs can stay hidden from the result:

```python
import pyarrow.compute as pc

from deltacat_client import Client, ComputedField

client = Client("http://localhost:8080")
action = pc.field("action:action")

steps = client.catalog.read(
    namespace="robotics",
    table="training_steps",
    read_as="pyarrow",
    columns=["episode_id", "step_index"],
    computed_fields=[
        ComputedField(
            name="action_42",
            expression=pc.list_element(action, 42),
            source_fields=("action:action",),
        ),
    ],
)

# episode_id, step_index, and the extracted float32 action_42 are returned;
# the fixed_size_list action:action input is not.
print(steps.schema)
```

See [Computed Fields](docs/computed-fields.md) for the PackDS range-extraction
schema-evolution example, expression composition, projection and plan-reuse
semantics, validation, performance, and current limits.

</details>

<details>

<summary><b>Listing / paginating partitions</b></summary>

```python
# Page through a table's partitions (e.g. to pick a PackDS episode partition).
token = None
while True:
    res = client.catalog.query_partition_stats(namespace="xdof", table="my_steps", limit=100, after_token=token)
    for row in res.rows:               # each row: partition id/values + stats (record_count, ...)
        print(row)
    token = res.next_token
    if not token:
        break
```

PackDS `_steps` reads must be scoped to a single `episode_id` partition (pass
`partition_filter=[<partition_id>]` to `read`/`plan`); paginating
`query_partition_stats` is how you discover the valid partition ids to scope to.

</details>

<details>

<summary><b>Fast metadata-only planning (counts, sizes, existence)</b></summary>

`plan(..., include_files=False)` returns a **metadata-only** plan: the server
skips manifest hydration and file enumeration entirely, answering from catalog
partition statistics instead of resolving per-file URIs. It is the right call
whenever you do **not** intend to read row data — existence checks, row/byte
counts, content-type discovery, and (for PackDS) episode counts — and it stays
fast on tables with thousands of partitions where full file enumeration would
otherwise be expensive.

```python
# Metadata-only: O(partitions) catalog stats, no per-file resolution.
plan = client.catalog.plan(namespace="robotics", table="steps", include_files=False)

has_data   = plan.total_records > 0          # existence check
row_count  = plan.total_records              # summed from partition stats
byte_size  = plan.total_bytes
fmt        = plan.content_type
episodes   = plan.episode_count              # PackDS partition (episode) count
assert plan.total_files == 0 and not plan.scan_tasks   # no files were enumerated
```

A metadata-only plan carries **no `scan_tasks` and no vended credentials**, so it
cannot be used to read data. When you actually need to materialize rows, request
files — `include_files=True` resolves the manifest into `scan_tasks` (file URIs,
per-file stats, and short-lived credentials) so a downstream `read(...)` /
`pl.scan_parquet(...)` has something to scan:

```python
# Data read: resolve files (scope PackDS reads to a partition / episode_id).
plan  = client.catalog.plan(
    namespace="robotics", table="steps",
    filter_predicate={"eq": ["episode_id", "ep_1"]},
    include_files=True,
)
steps = client.catalog.read(namespace="robotics", table="steps", plan=plan, read_as="polars")
```

Long-running training brokers should use the bounded preparation facade. It
deduplicates physical planning while preserving the exact logical episode order
and multiplicity, waits for one immutable snapshot, renews server authority,
and returns an executable authorized plan:

```python
import uuid

prepared = client.catalog.prepare_packds_episode_plan(
    namespace="robotics",
    table="steps",
    operation_id=f"prepare-{uuid.uuid4().hex}",
    authority_operation_id=f"authority-{uuid.uuid4().hex}",
    episode_ids=["ep_2", "ep_1", "ep_2"],
    as_of=1712697600000000000,
    columns=["episode_id", "step_index", "state", "action"],
    timeout_seconds=300,
    run_id="n2-training-20260712-001",
)
authorized_plan = prepared.composition.plan
assert prepared.composition.requested_episode_ids == ("ep_2", "ep_1", "ep_2")
assert prepared.composition.composed_episode_ids == ("ep_2", "ep_1")
rows = client.catalog.read(
    plan=authorized_plan,
    read_as="pyarrow",
    plan_authority_verification_keys=prepared.plan_authority_verification_keys,
)
```

Trusted brokers can prepare overlapping consumers with one structural union
while retaining exact authority boundaries:

```python
prepared_by_rank = client.catalog.prepare_packds_episode_plans(
    namespace="robotics",
    table="steps",
    operation_id=f"prepare-bundle-{uuid.uuid4().hex}",
    authority_operation_id=f"authority-bundle-{uuid.uuid4().hex}",
    episode_id_groups={
        "rank-0000": ["ep_2", "ep_1", "ep_2"],
        "rank-0001": ["ep_1", "ep_3"],
    },
    as_of=1712697600000000000,
    columns=["episode_id", "step_index", "state", "action"],
    timeout_seconds=300,
)
rank_zero = prepared_by_rank["rank-0000"]
assert rank_zero.composition.requested_episode_ids == (
    "ep_2",
    "ep_1",
    "ep_2",
)
```

Each named group receives an independently signed exact authority with a common
bundle expiration. The helper never slices or reuses a covering union authority;
the entire grouped refresh fails closed if any child cannot be vended, signed,
persisted, replayed, or reauthorized.

The caller supplies stable operation IDs so transport retries remain
idempotent. `run_id` is an optional, non-secret provenance identifier sent on
every preparation, poll, authority, and verification-key request. Keep it
stable across retries for coherent telemetry. It does not alter operation,
cache, preparation, or authority identity. Launchers can validate their final
Gear ID before submission with
`deltacat_client.validate_packds_fragment_run_id(...)`. Pass a
`threading.Event` as `cancellation_event` for bounded
cooperative cancellation. The low-level fragment facade raises
`UnsupportedFeatureError` explicitly for old servers; the bounded high-level
facade raises `PackDSEpisodePlanPreparationUnsupported` without retrying it, so
callers can opt into a narrow legacy fallback without treating transient HTTP
failures as unsupported servers.
Preparation telemetry contains fixed-cardinality timing and retry counts and
never embeds episode, fragment, credential, or token values.

The lower-level split/compose primitives remain available for trusted brokers
that manage fragment persistence and authority refresh themselves:

```python
from deltacat_client import (
    compose_packds_episode_plan,
    split_packds_episode_plan,
)

plan = client.catalog.plan(
    namespace="robotics",
    table="steps",
    as_of=1712697600000000000,
    filter_predicate={"in": ["episode_id", ["ep_1", "ep_2"]]},
    include_files=True,
)
fragments = split_packds_episode_plan(plan)
cache.update({episode_id: fragment.to_dict() for episode_id, fragment in fragments.items()})
combined = compose_packds_episode_plan([fragments["ep_2"], fragments["ep_1"]])
assert combined.credentials is None
```

Fragment composition is intentionally limited to complete, non-MOR,
nontransactional PackDS episode plans with an explicit `as_of`. The hashes detect
cache corruption; they are not server signatures. The credential-free composed
plan is deliberately non-executable. A trusted broker must obtain a fresh
Ed25519-signed server authority covering the exact fragment fingerprints,
snapshot, roots, bindings, and credentials, then use
`compose_authorized_packds_episode_plan(...)` with a trusted server verification
key before executing it.

The metadata-only fast path fails closed on stale or dirty partition statistics
(it raises rather than return a stale count) and falls back to exact planning when
stats are incomplete, so the numbers it returns are never silently out of date.

</details>

<details>

<summary><b>Reading PackDS video blobs</b></summary>

A PackDS `_steps` table stores each camera — and each resolution — as its own
`video_blob:<key>` column. Frames are packed into 16-frame H.264 groups whose
*leader* row carries the byte-range reference (the other 15 rows are empty); a
companion `video_blob_frame_index:<key>` column records each row's position
(0..15) within its group. Read blobs through the client: the server vends
short-lived, scoped credentials on demand, so you never need direct S3 (or
S3-Express) access.

`step_index` (a real `_steps` column) is the **canonical selection coordinate** —
group-leader recovery is value-based on it, so selection stays correct under
filtering/reordering. Select by `step_index=` (canonical) or `where=` (a deltacat
filter_predicate dict applied client-side).

```python
# 1. Scope to one episode partition (see "Listing / paginating partitions" for how
#    to discover a partition_id). read(read_as="pyarrow") attaches the scoped
#    storage-authority lease to the returned table.
plan = client.catalog.plan(namespace="xdof", table="my_steps", partition_filter=[partition_id])
steps = client.catalog.read(namespace="xdof", table="my_steps", plan=plan, read_as="pyarrow")

# 2. Discover which camera/resolution blobs the table carries.
#    Bare key = native resolution; the "_320_240" suffix = the downscaled copy.
keys = client.catalog.packds_video_blob_keys(steps)
# ['video_blob:left_camera-images-rgb', 'video_blob:left_camera-images-rgb_320_240', ...]

# 3. Read a SPECIFIC blob for a step by step_index (canonical) or a predicate.
#    The 16-frame group leader is auto-resolved, so any step works. Returns the
#    raw H.264 group slice bytes (no `av` needed):
native = client.catalog.read_packds_video_blob(
    steps, video_blob_key="left_camera-images-rgb", step_index=42, plan=plan)
by_anno = client.catalog.read_packds_video_blob(
    steps, video_blob_key="left_camera-images-rgb",
    where={"eq": ["annotation", "grasp"]}, plan=plan)

# 4. Batch-resolve several cameras across several steps in ONE credential vend
#    -> dict keyed by (row, column):
blobs = client.catalog.resolve_packds_video_blobs(
    steps, video_blob_keys=keys, step_indices=[10, 26, 42], plan=plan)

# 5. The pyarrow accessor reuses ONE plan + fs cache across calls and recovers
#    the lease from the table automatically:
vb = client.catalog.video_blobs(steps)
data   = vb.blob("left_camera-images-rgb", step_index=42)             # bytes
frames = vb.frames("left_camera-images-rgb", step_index=42)           # list[np.ndarray]
many   = vb.resolve(keys, step_indices=[10, 26])                      # dict
```

#### High-level episode readers

These do their own episode-scoped read, so the group leader is always present —
you do not pre-read the steps table:

```python
# A single frame by ABSOLUTE frame number (frame is offset via min(step_index);
# fail-closed if outside the episode's step range). Decodes -> np.ndarray:
frame = client.catalog.read_packds_video(
    namespace="xdof", table="my_steps", episode_id=ep_id,
    video_blob_key="top_camera-images-rgb", frame=128, decode=True)

# Or by canonical step_index / predicate (returns the 16-frame group bytes):
group = client.catalog.read_packds_video(
    namespace="xdof", table="my_steps", episode_id=ep_id,
    video_blob_key="top_camera-images-rgb", step_index=128)

# Walk the WHOLE episode's video IN ORDER (group leaders only, sorted by
# step_index) to reassemble it — yields each 16-frame group:
for group_bytes in client.catalog.iter_packds_episode_video(
        namespace="xdof", table="my_steps", episode_id=ep_id,
        video_blob_key="top_camera-images-rgb"):
    ...  # concatenate / write out in order
```

#### Polars: resolve a column lazily

With polars installed, a `packds` namespace resolves the group leader for **every
current row** lazily (only the rows you `.collect()` are resolved). It force-keeps
the `video_blob_frame_index:<key>` companion and fails closed if it was projected
away:

```python
df = client.catalog.read(namespace="xdof", table="my_steps", plan=plan, read_as="polars")
resolved = (
    df.lazy()
      .packds.with_resolved_video("top_camera-images-rgb", out_column="rgb")
      .collect()
)  # adds an `rgb` column of raw group-slice bytes
```

Pass `plan=` (the scoped read plan) to reuse the credentials it already vended; or
omit it and pass `namespace=`/`table=` and the client vends a plan for you. The same
`where=` filter_predicate dict has identical semantics server-side (in `plan(...)`,
prunes files) and client-side (against a materialized table, scans rows). Decoding
needs `pip install "deltacat-client[packds]"` (pulls `av`); the raw-bytes path does not.

</details>

<details>

<summary><b>Converting a dataset to PackDS</b></summary>

Turn any source robotics dataset into the shared PackDS training corpus with a small
**format plugin** + one command. By default every dataset/run converges into one unified
corpus in the auto-created `iris` namespace:

- **`iris.packds_steps`** — the unified steps table, identity-partitioned by `episode_id`;
  **every** conversion appends here.
- **`iris.packds_episodes`** — its per-episode companion (`source_format`, `embodiment`,
  length, duration, video-codec metadata, …).
- **`iris.<dataset>_crawl` / `_diff` / `_placement` / `_facts`** — per-`--dataset`
  incremental state: one set per source dataset, reused across that source's runs, so
  re-running only converts what changed.

mp4pack video chunks land in the chunks S3-Express bucket and are referenced by-ref from
`packds_steps`. To pull one dataset back out of the shared table, filter on the
`episode_id` dataset prefix (it's the partition key → prunes files) or the companion's
`source_format` / `embodiment`. Pass `--out-table` to carve a separate, isolated corpus
instead.

**1 — write a format plugin.** Subclass `FormatLoader` (`deltacat_io_core.packds_format`):

```python
from deltacat_io_core.packds_format import FormatLoader, register_format

@register_format("my_format")
class MyLoader(FormatLoader):
    @property
    def name(self) -> str: return "my_format"
    @property
    def control_freq(self) -> float: return 30.0            # Hz
    def discover_episodes(self, input_path):                # -> list[dict] episode descriptors
        ...
    def load_episode(self, descriptor):                     # -> EpisodeStream of PackDSStep
        ...   # each step carries decoded_video / state / action / annotations
```

`deltacat_io_core/packds_lerobot.py::LeRobotLoader` is a complete in-repo example to copy.
(The gr00t XDOF loader lives on a gr00t branch and ships as a staged payload module — it is
the default `--plugin-source` below; pass your own to convert a new format.)

**2 — create the pipeline (client-only).** One thin-client call declares the whole
autonomous cascade `crawl → diff → place → convert` and triggers it from a single root
crawler. `dataset` is a stable slug that names the per-dataset incremental tables
(re-running the same source reuses them); `plugin_source` is YOUR loader module. The `iris`
namespace is created on demand (idempotent), and every run converges into the unified
`iris.<out_table>_steps` / `_episodes` corpus:

```python
from deltacat_client import Client, create_packds_converter

client = Client(server_url="https://deltacat-api.example.com", bearer_token="...")
result = create_packds_converter(
    client,
    dataset="my_dataset",
    seed_prefixes=["s3://my-source-bucket/my/episodes/"],
    plugin_source="./my_loader.py",   # your FormatLoader, staged as a flat payload module
    crawl_schedule="0 0 * * *",       # cron: re-crawl nightly (default; pass None for manual-only)
)
print(result.pipeline_id, result.triggered_nodes)  # created + the single root trigger fanned out
```

`create_packds_converter` returns a `PackDSConverterResult` carrying the created node ids
(`crawler_id` / `diff_id` / `placement_id` / `conversion_id` / `pipeline_id`).
`crawl_schedule` is persisted on the crawler as a cron trigger (see Scheduled Processing
below) so the schedule scanner re-crawls on the cadence; the first run still fires
immediately (`trigger_now=True`). Pass `dry_run=True` to print the full plan with no server
calls. In prod, pass `conversion_cluster_affinity="xlarge"` for multi-TB or
million+ episode backfills so they can only be claimed by the xlarge conversion
cluster. The convert-only operator CLI (`python -m deltacat.compute.training.submit_x_to_packds`,
for an already-crawled facts table) remains available for that narrower case.

**Multiple converters, one unified corpus (multi-writer).** Several `create_packds_converter`
pipelines may feed the SAME unified `iris.<out_table>_steps` / `_episodes` tables — this is
the intended end state for consolidating many sources into one queryable corpus. Each
converter flags its shared `_steps` / `_episodes` sinks as co-writable, and the server only
relaxes its single-writer guard when EVERY writer of a table is a mutually-flagged packds
converter (a packds converter never silently collides with an unrelated general transform).
The per-dataset `_facts` table stays single-writer.

**Duplicate rejection.** Before any server call, `create_packds_converter` rejects creating a
TRUE duplicate — a pipeline that feeds the SAME `_steps` table from the SAME crawl source
(same seed-prefix set + crawl binding) — and raises `PackDSDuplicatePipelineError` naming the
offending pipeline / conversion / crawler ids. DISTINCT sources feeding the same unified table
is allowed (that is the multi-writer case above). Pass `force=True` (a.k.a.
`allow_duplicate=True`) to create it anyway; under `dry_run=True` the finding is printed
instead of raised.

**Redrive is disabled by default.** Because a redrive does a WHOLE-TABLE rollback per declared
sink, redriving one converter would reverse EVERY co-writer's contributions to the shared
`_steps` / `_episodes` tables. The conversion transform is therefore created with redrive
disabled: the server rejects `client.transforms.redrive(conversion_id)` (409
`transform_redrive_disabled`) and `client.pipelines.redrive(pipeline_id)` (409
`pipeline_redrive_disabled`). To replace episodes, drive the change through the SOURCE
instead — see [PackDS Episode Rewrites](#) below: update the crawled source so the diff emits
the affected `episode_id`s as REPLACE (per-episode `steps_write_mode=REPLACE`), or
delete/deprecate the offending episode-ID partitions. (Proper shared-table redrive is tracked
separately.)

**Run an existing converter on demand.** Re-crawl + reconvert the latest source files
whenever you want by running the converter's root crawler publication. Use the
`crawler_id` from the `PackDSConverterResult` (the cascade fans out from this single
trigger — diff → place → convert drain automatically):

```python
run_result = client.publications.run(result.crawler_id)
print(run_result.triggered_nodes)  # the downstream nodes the crawl completion fanned out to
```

**Update the configured crawl schedule.** Change the cron cadence of an existing
converter's crawler with a new `schedule` trigger; the server persists the new
`schedule_cron` and the schedule scanner picks up the next fire on the new cadence. A
manual `publications.run(...)` still works alongside the schedule:

```python
from deltacat_client import PublicationTrigger

# Re-crawl every 6 hours instead of nightly (pass cron=None / a different trigger to clear).
client.publications.update(
    result.crawler_id,
    trigger=PublicationTrigger.schedule(cron="0 */6 * * *", timezone="UTC"),
)
```

</details>

<details>

<summary><b>Writing Data</b></summary>

The client supports writing PyArrow tables, Pandas DataFrames, Polars DataFrames, Daft DataFrames, NumPy arrays, Ray Datasets, and local files (Parquet, CSV, TSV, PSV, Feather, JSON, ORC, AVRO, Lance).

```python
import pyarrow as pa

# Write data to a table. The defaults are mode="auto" (creates the
# table on first write, appends thereafter) and the client default Parquet format.
data = pa.table({"episode_id": [1, 2], "score": [0.95, 0.87]})
client.catalog.write(data, namespace="robotics", table="predictions")

# Append more data — same call, same defaults.
data2 = pa.table({"episode_id": [3, 4], "score": [0.91, 0.89]})
client.catalog.write(data2, namespace="robotics", table="predictions")

# Write a Pandas DataFrame
import pandas as pd
df = pd.DataFrame({"episode_id": [5], "score": [0.93]})
client.catalog.write(df, namespace="robotics", table="predictions")

# Override the defaults when you need a specific mode or format. Here we
# create a Lance-formatted table with explicit schema, partitioning, and
# sort order.
from deltacat_io_core.schemes import (
    PartitionKey,
    PartitionScheme,
    PartitionTransform,
    SortKey,
    SortOrder,
    SortScheme,
)

client.catalog.create_table(
    namespace="robotics",
    table="scored_episodes",
    schema=pa.schema([
        pa.field("episode_id", pa.int64()),
        pa.field("score", pa.float64()),
        pa.field("episode_day", pa.string()),
    ]),
    partition_scheme=PartitionScheme.of([
        PartitionKey(key=["episode_day"], transform=PartitionTransform.IDENTITY),
    ]),
    sort_scheme=SortScheme.of([
        SortKey(key=["episode_id"], sort_order=SortOrder.DESCENDING),
    ]),
    auto_create_namespace=True,
)

data = pa.table({
    "episode_id": [1, 2],
    "score": [0.95, 0.87],
    "episode_day": ["2025-01-15", "2025-01-15"],
})
client.catalog.write(data, namespace="robotics", table="scored_episodes")

# Evolve a table's schema after creation
from deltacat_io_core.schema_updates import add_field

client.catalog.alter_table(
    namespace="robotics",
    table="predictions",
    schema_updates=[add_field(pa.field("confidence", pa.float64()))],
)
```

See [Reading and Writing](docs/reading-and-writing.md) for write modes,
staged writes, path-based writes with `create_if_missing`, and the full
list of supported formats and schema inputs.
For PackDS training data writes, see [Training Data](docs/training-data.md).

</details>

<details>

<summary><b>Live Feature Enrichment</b></summary>

Declare one or more merge keys on the table schema and subsequent writes
update existing records in place.

```python
import pyarrow as pa
import deltacat as dc
from deltacat_client import Client

client = Client("http://localhost:8080")

# Fields tagged with is_merge_key=True identify the key columns used to
# reconcile updates.
schema = dc.Schema.of([
    dc.Field.of(pa.field("user_id", pa.int64()), is_merge_key=True),
    dc.Field.of(pa.field("name", pa.string())),
    dc.Field.of(pa.field("age", pa.int32())),
    dc.Field.of(pa.field("job", pa.string())),
])

# First write creates a Lance table with the merge-key schema.
initial = pa.table({
    "user_id": pa.array([1, 2, 3], type=pa.int64()),
    "name": ["Jim", "Dinah", "Bob"],
    "age":  pa.array([30, 28, 45], type=pa.int32()),
    "job":  ["Teacher", "Painter", "Sailor"],
})
client.catalog.write(
    initial,
    namespace="demo",
    table="users",
    format="lance",
    create_if_missing={"schema": schema, "auto_create_namespace": True},
)

# Upsert: user_id 1 + 3 are updated; 4 and 5 are inserted.
upsert = pa.table({
    "user_id": pa.array([1, 3, 4, 5], type=pa.int64()),
    "name": ["Cheshire", "Felix", "Tom", "Simpkin"],
    "age":  pa.array([3, 2, 5, 12], type=pa.int32()),
    "job":  ["Tour Guide", "Drifter", "Housekeeper", "Mouser"],
})
client.catalog.write(
    upsert,
    namespace="demo",
    table="users",
    mode="merge",
    format="lance",
)

print(client.catalog.read(namespace="demo", table="users", read_as="pyarrow"))

# Delete: only the merge-key columns need to be supplied.
client.catalog.write(
    pa.table({"user_id": pa.array([3, 5], type=pa.int64())}),
    namespace="demo",
    table="users",
    mode="delete",
    format="lance",
)

# user_id 3 and 5 are gone; 1, 2, 4 remain.
print(client.catalog.read(namespace="demo", table="users", read_as="pyarrow"))
```

PackDS v5 tables do not support DeltaCAT merge keys. For canonical
PackDS training data, use `episode_id` identity partitioning and
`REPLACE_PARTITION`. If you need row-level `MERGE`/`DELETE` semantics,
use a normal DeltaCAT table instead of `layout="packds"`.
See [Training Data](docs/training-data.md).

#### PackDS Episode Rewrites

PackDS v5 tables are identity-partitioned by `episode_id` and reject
merge keys. Live feature enrichment runs episode-at-a-time: discover the
episodes you want to enrich from the public `<table>__episodes` companion
alias, then rewrite each target episode with
`mode="replace_partition"`. The replaced partition is committed atomically;
sibling episodes are never touched.

```python
import pyarrow as pa
from deltacat_client import Client

client = Client("http://localhost:8080")

# 1. Initial PackDS write across multiple episodes. DeltaCAT auto-creates
#    the table with episode_id identity partitioning, no merge keys, and
#    the single managed `<table>_info__system` system companion table. Client
#    reads should use the `episodes_table` field from the read plan.
initial = pa.table({
    "episode_id": [
        "ep_0001", "ep_0001", "ep_0001",
        "ep_0002", "ep_0002",
        "ep_0003", "ep_0003", "ep_0003", "ep_0003",
    ],
    "step_index": [0, 1, 2, 0, 1, 0, 1, 2, 3],
    "task": [
        "pick", "pick", "pick",
        "place", "place",
        "stack", "stack", "stack", "stack",
    ],
    "annotation": [None] * 9,
})
client.catalog.write(
    initial,
    namespace="robotics",
    table="steps",
    format="lance",
    create_if_missing={
        "schema_def": [
            {"name": "episode_id", "type": "string"},
            {"name": "step_index", "type": "int64"},
            {"name": "task", "type": "string"},
            {"name": "annotation", "type": "string"},
        ],
        "layout": "packds",
        "default_content_type": "lance",
        "auto_create_namespace": True,
    },
)

# 2. Discover which episodes exist by reading the companion table.
#    Companion rows are maintained after PackDS commits by default, so
#    recently-written rows may be briefly absent while background maintenance
#    catches up.
#    A full companion enumeration must opt in explicitly. Production hot paths
#    should normally use a partition filter or a bounded episode_id predicate.
companion = client.catalog.read(
    namespace="robotics",
    table="steps__episodes",
    read_as="pyarrow",
    columns=["episode_id", "total_frames"],
    allow_full_companion_scan=True,
).sort_by("episode_id")

# Pick the first episode by id. The companion's read order is not
# guaranteed without a sort, so we sort first and pull both the id and
# the row count from the same row.
target_id = companion["episode_id"][0].as_py()
target_len = companion["total_frames"][0].as_py()

# 3. Enrich the target episode end-to-end via REPLACE_PARTITION. No
#    merge keys, no merge-on-read. The replacement is per-partition;
#    sibling episodes are unaffected. Note that ``layout="packds"`` is
#    only required at create time; subsequent writes pick the layout up
#    from the table's stored ``dataset_layout`` property.
enriched = pa.table({
    "episode_id": [target_id] * target_len,
    "step_index": list(range(target_len)),
    "task": ["pick"] * target_len,
    "annotation": [f"auto-labeled-step-{i}" for i in range(target_len)],
})
client.catalog.write(
    enriched,
    namespace="robotics",
    table="steps",
    mode="replace_partition",
    format="lance",
)

# 4. Scoped read of the enriched episode. PackDS reads must be scoped
#    to a partition filter or a bounded set of episode_ids (equality / IN,
#    capped by ``DELTACAT_PACKDS_EPISODE_ID_FILTER_MAX_FANOUT``, default
#    1024); unscoped full-table PackDS reads are rejected by default and
#    require ``allow_full_packds_scan=True`` (or
#    ``DELTACAT_ALLOW_FULL_PACKDS_SCAN=true`` server-side) to opt in.
print(client.catalog.read(
    namespace="robotics",
    table="steps",
    read_as="pyarrow",
    filter_predicate={"eq": ["episode_id", target_id]},
    columns=["episode_id", "step_index", "task", "annotation"],
))
```

Larger enrichment loops typically batch over filtered companion reads
(e.g. `filter_predicate={"in": ["episode_id", batch_ids]}` within the
server fanout cap, or partition-filter pagination for larger batches) so
each `replace_partition` write touches exactly one episode while the
worker pool processes many in parallel. Because each episode is its own
DeltaCAT partition, concurrent replace-partition writes to disjoint
episodes do not contend.

See [Training Data](docs/training-data.md) for the companion-table
schema and guidance on choosing between PackDS episode rewrites and a
normal merge-key table.

</details>

<details>

<summary><b>Transactions</b></summary>

Transactions provide atomic multi-step operations with automatic heartbeating and rollback on failure.

```python
import pyarrow as pa

with client.transaction(commit_message="Backfill predictions") as tx:
    client.catalog.write(
        pa.table({"episode_id": [10, 11], "score": [0.88, 0.92]}),
        namespace="robotics",
        table="predictions",
        mode="add",
    )

    # Reads within the transaction see uncommitted writes
    df = client.catalog.read(
        namespace="robotics",
        table="predictions",
        read_as="pandas",
    )
    print(f"Rows visible in transaction: {len(df)}")
# Transaction commits automatically on exit; aborts on exception
```

See [Transactions](docs/transactions.md) for time-travel reads, manual commit/abort, and transaction rules.

</details>

<details>

<summary><b>Jobs</b></summary>

DeltaCAT uses a durable job system for background work (compaction, data relay, subscription processing). The client can submit, monitor, and execute jobs.

```python
# List all jobs
jobs = client.jobs.list()
for job in jobs:
    print(f"{job.job_id}: {job.state}")

# Submit a compaction job
result = client.jobs.submit_compaction(table="predictions")
print(f"Submitted: {result.job_id}")

# Wait for it to complete
status = client.jobs.wait(result.job_id, timeout_seconds=120)
print(f"Final state: {status.state}")
```

Workers claim and execute jobs. Any process can act as a worker:

```python
# Claim the next pending job matching our worker tags
job = client.jobs.claim(worker_tags=["subscriber"])
if job:
    print(f"Claimed {job.job_id} (type: {job.context.get('job_type')})")

    # Report fine-grained progress and declare the deadline for the next chunk
    client.jobs.emit_event(
        job,
        event_name="batch_started",
        completed=1,
        expected=4,
        metadata={"batch": 1},
        heartbeat_timeout_seconds=300,
    )

    # Do the work...

    # Mark complete. Source-consuming jobs must include the advanced
    # watermark so the server can persist progress correctly.
    client.jobs.complete(
        job,
        records_processed=1000,
        watermark={
            "partition_watermarks": {"analytics.events": 42},
            "known_partitions": ["analytics.events"],
        },
    )
```

See [Jobs and Workers](docs/jobs-and-workers.md) for job types, worker
routing, heartbeat rules, retry semantics, and dispatch modes.

### Managed Ray Example

For subscription, transform, and publication jobs, `dispatch_mode="ray"`
submits the work onto DeltaCAT-managed Ray clusters instead of the shared
custom-worker claim pool. For large prebuilt environments, prefer
`docker.image` as the primary runtime artifact and use `payload` only for
small code/config overlays. The server still vends the DeltaCAT runtime from
its promoted runtime manifest so the remote cluster gets compatible
`deltacat`, `deltacat_client`, and `deltacat_io_core` packages.

```python
from deltacat_client import Client, DispatchMode, SubscriberType
from deltacat_io_core.triggers import SubscriptionTrigger

client = Client("http://localhost:8080", bearer_token="token")

payload = client.jobs.stage_ray_payload(
    py_modules=["./shared_logic"],
)

ray_dispatch_yaml = """
cluster_shutdown_policy: terminate
docker:
  image: "registry.example.com/ml/runtime@sha256:0123456789abcdef"
payload:
  py_modules:
    - {py_module}
head_node_type: ray.head.default
available_node_types:
  ray.head.default:
    node_config:
      InstanceType: m7i.2xlarge
  ray.worker.default:
    min_workers: 1
    max_workers: 4
    node_config:
      InstanceType: m7i.2xlarge
""".format(
    py_module=payload.payload["py_modules"][0],
)

client.subscriptions.create(
    subscriber_id="episode_processor",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    subscriber_type=SubscriberType.CUSTOM,
    dispatch_mode=DispatchMode.RAY,
    dispatch_config=ray_dispatch_yaml,
    trigger=SubscriptionTrigger.schedule(interval_seconds=300),
)
```

The scheduler now auto-triggers the subscription every five minutes, DeltaCAT
auto-provisions the Ray cluster when a run starts, and
`cluster_shutdown_policy: terminate` tears the cluster back down after each run
by default. If the deployment advertises a compatible image profile, DeltaCAT
will execute inside the requested `docker.image` while keeping the promoted
DeltaCAT runtime bundle as the compatibility layer.

For reference, the Ray launcher YAML above is:

```yaml
cluster_shutdown_policy: terminate
docker:
  image: registry.example.com/ml/runtime@sha256:0123456789abcdef
payload:
  py_modules:
    - s3://.../shared_logic.tar.gz
head_node_type: ray.head.default
available_node_types:
  ray.head.default:
    node_config:
      InstanceType: m7i.2xlarge
  ray.worker.default:
    min_workers: 1
    max_workers: 4
    node_config:
      InstanceType: m7i.2xlarge
```

Use `payload` for small overlays only:

- experiment modules
- config bundles
- supplemental wheels

Do not treat `payload` as the primary transport for a large application image.

After launch, owner/admin callers can inspect cluster metadata and then use
the managed-Ray access helpers against a specific job:

```python
# `access` describes the currently discovered cluster shape for the job:
# region, cluster name, head instance id/private IP, worker ids, and the
# launch template metadata DeltaCAT used for the cluster.
access = client.jobs.get_managed_ray_access("job-id")
print(access.head_instance_id, access.head_private_ip)

# By default the SSM session starts on the Ray head node.
# The AWS CLI / Session Manager plugin will open an interactive session in
# your terminal, and the helper asks SSM to launch `bash -l` there so you land
# directly in a login shell on the instance.
client.jobs.start_managed_ray_ssm_session("job-id")

# For non-interactive inspection across the whole cluster, use send-command.
client.jobs.send_managed_ray_ssm_command("job-id", commands=["hostname"], target="all")
```

</details>

<details>

<summary><b>Publications</b></summary>

Publications are incremental producers that write new data into the
DeltaCAT Lakehouse. They sit at the root of a pipeline DAG and can be
triggered manually or fired by an upstream event.

```python
# Create a publication that writes to a sink table
client.publications.create(
    publication_id="episode_publisher",
    name="Episode Publisher",
    sink_tables=[{"namespace": "robotics", "table": "clean_episodes"}],
    dispatch_mode="local",
)

# Run the publication
result = client.publications.run("episode_publisher")
print(f"Published: {result}")
```

See [Pipelines](docs/pipelines.md) for publication configuration and DAG construction.

</details>

<details>

<summary><b>Subscriptions</b></summary>

Subscriptions are incremental consumers of DeltaCAT tables. They sit at
the leaves of a pipeline DAG, tracking a per-partition watermark so each
run picks up only new data.

```python
# Create a subscription that watches for new data in "raw_episodes"
client.subscriptions.create(
    subscriber_id="episode_processor",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    subscriber_type="custom",
    dispatch_mode="custom",
)

# Trigger processing (dispatches a job to a subscriber worker)
client.subscriptions.trigger("episode_processor")

# Check watermark state
wm = client.subscriptions.get_watermark("episode_processor")
print(f"Watermark: {wm.watermark}")

# Pause / resume / delete
client.subscriptions.pause("episode_processor")
client.subscriptions.resume("episode_processor")
client.subscriptions.delete("episode_processor")
```

See [Pipelines](docs/pipelines.md) for subscription modes (delta vs. version), triggers, and redrive.

</details>

<details>

<summary><b>Transforms</b></summary>

Transforms are the intermediate nodes of a pipeline DAG. Each transform
reads from one or more source tables, applies processing logic, and
writes to one or more sink tables.

```python
# Create a transform: raw_episodes -> clean_episodes
client.transforms.create(
    transform_id="episode_cleaner",
    name="Episode Cleaner",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    sink_tables=[{"namespace": "robotics", "table": "clean_episodes"}],
    dispatch_mode="custom",
)

# Trigger transform processing
client.subscriptions.trigger("episode_cleaner")

# Pause / resume
client.transforms.pause("episode_cleaner")
client.transforms.resume("episode_cleaner")
```

See [Pipelines](docs/pipelines.md) for transform configuration, redrive, and rollback.

</details>

<details>

<summary><b>Pipelines</b></summary>

Pipelines wire publications, transforms, and subscriptions into a
connected DAG. When an upstream node completes, downstream nodes are
triggered automatically.

```python
# First, create connected pipeline nodes
client.publications.create(
    publication_id="ingest_pub",
    name="Raw Ingest Publisher",
    sink_tables=[{"namespace": "robotics", "table": "raw_data"}],
    dispatch_mode="local",
)
client.transforms.create(
    transform_id="clean",
    name="Data Cleaner",
    source_tables=[{"namespace": "robotics", "table": "raw_data"}],
    sink_tables=[{"namespace": "robotics", "table": "clean_data"}],
    dispatch_mode="custom",
)
client.subscriptions.create(
    subscriber_id="consume_clean",
    source_tables=[{"namespace": "robotics", "table": "clean_data"}],
    subscriber_type="custom",
    dispatch_mode="custom",
)

# Preview the connected pipeline from an interior seed node
preview = client.pipelines.discover(seed_node_ids=["clean"])
print(preview.execution_order)

# Option A: persist exactly the previewed node_ids
client.pipelines.create(
    pipeline_id="etl_pipeline_pinned",
    name="ETL Pipeline (Pinned)",
    node_ids=preview.node_ids,
)

# Option B: create directly from seed node ids
client.pipelines.create(
    pipeline_id="etl_pipeline_seeded",
    name="ETL Pipeline (Seeded)",
    seed_node_ids=["clean"],
)

# Check pipeline status
status = client.pipelines.status("etl_pipeline_seeded")

# Pause / resume all nodes at once
client.pipelines.pause("etl_pipeline_seeded")
client.pipelines.resume("etl_pipeline_seeded")
```

See [Pipelines](docs/pipelines.md) for DAG construction, discovery
semantics, redrive, rollback, and stored-order validation.

</details>

<details>

<summary><b>Data Placement and Replication</b></summary>

DeltaCAT catalogs can span multiple storage backends (S3, SwiftStack, Lustre). Data placement lets you replicate tables across roots so readers access data from the closest location.

```python
# List available data roots for the catalog. The keys of this dict
# are the valid values to pass as `roots=` and `root=` below.
available_roots = client.catalog.list_data_roots("default")
for root_name, root_info in available_roots.items():
    print(root_name, root_info)

# Example output (dict[str, DataRootInfoSummary]):
# aws_s3_iad       {'root': 's3://bucket-iad/',        'storage_type': 's3',
#                   'region': 'us-east-1', 'endpoint_url': None}
# aws_s3_pdx       {'root': 's3://bucket-pdx/',        'storage_type': 's3',
#                   'region': 'us-west-2', 'endpoint_url': None}
# swiftstack_pdx   {'root': 's3://swiftstack-bucket/', 'storage_type': 's3',
#                   'region': None,        'endpoint_url': 'https://pdx.swiftstack.example.com'}

# Pick a target root from what the catalog actually advertises.
# In production you'd choose based on region, storage class, or policy.
if not available_roots:
    print("This catalog uses a single default root; no placement root to select.")
else:
    target_root = next(iter(available_roots))

    # Place a table on an additional storage root for replication
    client.catalog.place(
        namespace="robotics",
        table="episodes",
        roots=[target_root],         # Replicate to this root
        backfill=True,               # Copy existing data too
    )

    # Check replication status
    status = client.catalog.replication_status(namespace="robotics", table="episodes")
    print(f"Roots: {status}")

    # Read from the closest root (server resolves automatically)
    df = client.catalog.read(
        namespace="robotics",
        table="episodes",
        read_as="pandas",
        root=target_root,            # Prefer this root for file paths
    )

    # Remove a replication target
    client.catalog.unplace(namespace="robotics", table="episodes", root=target_root)
```

New writes are automatically replicated to all placed roots via a background subscriber. Reads with `root=` get file paths rewritten through the preferred root when data is available there.

See [Configuration](docs/configuration.md) for data root setup and multi-root catalog configuration.

</details>

<details>

<summary><b>Discovery Root Administration</b></summary>

Catalog admins can publish read-only discovery roots for system crawler, GC,
relay, replication, and placement jobs. Discovery roots are separate from
managed data roots, so they can describe authority roots such as `s3://`
without becoming valid table write targets.

```python
client.catalog.create_storage_binding(
    "default",
    name="swiftstack_pdx_root",
    storage_type="s3",
    storage_class="discovery",
    uri_scheme="s3",
    endpoint_url="https://pdx.swiftstack.example.com",
    region="us-west-2",
    credential_source={"kind": "aws_profile", "profile": "pdx"},
)

client.catalog.create_discovery_root(
    "default",
    name="swiftstack_pdx",
    binding_name="swiftstack_pdx_root",
    root_uri="s3://",
    authorization={"principals": ["service:system-crawler"]},
)

roots = client.catalog.list_discovery_roots(
    "default",
    include_disabled=True,
    include_history=True,
)
details = client.catalog.describe_discovery_root(
    "default",
    "swiftstack_pdx",
    include_history=True,
)

# Updates create a new immutable discovery-root version.
client.catalog.update_discovery_root(
    "default",
    "swiftstack_pdx",
    binding_name="swiftstack_pdx_root",
    root_uri="s3://",
    description="PDX authority root",
    authorization={"principals": ["service:system-crawler"]},
)

# Deletes are logical disables and remain visible with include_disabled=True.
client.catalog.delete_discovery_root("default", "swiftstack_pdx")
```

Discovery-root detail responses include the effective binding revision,
endpoint, region, authorization summary, lifecycle state, sanitized
credential-source metadata, and effective storage snapshot.

</details>

<details>

<summary><b>Scheduled Processing</b></summary>

Subscriptions and transforms can run on a schedule instead of being triggered manually. DeltaCAT supports interval-based and cron-based scheduling.

```python
# Process new data every 5 minutes
client.subscriptions.create(
    subscriber_id="metrics_ingester",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "schedule", "schedule": {"interval_seconds": 300}},
)

# Process at 2am UTC daily using a cron expression
client.subscriptions.create(
    subscriber_id="nightly_aggregator",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "schedule", "schedule": {"cron": "0 2 * * *", "timezone": "UTC"}},
)

# Event-driven: only runs when triggered manually or by an upstream pipeline node
client.subscriptions.create(
    subscriber_id="on_demand_processor",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "event"},
)
```

See [Configuration](docs/configuration.md) for trigger options and scheduling details.

</details>


## Agentic Access (MCP)

DeltaCAT ships with a built-in [Model Context
Protocol](https://modelcontextprotocol.io) server so AI agents (for
example, Claude Code) can browse catalogs, inspect schemas, plan reads,
and stage writes through natural language instead of hand-written Python.

Most hand-written code should just use the REST `Client(...)` shown
above. Reach for MCP when you want to:

- let an agent explore and operate on your catalog conversationally
- embed DeltaCAT as a tool in an agentic application
- use a typed async Python wrapper over the MCP HTTP surface
  (`deltacat-client[mcp]`) from code that is already agentic in shape

See the [MCP Server guide](../deltacat/docs/mcp/README.md) for the full
tool reference, the typed async client, and recipes for agent-driven
catalog workflows.


## Server Setup

The DeltaCAT client connects to a DeltaCAT API server. For setup instructions, see:

- **[Server Setup Guide](docs/server-setup.md)** -- start a local or production server
- **[REST API Reference](../deltacat/docs/server/README.md)** -- full REST endpoint documentation
- **[MCP Server Guide](../deltacat/docs/mcp/README.md)** -- agentic access via MCP


## Additional Resources

| Guide | Description |
|-------|-------------|
| [Reading and Writing](docs/reading-and-writing.md) | Read plans, write modes, staged writes, supported formats |
| [Computed Fields](docs/computed-fields.md) | Persisted and request-time PyArrow expressions, schema evolution, projection rules, and validation |
| [Training Data](docs/training-data.md) | PackDS v5 tables, episodes companion, scoped training reads |
| [Transactions](docs/transactions.md) | Transaction lifecycle, time travel, rules and limitations |
| [Jobs and Workers](docs/jobs-and-workers.md) | Job types, claiming, heartbeat, worker routing, authentication |
| [Pipelines](docs/pipelines.md) | Publications, transforms, subscriptions, DAGs, redrive |
| [Configuration](docs/configuration.md) | Auth, dispatch modes, triggers, data placement |
| [Maintainer Workflow](docs/maintainer-workflow.md) | Relationship between REST and MCP, generated bindings, facade updates, and validation guards |
| [Architecture](docs/architecture.md) | Package boundary, generated client, development notes |
| [Client Compatibility Runbook](../deltacat/docs/deploy/CLIENT_COMPATIBILITY_RUNBOOK_2026-04-12.md) | Promotion-time packaged client/server compatibility validation |

For the core DeltaCAT data model, storage architecture, and catalog APIs, see the [DeltaCAT documentation](../README.md#overview).
