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, never the web
app in a browser; 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 or a
   `recommendation`, print exactly this and wait until I say done: "In the
   Mammoth web app open Workspace settings → API Tokens → Create token (no
   project, so I can create projects), copy the token (it starts with mm_ and
   is shown once), then run in your own terminal: mammoth auth login". Or run
   `mammoth auth login --device` yourself, show me the link and code it prints and wait
   for it to finish — never ask for, read, or pass the token yourself, and never run any
   other auth login method.
3. Require `mammoth doctor` to pass, then run `mammoth skill show` and
   follow the guide it prints.
Work inside `mammoth project ensure 'PROJECT NAME'` unless I name a project;
take ids only from reads; find a command with `schema find "GOAL IN PLAIN
WORDS"` or `view transform --help`, and `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

The SDK is async: every call that reaches the API is a coroutine and must be awaited. A complete script:

import asyncio

from mammoth import MammothClient


async def main() -> None:
    async with MammothClient(api_token="mm_your-token") as client:
        client.set_project_id(10)
        view = await client.views.get(1039)
        print(view.display_names)


asyncio.run(main())

The other examples show the calls on their own, with top-level await as in a Jupyter cell or python -m asyncio. In a script, put them inside an async def like main above.

from mammoth import MammothClient

client = MammothClient(
    api_token="mm_your-token",
)
client.set_project_id(42)

# Get a view and inspect its columns
view = await 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:
await 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 = await 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_token="mm_your-token",
)
client.set_project_id(ids["project_id"])
view = await 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.

await 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
await view.filter_rows(Condition("Sales", Operator.GTE, 1000))

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

Set Values (Conditional Labeling)

from mammoth import SetValue, ColumnType

await 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
await view.math("Price * Quantity", new_column="Revenue")
await view.math("(Price + Tax) * 1.1", new_column="Grand Total")

Join

from mammoth import JoinType, JoinKeySpec

other_view = await client.views.get(2050)

await 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

await 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

await 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
await view.text_transform(["Customer Name"], case=TextCase.UPPER)

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

# Split column
from mammoth import SplitColumnSpec

await 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
await view.extract_date("Order Date", DateComponent.YEAR, new_column="Order Year")

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

# Add 30 days to a date
from mammoth import DateDelta

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

Column Operations

from mammoth import CopySpec, ConversionSpec

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

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

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

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

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

Row Operations

from mammoth import FillDirection

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

# Keep top 100 rows
await view.limit_rows(100)

# Remove duplicates
await view.discard_duplicates()

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

AI and SQL

# AI-powered transformation
await 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 = await view.generate_sql("count customers by region")

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

Pipeline Management

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

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

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

All Transformation Methods

Method	Description
filter_rows()	Filter rows by condition
set_values()	Label/insert values with conditional logic
math()	Arithmetic expressions
small_large()	Nth smallest or largest value across columns/constants
join()	Join with another view
pivot()	Group by and aggregate
window()	Window functions (rank, lag, running sum, etc.)
crosstab()	Pivot table
text_transform()	Change case, trim whitespace
replace_values()	Find and replace
bulk_replace()	Bulk find-and-replace with mapping
split_column()	Split by delimiter
substring()	Extract text by position or regex
extract_date()	Extract date components
date_diff()	Date difference
increment_date()	Add/subtract from dates
add_column()	Add empty column
delete_columns()	Remove columns
copy_columns()	Duplicate columns
combine_columns()	Concatenate columns
convert_type()	Change column data type
fill_missing()	Fill gaps forward/backward
limit_rows()	Keep top/bottom N rows
discard_duplicates()	Remove duplicate rows
unnest()	Unpivot columns to rows
lookup()	Lookup values from another view

Parameter Spec Dataclasses

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

Dataclass	Used by
CopySpec	copy_columns()
ConversionSpec	convert_type()
AggregationSpec	pivot()
CrosstabSpec	crosstab()
JoinKeySpec	join() on
JoinSelectSpec	join() select
JsonExtractionSpec	json_extract()

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
await view.filter_rows(priority)
await view.set_values(
    new_column="Flag",
    column_type=ColumnType.TEXT,
    values=[
        SetValue("Priority", condition=high_sales & west_region),
        SetValue("Normal"),
    ],
)
await view.math("Sales * 1.1", new_column="Adjusted", condition=west_region)

Supported Operators

Operator	Description
EQ, NE	Equal, not equal
GT, GTE, LT, LTE	Comparison
IN_LIST, NOT_IN_LIST	Value in/not in list
IN_RANGE	Between two bounds (inclusive); value is a 2-element list
CONTAINS, NOT_CONTAINS	Text contains/not contains
ICONTAINS	Case-insensitive text contains
STARTS_WITH, ENDS_WITH	Text prefix/suffix
NOT_STARTS_WITH, NOT_ENDS_WITH	Negated prefix/suffix
IS_EMPTY, IS_NOT_EMPTY	Null check
IS_MAXVAL, IS_NOT_MAXVAL	Max value in column
IS_MINVAL, IS_NOT_MINVAL	Min value in column

IN_RANGE — numeric or date range

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

ICONTAINS — case-insensitive substring match

# Match "York", "york", "YORK" in City column
await 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
await 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
await 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
await 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 = await client.files.upload("sales_data.csv")

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

# Upload an entire folder
dataset_ids = await 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 = await client.files.upload("sales_data.csv")
views = await 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 = await view.export.to_csv("output.csv")

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

Export to S3

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

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

Export to Database

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

# MySQL
await 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 = await view.export.to_dataset(dataset_name="Q1 snapshot")

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

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

Export to FTP / SFTP

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

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

# SFTP — private-key auth
await 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`)
await view.export.to_email(
    emails=["[email protected]"],
    subject="Q1 Sales Report",
)

BigQuery Export

from mammoth import BigQueryExportType

await view.export.to_bigquery(
    selected_profile={"name": "my_dataset", "value": [["project_id", "dataset_id"]]},
    selected_identity={"identity_config": {...}, "host": "[email protected]"},
    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
await view.export.publish_to_db(table="sales_dashboard", odbc_type=OdbcType.POSTGRES)

REST API Export

from mammoth import RestAuthType, HttpMethod

await 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

await view.export.to_redshift(...)
await view.export.to_elasticsearch(...)
await view.export.to_azure_blob(...)
await view.export.to_sharepoint(...)
await view.export.to_onedrive(...)
await view.export.to_tableau(...)
await view.export.to_powerbi(...)
await 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_token="mm_...")
client.set_project_id(42)
view = await client.views.get(1039)

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

try:
    await 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 = await client.external_keys.create(
    key_type=ExternalKeyType.ANTHROPIC,
    key_name="My Claude key",
    secure_key="sk-ant-...",
)
await client.external_keys.delete(key["id"])

Addons (storage, users, connectors)

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

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

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

Dashboards

dashboard = await 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 = await 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 = await 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
await client.workspaces.update(patches=[
    WorkspacePatchOp(op="replace", path=WorkspacePatchPath.NAME, value="My Workspace"),
])

# Change a user's role
await 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 = await 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 = await 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

Mammoth runs a hosted MCP (Model Context Protocol) server for AI assistants: https://mcp.mammoth.io/mcp. Add it as a remote MCP server in your client and sign in with your Mammoth account (OAuth). There is nothing to install.

To run the server yourself, install mammoth-mcp-server:

pip install mammoth-mcp-server

It needs the environment variable MCP_ENCRYPTION_KEY set to a Fernet key (and a Redis for MCP_REDIS_URL). Generate a key with python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())". See the mammoth-mcp-server README for every setting. The package name mammoth-mcp is not this server.

Releasing

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