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],
)

print(result.experiment_id)

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.

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)