Zamp SDK

The official Python SDK for executing actions on the Zamp platform.

Installation

pip install zamp-sdk

Or with Poetry:

poetry add zamp-sdk

Quick Start

import asyncio
from zamp_sdk import ActionExecutor

async def main():
    result = await ActionExecutor.execute(
        "send_invoice",
        {"invoice_id": "inv_123"},
        base_url="https://api.zamp.ai",
        auth_token="your-api-token",
    )
    print(result)

asyncio.run(main())

Using environment variables

Set ZAMP_BASE_URL and ZAMP_AUTH_TOKEN in your environment, then call without explicit config:

result = await ActionExecutor.execute("send_invoice", {"invoice_id": "inv_123"})

API Reference

ActionExecutor.execute()

@staticmethod
async def execute(
    action_name: str,
    params: Any,
    *,
    base_url: str | None = None,
    auth_token: str | None = None,
    summary: str | None = None,
    return_type: type | None = None,
    action_retry_policy: RetryPolicy | None = None,
    action_start_to_close_timeout: timedelta | None = None,
) -> Any

RetryPolicy

from zamp_sdk import RetryPolicy

policy = RetryPolicy(
    initial_interval=timedelta(seconds=30),
    maximum_attempts=3,
    maximum_interval=timedelta(minutes=15),
    backoff_coefficient=1.5,
)

# Or use the default configuration:
policy = RetryPolicy.default()

When action_retry_policy is omitted from ActionExecutor.execute(), the SDK applies RetryPolicy.default() automatically — actions fail fast (2 retries + 1 initial attempt) so script failures surface quickly. Pass an explicit policy to opt into longer retries when needed.

Configuration

Explicit parameters passed to ActionExecutor.execute() take precedence over environment variables.

Databases (zamp_sdk.db)

Access the organization's datasets by writing ordinary SQLAlchemy. The SDK compiles your expression and ships it to the platform — no DSN, no connection, no credential ever reaches your script.

import sqlalchemy as sa
from sqlalchemy import select, insert, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from zamp_sdk.db import datasets

# Existing dataset: one line, and you get a real sqlalchemy.Table
invoices = await datasets.table("invoices")

# From here down it is stock SQLAlchemy
rows = await datasets.execute(
    select(invoices).where(invoices.c.status == "open").limit(50)
)

# Several statements, one transaction — all land or none do
async with datasets.transaction() as tx:
    tx.add(insert(invoices).values(vendor="Acme", amount=1200.50))
    tx.add(
        update(invoices).where(invoices.c.id == 7).values(status="closed"),
        expected_rows=1,          # 0 → someone else changed it → whole body rolls back
    )

# Large reads page with a keyset cursor
async for page in datasets.stream(select(invoices), page_size=10000):
    process(page)

The six calls

Renaming, altering, sharing, revoking and listing are called directly via ActionExecutor (agent_db_rename_dataset, agent_db_alter_dataset, agent_db_share_dataset_access, agent_db_revoke_dataset_access, agent_db_list_datasets) — SQLAlchemy has no construct to compile from for those, so a wrapper would add a surface without adding anything.

Creating: use the returned table

Every dataset gets an auto-injected id primary key (added by the platform). create() returns the table as the platform describes it after creating, so table.c.id exists on the object you go on to use:

customers = sa.Table("customers", sa.MetaData(),
                     sa.Column("name", sa.Text, nullable=False))

customers = await datasets.create(customers, if_exists="skip")   # ← reassign
await datasets.execute(select(customers.c.id, customers.c.name))

Declaring your own id is safe and wins — the injection is skipped.

Errors

One exception type, AgentDbError, carrying Postgres's own vocabulary:

from zamp_sdk import AgentDbError

try:
    await datasets.execute(stmt)
except AgentDbError as e:
    print(e.sqlstate)          # "23505" — None if it never reached Postgres
    print(e.statement_index)   # which statement in the body failed

sqlstate is None is meaningful: it says the failure happened before Postgres saw the statement (a gate rejection, an authorization refusal). A TimeoutError propagates unwrapped, because the statement may have committed.

Writes that may run twice

A script can be re-run, so say in the SQL when a write must be safe to repeat: on_conflict_do_update is idempotent by construction, and a guarded update(...).where(status == "pending") is safe to repeat. Both need a unique constraint, which you declare in the CREATE TABLE or add later with CREATE UNIQUE INDEX.

Error Handling

from zamp_sdk.action_executor.utils import HttpClientError

try:
    result = await ActionExecutor.execute("my_action", params)
except HttpClientError as e:
    print(f"HTTP error {e.status_code}: {e.message}")
except RuntimeError as e:
    print(f"Action failed: {e}")
except TimeoutError as e:
    print(f"Timed out: {e}")

Development

# Clone and install
git clone https://github.com/Zampfi/zamp-sdk.git
cd zamp-sdk
make install

# Run all checks (lint + type-check + tests)
make check

# Individual targets
make lint          # ruff check + format check
make lint-fix      # auto-fix lint issues
make format        # format code
make type-check    # mypy
make test          # pytest with coverage
make clean         # remove build artifacts

Contributing

1. Create a feature branch from main
2. Make your changes
3. Run make check to verify lint, type-check, and tests pass
4. Open a pull request

Pre-commit hooks are configured -- install them with:

poetry run pre-commit install

License

MIT