Metadata-Version: 2.4
Name: atlanai
Version: 0.5.2
Summary: Unified Atlan Agent Gateway management and tracing SDK
License-Expression: Apache-2.0
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: pydantic<3,>=2.10
Requires-Dist: python-dateutil<3,>=2.8.2
Requires-Dist: typing-extensions<5,>=4.12
Requires-Dist: urllib3<3,>=2.3
Provides-Extra: tracing
Requires-Dist: opentelemetry-api<2,>=1.39.0; extra == "tracing"
Requires-Dist: opentelemetry-sdk<2,>=1.39.0; extra == "tracing"
Requires-Dist: opentelemetry-exporter-otlp-proto-http<2,>=1.39.0; extra == "tracing"
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3; extra == "langchain"

# atlanai

Python SDK for the Atlan Agent Gateway: manage agents, skills, sessions, and
workspaces, and optionally trace what your agents do.

## Install

```bash
pip install atlanai
```

Tracing is a separate extra — it pulls in OpenTelemetry, which a management-only
install doesn't need:

```bash
pip install 'atlanai[tracing]'
```

## Quickstart

```python
from atlanai import AtlanClient

client = AtlanClient(
    "https://<your-gateway-host>",
    bearer_token="<your-api-token>",
)

agent = client.agents.create({
    "name": "support-triage",
    "workspace_id": "workspace_01example",
})

page = client.agents.list(limit=10)
print(f"{len(page.items)} agents")
```

Methods read as `client.<resource>.<action>` — `client.agents.get(agent_id)`,
`client.sessions.messages.create(session_id, {...})`, and so on. Every public
Agent Gateway operation is reachable this way; nothing requires reaching into
a generated client directly.

Pass a default workspace once instead of repeating it on every call:

```python
client = AtlanClient(gateway_url, bearer_token=token, workspace="workspace_01example")
```

## Run an eval

`Eval()` executes the task and scorers, creates the Registry experiment, emits
one trace per case, uploads trace-linked results, and finalizes the run. Registry
derives the durable score summary during that finalization:

Install `atlanai[tracing]`, not the management-only package, before using this
API because every result is required to have an exported trace.

```python
from atlanai import Eval


def accuracy(*, output: str, expected: str, **_):
    return float(output == expected)


result = Eval(
    "Anthropic Evaluation",
    data=lambda: [
        {"input": "What is 2+2?", "expected": "4"},
        {"input": "What is the capital of France?", "expected": "Paris"},
    ],
    task=call_model,
    scores=[accuracy],
    subject_kind="agent",
    subject_id="agent_01example",
)

print(result.experiment_id)
```

`subject_kind` and `subject_id` link the run to the agent it tested, which is
how the Evals UI finds it; the agent ID (`agent_...`) is the `id` that
`client.agents.list()` returns. A run with neither is still saved, but lists
under no agent, and `Eval()` warns when it creates one.

Set `ATLAN_API_KEY` and `ATLAN_WORKSPACE_ID`. Use `await async_eval(...)` in an
async application. Replace inline `data` with `dataset="dataset_..."` or one
exact Registry dataset name. After flushing, the runner reads every case trace
back through the experiment filter before it uploads results; a missing trace
marks the experiment failed. A run that produced no usable score at all is
finalized and then refused; pass `require_scores=False` if an empty score set
is expected.

### Scorer return values

A scorer returns one of these. The scorer's own name is the metric name unless
the value carries one.

```python
from atlanai import ScoreValue

lambda **_: 0.8                                         # a number (a bool counts as 0 or 1)
lambda **_: True                                        # a bool records 1 or 0
lambda **_: None                                        # abstain: recorded, but not averaged
lambda **_: ScoreValue(score=1, name="exact", metadata={"why": "match"})
lambda **_: {"score": 1, "name": "exact", "metadata": {}}   # a mapping with a "score" key
lambda **_: {"exact": 1, "length": 0.5}                 # a {name: score} mapping
lambda **_: [ScoreValue(score=1, name="a"), {"score": 0, "name": "b"}]  # a list of ScoreValue objects or mappings with a "score" key
```

A list holds ScoreValue objects or mappings with a `"score"` key, not bare
numbers.

By default a scorer that raises marks its whole case as errored. Pass
`scorer_errors="score"` to keep the failure to that scorer: its metric is
recorded as `None`.
The message is stored on the case result (and under
`source_metadata.atlan_eval.scorer_errors` on the persisted row), and the case
stays successful unless no scorer produced a number. Only an exception raised
by the scorer counts: an unsupported return value, a duplicate or colliding
score name, or a failed registration still errors the case. A scorer that fails
on every case produces no numeric value for that metric, so check
`summary.metrics.scorer_errors`. The CI Action does not yet gate on it.

### Run part of a dataset

