Mammothsdk MCP Server

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

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
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

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=["team@example.com"],
    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": "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
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.