neosigma-sdk

Trace your AI agents and ship the results to NeoSigma. Add a few lines, run your agents as usual, and every run's model calls, tool calls, and token usage lands in NeoSigma as a structured OpenTelemetry trace.

• Dark by default: with no API key (and NEOSIGMA_CONSOLE_EXPORT=false) the SDK is a complete no-op and never touches your application's own OpenTelemetry setup, so it's safe to leave in place.
• Provider-agnostic: a small core with thin adapters that wrap the agent framework you already use, with no hard dependency on any provider SDK.

Install

pip install neosigma-sdk
# or:  uv add neosigma-sdk

Quickstart

import neosigma_sdk as neosigma

neosigma.init()        # reads NEOSIGMA_API_KEY from the environment
# ... trace your agent with the decorators or an adapter (below) and run as usual ...
neosigma.shutdown()    # flush before exit (long-running servers flush in the background)

Tracing your agents

A few ways to produce spans, and they compose: anything traced while an interaction is active nests under it, so one run is one trace.

• Decorators mark a run and its steps, with no framework required. @interaction is the run; @tool is a step inside it.

  @neosigma.tool()
  def search(query: str) -> list[str]:
      ...
  
  @neosigma.interaction()
  def answer(question: str) -> str:
      hits = search(question)   # nested under the interaction
      ...
  

• turn() / finish() track a run whose lifecycle spans multiple functions, where a single decorator cannot wrap the whole thing. One user message is one trace: turn() always opens a fresh root, tied to a session_id and a turn_id (minted, or supplied) that is also the join key for capture() events below.

  t = neosigma.turn(session_id="sess_123", user_message=question, distinct_id="user_123")
  t.set_attributes({"plan": "pro"})   # attach metadata/tags to the run
  reply = run_agent(question)         # tool and LLM calls nest under this run
  t.finish(output=reply)
  

• Auto-instrumentation traces raw LLM clients (Anthropic, OpenAI) with no per-call code: install the instrumentation extra and call neosigma.init(tracing_enabled=True).

  import anthropic
  
  neosigma.init(tracing_enabled=True)   # turn on the off-the-shelf instrumentors
  client = anthropic.Anthropic()
  client.messages.create(...)           # this call is now a traced span
  

See the documentation for the full API and configuration.

Product events and the turn_id spine

Agent traces tell you what the model did. Product events tell you what the user did (a button click, a feature used, a conversion). NeoSigma joins the two streams on a single id, the turn_id, so you can go from "this user clicked rewind" to "here is the exact agent trace behind it" without stitching timestamps.

turn_id is the durable correlation key. One turn is one user message plus everything the agent did in response; a session is a series of turns. You supply the id, bind it once, and from then on:

• every span opened inside the turn carries it (stamped by the CorrelationSpanProcessor, so adapters, auto-instrumented LLM clients, and your own spans all pick it up with no per-framework code), and
• every product event you capture() inside the turn carries the same value.

Both land in NeoSigma keyed on turn_id, and join there.

Binding the turn

Use trace() when the span is produced elsewhere (an adapter or auto-instrumented client), or begin() / @interaction to also open a root span. Both bind the same ambient ids:

import neosigma_sdk as neosigma

neosigma.init()

with neosigma.trace(turn_id="turn_abc", distinct_id="user_123"):
    reply = run_agent(question)            # any spans here carry turn_abc
    neosigma.capture("agent_answered",     # this event carries turn_abc too
                     {"helpful": True, "latency_ms": 820})

Contextvars propagate across await within a task, but not across a process or queue hop. Across such a boundary, thread the turn_id into the job payload and re-bind it on the far side (with neosigma.trace(turn_id=...) or begin(turn_id=...)).

capture() and identify()

• capture(event_name, properties=None) emits a product event stamped with the ambient turn_id / distinct_id / session_id (and the active span's trace_id, best effort). Property values are scalar (str, int, float, bool). Each event gets an event_uuid idempotency key, so a retried delivery de-dupes rather than double-counts.
• identify(distinct_id, properties=None) binds distinct_id (the analytics actor) for every later event and span in the task, and emits an $identify event. Call it once at login; a per-turn trace() that omits distinct_id will not clobber it.

neosigma.identify("user_123", {"plan": "pro"})
# ... later, anywhere in the same task ...
neosigma.capture("rewind_clicked", {"surface": "chat"})   # distinct_id rides along

Both calls are fail-open: a telemetry failure drops the event, it never raises into your application.

Where events go: the EventSink

capture() hands each built ProductEvent to the active EventSink, it never writes a datastore directly (the SDK runs in your process and has no such access). When you call init() with an API key, the SDK installs an HttpEventSink that batches events on a background daemon thread and POSTs them to the events endpoint with your API key, the same auth path traces use. It is bounded and fail-open: a full queue drops newest, an unreachable ingest is swallowed, and your hot path never blocks. With no API key, a default in-process BufferSink keeps capture() usable (and testable) but ships nothing. shutdown() stops the flush thread and drains anything queued, so call it before exit.

Already using PostHog or Mixpanel?

If your product is already instrumented with PostHog or Mixpanel, you do not need to re-instrument. Wrap the client once and every event you already send also flows into NeoSigma, sharing the same turn_id spine. Your existing provider keeps receiving every event unchanged (this mirrors, it does not redirect):

import posthog
import neosigma_sdk as neosigma

neosigma.init()
ph = neosigma.wrap_posthog(posthog)          # the posthog module or a Posthog() instance

# Use it exactly as before. Each call ALSO reaches NeoSigma.
ph.capture("user_123", "rewind_clicked", {"surface": "chat"})
ph.identify("user_123", {"plan": "pro"})

Mixpanel works the same way via wrap_mixpanel(Mixpanel(token)): track(...) mirrors to capture() and people_set(...) to identify(). Both wraps are transparent (all other attributes delegate unchanged), duck-typed (the SDK never imports posthog / mixpanel, so no new dependency), and fail-open (the mirror is best-effort and can never break your analytics call). An event fired inside a trace() / begin() block joins to that agent trace on turn_id; one fired outside is still a valid event, joinable by distinct_id.

Adapters

Thin wrappers that trace an agent framework you already use, feeding the same trace contract as the decorators. More are on the way.

• Anthropic Managed Agents: wrap_managed_agents(client) traces a session's model and tool calls (sync Anthropic and AsyncAnthropic).
• Claude Agent SDK: trace_claude(stream) traces a query(...) run; for the stateful ClaudeSDKClient, ClaudeTracingProcessor().configure() is the zero-touch option.

Example: Anthropic Managed Agents

import anthropic
import neosigma_sdk as neosigma

neosigma.init()
client = neosigma.wrap_managed_agents(anthropic.Anthropic())

# Build and run a Managed Agents session as you normally would. Streaming the
# session produces one NeoSigma trace: model calls, tool calls, and token usage.
session = client.beta.sessions.create(agent=agent, environment_id=environment.id)
with client.beta.sessions.events.stream(session_id=session.id) as stream:
    for event in stream:
        ...

neosigma.shutdown()

AsyncAnthropic works the same way (async with / async for).

Configuration

Common settings read from a NEOSIGMA_* environment variable, or can be passed to init(...):

See the NeoSigma documentation for the complete configuration reference and API docs.

License

Released under the MIT License.