mammoth-io

Python SDK for the Mammoth Analytics platform. Build data pipelines, apply transformations, and export results -- all from Python.

Prefer the terminal? This repository also ships mammoth-cli, a command-line interface for people, scripts, and schema-driven agent runs. Install the CLI without a preinstalled Python tool manager:

curl -fsSL https://raw.githubusercontent.com/EdgeMetric/mammothsdk/main/mammoth-cli/installers/mammoth-install.sh | bash

See the full mammoth-cli guide.

Copy-paste prompt for a shell-capable agent (fill PROJECT NAME and TASK; it installs the CLI and walks you through the one-time login; the long form with the reasoning is in docs/agent-prompt.md):

Use Mammoth Analytics only through the `mammoth` CLI in bash; run
`export MAMMOTH_OUTPUT=json MAMMOTH_NO_INPUT=1` once. Onboard me first:
1. If `mammoth --version` fails, install, then re-check (new shell if needed):
   curl -fsSL https://raw.githubusercontent.com/EdgeMetric/mammothsdk/main/mammoth-cli/installers/mammoth-install.sh | bash
2. Run `mammoth auth status`. If it shows no credentials, print exactly this
   and wait until I say done: "In the Mammoth web app open account settings,
   create an API key (key + secret) and note your workspace id, then run in
   your own terminal: mammoth auth login" — never ask for, read, or pass a key
   or secret yourself, and never run auth login.
3. Require `mammoth doctor` to pass, then `cat` the SKILL.md at
   `mammoth skill path` (data.canonical) and follow it.
Work inside `mammoth project ensure 'PROJECT NAME'` unless I name a project;
take ids only from reads; `schema get COMMAND_ID` before a new command; read
results back before reporting.
TASK: <what to achieve, and how you will know it is done>

Installation

pip install -U mammoth-io

Requires Python 3.12 through 3.14.

Quick Start

from mammoth import MammothClient

client = MammothClient(
    api_key="your-api-key",
    api_secret="your-api-secret",
    workspace_id=11,
)
client.set_project_id(42)

# Get a view and inspect its columns
view = client.views.get(1039)
print(view.display_names)   # ["Customer", "Region", "Sales", "Date"]
print(view.column_types)    # {"Customer": "TEXT", "Region": "TEXT", "Sales": "NUMERIC", ...}

# After any transformation, display_names is automatically refreshed
# (including pipeline-added columns like those created by math/set_values/add_column).
# Use get_metadata() to inspect the full list:
view.math("Sales * 1.1", new_column="Revenue")
print(view.display_names)   # now includes "Revenue"
meta = view.get_metadata()  # [{"display_name": "Revenue", "type": "NUMERIC"}, ...]

# Fetch data — returns {"data": [rows...], "paging": {...}}
result = view.data(limit=100)
rows = result["data"]

You can also extract IDs directly from a Mammoth URL:

from mammoth import MammothClient, parse_path

ids = parse_path("https://app.mammoth.io/#/workspaces/11/projects/42/views/1039")
# {"workspace_id": 11, "project_id": 42, "dataview_id": 1039}

client = MammothClient(
    api_key="your-api-key",
    api_secret="your-api-secret",
    workspace_id=ids["workspace_id"],
)
client.set_project_id(ids["project_id"])
view = client.views.get(ids["dataview_id"])

Views & Transformations

The View object is the central interface. It wraps a single dataview and exposes 25+ transformation methods. Each method sends a pipeline task to the API and automatically refreshes the view metadata — including any new columns added by the transformation.

view.math(expression="Price * Quantity", new_column="Revenue")
print("Revenue" in view.display_names)   # True — refreshed automatically

# Inspect full column list (display_name and type)
for col in view.get_metadata():
    print(col)

Filter Rows

from mammoth import Condition, Operator, FilterType

# Keep rows where Sales >= 1000
view.filter_rows(Condition("Sales", Operator.GTE, 1000))

# Remove rows where Region is empty
view.filter_rows(
    Condition("Region", Operator.IS_EMPTY),
    filter_type=FilterType.REMOVE,
)

Set Values (Conditional Labeling)

from mammoth import SetValue, ColumnType

