Jasmind Documentation

# 1. Start the server
git clone https://github.com/jasmind-ai/jasmind
docker-compose up -d

# 2. Install the SDK
pip install jasmind

from jasmind import Jasmind, wrap_openai
from openai import OpenAI

# Connect to your local Jasmind instance
jm = Jasmind(endpoint="http://localhost:8080")

# Wrap the OpenAI client — every call is now captured
client = wrap_openai(OpenAI(), tracer=jm)


@jm.trace(name="research_task")
def run_agent(user_input: str) -> str:
    # your agent logic — completely unchanged
    for step in range(10):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": user_input}],
        )
        # ↑ Observation: cost, tokens, latency, loop detection
        # ↑ Security: injection, PII, secrets — scanned in parallel
        ...
    return result

version: '3.8'
services:
  backend:
    build: ./backend
    ports:
      - "8080:8080"
    volumes:
      - ./data:/app/data
    environment:
      - DATABASE_URL=sqlite:///./data/jasmind.db

  frontend:
    build: ./frontend
    ports:
      - "3000:3000"
    environment:
      - VITE_API_URL=http://localhost:8080

docker-compose up -d

# Collector API → http://localhost:8080
# Dashboard     → http://localhost:3000

from jasmind import Jasmind

jm = Jasmind(endpoint="http://localhost:8080")

@jm.trace(name="classify_email")
def classify_email(email_body: str) -> str:
    # your function — unchanged
    ...

from jasmind import Jasmind, wrap_openai
from openai import OpenAI

jm = Jasmind(endpoint="http://localhost:8080")
client = wrap_openai(OpenAI(), tracer=jm)

# Use client exactly as before — nothing else changes
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": prompt}],
)

import os
from openai import OpenAI

# Change only base_url — one line
client = OpenAI(
    base_url="http://localhost:8080/proxy/openai",
    api_key=os.environ["OPENAI_API_KEY"],
)

# Everything below stays exactly the same
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": prompt}],
)

curl -X POST http://localhost:8080/api/v1/spans \
  -H "Content-Type: application/json" \
  -d '{
    "trace_id":         "abc123",
    "name":             "call_gpt4o",
    "span_type":        "llm",
    "model":            "gpt-4o",
    "prompt_tokens":    1200,
    "completion_tokens": 340,
    "cost_usd":         0.0092,
    "started_at":       "2026-06-27T10:00:00Z",
    "ended_at":         "2026-06-27T10:00:02Z"
  }'

from collections import Counter

def detect_loop(spans):
    name_counts = Counter(s.name for s in spans)
    return any(count > 5 for count in name_counts.values())

LONG_CONTEXT_THRESHOLD = 8000  # prompt_tokens

# hash(tool_name + tool_input) duplicates within one trace
def detect_repeated_calls(spans):
    seen = set()
    for span in spans:
        key = hash((span.tool_name, str(span.tool_input)))
        if key in seen:
            span.obs_flags.append("repeated_call")
        seen.add(key)

INJECTION_PATTERNS = [
    r"ignore (all |previous |above |prior )?instructions?",
    r"forget (everything|what you were told)",
    r"you are now",
    r"new (system |)prompt",
    r"(act|pretend|roleplay) as",
    r"DAN",
    r"jailbreak",
    r"override",
]

PII_PATTERNS = {
    "ssn":         r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",
    "email":       r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    "phone":       r"\b(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b",
}

SECRET_PATTERNS = {
    "openai_key":    r"sk-[A-Za-z0-9]{32,}",
    "anthropic_key": r"sk-ant-[A-Za-z0-9\-]{32,}",
    "aws_key":       r"AKIA[0-9A-Z]{16}",
    "github_token":  r"ghp_[A-Za-z0-9]{36}",
    "generic_key":   r"(api[_-]?key|secret|token)\s*[:=]\s*['\"]?[A-Za-z0-9\-_.]{16,}",
}

RISKY_TOOL_SEQUENCES = [
    ["read_file", "send_email"],            # exfiltration
    ["read_file", "write_file", "delete"], # read-modify-destroy
    ["search_web", "execute_code"],        # search-then-exec
]

def calculate_security_score(alerts):
    score = 100
    deductions = {
        "critical": 40,
        "high":     20,
        "medium":   10,
        "low":       5,
    }
    for alert in alerts:
        score -= deductions[alert.severity]
    return max(0, score)

CREATE TABLE traces (
    id              TEXT PRIMARY KEY,
    name            TEXT,
    status          TEXT,   -- running / completed / failed / loop_detected
    started_at      DATETIME,
    ended_at        DATETIME,
    total_tokens    INTEGER,
    total_cost_usd  REAL,
    security_score  INTEGER, -- 0–100
    metadata        JSON
);

CREATE TABLE spans (
    id                   TEXT PRIMARY KEY,
    trace_id             TEXT REFERENCES traces(id),
    parent_id            TEXT,
    name                 TEXT,
    span_type            TEXT,   -- llm / tool / chain / agent
    model                TEXT,
    started_at           DATETIME,
    ended_at             DATETIME,
    latency_ms           INTEGER,
    prompt_tokens        INTEGER,
    completion_tokens    INTEGER,
    cost_usd             REAL,
    prompt_preview       TEXT,   -- first 500 chars
    completion_preview   TEXT,
    tool_name            TEXT,
    tool_input           JSON,
    tool_output          TEXT,
    obs_flags            JSON,   -- ["loop", "long_context", "repeated_call"]
    sec_flags            JSON    -- ["prompt_injection", "pii_leak", "secret_leak", ...]
);

CREATE TABLE security_alerts (
    id          TEXT PRIMARY KEY,
    trace_id    TEXT REFERENCES traces(id),
    span_id     TEXT REFERENCES spans(id),
    created_at  DATETIME,
    severity    TEXT,   -- critical / high / medium / low
    alert_type  TEXT,   -- prompt_injection / pii_leak / secret_leak / behavioral_anomaly
    description TEXT,
    evidence    TEXT,   -- redacted
    status      TEXT DEFAULT 'open' -- open / acknowledged / resolved
);

CREATE TABLE insights (
    id                   TEXT PRIMARY KEY,
    trace_id             TEXT REFERENCES traces(id),
    created_at           DATETIME,
    wasted_cost_usd      REAL,
    waste_reason         TEXT,   -- loop_detected / oversized_context / repeated_calls
    suggestion_type      TEXT,   -- cheaper_model / reduce_context / add_caching
    suggestion_detail    TEXT,
    estimated_saving_usd REAL,
    quality_impact       TEXT,   -- minimal / moderate / significant
    risk_change          TEXT    -- lower / same / higher
);