`record_ids`, `where` and `limit` run a subset of a Registry dataset. The run
stays linked to the dataset, so each result keeps its `dataset_record_id` and
the run can be compared with a full run of the same dataset on the records both
ran. They apply in that order to the experiment's frozen snapshot, and a resume
must pass the same selectors. `where` must be a synchronous function that
returns a bool.

```python
Eval(
    "Support smoke",
    dataset="support-daily-tasks",
    task=task,
    scores=scores,
    where=lambda case: "billing" in (case.tags or ()),
    limit=10,
)
```

## Control an eval lifecycle directly

Resolve an existing dataset by artifact ID or exact name, then create the
Registry experiment before another runner emits any traces:

```python
from atlanai import ContextItem, ContextManifest, start_experiment

context = ContextManifest([
    ContextItem(
        kind="file",
        name="CLAUDE.md",
        version="git:0123456789abcdef0123456789abcdef01234567",
        digest="sha256:" + "0" * 64,
    ),
])

run = start_experiment(
    client,
    "conversational-studio-daily",  # exact name, or dataset_... ID
    {"name": "candidate-run", "config": {"model": "example-model"}},
    context_manifest=context,
)

with run.trace():
    output = existing_runner()
```

`run.experiment` is the generated create response and `run.id` is its
`experiment_id`. The helper pins the resolved dataset version and context
manifest in the immutable experiment config. `run.trace()` stamps the
experiment join and context-manifest digest on spans. Use it when another
harness owns execution, result upload, and finalization.

## Tracing

```python
from atlanai.tracing import auto_instrument, current_span, init_logger, traced

auto_instrument()
logger = init_logger(project="support-agent")


@traced(name="handle-request", type="task")
def handle_request(input: str):
    output = call_model(input)
    current_span().log(input=input, output=output)
    return output


handle_request("hello")
logger.flush()
```

`auto_instrument()` discovers installed AI OpenTelemetry instrumentors. Call it
before importing provider clients. Tracing has its own lifecycle and does not
reuse the management client's transport.

See [evals and tracing](../docs/evals-and-tracing.md) for framework setup,
context manifests, async delivery, and the tested integration matrix.

### LiteLLM deployment id

A LiteLLM proxy names the deployment that served a call in the
`x-litellm-model-id` response header. After `init()` or `init_logger()`, the SDK
records that value as `atlan.model.ref` on the call's LLM span, and
`x-litellm-call-id` as `atlan.model.call_id`, so Atlan can price the call by
deployment. No client code changes.

It is on by default and hooks the `send` method of every `httpx` and `httpx2`
client in the process, including clients built before `init()`. Only 2xx
responses are read, response bodies never are, and a value the span already
carries is kept. The value needs an LLM span to land on: an `llm` span of your
own, a provider instrumentor's span, or the LangChain `CallbackHandler` span.

Turn it off with `init(..., capture_model_ref=False)` or
`ATLAN_CAPTURE_MODEL_REF=false`. Off means the hook is not installed and the
LangChain `CallbackHandler` does not record the headers; it does not remove a
hook that an earlier `init()` in the same process installed.

With LangChain async calls (`ainvoke`, async chains), the HTTP hook cannot find
the handler's span, so the handler reads the header from the response instead.
Ask the chat model to keep response headers:

```python
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4.1-mini", base_url=litellm_url, include_response_headers=True)
```

For a call the hook cannot see, record the value yourself.
`record_model_ref(model_ref, *, call_id=None, span=None)` writes to the current
LLM span unless you pass one, returns `True` when it wrote anything, and never
raises:

```python
import requests
from atlanai.tracing import record_model_ref, traced


@traced(name="chat", type="llm")
def chat(payload: dict):
    response = requests.post(f"{litellm_url}/chat/completions", json=payload, timeout=60)
    if response.ok:
        record_model_ref(
            response.headers.get("x-litellm-model-id"),
            call_id=response.headers.get("x-litellm-call-id"),
        )
    return response.json()
```

Known limitations:

- Clients that subclass `httpx.Client` and override `send`, or that use another
  transport (`requests`, `aiohttp`), are not covered. Use `record_model_ref()`.
- Async LangChain without `include_response_headers=True` is not attributed.
- A task that keeps a stream open and starts a second call records nothing for
  the second call.
- A call with no LLM span at all (no app span, handler, or instrumentor) has
  nowhere to record.

## Errors

Every non-2xx response raises `AtlanAPIError`, with `.status`, `.code`, and
(where the gateway includes one) `.trace_id`. A failure before any response is
also normalized as `AtlanAPIError` with `status == 0` and an actionable message:

```python
from atlanai import AtlanAPIError

try:
    client.agents.get("agent_does_not_exist")
except AtlanAPIError as error:
    print(error.status, error.code)
```
