naggy 0.4.0

NaggyAI Python SDK — instrument LLM calls and query the NaggyAI platform

naggy

Python SDK for the NaggyAI platform.
Track LLM calls, query costs, get nags, and surface insights — from any Python app.

---

Install

pip install naggy

Requires Python ≥ 3.9. Dependencies: httpx>=0.27, pydantic>=2.

---

Quick start

import naggy

naggy.init("nag_sk_...")   # once at startup

Auto-instrument Anthropic

import anthropic, naggy

naggy.init("nag_sk_...")
client = naggy.instrument_anthropic(anthropic.Anthropic())

# Every client.messages.create() now sends a telemetry event automatically
response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

Auto-instrument OpenAI

import openai, naggy

naggy.init("nag_sk_...")
client = naggy.instrument_openai(openai.OpenAI())

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Pass environment="staging" to either instrument call to tag events accordingly.

@trace decorator

Wraps a function and sends one event per call. Tokens are auto-extracted from Anthropic/OpenAI response usage objects.

@naggy.trace("anthropic", "claude-opus-4-7")
def summarize(text: str) -> str:
    return client.messages.create(...).content[0].text

Optional kwargs: client= (override default client), environment="production".

trace_context — fine-grained control

with naggy.trace_context("anthropic", "claude-opus-4-7") as ctx:
    response = client.messages.create(...)
    ctx.input_tokens = response.usage.input_tokens
    ctx.output_tokens = response.usage.output_tokens
    ctx.cost_usd = 0.0042

TraceContext attributes you can set inside the block:

Attribute	Type	Default
input_tokens	int	0
output_tokens	int	0
cost_usd	float	0.0
status	str	"success"

ctx.trace_id (str) is auto-generated — read it to link a feedback event to this trace.

Full signature:

trace_context(
    provider: str,
    model: str,
    *,
    client=None,             # override default NaggyClient
    environment="production",
    trace_id=None,           # supply your own UUID or let the SDK generate one
)

---

Async (FastAPI / asyncio)

AsyncNaggyClient mirrors the full API surface with async/await:

from naggy import AsyncNaggyClient

client = AsyncNaggyClient("nag_sk_...")

# any method from NaggyClient is available as a coroutine
score = await client.score()
await client.ingest([llm_event(provider="openai", model="gpt-4o")])
insights = await client.list_insights()

FastAPI example:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from naggy import AsyncNaggyClient

naggy_client: AsyncNaggyClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    global naggy_client
    naggy_client = AsyncNaggyClient("nag_sk_...")
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/score")
async def get_score():
    return await naggy_client.score()

---

High-throughput batching

For production apps that call ingest frequently, use a queue to buffer events and flush them in batches rather than one HTTP request per call.

Sync — NaggyQueue

from naggy import NaggyClient, NaggyQueue

client = NaggyClient("nag_sk_...")
queue = NaggyQueue(client, max_size=100, flush_interval=5.0)

# non-blocking — flushes automatically when buffer hits max_size or every 5 s
queue.push(llm_event(provider="openai", model="gpt-4o", input_tokens=200))

queue.flush()   # manual flush
queue.close()   # flush remaining + stop background thread

Or as a context manager (calls close() on exit):

with NaggyQueue(client, max_size=100, flush_interval=5.0) as queue:
    for event in events:
        queue.push(event)

Async — AsyncNaggyQueue

from naggy import AsyncNaggyClient, AsyncNaggyQueue

client = AsyncNaggyClient("nag_sk_...")

async with AsyncNaggyQueue(client, max_size=100, flush_interval=5.0) as queue:
    await queue.push(llm_event(provider="anthropic", model="claude-opus-4-7"))

Both queues register an atexit / cancellation handler so buffered events are sent on shutdown. Flush failures are logged as warnings and dropped — they never raise inside your application code.

Parameters:

Parameter	Default	Description
max_size	100	Flush immediately when buffer reaches this size
flush_interval	5.0	Seconds between background flushes

---

Auth

