neosigma

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.

Using JavaScript or TypeScript? See the TypeScript SDK.

Use cases

• Agent observability. See every model call, tool call, and token count from a run as one trace, without hand-instrumenting each call.
• Product analytics joined to agent behavior. capture() events and agent traces share one turn_id, so a product signal (a click, a conversion) links to the exact run behind it.
• Keep your existing stack. Dual-export the same traces to LangSmith, Braintrust, or Langfuse, and mirror PostHog or Mixpanel events, with no migration.
• Move history in bulk. Import past traces from another provider, or export your own.

How it works

The SDK runs inside your application process. It builds spans on a private OpenTelemetry provider (never the global one, unless you opt in), stamps each with the ambient turn_id, and ships them to NeoSigma over OTLP/HTTP with your API key. Product events take a parallel path through a background event sink. Both are bounded and fail-open, so telemetry never blocks or breaks your app, and with no API key the SDK is a complete no-op. The sections below cover each piece in detail.

Install

pip install neosigma
# or:  uv add neosigma

Quickstart

import 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)

Managed agents

The same pip install neosigma distribution includes the Managed Agents API client. Python packages do not use npm-style scoped names such as @neosigma/managed-agents; import this client as neosigma.agents or from the neosigma package root.

Set the API key once:

export NEOSIGMA_API_KEY="ns_live_..."

Then create the three durable resources and send work to the session:

from neosigma import NeoSigma

client = NeoSigma()

environment = client.environments.create(
    project_id="YOUR_PROJECT_ID",
    name="python-dev",
)

agent = client.agents.create(
    name="Coding assistant",
    instructions="Make focused changes and run the relevant tests.",
    model={"provider": "anthropic", "id": "claude-sonnet-4-5"},
    harness="claude",
    task_type="create_pr",
)

session = client.sessions.create(
    agent_id=agent.agent.id,
    environment_id=environment.environment.id,
)

client.sessions.events.send(
    session_id=session.session.id,
    messages=[{"content": "Fix the failing unit test."}],
)

events = client.sessions.events.list(session_id=session.session.id)
for event in events.items:
    print(event.type, event.payload)

Creating an agent also creates its backing project and trace project. Managed session runs are traced automatically to that project. Open Traces in NeoSigma and select the project named after the agent. Your application does not initialize tracing or configure a trace project.

The resource layout follows the API operation names:

• client.agents.create, update, get, list, and versions.list
• client.environments.create, update, get, list, and versions.list
• client.sessions.create, get, list, interrupt, events.list, and events.send

List methods return items and an optional next_cursor. Send that cursor back unchanged to get the next page. AsyncNeoSigma exposes the same resource layout with await. The production API is the default. To test against the development contract, pass base_url="https://api.dev.neosigma.ai" or set NEOSIGMA_MANAGED_AGENTS_BASE_URL.

Updating the generated API layer

api/agents.json at the repository root is the committed source for the private generated models and endpoint functions under neosigma.agents._generated. The public resource classes are thin, stable argument-shaping wrappers over those functions. Do not edit generated files directly.

Refresh the scoped snapshot from the live specification and regenerate:

uv run python scripts/generate_agents_sdk.py \
  --source https://api.dev.neosigma.ai/openapi.json

Check for generated drift without writing files:

uv run python scripts/generate_agents_sdk.py --check

The refresh keeps only agents.*, environments.*, and sessions.* paths and their transitively referenced schemas. CI generates from the committed snapshot, not the network, and fails when generated files are stale.

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 turn() / @interaction to also open a root span. Both bind the same ambient ids:

import 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 neosigma.turn(turn_id=...)).

Routing a turn to a project

Declare a project on the turn and every span in it carries neosigma.project, which takes precedence over the process-wide project set at init(). Use it when one process serves several projects.

with neosigma.turn(project="acme-support", session_id=session_id) as current:
    reply = run_agent(question)

# Or when the project is only known after the turn opens:
with neosigma.turn(session_id=session_id) as current:
    current.set_project(resolve_project(question))

Four things worth knowing:

• A trace has one project, and the first declaration wins. A second, differing one is a conflict rather than an update, so it warns and the first one stands. This keeps a framework default from silently overwriting a project you declared yourself.
• A nested turn() inside a trace that has no project yet declares for the whole trace, which is how you declare when a framework or middleware owns the outer turn. Inside a trace that already has one it warns and inherits.
• Spans that started before the declaration stay unlabelled. They cannot be re-stamped once open, and labelling them later would leave one trace disagreeing with itself. The trace root is still open, so it does take the project. Prefer turn(project=...) when you can.
• set_attributes refuses neosigma.project. Routing comes from a declaration, never from forwarded metadata, so user-supplied data cannot pick the project.

