WaveAssist SDK & CLI 🌊

WaveAssist SDK makes it simple to store and retrieve data in your automation workflows. Access your projects through our Python SDK or CLI.

✨ Features

• 🔐 One-line init() to connect with your WaveAssist project
• ⚙️ Automatically works on local and WaveAssist Cloud (worker) environments
• 📦 Store and retrieve data (DataFrames, JSON, strings)
• 🧠 LLM-friendly function names (init, store_data, fetch_data)
• 📁 Auto-serialization for common Python objects
• 📊 Publish HTML dashboards with shareable public URLs
• 🤖 LLM integration with structured outputs via Instructor and OpenRouter
• 💳 Credit management and automatic email notifications
• 🖥️ Command-line interface for project management
• ✅ Built for automation workflows, cron jobs, and AI pipelines

🚀 Getting Started

1. Install

pip install waveassist

2. Initialize the SDK

import waveassist

# Option 1: Use no arguments (recommended)
waveassist.init()

# Option 2: With explicit parameters
waveassist.init(
    token="your-user-id",
    project_key="your-project-key",
    environment_key="your-env-key",  # optional
    run_id="run-123",  # optional
    check_credits=True  # optional: raises RuntimeError if credits_available is "0"
)

# When check_credits=True, a missing credits_available key is treated as credits available (default "1").
# Will auto-resolve from:
# 1. Explicit args (if passed)
# 2. .env file (uid, project_key, environment_key)
# 3. Worker-injected credentials (on [WaveAssist Cloud](https://waveassist.io))

🛠 Setting up .env (for local runs)

uid=your-user-id
project_key=your-project-key

# optional
environment_key=your-env-key

This file will be ignored by Git if you use our default .gitignore.

3. Store Data

Data is serialized by type. You can rely on type-safe storage and retrieval.

🧾 Store a string

waveassist.store_data("welcome_message", "Hello, world!")

📊 Store a DataFrame

import pandas as pd

df = pd.DataFrame({"name": ["Alice", "Bob"], "score": [95, 88]})
waveassist.store_data("user_scores", df)

🧠 Store JSON/dict/array

profile = {"name": "Alice", "age": 30}
waveassist.store_data("profile_data", profile)

📌 Optional: Force storage type

Store data as a specific type regardless of input:

# Store a dict as a string
waveassist.store_data("config", {"a": 1}, data_type="string")

# Store a string as JSON (wraps as {"value": "..."})
waveassist.store_data("greeting", "Hello", data_type="json")

Parameters: store_data(key, data, run_based=False, data_type=None). Use data_type="string", "json", or "dataframe" to force that storage type.

4. Fetch Data

result = waveassist.fetch_data("user_scores")

# Returns the correct type:
# - pd.DataFrame (if stored as dataframe)
# - dict or list (if stored as JSON)
# - str (if stored as string)

Use a default when the key might be missing:

# Return a default if key is missing or API fails
count = waveassist.fetch_data("failure_count", default=0)
greeting = waveassist.fetch_data("welcome", default="Hello")
df = waveassist.fetch_data("results", default=pd.DataFrame())  # empty DataFrame

Parameters: fetch_data(key, run_based=False, default=None).

5. Send Email

Send emails (e.g. for notifications) via the WaveAssist backend:

waveassist.send_email(
    subject="Daily report",
    html_content="<p>Summary: ...</p>",
    attachment_file=open("report.pdf", "rb"),  # optional
    raise_on_failure=True  # default: raise WaveAssistEmailError on validation or API failure
)

• Validation: Subject and HTML body must be non-empty (and within length limits). Attachments must be file-like with a .read() method.
• Retry: The SDK retries the send once on transient failure.
• Errors: By default, validation failures raise ValueError and API failures raise RuntimeError. Pass raise_on_failure=False to return False instead.

6. Publish an HTML Dashboard

Generate a shareable public URL for an HTML dashboard. Anyone with the link can view it — no login required.

import waveassist

waveassist.init()

html = """
<html>
<head><title>Weekly Report</title></head>
<body>
  <h1>Sales Summary</h1>
  <p>Total revenue: $12,340</p>
</body>
</html>
"""

# Publish and get a shareable URL
url = waveassist.publish_dashboard(html)
print(url)  # https://api.waveassist.io/d/a3f8c2d19e7b4f2ab5c8.../

# Email the link to yourself
waveassist.send_email(
    subject="Your Dashboard is Ready",
    html_content=f'<p>View your dashboard: <a href="{url}">{url}</a></p>'
)

Static vs Per-Run Dashboards:

# Static — same data_key, URL always shows the latest content
url = waveassist.publish_dashboard(html, data_key="weekly_report")

# Per-run — each run gets its own unique dashboard URL
url = waveassist.publish_dashboard(html, run_based=True)

Parameters: publish_dashboard(html_content, data_key="dashboard_html", run_based=False). Returns the public URL string, or None on failure.

7. Check Credits and Notify

Check OpenRouter credits and automatically send email notifications if insufficient credits are available:

# Check if you have enough credits for an operation
has_credits = waveassist.check_credits_and_notify(
    required_credits=10.5,
    assistant_name="WavePredict"
)