view.set_values(
    new_column="Risk Level",
    column_type=ColumnType.TEXT,
    values=[
        SetValue("High", condition=Condition("Sales", Operator.GTE, 10000)),
        SetValue("Medium", condition=Condition("Sales", Operator.GTE, 5000)),
        SetValue("Low"),
    ],
)

Math

# String expressions are parsed automatically
view.math("Price * Quantity", new_column="Revenue")
view.math("(Price + Tax) * 1.1", new_column="Grand Total")

Join

from mammoth import JoinType, JoinKeySpec

other_view = client.views.get(2050)

view.join(
    foreign_view=other_view,
    join_type=JoinType.LEFT,
    on=[JoinKeySpec(left="Customer ID", right="Customer ID")],
    select=["Category", "Tier"],
)

Pivot (Group By / Aggregate)

from mammoth import AggregateFunction, AggregationSpec

view.pivot(
    group_by=["Region"],
    aggregations=[
        AggregationSpec(column="Sales", function=AggregateFunction.SUM, as_name="Total Sales"),
        AggregationSpec(column="Sales", function=AggregateFunction.AVG, as_name="Avg Sales"),
    ],
)

Window Functions

from mammoth import WindowFunction, SortDirection

view.window(
    function=WindowFunction.ROW_NUMBER,
    new_column="Rank",
    partition_by=["Region"],
    order_by=[["Sales", SortDirection.DESC]],
)

Text Operations

from mammoth import TextCase

# Change case
view.text_transform(["Customer Name"], case=TextCase.UPPER)

# Find and replace
view.replace_values(columns=["Status"], find="Pending", replace="In Progress")

# Split column
from mammoth import SplitColumnSpec

view.split_column(
    "Full Name",
    delimiter=" ",
    new_columns=[SplitColumnSpec(name="First"), SplitColumnSpec(name="Last")],
)

Date Operations

from mammoth import DateComponent, DateDiffUnit

# Extract year from a date column
view.extract_date("Order Date", DateComponent.YEAR, new_column="Order Year")

# Calculate difference between two dates
view.date_diff(DateDiffUnit.DAY, start="Start Date", end="End Date", new_column="Duration")

# Add 30 days to a date
from mammoth import DateDelta

view.increment_date("Ship Date", delta=DateDelta(days=30), new_column="Expected Arrival")

Column Operations

from mammoth import CopySpec, ConversionSpec

# Add an empty column
view.add_column("Notes", ColumnType.TEXT)

# Delete columns
view.delete_columns(["Temp1", "Temp2"])

# Copy a column
view.copy_columns([CopySpec(source="Sales", as_name="Sales Backup", type=ColumnType.NUMERIC)])

# Combine (concatenate) columns
view.combine_columns(["First Name", "Last Name"], new_column="Full Name", separator=" ")

# Convert column type
view.convert_type([ConversionSpec(column="ZipCode", to=ColumnType.TEXT)])
view.convert_type([ConversionSpec(column="Order Date", to=ColumnType.DATE, format="MM/DD/YYYY")])

Row Operations

from mammoth import FillDirection

# Fill missing values
view.fill_missing("Revenue", direction=FillDirection.LAST_VALUE)

# Keep top 100 rows
view.limit_rows(100)

# Remove duplicates
view.discard_duplicates()

# Unpivot columns to rows
view.unnest(["Q1", "Q2", "Q3", "Q4"], label_column="Quarter", value_column="Revenue")

AI and SQL

# AI-powered transformation
view.gen_ai(
    prompt="Classify the sentiment of the review as Positive, Negative, or Neutral",
    context_columns=["Review Text"],
    new_column="Sentiment",
)

# Generate SQL from natural language (also adds pipeline task)
sql_query = view.generate_sql("count customers by region")

# Add a raw SQL query as a pipeline task
view.add_sql('SELECT region, COUNT(*) AS cnt FROM "view:123" GROUP BY region')

Pipeline Management

# List all tasks on a view
tasks = view.list_tasks()

# Delete a specific task
view.delete_task(task_id=123)

# Preview a task before applying
preview = view.preview_task({"MATH": {"EXPRESSION": [...]}})

All Transformation Methods

Parameter Spec Dataclasses

Methods that accept structured parameters use typed dataclasses for IDE autocomplete:

Conditions

The Condition class supports Python's & (AND), | (OR), and ~ (NOT) operators for composing filter logic.

from mammoth import Condition, Operator