Across a process or queue hop, thread the resolved project alongside the turn_id and pass it back in (neosigma.trace(turn_id=..., project=...)). Pass the value you were given rather than recomputing it on the far side, whose configuration may differ.

Serving many projects from one HTTP service? The FastAPI middleware owns the turn, so declare the project there rather than in the handler:

app.add_middleware(
    NeosigmaTurnMiddleware,
    project_resolver=lambda request: request.headers.get("x-tenant", ""),
)

project_resolver takes precedence over the static project= argument, and falls back to it when the resolver returns "" or raises. A failing tenant lookup costs the label, never the request.

An adapter that opens its own traces takes a project too, for a process that runs several agents belonging to different projects:

async for message in neosigma.trace_claude(query(prompt=...), project="agent-one"):
    ...

client = neosigma.wrap_managed_agents(anthropic_client, project="agent-two")

That binds the project for every trace the adapter opens, so it fits one agent per project, not one project per request. For a multi-tenant service sharing a client, declare on the enclosing turn instead. Omitted, an adapter inherits whatever turn encloses it.

Product events carry the project too. capture() stamps the ambient turn's project, and takes an explicit project= for a caller that knows the routing without being inside the turn that owns it. An event emitted outside any turn (telemetry-only mode, a wrapped analytics client, identify() at login) names no project and resolves to the default.

Work handed to a thread follows the usual contextvar rules. asyncio.to_thread copies the context, so the work stays in the turn. A bare threading.Thread or a ThreadPoolExecutor.submit does not, so spans there are unlabelled. Re-bind explicitly with neosigma.trace(turn_id=..., project=...) inside the worker if you need them attributed.

A turn that is never finished releases its project once it is unreachable, so later spans are unlabelled rather than mislabelled. One exception: an adapter stream abandoned mid-iteration (a bare for ... break rather than with) does not release its turn, and spans opened afterwards in that context can still carry it. Use with / async with around adapter streams, which is the documented pattern anyway.

An unlabelled span is not an error. It routes to the default project, which is recoverable in a way a wrong label is not.

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). Anything else is dropped and the event still ships. 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.

One limitation to know about. A distinct_id scoped to a trace() block or passed to turn() is unbound when that scope closes, but only in the context that opened it. Contextvars are isolated per task, so if you open a turn on one task and finish it from another while the opening task keeps running, that task holds the turn's actor until it ends. Nothing can reach into another context to clear it, and writing the value into the finishing task instead would corrupt whatever actor that task legitimately has. Finish a turn on the task that opened it, or use identify() if the actor really is sticky.

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

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

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

The PostHog wrap mirrors capture() only, since the current posthog package has no identify() to mirror. Mixpanel mirrors both, via wrap_mixpanel(Mixpanel(token)): track(...) 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() / turn() 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

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).

LangChain

Install the extra and pass the handler in LangChain's callbacks.

pip install "neosigma[langchain]"

import neosigma
from neosigma.integrations.langchain import neosigma_callback_handler

neosigma.init()
handler = neosigma_callback_handler()

with neosigma.turn(user_message="what is the weather in Paris?"):
    chain.invoke({"question": "what is the weather in Paris?"}, config={"callbacks": [handler]})

neosigma.shutdown()

Each LangChain run becomes a span. Model calls become chat spans carrying the model, prompt, completion, and token usage. Tool calls become execute_tool spans. Chains and runnables become structural spans that hold the nesting. Everything nests under the enclosing turn(), so one trace covers the whole request.

One handler can be shared across turns and baked into your model or chain. Reusing it across concurrent runs is safe, including the thread-parallel ones that RunnableParallel and .batch() produce.

ainvoke needs nothing extra. The same handler serves sync and async runs.

The handler is the only supported LangChain path

Do not also enable auto-instrumentation for a LangChain model that wraps a provider SDK we instrument, meaning langchain-anthropic over anthropic or langchain-openai over openai. Both would trace the same call, producing two chat spans and double the reported token usage. For that reason LangChain is deliberately absent from the auto-instrumentation registry, in this SDK and in the TypeScript one.

A model that does not wrap one of those SDKs is unaffected, since the handler is its only tracer either way.

Configuration

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

Variable	Default	Purpose
NEOSIGMA_API_KEY	(none)	Your ns_live_... key. Required to export, without it the SDK stays dark.
NEOSIGMA_PROJECT	default	Logical project name, attached to every trace.
NEOSIGMA_OTEL_ENDPOINT	NeoSigma cloud	OTLP/HTTP endpoint agent traces ship to. Override to target another environment.
NEOSIGMA_EVENTS_ENDPOINT	NeoSigma cloud	HTTP endpoint product events (capture()) ship to. Override alongside NEOSIGMA_OTEL_ENDPOINT when targeting another environment, otherwise traces move but events keep going to the default cloud.
NEOSIGMA_CONSOLE_EXPORT	false	Also print spans to stdout, for local debugging.
NEOSIGMA_PRIVATE_PROVIDER	true	Use a dedicated TracerProvider that is never registered as the OTel global (the default), so NeoSigma coexists with any OTel setup you already have. Set false to own the process-global provider and capture everything global-routed.

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