if has_credits:
    # Proceed with your operation
    print("Credits available, proceeding...")
else:
    # Credits insufficient - email notification sent (max 3 times)
    print("Insufficient credits, operation skipped")

Features:

• Automatically checks OpenRouter credit balance
• Sends email notification if credits are insufficient (max 3 times)
• Resets notification count when credits become sufficient
• Stores credit availability status for workflow control

8. Call LLM with Structured Outputs

Get structured responses from LLMs via OpenRouter with Pydantic models:

from pydantic import BaseModel

# Define your response structure
class UserInfo(BaseModel):
    name: str
    age: int
    email: str

# Call LLM with structured output
result = waveassist.call_llm(
    model="gpt-4o",
    prompt="Extract user info: John Doe, 30, john@example.com",
    response_model=UserInfo
)

print(result.name)  # "John Doe"
print(result.age)    # 30
print(result.email)  # "john@example.com"

Setup:

1. Store your OpenRouter API key:

waveassist.store_data('open_router_key', 'your_openrouter_api_key')

2. Use call_llm() with any Pydantic model for structured outputs

Advanced Usage:

result = waveassist.call_llm(
    model="anthropic/claude-3-opus",
    prompt="Analyze this data...",
    response_model=MyModel,
    should_retry=True,  # retry once on JSON/format errors
    max_tokens=3000,
    extra_body={"web_search_options": {"search_context_size": "medium"}}
)

Errors: RuntimeError (API/network failure), ValueError (invalid or non-JSON response). Transport errors are retried once automatically.

Local Testing with Claude CLI

Route call_llm through the Claude Code CLI for free local testing — no OpenRouter credits needed. Requires claude CLI installed and authenticated (works with Claude Max, Pro, or an Anthropic API key).

LLM_PROVIDER=claude_cli        # activates CLI routing
CLAUDE_CLI_MODEL=claude-sonnet-4-6  # optional: pin a specific model

No code changes needed — OpenRouter model names (e.g. anthropic/claude-sonnet-4.6) are auto-converted. Non-Claude models require CLAUDE_CLI_MODEL to be set. Unset LLM_PROVIDER to switch back to OpenRouter.

Claude on the worker fleet (setup token)