# Simple conditions
high_sales = Condition("Sales", Operator.GTE, 10000)
west_region = Condition("Region", Operator.EQ, "West")
active = Condition("Status", Operator.IN_LIST, ["Active", "Pending"])
has_email = Condition("Email", Operator.IS_NOT_EMPTY)

# Combine with & (AND), | (OR), and ~ (NOT)
priority = high_sales & west_region          # Both must be true
either = high_sales | west_region            # At least one true
not_active = ~active                         # Negate a condition
complex_filter = (high_sales & west_region) | ~active  # Nested logic

# Use anywhere conditions are accepted
view.filter_rows(priority)
view.set_values(
    new_column="Flag",
    column_type=ColumnType.TEXT,
    values=[
        SetValue("Priority", condition=high_sales & west_region),
        SetValue("Normal"),
    ],
)
view.math("Sales * 1.1", new_column="Adjusted", condition=west_region)

Supported Operators

IN_RANGE — numeric or date range

# Keep rows where Amount is between 100 and 500 (inclusive)
view.filter_rows(Condition("Amount", Operator.IN_RANGE, [100, 500]))

ICONTAINS — case-insensitive substring match

# Match "York", "york", "YORK" in City column
view.filter_rows(Condition("City", Operator.ICONTAINS, "york"))

Date-relative function operands via DateFunction

Pass a DateFunction enum member as the value with value_is_date_fn=True to compare against a dynamic date/time:

from mammoth import Condition, Operator, DateFunction

# Rows where Order Date is after today
view.filter_rows(Condition("Order Date", Operator.GT, DateFunction.TODAY, value_is_date_fn=True))

# Rows where Timestamp is on or after the current instant
view.filter_rows(Condition("Timestamp", Operator.GTE, DateFunction.NOW, value_is_date_fn=True))

# Rows where Date equals the maximum date value in that column
view.filter_rows(Condition("Date", Operator.EQ, DateFunction.MAX, value_is_date_fn=True))

Available DateFunction values: NOW, TODAY, MAX, MIN, SYSTEM_DATE, SYSTEM_TIME.

File Upload

# Upload a single file (returns dataset ID)
dataset_id = client.files.upload("sales_data.csv")

# Upload multiple files
dataset_ids = client.files.upload(["sales.csv", "customers.xlsx"])

# Upload an entire folder
dataset_ids = client.files.upload_folder("./data/")

Supported formats: CSV, TSV, PSV, XLS, XLSX, ZIP, BZ2, GZ, TAR, 7Z, PDF, TIFF, JPEG, PNG, HEIC, WEBP. Maximum file size: 50 MB.

After upload, get a view for the new dataset:

dataset_id = client.files.upload("sales_data.csv")
views = client.views.list(dataset_id=dataset_id)
view = next(v for v in views if v.dataset_id == dataset_id)
print(view.display_names)

Exports

Download as CSV

# From a View object
path = view.export.to_csv("output.csv")

# From client with a known dataview ID
path = client.exports.to_csv(dataview_id=1039, output_path="output.csv")

Export to S3

# From a View object
result = view.export.to_s3(file_name="monthly_report.csv")

# From client with a known dataview ID (parameter name is `file=`, not `file_name=`)
result = client.exports.to_s3(dataview_id=1039, file="monthly_report.csv")

Export to Database

# PostgreSQL
view.export.to_postgres(
    host="db.example.com",
    port=5432,
    database="analytics",
    table="sales_summary",
    username="user",
    password="pass",
)

# MySQL
view.export.to_mysql(
    host="db.example.com",
    port=3306,
    database="analytics",
    table="sales_summary",
    username="user",
    password="pass",
)

Branch Out (Export to Another Dataset)

# From a View object — creates a new dataset named "Q1 snapshot"
new_dataset_id = view.export.to_dataset(dataset_name="Q1 snapshot")

# Shorthand
new_dataset_id = view.branch_out(dataset_name="Q1 snapshot")

# Append into an existing dataset
view.branch_out(dataset_name="Sales Archive", target_ds_id=500)

Export to FTP / SFTP

# FTP — parameters: domain, directory, file, username, password
view.export.to_ftp(
    domain="ftp.example.com",
    directory="/exports",
    file="sales.csv",
    username="user",
    password="pass",
)