Dual export: send to NeoSigma and another backend

NeoSigma is built on OpenTelemetry, so you can send the same traces to NeoSigma and to another backend at once. One TracerProvider holds several span processors, and every span fans out to all of them.

By default neosigma.init() uses a private provider and does not touch your OTel global, so NeoSigma already coexists with another backend with no configuration.

To also send NeoSigma's agent traces to that other backend, build one TracerProvider that carries NeoSigma's processors and your other backend's exporter, and hand it to init(tracer_provider=...). All three backends below accept OpenTelemetry GenAI spans, which is what NeoSigma emits, so your traces render in both places with no translation.

LangSmith

import os
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
import neosigma
from neosigma import CorrelationSpanProcessor, NeoSigmaSpanProcessor

# One provider carrying NeoSigma's processors plus your other backend's exporter.
# CorrelationSpanProcessor goes first so it stamps turn/session ids before export.
provider = TracerProvider()
provider.add_span_processor(CorrelationSpanProcessor())
provider.add_span_processor(NeoSigmaSpanProcessor(api_key="ns_live_..."))
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://api.smith.langchain.com/otel/v1/traces",
    headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]},
)))

# NeoSigma emits through your provider and owns nothing.
neosigma.init(tracer_provider=provider)

Braintrust

Braintrust requires an x-bt-parent header naming the destination project. Add this processor to the same provider from the LangSmith example, before calling neosigma.init(tracer_provider=provider):

provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://api.braintrust.dev/otel/v1/traces",
    headers={
        "Authorization": f"Bearer {os.environ['BRAINTRUST_API_KEY']}",
        "x-bt-parent": f"project_id:{os.environ['BRAINTRUST_PROJECT_ID']}",
    },
)))

Langfuse

Langfuse uses HTTP Basic auth built from your public and secret keys. Add this processor to the same provider from the LangSmith example, before calling neosigma.init(tracer_provider=provider):

import base64

public_key = os.environ["LANGFUSE_PUBLIC_KEY"]
secret_key = os.environ["LANGFUSE_SECRET_KEY"]
auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()

provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    # EU region shown. US region: https://us.cloud.langfuse.com/api/public/otel/v1/traces
    endpoint="https://cloud.langfuse.com/api/public/otel/v1/traces",
    headers={
        "Authorization": f"Basic {auth}",
        "x-langfuse-ingestion-version": "4",
    },
)))

NeoSigma as the primary provider

Pass private_provider=False to opt out of the private default and let NeoSigma build and register the process-global provider instead. Then add the other backend's processor to that same global provider:

import os
import neosigma
from opentelemetry import trace
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

neosigma.init(api_key="ns_live_...", private_provider=False)  # NeoSigma owns the global provider
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://api.smith.langchain.com/otel/v1/traces",
    headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]},
)))

If a global provider is already registered when init() runs, private_provider=False falls back to a private provider instead of replacing it.

Bulk import / export

Move traces in bulk: pull historical traces in from another provider, or pull your own NeoSigma traces out. Both return a job handle you can poll with .wait().

import neosigma

neosigma.init(api_key="ns_live_...")

job = neosigma.import_traces(
    "langsmith",
    destination="my-neosigma-project",
    source_project="my-langsmith-project",
)
job.wait()
print(job.status, job.spans_done)

export = neosigma.export_traces(project="my-neosigma-project")
export.wait()
print(export.download_url)

import_traces(source, *, destination, source_project=None, since=None, until=None) starts a bulk import from source (an opaque string, for example "langsmith" or "braintrust"). destination is the NeoSigma project the imported traces land in and is required. source_project is the provider's own project to pull from, a separate thing from destination. export_traces(*, project=None, since=None, until=None) starts a bulk export of your own traces matching the given filters. Here project is your NeoSigma project to export from. Both accept since/until as datetime objects or ISO-8601 strings, and return immediately with a pending job. Call .wait(timeout=...) to block until the job reaches a terminal status (complete, failed, or cancelled, read from job.status), or .refresh() to poll once. get_import_job(job_id) / get_export_job(job_id) re-fetch a handle by id.

This call uses your NeoSigma API key (the same one init() reads), and the import source plus filters are validated server-side.

License

Released under the MIT License.