elasticdash-sdk 0.1.3

Elasticdash AI test framework for Python

ElasticDash SDK (Python)

Observability and tracing SDK for AI agent workflows. Captures LLM calls, tool invocations, and workflow traces — enabling debugging, reruns, and evaluation through the ElasticDash MCP and dashboard.

• Automatic AI interception — captures OpenAI, Gemini, and Grok calls via httpx/requests without code changes
• Tool tracing with @ed_tool — decorator-based tool registration with telemetry, rerun support, and MCP integration
• Workflow tracing — group LLM and tool calls into named traces with start_trace / end_trace
• Branch-aware traces — auto-detects git branch to separate prod vs dev traces
• MCP rerun support — traced tools can be re-executed locally via CLI for behavior validation
• CI/CD runner — fetch test groups from ElasticDash, execute locally, submit results

---

Quick Links

• Installation
• Quick Start
• Tool Tracing
• AI Call Interception
• Workflow Tracing
• MCP Integration
• CI/CD Runner
• Configuration

---

Installation

pip install elasticdash-sdk

With observability extras (socket.io for real-time backend connection):

pip install elasticdash-sdk[observability]

Requires Python 3.10+.

Cloud Setup

Add these to your .env (or CI secrets):

ELASTICDASH_SERVER_URL=https://server.elasticdash.com
ELASTICDASH_API_KEY=ed_your_api_key_here

• ELASTICDASH_SERVER_URL -- The ElasticDash cloud backend URL. For cloud users this is always https://server.elasticdash.com. For self-hosted instances, use your own backend URL.
• ELASTICDASH_API_KEY -- Your project API key. Find it in the ElasticDash dashboard under project settings.

---

Quick Start

1. Add @ed_tool to your tools

from elasticdash_sdk import ed_tool

@ed_tool
async def web_search(query: str) -> str:
    """Search the web and return results."""
    # your tool implementation
    return results

2. Initialize observability in your entrypoint

import os
from elasticdash_sdk.observability import (
    init_observability, shutdown_observability,
    ObservabilityOptions, start_trace, end_trace,
)

async def main():
    handle = await init_observability(ObservabilityOptions(
        server_url=os.environ.get("ELASTICDASH_SERVER_URL"),
        api_key=os.environ.get("ELASTICDASH_API_KEY"),
    ))

    try:
        start_trace("my_workflow")
        result = await run_my_agent("user query")
        end_trace()
    finally:
        await handle.shutdown()

3. Traces appear in ElasticDash

Every LLM call and @ed_tool invocation within a trace is captured and pushed to the backend. Use the MCP or dashboard to search, inspect, and rerun steps.

---

Tool Tracing

@ed_tool decorator

Register tools from anywhere in your codebase. Decorated functions are automatically wrapped with telemetry and registered for MCP rerun discovery.

from elasticdash_sdk import ed_tool

@ed_tool
def fetch_order(order_id: str) -> dict:
    """Fetch order details from the database."""
    return db.orders.find(order_id)

@ed_tool(name="custom_name")
async def search_docs(query: str, limit: int = 5) -> list:
    """Search documentation."""
    return await doc_service.search(query, limit)

Both sync and async functions are supported. The decorator:

• Records input, output, duration, and errors as trace events
• Registers the tool in a global registry for CLI rerun and MCP discovery
• Preserves the original function signature and docstring

ed_tools.py (simple alternative)

For projects that prefer a single file, define tools in ed_tools.py at the project root. Public functions are discovered automatically via AST scanning -- no decorator needed.

# ed_tools.py
def fetch_order(order_id: str) -> dict:
    return db.orders.find(order_id)

async def search_docs(query: str, limit: int = 5) -> list:
    return await doc_service.search(query, limit)

Both approaches work side by side. @ed_tool is recommended for tools spread across the codebase; ed_tools.py works for simple projects.

Running a tool via CLI

elasticdash run-tool fetch_order --input '{"order_id": "ord-123"}'
elasticdash run-tool search_docs --input '{"query": "auth", "limit": 3}'

This is used by the MCP server to rerun traced tool steps for behavior validation.

---

AI Call Interception

The SDK automatically intercepts and records HTTP calls to:

Provider	Endpoints
OpenAI	api.openai.com/v1/chat/completions, /v1/responses
Gemini	generativelanguage.googleapis.com/.../generateContent
Grok (xAI)	api.x.ai/v1/chat/completions

No code changes needed -- init_observability() installs the interceptors automatically. Each captured AI event includes:

• Model name and provider
• Input (messages, system prompt, or Responses API input)
• Output (completion text)
• Token usage (input/output/total)
• Duration

Manual AI recording

For providers not covered by automatic interception:

from elasticdash_sdk.interceptors import wrap_ai

my_llm_call = wrap_ai("my-model", async_llm_function)
result = await my_llm_call(messages)

---

Workflow Tracing

Group related LLM and tool calls into named traces using start_trace / end_trace:

from elasticdash_sdk.observability import start_trace, end_trace