# SFTP — password auth
view.export.to_sftp(
    host="sftp.example.com",
    username="user",
    password="pass",
    directory="/exports",
    file_name="data.csv",
)

# SFTP — private-key auth
view.export.to_sftp(
    host="sftp.example.com",
    username="user",
    ssh_key_authentication=True,
    private_key="-----BEGIN OPENSSH PRIVATE KEY-----\n...",
)

Email Export

# `emails` is a list of recipient addresses (not `recipients`)
view.export.to_email(
    emails=["team@example.com"],
    subject="Q1 Sales Report",
)

BigQuery Export

from mammoth import BigQueryExportType

view.export.to_bigquery(
    selected_profile={"name": "my_dataset", "value": [["project_id", "dataset_id"]]},
    selected_identity={"identity_config": {...}, "host": "sa@project.iam.gserviceaccount.com"},
    table="sales_summary",
    export_type=BigQueryExportType.REPLACE,
)

Publish to Managed DB (for Dashboards)

from mammoth import OdbcType

# Publishes to a Mammoth-managed Postgres or BigQuery connection
view.export.publish_to_db(table="sales_dashboard", odbc_type=OdbcType.POSTGRES)

REST API Export

from mammoth import RestAuthType, HttpMethod

view.export.to_rest_api(
    base_url="https://api.example.com",
    endpoint_path="/v1/records",
    auth_type=RestAuthType.BEARER,
    http_method=HttpMethod.POST,
    auth={"token": "my-bearer-token"},
    batch_size=500,       # 1–10 000
    timeout_seconds=30,   # 5–300
)

Other Export Targets

view.export.to_redshift(...)
view.export.to_elasticsearch(...)
view.export.to_azure_blob(...)
view.export.to_sharepoint(...)
view.export.to_onedrive(...)
view.export.to_tableau(...)
view.export.to_powerbi(...)
view.export.to_mssql(...)

Error Handling

The SDK validates arguments before making any network call and raises specific exceptions so failures are caught early with an actionable message.

Exception hierarchy

MammothError                    # base; all SDK exceptions inherit from this
├── MammothValidationError      # invalid SDK arguments (raised before any API call)
├── MammothAPIError             # API returned an error response
│   └── MammothAuthError        # HTTP 401 — invalid credentials
├── MammothColumnError          # column display name not found in view metadata
├── MammothExportError          # export submitted but result dataset id didn't resolve
├── MammothJobTimeoutError      # async job polling exceeded the timeout
├── MammothJobFailedError       # async job completed with a failure status
└── MammothTransformError       # pipeline transformation task failed

All exceptions expose a .message attribute. MammothAPIError additionally exposes .status_code (int | None) and .response_body (dict).

Example

from mammoth import (
    MammothClient,
    MammothValidationError,
    MammothAPIError,
)

client = MammothClient(api_key="...", api_secret="...", workspace_id=11)
client.set_project_id(42)
view = client.views.get(1039)

try:
    # Validation fires before any network call
    view.export.to_email(emails=[])          # raises MammothValidationError
except MammothValidationError as e:
    print("Bad arguments:", e.message)

try:
    client.views.get(99999)                  # raises MammothAPIError if not found
except MammothAPIError as e:
    print(f"API error {e.status_code}: {e.response_body}")

Resource APIs

The SDK exposes typed resource APIs on the client for managing workspace entities. All public types are importable directly from mammoth.

External Keys

from mammoth import ExternalKeyType

key = client.external_keys.create(
    key_type=ExternalKeyType.ANTHROPIC,
    key_name="My Claude key",
    secure_key="sk-ant-...",
)
client.external_keys.delete(key["id"])

Addons (storage, users, connectors)

# Add 50 GB storage
client.addons.add_storage(additional_storage_gb=50)

# Add 5 user seats
client.addons.add_users(user_count=5)

# Add a connector addon (single or bulk)
client.addons.add_connector(connector_id=42)
client.addons.add_connector(connector_ids=[42, 43])

Dashboards

dashboard = client.dashboards.create(
    intent="Show quarterly revenue by region and product",
    source=[1039, 1040],       # dataview IDs
    enable_filters=True,
)

Automations

from mammoth import (
    AutomationTaskSpec, AutomationTaskType,
    TaskDetailsSpec, DataRefreshConfig,
)