The same claude -p path, but authenticated headlessly by a setup token so it runs on the WaveAssist workers (where there's no interactive claude login). The token draws on your Claude subscription — it is not an API key and is not billed per-token.

# 1. On a machine logged into Claude, mint a ~1-year token:
#    $ claude setup-token   ->   sk-ant-oat01-...
# 2. Store it as a Variable and select the provider (per project):
waveassist.store_data("claude_setup_token", "sk-ant-oat01-...")
waveassist.store_data("llm_provider", "claude_cli_token")

On each call_llm, the token is injected as CLAUDE_CODE_OAUTH_TOKEN into the claude subprocess for that call only, with an isolated CLAUDE_CONFIG_DIR and conflicting auth (ANTHROPIC_API_KEY) scrubbed so the subscription is used. Subject to your subscription's rate limits and Anthropic's terms.

Per-model providers (llm_models registry)

To mix providers in one project — say one model on Azure, another on a Claude subscription, the rest on OpenRouter, or just a lighter model for a cheaper step — store an llm_models registry keyed by the model name your code passes to call_llm. Each entry is self-contained: it carries its own provider, real model id, credentials, and any provider-specific knobs (Azure api_type, api_base). A model not in the registry falls back to the legacy global resolution, which defaults to OpenRouter — so the common path needs no registry at all.

waveassist.store_data("llm_models", {
    # Azure reasoning model on the Responses API
    "gpt-5.4-pro": {"provider": "azure", "model": "gpt-5.4-pro", "api_type": "responses",
                    "api_base": "https://<resource>.openai.azure.com/", "api_key": "..."},
    # Azure chat model on the same account, different API surface
    "gpt-5.4":     {"provider": "azure", "model": "gpt-5.4", "api_type": "completions",
                    "api_base": "https://<resource>.openai.azure.com/", "api_key": "..."},
    # Claude subscription via setup token (headless / cloud)
    "haiku":       {"provider": "claude_cli_token", "model": "claude-haiku-4-5", "token": "sk-ant-oat01-..."},
    # Local Claude login (dev)
    "sonnet":      {"provider": "claude_cli", "model": "claude-sonnet-4-6"},
    # Explicit OpenRouter entry (optional — or just omit and rely on the default)
    "gpt-4o":      {"provider": "openrouter", "model": "openai/gpt-4o", "api_key": "..."},
})

Nodes keep passing the alias as usual, so different roles can use different models with no code change:

waveassist.store_data("model_name", "gpt-5.4-pro")   # main model
waveassist.store_data("brain_model", "gpt-5.4")      # a lighter model for a cheaper step

Shared credentials (optional). When several models share one account/key, store the secret once and reference it; inline entry fields override the shared block:

waveassist.store_data("llm_credentials",
    {"acme_azure": {"api_key": "...", "api_base": "https://<resource>.openai.azure.com/"}})
waveassist.store_data("llm_models", {
    "gpt-5.4-pro": {"provider": "azure", "model": "gpt-5.4-pro", "api_type": "responses",   "credential": "acme_azure"},
    "gpt-5.4":     {"provider": "azure", "model": "gpt-5.4",     "api_type": "completions", "credential": "acme_azure"},
})

Resolution order: LLM_PROVIDER env override (local Claude CLI) → llm_models registry → legacy llm_provider / azure_openai_config → OpenRouter default. Existing projects keep working unchanged; the registry is purely additive.

🖥️ Command Line Interface

WaveAssist CLI comes bundled with the Python package. After installation, you can use the following commands:

🔑 Authentication

waveassist login

This will open your browser for authentication and store the token locally.

📤 Push Code

waveassist push PROJECT_KEY [--force]

Push your local Python code to a WaveAssist project.

📥 Pull Code

waveassist pull PROJECT_KEY [--force]

Pull Python code from a WaveAssist project to your local machine.

ℹ️ Version Info

waveassist version

Display CLI version and environment information.

🧪 Running Tests

Run with pytest (recommended) or the test scripts:

# All tests (use project Python if you have a venv)
python -m pytest tests/ -v

# Or run modules directly
python tests/test_core.py
python tests/test_json_generate.py
python tests/test_json_extract.py

✅ Includes tests for:

• String, JSON, and DataFrame roundtrips; store_data with explicit data_type; fetch_data with default
• send_email validation, attachments, and raise_on_failure
• Error handling when init() is not called (RuntimeError)
• Environment variable and .env file resolution
• JSON template generation for Pydantic models
• JSON extraction from various formats (pure JSON, markdown code blocks, embedded text)
• Soft parsing with missing required fields (safety fallback for LLM responses)
• Type coercion and nested model handling
• Resilience / never-crash tests (test_never_crash.py): 45 tests ensuring fetch_data, store_data, send_email, publish_dashboard, and fetch_openrouter_credits never throw unhandled exceptions — they return defaults (default, False, None, {}) on garbage inputs, corrupted API responses, and network failures

🛠 Project Structure

WaveAssist/
├── waveassist/
│   ├── __init__.py          # Public API: init(), store_data(), fetch_data(), publish_dashboard(),
│   │                         # send_email(), check_credits_and_notify(), call_llm()
│   ├── _config.py            # Global config and version
│   ├── constants.py         # API_BASE_URL, OpenRouter, dashboard URLs
│   ├── utils.py             # API helpers, JSON parsing, soft_parse, exception classes
│   ├── core.py              # CLI: login, pull, push (uses constants for URLs)
│   └── cli.py               # Command-line entry (waveassist login/push/pull/version)
├── tests/
│   ├── test_core.py         # Core SDK + send_email tests
│   ├── test_json_generate.py # JSON template generation tests
│   ├── test_json_extract.py  # JSON extraction/parsing tests
│   ├── test_llm_call.py     # call_llm integration tests (skipped without API key)
│   └── test_never_crash.py  # Resilience tests: no SDK function crashes on bad input

📌 Notes

• Data is stored in your WaveAssist backend (e.g. MongoDB) as serialized content
• store_data() auto-detects the object type and serializes it (dataframe/JSON/string), or use data_type to force a type
• fetch_data() returns the correct Python type and supports a default when the key is missing or the API fails
• Logging: The SDK uses the standard library logging module (logger name "waveassist"). Configure level or handlers to control or suppress SDK messages (e.g. logging.getLogger("waveassist").setLevel(logging.WARNING)).
• Errors: The SDK raises standard exceptions: ValueError for bad or missing input (e.g. missing uid/project_key in init, email validation, OpenRouter key not found, LLM JSON/format failure) and RuntimeError for invalid state or API failures (e.g. not initialized, credits not available, send email failed, LLM API/network failure). Catch ValueError or RuntimeError (or Exception) as needed.

Failure behavior per function:

🧠 Example Use Cases

Basic Data Storage

import waveassist
waveassist.init()  # Auto-initialized from .env or worker

# Store GitHub PR data
waveassist.store_data("latest_pr", {
    "title": "Fix bug in auth",
    "author": "alice",
    "status": "open"
})

# Later, fetch it for further processing
pr = waveassist.fetch_data("latest_pr")
print(pr["title"])

LLM Integration with Credit Management

import waveassist
from pydantic import BaseModel

waveassist.init()

# Store OpenRouter API key
waveassist.store_data('open_router_key', 'your_api_key')

# Check credits before expensive operation
required_credits = 5.0
if waveassist.check_credits_and_notify(required_credits, "MyAssistant"):
    # Use LLM with structured output
    class AnalysisResult(BaseModel):
        summary: str
        confidence: float
        recommendations: list[str]

    result = waveassist.call_llm(
        model="gpt-4o",
        prompt="Analyze this data and provide recommendations...",
        response_model=AnalysisResult
    )

    # Store the structured result
    waveassist.store_data("analysis_result", result.dict())

🤝 Contributing

Want to add formats, features, or cloud extensions? PRs welcome!

📬 Contact

Need help or have feedback? Reach out at connect@waveassist.io, visit WaveAssist.io, or open an issue.

© 2026 WaveAssist