atlanai

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

Install

pip install atlanai

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

pip install 'atlanai[tracing]'

Quickstart

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:

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.

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.

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.

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:

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

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 for framework setup, context manifests, async delivery, and the tested integration matrix.

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:

from atlanai import AtlanAPIError

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