automation = client.automations.create(
    name="Nightly refresh",
    description="Pulls cloud data every night",
    tasks=[
        AutomationTaskSpec(
            task_type=AutomationTaskType.RUN_DATA_RETRIEVAL,
            details=TaskDetailsSpec(
                ds_details=[DataRefreshConfig(ds_id=42)],
            ),
        )
    ],
)

Schedules

from datetime import datetime
from mammoth import ScheduleCreateSpec, RruleSpec, RruleFrequency

schedule = client.automations.create_schedule(
    spec=ScheduleCreateSpec(
        rrule=RruleSpec(
            frequency=RruleFrequency.DAILY,
            start=datetime(2025, 1, 1, 6, 0),
        ),
    )
)

Workspace — update settings and user roles

from mammoth import WorkspacePatchOp, WorkspacePatchPath, UserRolePatchOp, WorkspaceRoleType

# Rename the workspace
client.workspaces.update(patches=[
    WorkspacePatchOp(op="replace", path=WorkspacePatchPath.NAME, value="My Workspace"),
])

# Change a user's role
client.workspaces.update_user(
    user_id="user-uuid-here",
    patches=[UserRolePatchOp(op="replace", path="role", value=WorkspaceRoleType.WORKSPACE_ADMIN)],
)

Connectors — create a connection and data source config

# Create a PostgreSQL connection (config shape varies by connector_key)
conn = client.connectors.create_connection(
    connector_key="postgres",
    config={
        "hostname": "db.example.com",
        "port": 5432,
        "database": "analytics",
        "username": "user",
        "password": "pass",
    },
)

# Create a data source config (query-based for DB connectors)
ds_config = client.connectors.create_ds_config(
    connector_key="postgres",
    connection_key=conn["connection_key"],
    query="SELECT * FROM sales WHERE year = 2024",
    validate=True,
)

Command-line interface (mammoth-cli)

mammoth-cli is the terminal companion to this SDK. Use it to inspect and move data, run transformations, organize work, and automate repeatable operations. It renders for people in a terminal and switches to a versioned JSON envelope when output is piped, making the same command useful in CI and agent workflows. For unattended work, an agent should discover the capability and schema, resolve the exact workspace/project/dataset/view scope, compose operations from observed IDs, verify the result, and record a checkpoint before handing the task over. View and transformation inputs use the display names shown by Mammoth; agents do not need to find or manufacture backend column identifiers. A dataset is not assumed to have a usable default view: list views and choose one explicitly.

Install the published CLI without a preinstalled Python tool manager:

curl -fsSL https://raw.githubusercontent.com/EdgeMetric/mammothsdk/main/mammoth-cli/installers/mammoth-install.sh | bash

Open a new shell if needed so the installer-added tool directory is on PATH, then run:

mammoth skill list
mammoth skill path
# Read the installed SKILL.md before operating.
mammoth auth status
# If the profile or stored credentials are absent, human terminal only:
mammoth auth login
mammoth doctor
mammoth schema find "TASK OR RESOURCE"
mammoth schema get COMMAND_ID

For an agent or CI on POSIX, use a private owner-only (0600) JSON file outside the repository: mammoth auth login --input /private/path/credentials.json --storage file. Do not put secrets in chat, prompts, or command arguments. On Windows, use the approved OS keyring instead; do not use a file fallback unless its ACL hardening is approved. An agent that finds no credentials asks the operator to run the hidden-prompt login in their own terminal; the CLI does not read credentials from environment variables. The installer also installs the bundled agent skill. Start with the CLI README, which indexes the agent guide, portable handoff format, commands, and capability matrix.

The CLI's examples are deliberately nonexhaustive. Use mammoth schema find QUERY and mammoth schema get COMMAND_ID for the complete installed command surface; mammoth capability list is an API-binding inventory and can omit typed or local CLI routes. Treat a timeout or transport failure on a mutation as an unknown outcome until a read or job inspection proves what happened. Exit 7 is not a blanket permission to replay a write.

MCP Server

The SDK includes a companion MCP (Model Context Protocol) server that lets AI assistants interact with Mammoth directly. Install it separately:

pip install mammoth-mcp

See the mammoth-mcp directory for configuration and usage details.

Releasing

Maintainer release process (SDK + CLI, PyPI + GitHub releases, tag-triggered CI and the local fallback) is documented in RELEASING.md.