async def handle_request(user_input: str):
    start_trace("chat_handler")
    try:
        plan = await planner_agent.run(user_input)
        results = await search_tool(plan.query)
        response = await writer_agent.run(results)
        return response
    finally:
        end_trace()

All @ed_tool calls and intercepted AI calls between start_trace and end_trace are grouped under the same trace ID.

Branch-aware tracing

The SDK auto-detects the current git branch and includes it in trace metadata. This lets you filter traces by environment (prod vs dev) in the MCP and dashboard.

# Auto-detected from local git or CI environment
handle = await init_observability(ObservabilityOptions(
    server_url="https://server.elasticdash.com",
    api_key="ed_xxx",
))

# Or explicitly set
handle = await init_observability(ObservabilityOptions(
    server_url="https://server.elasticdash.com",
    api_key="ed_xxx",
    branch="staging",
))

---

MCP Integration

The ElasticDash MCP enables coding agents (Claude Code, Cursor, etc.) to search and debug traces from your agent's runtime behavior.

How it works

1. Your agent runs with the SDK installed -- traces are pushed to the backend
2. A user reports unexpected agent behavior
3. The coding agent uses MCP tools to investigate:
   • search_traces -- find traces by keyword, time range, tool names, branch
   • get_trace_details -- inspect individual events in a trace
   • rerun_step -- re-execute a suspicious step to check reproducibility

Tool reruns

For AI steps, the MCP re-executes the LLM call directly with the original input.

For tool steps, the MCP returns the tool name and original input, then the coding agent calls it via:

elasticdash run-tool <tool_name> --input '<original_input_json>'

This requires the tool to be registered via @ed_tool or defined in ed_tools.py.

---

CI/CD Runner

Run ElasticDash test groups from CI pipelines:

elasticdash ci --server-url $ELASTICDASH_SERVER_URL --api-key $ELASTICDASH_API_KEY

How it works

1. Fetch -- retrieves active test groups from the backend
2. Execute -- runs each test locally (tool reruns, AI reruns, full workflows)
3. Evaluate -- checks expectations (output-contains, latency, token budget, LLM-judge)
4. Submit -- posts results with git metadata (branch, commit, PR number)

GitHub Actions example

name: AI Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - run: pip install -e .

      - name: Run ElasticDash CI tests
        run: elasticdash ci
        env:
          ELASTICDASH_SERVER_URL: ${{ secrets.ELASTICDASH_SERVER_URL }}
          ELASTICDASH_API_KEY: ${{ secrets.ELASTICDASH_API_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

Git branch, commit SHA, and PR info are auto-detected from CI environment variables (GitHub Actions, GitLab CI, CircleCI, Buildkite).

All CLI flags

Flag	Env Var	Description
--server-url	ELASTICDASH_SERVER_URL	Backend API URL (required)
--api-key	ELASTICDASH_API_KEY	Project API key (required)
--workflow	--	Filter test groups by workflow name
--tags	--	Filter by tags (comma-separated)
--triggered-by	--	Trigger source label (default: ci)
--git-branch	Auto-detected	Git branch name
--git-commit	Auto-detected	Git commit SHA
--git-pr-number	Auto-detected	PR number

---

Testing

The SDK also includes a test runner for AI workflow assertions:

elasticdash test              # discover all *.ai_test.py files
elasticdash run my_flow.ai_test.py  # run a single file

from elasticdash_sdk import ai_test, expect

@ai_test("checkout flow")
async def test_checkout(ctx):
    await run_checkout(ctx)

    expect(ctx.trace).to_have_llm_step(model="gpt-4o", contains="order confirmed")
    expect(ctx.trace).to_call_tool("chargeCard")

See the test writing guidelines for full documentation on matchers and test patterns.

---

Configuration

Optional elasticdash.config.py at the project root:

config = {
    "test_match": ["**/*.ai_test.py"],
}

CLI commands

Command	Description
elasticdash test [path]	Discover and run test files
elasticdash run <file>	Run a single test file
elasticdash run-tool <name> --input '<json>'	Execute a tool by name
elasticdash ci	Run CI/CD test pipeline
elasticdash observe	Start observability mode
elasticdash dashboard	Open workflows dashboard
elasticdash portal	Start portal server for remote execution
elasticdash init-guide	Generate starter config files

---

Programmatic API

from elasticdash_sdk.observability import (
    init_observability, shutdown_observability,
    ObservabilityOptions, start_trace, end_trace,
)
from elasticdash_sdk import ed_tool

# Observability
handle = await init_observability(ObservabilityOptions(
    server_url="https://server.elasticdash.com",
    api_key="ed_xxx",
))

start_trace("my_workflow")
# ... your agent code with @ed_tool functions ...
end_trace()

await handle.shutdown()

# Tool registration
from elasticdash_sdk import ed_tool

@ed_tool
async def my_tool(query: str) -> str:
    return await do_something(query)

# CI runner
from elasticdash_sdk.ci.runner import run_ci
from elasticdash_sdk.ci.types import CIRunConfig

summary = await run_ci(CIRunConfig(
    server_url="https://server.elasticdash.com",
    api_key="ed_xxx",
))
print(f"{summary.passed}/{summary.total} passed")

---

License

MIT