Metadata-Version: 2.4
Name: neosigma-sdk
Version: 0.4.0
Summary: NeoSigma tracing SDK: wrap your agents and emit OpenTelemetry spans to NeoSigma
Author: NeoSigma Team
License: MIT
Project-URL: Homepage, https://neosigma.ai
Project-URL: Documentation, https://docs.neosigma.ai
Keywords: opentelemetry,observability,tracing,llm,agents,anthropic,gen-ai
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: opentelemetry-api>=1.27.0
Requires-Dist: opentelemetry-sdk>=1.27.0
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.27.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: pydantic-settings>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-mock>=3.12; extra == "dev"
Provides-Extra: instrumentation
Requires-Dist: opentelemetry-instrumentation-anthropic>=0.61.0; extra == "instrumentation"
Requires-Dist: opentelemetry-instrumentation-openai>=0.61.0; extra == "instrumentation"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == "fastapi"
Dynamic: license-file

# neosigma-sdk

Trace your AI agents and ship the results to [NeoSigma](https://neosigma.ai). 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

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

## Quickstart

```python
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.

  ```python
  @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.

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

  ```python
  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](https://docs.neosigma.ai) 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:

```python
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 `neosigma.turn(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.

```python
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):

```python
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 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

```python
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(...)`:

| 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](https://docs.neosigma.ai) 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

```python
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_sdk as neosigma
from neosigma_sdk 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)`:

```python
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)`:

```python
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:

```python
import os
import neosigma_sdk as 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.

### Migrating to 0.4.0

0.4.0 makes the private provider the default, so `neosigma.init()` no longer
registers or touches your OTel global. The `attach_to_existing_provider` flag is
removed.

To keep the previous behavior where NeoSigma owns the global provider, pass
`private_provider=False`. For dual export, build one provider with NeoSigma's
processors and your other backend's exporter and pass it to
`init(tracer_provider=...)`, shown above.

## 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()`.

```python
import neosigma_sdk as neosigma

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

job = neosigma.import_traces("langsmith", 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, *, project=None, since=None, until=None)` starts a bulk
import from `source` (an opaque string, for example `"langsmith"` or
`"braintrust"`).
`export_traces(*, project=None, since=None, until=None)` starts a bulk export of
your own traces matching the given filters. 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, 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.
