Metadata-Version: 2.4
Name: neosigma
Version: 0.18.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"
Provides-Extra: langchain
Requires-Dist: langchain-core>=1.0; extra == "langchain"
Provides-Extra: harbor
Requires-Dist: harbor>=0.22.0; extra == "harbor"
Dynamic: license-file

# neosigma

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.

Using JavaScript or TypeScript? See the [TypeScript SDK](https://www.npmjs.com/package/neosigma).

## 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

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

## Quickstart

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

Running a Harbor benchmark? See the [Harbor integration](examples/harbor/README.md)
for the optional plugin, Python API example, and a three-tool-call smoke task.

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:

```bash
export NEOSIGMA_API_KEY="ns_live_..."
```

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

```python
from neosigma import NeoSigma

client = NeoSigma()

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

agent = client.agents.create(
    workspace_id="YOUR_WORKSPACE_ID",
    name="Coding assistant",
    instructions="Make focused changes and run the relevant tests.",
    model={"provider": "anthropic", "id": "sonnet"},
)

# Agent listings must be scoped to one workspace or project.
agents = client.agents.list(workspace_id="YOUR_WORKSPACE_ID")

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

New agents use Pi when `harness` is omitted. Set `harness="claude"` for Claude
Code or `harness="codex"` for Codex. The selected model must support that harness.
`agents.update` requires an explicit harness and creates a new immutable version.
Existing sessions keep their pinned version.

Use the server catalogs to discover compatible models and execution limits:

```python
models = client.models.list(limit=25)
for model in models.items:
    print(model.id, model.provider, model.harnesses)

harnesses = client.harnesses.list(limit=25)
for harness in harnesses.items:
    print(harness.id, harness.supports_mcp_tool_filters)
```

Both methods return `items` and `next_cursor`. Pass `next_cursor` as `cursor` to
read the next page until it is `None`. The same methods are available on
`AsyncNeoSigma` with `await`. Choose a harness from the selected model's
`harnesses` list. Native harnesses reject per-server MCP tool filters before
execution; use Pi when those filters are required.

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.models.list` and `client.harnesses.list`
- `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`. Pass `vault_ids` to `sessions.create` when the session needs
  credentials from workspace vaults.

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

Agent skills are typed references. For a workspace skill, pass
`AgentCustomSkill(name="my-skill")`. You can pin a version with
`AgentCustomSkill(name="my-skill", version=3)`.

### 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:

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

Check for generated drift without writing files:

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

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

  With `NEOSIGMA_CAPTURE_CONTENT=false`, `set_attributes()` and the `attributes=` option
  of `neosigma.span()` write nothing, since the values are your data.

- **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

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.

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

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

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

### Session metadata

Pass `session_metadata` to `turn()` to attach details about the session, such as a plan
or a customer tier. Each entry is stamped on the turn's root span as
`neosigma.session.<key>`.

```python
with neosigma.turn(
    session_id=session_id,
    session_metadata={"plan": "pro", "seats": 3, "cohort": {"name": "beta"}},
):
    reply = run_agent(question)
```

- At most 50 keys. A key may have at most 128 characters, and a value at most 1,000
  characters once it is text. An entry over a limit is dropped, and the SDK warns the
  first time and then once per 100 drops.
- Strings, numbers, booleans, and lists of one of those (`None` entries allowed) stay
  typed. A mapping or a mixed list is stored as JSON text. A `None` value is skipped.
  Any other object, such as a `datetime`, `UUID`, `Decimal`, or `set`, is dropped, and
  counts toward the drop warning. `set_attributes` follows the same rules.
- Lengths count characters (code points), so an emoji counts as one.
- With `NEOSIGMA_CAPTURE_CONTENT=false`, no session metadata is recorded, plain strings
  and numbers included. The SDK drops it without a warning.
- Only the root turn reads `session_metadata`. A nested `turn()` ignores it and warns,
  the same as it does for a different `turn_id` or `session_id`.

### Uploading files

Call `upload_artifact()` inside the span that produced a local file or in-memory
bytes. The SDK records the managed artifact on that exact span right away, then
uploads the data to NeoSigma storage on background threads, up to four at a time.
It uses the active turn's project, or the project passed to `init()`.

```python
with neosigma.turn(project="acme-support", session_id=session_id):
    report_path = build_report()
    uploaded = neosigma.upload_artifact(
        report_path,
        name="Support report.csv",
    )
    print(uploaded.location)
```

In-memory bytes use the same operation and require a filename. A `bytearray` or
`memoryview` is copied when you call:

```python
uploaded = neosigma.upload_artifact(
    b"report contents",
    name="Support report.txt",
)
```

For paths, the display name defaults to the file name. Bytes require `name`. The
content type is inferred when omitted. The call makes no network request. It
raises `ArtifactUploadError` only for problems it can see locally, such as a
missing or empty file, bytes larger than `NEOSIGMA_UPLOADS_MAX_QUEUE_BYTES`, an
invalid name, or a span that already holds 100 artifacts.

The returned `UploadedArtifact` names the file at once, but the file may still be
uploading. NeoSigma links a file to its span when the span arrives, so the SDK
keeps a finished span in memory until that span's uploads finish. Your code never
waits for this. A span waits at most `NEOSIGMA_UPLOADS_FLUSH_TIMEOUT_SECONDS`;
after that it is sent once each of its files exists, and after twice that time it
is sent anyway, with a warning.

`neosigma.flush()` waits for queued uploads, then sends the spans held for them.
It sends held spans only on a provider the SDK built. On a provider you pass to
`init(tracer_provider=...)`, your own `provider.force_flush()` sends them, or each
is sent when its uploads end.

`neosigma.shutdown()` waits until each queued file exists, sends the spans, then
waits for the remaining uploads and stops the upload threads, all within
`NEOSIGMA_UPLOADS_FLUSH_TIMEOUT_SECONDS`. On a provider you pass to
`init(tracer_provider=...)`, it only waits for the uploads and stops the threads;
your provider sends its spans when you shut it down. At interpreter exit, the
`NeoSigmaSpanProcessor` shutdown that OpenTelemetry runs does the same.

Upload failures never raise into your code. The SDK makes up to three attempts
at each upload step when a failure is temporary, then logs a warning. A
permanent failure, such as a file over the server's size limit, is logged
without a retry. When the queue is full (1,000 uploads waiting to start), a new
upload is dropped with a warning, and its reference stays on the span. In-memory
data larger than `NEOSIGMA_UPLOADS_MAX_QUEUE_BYTES` can never fit, so
`upload_artifact()` raises `ArtifactUploadError` for it and records nothing.

NeoSigma refuses a file larger than the deployment's limit, 32 MiB by default,
when the background upload starts. The call has already returned by then, so the
SDK logs a warning, the span keeps its reference, and the file opens as not
available. Check file sizes before uploading if that matters.

For a path, the queue holds the path, not the bytes. Keep the file in place and
unchanged until `flush()` returns or the upload finishes. A file that changes
size or modification time first, or is deleted, is skipped with a warning, and
the span keeps a reference to a file that never uploads.

By default a file does not expire. Pass `retention_days` (an integer of at least 1) to
`upload_artifact()` or `upload_artifact_from_url()`, or set `NEOSIGMA_UPLOADS_RETENTION_DAYS`
for a default, and NeoSigma expires the file that many days after the call:

```python
neosigma.upload_artifact(report_path, name="Support report.csv", retention_days=30)
```

The expiry is fixed when the upload is queued, so a retry never moves it. A value
that is not an integer of at least 1 raises `ArtifactUploadError`. The SDK sends an
expiry only when you set one, and it needs a NeoSigma server that supports file
expiry.

To copy a file from a short-lived signed HTTPS URL, use
`upload_artifact_from_url()`. An upload thread streams the source directly into
NeoSigma storage, so the URL must stay valid until the upload runs. The source
URL is not sent to NeoSigma or recorded on the span.

```python
uploaded = neosigma.upload_artifact_from_url(
    signed_url,
    name="Support report.csv",
)
```

The source must return HTTP 200 with a non-zero `Content-Length` and an unencoded
body. Redirects are rejected. `name` must be a single filename of at most 256
characters.

### Inputs and outputs

A recorded file is an output by default: something the span produced. Outputs are
recorded on the `neosigma.artifacts` span attribute. Pass `direction="input"` for a
file the span read, such as an image a user sent. Inputs are recorded on
`neosigma.input_artifacts`, and inputs and outputs are capped separately per span.

```python
neosigma.artifact("s3://uploads/photo.png", "photo.png", direction="input")
```

`upload_artifact()` and `upload_artifact_from_url()` take the same keyword. Any value
other than `"input"` or `"output"` makes `artifact()` drop the reference with a
warning, and makes an upload raise `ArtifactUploadError`.

### Recording files well

- Wrap each tool with `@neosigma.tool()` so the files it touches are recorded on the
  tool's span.
- Pass `direction="input"` on the paths a tool reads. Without it a file is recorded as an
  output.
- Record a file the user attached on the turn itself, with `direction="input"`, so it
  sits next to the message that carried it.
- Upload a file once. Later spans that use it call `neosigma.artifact(uploaded.location,
  name)` instead of uploading it again.
- Give a signed URL at least 15 minutes of lifetime. `upload_artifact_from_url()` fetches
  it in the background, possibly after a queue wait and a retry.

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

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

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

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

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

```bash
pip install "neosigma[langchain]"
```

```python
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. |
| `NEOSIGMA_CAPTURE_CONTENT` | `true` | Record prompt, completion, and tool text, and your own attributes (`set_attributes()`, `attributes=` on `span()`, and `session_metadata`). Set `false` to omit all of them, plain strings and numbers included. Ids and names are still sent. |
| `NEOSIGMA_UPLOADS_MAX_QUEUE_BYTES` | `268435456` (256 MiB) | In-memory upload data held by unfinished uploads. A new upload past this is dropped with a warning. |
| `NEOSIGMA_UPLOADS_RETENTION_DAYS` | (none) | Days an uploaded file is kept before it expires, an integer of at least 1. Unset means no expiry. A per-call `retention_days` overrides it. |
| `NEOSIGMA_UPLOADS_FLUSH_TIMEOUT_SECONDS` | `30` | How long `flush()` and `shutdown()` wait for queued uploads, and how long a finished span waits for its uploads before it is sent. |

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
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.
# NeoSigmaSpanProcessor goes last: its force_flush() first waits for artifact
# uploads, and the provider gives each processor only the time left.
provider = TracerProvider()
provider.add_span_processor(CorrelationSpanProcessor())
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://api.smith.langchain.com/otel/v1/traces",
    headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]},
)))
provider.add_span_processor(NeoSigmaSpanProcessor(api_key="ns_live_..."))

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

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