Every API call requires a nag_sk_... bearer key.
Create one in the Parallaxed dashboard or via the API:

client = naggy.get_client()
key = client.create_key(project_id="proj_...")
print(key["api_key"])   # nag_sk_...

---

API reference

Module-level helpers

naggy.init(api_key, *, base_url="https://parallaxed.app", timeout=15.0) -> NaggyClient
naggy.get_client() -> NaggyClient   # raises RuntimeError if init() not yet called

You can also instantiate directly:

from naggy import NaggyClient, AsyncNaggyClient

client = NaggyClient("nag_sk_...", base_url="https://parallaxed.app", timeout=15.0)
async_client = AsyncNaggyClient("nag_sk_...", base_url="https://parallaxed.app", timeout=15.0)

All methods below exist on both NaggyClient (sync) and AsyncNaggyClient (async — prefix each call with await).

Events

client.ingest(events: list[dict]) -> dict
# returns {"accepted": N}

client.list_events(
    *,
    provider: str | None = None,
    model: str | None = None,
    status: str | None = None,   # "success" | "error"
    limit: int = 50,
    offset: int = 0,
) -> list[dict]

Overview & config

client.overview() -> dict
client.config()   -> dict

Stats

client.cost_by_provider() -> list[dict]   # [{provider, calls, total_cost_usd, ...}]
client.cost_by_model()    -> list[dict]
client.error_stats()      -> dict         # {total_calls, total_errors, error_rate}

Naggy Score

client.score()                     -> dict         # {score, grade, ...}
client.score_history(*, limit=30)  -> list[dict]

Insights

client.list_insights(*, status="active", limit=50, offset=0)  -> list[dict]
client.generate_insights()                                     -> list[dict]
client.dismiss_insight(insight_id: str)                        -> dict
client.apply_insight(insight_id: str)                          -> dict

Alerts

client.list_alert_rules(*, limit=50, offset=0)  -> list[dict]
client.evaluate_alerts()                         -> list[dict]   # triggered rules
client.alert_history(*, limit=50)                -> list[dict]
client.intelligence()                            -> dict

Nags

client.nag_history(*, status="", limit=50, offset=0)  -> list[dict]
client.acknowledge_nag(nag_item_id: str)               -> dict

Traces

client.list_traces(*, limit=50, offset=0)  -> list[dict]
client.get_trace(trace_id: str)            -> dict

Feedback

client.submit_feedback(
    trace_id: str,
    *,
    score: float,              # 0.0–1.0
    label: str = "",           # e.g. "thumbs_up" | "thumbs_down"
    comment: str = "",
    environment: str = "production",
) -> dict

client.list_feedback(
    *,
    trace_id: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[dict]

Daily standup

client.standup()                     -> dict
client.standup_history(*, limit=30)  -> list[dict]

API key management

client.create_key(project_id: str, *, name: str = "default") -> dict   # {api_key, key_id}
client.list_keys(project_id: str)                             -> list[dict]
client.revoke_key(key_id: str)                                -> dict

Health

client.health() -> dict   # {status, active_projects, last_event_at} — no auth required

---

Event builders

Use these to construct event payloads for client.ingest() or queue.push().

from naggy import llm_event, feedback_event

event = llm_event(
    provider="anthropic",
    model="claude-opus-4-7",
    input_tokens=512,
    output_tokens=128,
    cost_usd=0.0063,
    latency_ms=820,
    status="success",                    # or "error"
    trace_id="...",                      # auto-generated UUID if omitted
    environment="production",
    metadata={"user_id": "u_123"},       # merged into the event dict
)

fb = feedback_event(
    trace_id="...",
    score=1.0,
    label="thumbs_up",
    comment="Great answer",
    environment="production",
)

client.ingest([event, fb])

---

Error handling

from naggy import NaggyHTTPError

try:
    client.score()
except NaggyHTTPError as e:
    print(e.status_code, e.detail)

Telemetry failures inside @trace, trace_context, auto-instrumentation, and queue flushes are silently swallowed — the SDK never raises inside your application code.