eddytor-sdk 2.0.0

Python SDK for Eddytor - Connect via Apache Arrow Flight SQL, REST API, and MCP

Eddytor Python SDK

Python SDK for connecting to Eddytor — Master Data Management powered by Apache Arrow and Delta Lake.

Built for developer experience: one client, automatic table discovery, pandas/PyArrow in and out, and first-class use from notebooks (Databricks, Jupyter, Hex, …).

Installation

pip install eddytor-sdk
# Optional: persist OAuth tokens in the OS keychain
pip install eddytor-sdk[auth]

Connecting

The SDK talks to your Eddytor server (default: a self-hosted install on http://localhost:8080). Pass url= or set EDDYTOR_API_URL:

from eddytor_sdk import EddytorClient

# API key (mint one with `eddytor api-keys create` or via the web UI)
client = EddytorClient(api_key="edd_live_xxx", url="http://localhost:8080")

# …or sign in interactively (OAuth 2.1 device-code flow, no key needed)
client = EddytorClient.login("http://localhost:8080")

The server proxies everything the SDK needs:

Protocol	Default port	Override
REST + native gRPC	URL port (8080)	EDDYTOR_API_PORT / EDDYTOR_GRPC_PORT
Flight SQL	8082	EDDYTOR_FLIGHT_PORT

TLS follows the URL scheme (https:// → TLS on). For TLS with a self-signed edge certificate set EDDYTOR_SKIP_VERIFY=true.

OAuth 2.1 Device-Code Login

Instead of copying an API key around, sign in interactively. The SDK opens your browser, you authenticate, and tokens are cached in the OS keychain so subsequent runs skip the prompt.

from eddytor_sdk import EddytorClient, device_login

tokens = device_login("http://localhost:8080")   # or any Eddytor server URL
with EddytorClient(api_key=tokens.access_token, url=tokens.api_url) as client:
    for t in client.tables():
        print(t.fqn)

EddytorClient.login(url) wraps these two steps. Force a fresh interactive login with clear_cached(api_url), refresh with tokens.refresh() when tokens.is_expired() returns True (refreshed tokens are persisted to the same keychain entry). Note: your user must belong to an organisation — provision the first admin with eddytoradm setup on a fresh install.

Databricks Notebooks

Use an API key from a Databricks secret scope — no interactive login needed on a cluster:

from eddytor_sdk import EddytorClient

client = EddytorClient(
    api_key=dbutils.secrets.get("eddytor", "api_key"),
    url="https://eddytor.example.com",
)

# Pull master data into a Spark DataFrame
tables = client.tables()
customers = next(t for t in tables if "customers" in t.name)
spark_df = spark.createDataFrame(customers.query_all())

# Append new rows (auto-aligns to the table schema)
customers.insert(new_rows_df.toPandas())

# Or upsert: tag each row with an _operation (INSERT / UPDATE / DELETE)
curated = spark_df.toPandas()
curated["_operation"] = "UPDATE"
customers.merge(curated)

Understanding Table Names (FQN)

Every table in Eddytor has a three-part Fully Qualified Name (FQN):

eddytor.cfg_<config_id_hex>.<table_id>_<table_name>

Part	Value	Example
Catalog	Always "eddytor" (fixed)	eddytor
Schema	cfg_ + storage config UUID (dashes removed)	cfg_550e8400e29b41d4a716446655440000
Table	Stable table ID + human-readable name	a1b2c3d4e5f67890abcdef1234567890_customers

The table ID is the stable UUID from the Delta protocol's metaData.id field, assigned at table creation. It does not change when a table is moved or renamed, so the FQN remains constant.

You don't need to construct these yourself. Use client.tables() to discover all available tables with FQNs already resolved.

Quick Start

from eddytor_sdk import EddytorClient

with EddytorClient(api_key="edd_live_xxx") as client:
    # Step 1: Discover your tables
    tables = client.tables()
    for t in tables:
        print(f"{t.name} -> {t.fqn}")
    # Output: a1b2c3d4..._customers -> `eddytor`.`cfg_550e8400...`.`a1b2c3d4..._customers`

    # Step 2: Work with a table
    customers = [t for t in tables if "customers" in t.name][0]
    df = customers.query_all(limit=10)
    print(df)

    # Or query with raw SQL using the FQN
    df = client.query(f"SELECT * FROM {customers.fqn} WHERE status = 'active'")

Flight SQL Operations

# Discover tables first
tables = client.tables()
my_table = tables[0]

# Query as pandas DataFrame
df = client.query(f"SELECT * FROM {my_table.fqn} LIMIT 10")

# Query as PyArrow Table
arrow_table = client.query_arrow(f"SELECT * FROM {my_table.fqn}")

# Execute and get raw tuples
rows = client.execute(f"SELECT id, name FROM {my_table.fqn}")

# DML operations (INSERT, UPDATE, DELETE)
count = client.execute_dml(f"INSERT INTO {my_table.fqn} VALUES (1, 'hello')")
count = client.execute_dml(f"DELETE FROM {my_table.fqn} WHERE id = 1")

# Bulk ingest (Arrow or pandas) — uses the table's catalog/schema/name parts
import pyarrow as pa
data = pa.table({"id": [1, 2], "name": ["Alice", "Bob"]})
client.ingest(my_table.name, data, mode="append", catalog=my_table.catalog, schema=my_table.schema)

# List schemas and tables (raw SQL)
schemas = client.list_schemas()
tables_in_schema = client.list_tables(schema="cfg_550e8400e29b41d4a716446655440000")

# Count rows
total = client.count(my_table.fqn)
filtered = client.count(my_table.fqn, "status = 'active'")

# Interactive SQL session
client.interactive()

DDL via Flight SQL

import pyarrow as pa

# Create a new table (provide the object store location)
schema = pa.schema([
    pa.field("id", pa.int64(), nullable=False),
    pa.field("name", pa.string()),
    pa.field("created_at", pa.timestamp("us", tz="UTC")),
])
client.create_table("my_table", "s3://bucket/my_table", schema)

# Add columns — use a discovered table's catalog/schema
tables = client.tables()
my_table = [t for t in tables if "my_table" in t.name][0]
new_cols = pa.schema([pa.field("email", pa.string())])
client.add_column(my_table.name, new_cols, catalog=my_table.catalog, schema=my_table.schema)

REST API Operations

Access REST APIs via client.rest:

Table Management

# Discover tables first (or use a Table handle — see Table Handles section)
tables = client.tables()
t = tables[0]  # catalog="eddytor", schema="cfg_550e...", name="abc123_customers"

# Get table metadata
metadata = client.rest.get_table_metadata(t.catalog, t.schema, t.name)
print(f"Columns: {[c.name for c in metadata.columns]}")

# Get table history (version history)
history = client.rest.get_table_history(t.catalog, t.schema, t.name)
for entry in history.entries:
    print(f"Version {entry.version}: {entry.operation} at {entry.timestamp}")

# Rollback to a specific version
client.rest.rollback_table(t.catalog, t.schema, t.name, version=5)

# Add constraints
client.rest.add_constraints(t.catalog, t.schema, t.name, [
    {"type": "NOT_NULL", "column": "id"},
])

# Update field metadata
client.rest.update_field_metadata(
    t.catalog, t.schema, t.name,
    field_name="status",
    metadata={"description": "Order status"}
)

# Delete a table (irreversible)
client.rest.delete_table(t.catalog, t.schema, t.name)

# Move a table to a different storage configuration
client.rest.move_table(
    t.catalog, t.schema, t.name,
    destination_config_id="target-config-uuid",
    destination_path="new/path/",
)

AI-Powered Analysis

from eddytor_sdk import AIProvider, AIAction

# Use a discovered table
tables = client.tables()
t = tables[0]

# Summarize table contents
result = client.rest.magic_dust(
    t.catalog, t.schema, t.name,
    provider=AIProvider.CLAUDE,
    action=AIAction.SUMMARY,
    model="claude-sonnet-4-6",
    sample_size=1000,
)
print(result.content)

# Detect anomalies
result = client.rest.magic_dust(
    t.catalog, t.schema, t.name,
    provider=AIProvider.OPENAI,
    action=AIAction.DETECT_ANOMALIES,
    model="gpt-5.2",
)

# Explain rows with tagged cell references
result = client.rest.explain_rows(
    t.catalog, t.schema, t.name,
    provider=AIProvider.CLAUDE,
    model="claude-sonnet-4-6",
    sample_size=20,
)
print(result.explanation)
for ref in result.references:
    print(f"  {ref.display}: row {ref.row_index}, column {ref.column}")

Supported AI providers: Claude, OpenAI, Gemini, Mistral, Ollama (names are case-insensitive)

AI Credentials & Model Catalog

Magic Dust runs with your provider keys. Manage them via the SDK:

# Store a provider key (encrypted at rest, never returned)
client.rest.set_ai_credential("claude", "sk-ant-...")

# Self-hosted Ollama needs a base URL, the key can be anything
client.rest.set_ai_credential("ollama", "unused", base_url="http://localhost:11434")

# Which providers are configured?
for cred in client.rest.get_ai_credentials():
    print(cred.provider, cred.base_url)

# Full model catalog, grouped by provider, with configuration status
for group in client.rest.get_ai_models():
    marker = "✓" if group.configured else "✗"
    print(f"{marker} {group.provider}: {[m.id for m in group.models]}")

# Remove a key
client.rest.delete_ai_credential("claude")

Storage Operations

# Get all storage configurations
configs = client.rest.get_storage_configs()
for cfg in configs:
    print(f"{cfg.name}: {cfg.path} ({cfg.scheme_type})")

# Register S3 storage
config_id = client.rest.register_s3_storage(
    bucket="my-bucket",
    region="us-east-1",
    access_key_id="AKIAXXXXXXXX",
    secret_key="secret",
)

# Register Azure storage
config_id = client.rest.register_azure_storage(
    account_name="myaccount",
    container="mycontainer",
    access_key="key...",
)

# Register Google Cloud Storage
config_id = client.rest.register_gcs_storage(
    bucket="my-bucket",
    service_account_key='{"type": "service_account", ...}',
)

# Update name / discovery settings
client.rest.update_storage_config(config_id, name="lakehouse", discover_iceberg=True)

# Seed a demo table to explore on a fresh install
client.rest.create_demo_table(config_id)

# List objects in storage
result = client.rest.list_objects(config_id, path="data/", extensions="csv,parquet")
for obj in result.objects:
    print(f"{obj.name}: {obj.size} bytes")

# Download / delete objects
data, filename = client.rest.download_object(config_id, "data/file.csv")
client.rest.delete_object(config_id, "data/old_file.csv")

# Create folders
client.rest.create_folder(config_id, "data/new/")

# Move objects between storage configurations
client.rest.move_objects(
    source_config_id="cfg_aaa",
    source_path="old/path/",
    destination_config_id="cfg_bbb",
    destination_path="new/path/",
)

# Delete storage configuration
client.rest.delete_storage_config(config_id)

Column Domains

# Use a discovered table (or Table handle)
tables = client.tables()
t = tables[0]

# Get domain configuration for a column
domain = client.rest.get_column_domain(t.catalog, t.schema, t.name, "status")

# Set a fixed domain (enum values)
client.rest.set_fixed_domain(
    t.catalog, t.schema, t.name, "status",
    values=["pending", "active", "completed", "cancelled"]
)

# Or use the Table/Column handle (simpler):
t.column("status").set_fixed_domain(["pending", "active", "completed", "cancelled"])

# Set hierarchical domain (parent-child relationships)
client.rest.set_hierarchical_inline_domain(
    t.catalog, t.schema, t.name, "subcategory",
    parent_column="category",
    mappings={
        "uuid-of-electronics": ["Phones", "Laptops", "Tablets"],
        "uuid-of-clothing": ["Shirts", "Pants", "Shoes"],
    }
)

# Set a reference domain — follow another table's column domain
departments = [x for x in tables if "departments" in x.name][0]
t.column("dept_id").set_reference_domain(departments, "DepartmentID")

# Get allowed values (optionally filtered by parent)
allowed = client.rest.get_allowed_values(
    t.catalog, t.schema, t.name, "subcategory",
    parent_value="Electronics"
)
print(allowed.values)  # ["Phones", "Laptops", "Tablets"]

# Find values that violate their column's domain (with fix suggestions)
report = t.domain_mismatches()
for col in report.columns:
    for m in col.mismatches:
        print(f"{col.column}: {m.value!r} -> did you mean {m.suggestions}?")

Table and Column Handles

The recommended way to get table handles is via client.tables() (discovery):

# Discover all tables (recommended)
tables = client.tables()
customers = [t for t in tables if "customers" in t.name][0]

print(customers.fqn)       # `eddytor`.`cfg_550e...`.`abc123_customers`
print(customers.count())
print(customers.history())
print(customers.metadata())

df = customers.query_all(limit=100)

# Or construct manually if you know the exact FQN parts
table = client.table("eddytor", "cfg_550e8400e29b41d4a716446655440000", "abc123_customers")

# Or resolve from a stable table ID (survives rename/move — ideal to persist in notebooks)
table = client.resolve_table("a1b2c3d4e5f67890abcdef1234567890")

# Column handle
col = customers.column("status")
col.set_fixed_domain(["Active", "Inactive"])
print(col.allowed_values())

Storage Handle

store = client.storage("cfg_xxx")
objects = store.list_objects(path="data/")
store.create_folder("data/new/")

MCP Integration

Eddytor supports the Model Context Protocol for AI agent integration. Add to your Claude Desktop config (claude_desktop_config.json):

{
  "mcpServers": {
    "eddytor": {
      "url": "http://localhost:8080/mcp",
      "headers": {
        "Authorization": "Bearer edd_live_xxx"
      }
    }
  }
}

Replace the URL with your server's public URL when not running locally.

This gives AI agents access to query, insert, merge, delete, create tables, manage schemas, run aggregations, and more.

Error Handling

from eddytor_sdk import EddytorClient, EddytorConnectionError, EddytorQueryError, EddytorError

try:
    client = EddytorClient(api_key="invalid_key")
except EddytorConnectionError as e:
    print(f"Connection failed: {e}")

try:
    df = client.query("SELECT * FROM nonexistent_table")
except EddytorQueryError as e:
    print(f"Query failed: {e}")

try:
    metadata = client.rest.get_table_metadata("invalid", "schema", "table")
except EddytorError as e:
    print(f"API error: {e}")

Other SDKs

SDK	Package	Install
Python	eddytor-sdk	pip install eddytor-sdk
Go	github.com/eddytor/eddytor-go-sdk	go get github.com/eddytor/eddytor-go-sdk

API Reference

EddytorClient

EddytorClient(api_key: str, url: str | None = None)  # API key or OAuth access token
EddytorClient.login(url: str | None = None)          # Interactive device-code sign-in

Flight SQL Methods

Method	Returns	Description
query(sql)	pd.DataFrame	Execute SQL, return pandas DataFrame
query_arrow(sql)	pa.Table	Execute SQL, return PyArrow Table
execute(sql)	list[tuple]	Execute SQL, return raw tuples
execute_dml(sql)	int	Execute DML, return affected row count
ingest(table, data, mode, catalog, schema)	int	Bulk ingest data
create_table(name, location, schema, ...)	None	Create a new table
add_column(table, columns, ...)	None	Add columns to a table
list_schemas()	list[str]	List all schemas
list_tables(schema=None)	list[str]	List tables (optionally in schema)
describe_table(table)	pd.DataFrame	Get table schema
table_exists(table)	bool	Check if table exists
count(table, where=None)	int	Count rows in table
tables(with_discovery=True)	list[Table]	Discover all tables (recommended starting point)
table(catalog, schema, table)	Table	Get a Table handle by exact FQN parts
resolve_table(table_id)	Table	Get a Table handle from a stable table ID
storage(config_id)	Storage	Get a Storage handle
interactive()	None	Start interactive SQL session
close()	None	Close connection

REST API Methods (via client.rest)

Table API

Method	Description
get_table_metadata(catalog, schema, table)	Get table metadata
get_table_history(catalog, schema, table)	Get version history
rollback_table(catalog, schema, table, version)	Rollback to version
delete_table(catalog, schema, table)	Delete table permanently
move_table(...)	Move table to different storage
add_constraints(...)	Add table constraints
drop_constraint(...)	Drop a constraint
update_field_metadata(...)	Update field metadata
infer_schema(file_path, has_header, delimiter)	Infer schema from CSV
resolve_table(table_id)	Resolve stable table ID to FQN parts
get_domain_mismatches(...)	Find values violating column domains
magic_dust(...)	AI-powered analysis
explain_rows(...)	AI row explanations with cell references

Storage API

Method	Description
get_storage_configs()	List storage configurations
register_s3_storage(...)	Register S3 storage
register_azure_storage(...)	Register Azure storage
register_gcs_storage(...)	Register Google Cloud Storage
update_storage_config(...)	Update name / discovery settings
create_demo_table(config_id)	Seed a demo table
delete_storage_config(config_id)	Delete storage config
get_registered_tables(with_discovery)	Get registered tables
list_objects(config_id, ...)	List objects in storage
download_object(config_id, path)	Download object
delete_object(config_id, path)	Delete object
create_folder(config_id, path)	Create folder
move_objects(...)	Move objects between configs
upload_files(files, path)	Upload files

Column Domain API

Method	Description
get_column_domain(...)	Get domain configuration
set_fixed_domain(...)	Set fixed (enum) domain
set_hierarchical_inline_domain(...)	Set hierarchical domain
set_hierarchical_derived_domain(...)	Link domain to another column
set_hierarchical_table_domain(...)	Domain from another table
set_reference_domain(...)	Follow another table's column domain
delete_column_domain(...)	Remove domain
add_fixed_value(...)	Add value to fixed domain
set_fixed_values(...)	Replace all fixed values
remove_fixed_value(...)	Remove value from fixed domain
get_allowed_values(...)	Get allowed values

AI Credentials API

Method	Description
get_ai_credentials()	List configured providers (keys never returned)
set_ai_credential(provider, api_key, base_url)	Store/replace a provider key
delete_ai_credential(provider)	Delete a provider key
get_ai_models()	Model catalog grouped by provider

Requirements

• Python 3.10+
• adbc-driver-flightsql >= 0.10.0
• pyarrow >= 15.0.0
• pandas >= 2.0.0
• requests >= 2.28.0