Server-side and Python

Some moments only your server knows: a failed login and its reason, a policy bound, a payment captured. Post them to the same ingest endpoint the SDKs use. They join the person’s browser sessions through the user_id.

POST /api/ingest/batch

curl -s https://clickiq.example.com/api/ingest/batch \
  -H 'Content-Type: application/json' -H 'Origin: https://app.example.com' \
  -d '{"write_key": "ciq_…", "sent_at": 1761230000000, "events": [{
        "type": "custom", "name": "Policy bound", "ts": 1761229999000,
        "anonymous_id": "server_u_123", "user_id": "u_123",
        "url": "https://app.example.com/", "props": {"lob": "Property", "source": "server"},
        "value": 12500, "geo": {"country": "US", "region": "Texas", "city": "Dallas"} }]}'
# 200 {"accepted": 1, "rejected": 0}

Fields that matter from a server

The Python SDK

insightclickiq-sdk (import clickiq) has no runtime dependencies. Events are batched on a daemon thread (up to 50 per request, every 10 s), retried with exponential backoff on 429, 5xx and network errors (up to 5 minutes), dropped on other 4xx, capped at 10,000 waiting, and flushed when the interpreter exits. Nothing it does can raise into your request.

pip install insightclickiq-sdk                 # or "insightclickiq-sdk[langchain]"

import clickiq
clickiq.init("ciq_…", "https://clickiq.example.com")

clickiq.track("Policy bound", {"lob": "Property", "premium": 1200}, user_id="u_42")

# the browser's anonymous id (cookie clickiq_aid) joins the person: one person across browser, app and server
clickiq.identify("u_42", {"team": "Claims", "manager": "Ann"}, anonymous_id=request.cookies.get("clickiq_aid"))

clickiq.flush()       # send now; clickiq.shutdown() flushes and stops (also registered at exit)

init(key, host, batch_size=50, interval=10.0, default_user_id=None, device={...}, capture_text=False, enabled=True, debug=False). Server events carry platform = other and sdk = clickiq-python, so you can filter them.

Tracing your AI agents

with clickiq.ai.trace(agent="claims-copilot", model="claude-sonnet-4-6", provider="bedrock", user_id="u_42") as t:
    with t.tool("policy_lookup"):                 # a timed $ai_tool_call
        policy = lookup(claim)
    reply = llm.invoke(prompt)
    t.set(input_tokens=reply.usage.input, output_tokens=reply.usage.output, cost_usd=0.004)
# on exit: one $ai_generation with latency_ms and tool_calls; status "error" and the exception if the block raised (re-raised)
t.feedback("up")                                  # $ai_feedback for the same trace

# or record calls yourself
clickiq.ai.generation(agent="claims-copilot", model="claude-haiku-4-5", input_tokens=812, output_tokens=240,
                      latency_ms=1830, status="ok", trace_id=tid, user_id="u_42")
clickiq.ai.tool_call(agent="claims-copilot", tool="policy_lookup", latency_ms=120, status="ok", trace_id=tid)
clickiq.ai.feedback(agent="claims-copilot", trace_id=tid, rating="down", comment="wrong policy")

LangChain

from clickiq.langchain import ClickIQCallbackHandler

chain.invoke({"question": q},
             config={"callbacks": [ClickIQCallbackHandler(agent="claims-copilot", user_id="u_42")]})

Every LLM or chat-model call becomes an $ai_generation (model from the invocation, provider from the class, tokens from the usage metadata, latency, tool calls, and the chain’s root run as trace_id); every tool run becomes an $ai_tool_call. Prompt and response text is never sent unless you pass capture_text=True, and the server keeps it only when the project’s ai_capture_text setting is on. Once events arrive, create the AI agents dashboard from its template.

Without an SDK: plain HTTP

import os, threading, time, logging, requests
log = logging.getLogger(__name__)

def _send(event):
    try:
        r = requests.post(f"{os.environ['CLICKIQ_HOST']}/api/ingest/batch",
                          json={"write_key": os.environ["CLICKIQ_WRITE_KEY"], "sent_at": int(time.time() * 1000), "events": [event]},
                          timeout=4)
        if r.status_code >= 400:
            log.warning("clickiq rejected %s %s", r.status_code, r.text[:200])
    except Exception as exc:            # analytics must never raise into the request
        log.warning("clickiq send failed: %s", exc)

def track(user_id, name, props=None, value=None, geo=None):
    event = {"type": "custom", "name": name, "ts": int(time.time() * 1000),
             "anonymous_id": f"server_{user_id or 'system'}", "user_id": user_id,
             "url": "https://app.example.com/", "props": {**(props or {}), "source": "server"}, "value": value}
    if geo:
        event["geo"] = geo
    threading.Thread(target=_send, args=(event,), daemon=True).start()

track("u_123", "Login failed", {"method": "password", "reason": "bad_password"})

This is the pattern from the insurance client-360 instrumentation guide, for when you would rather not add a package: a daemon thread, a short timeout, never on the request path. The same approach works from any language with an HTTP client.