Skip to content

Server-Side Integration

The Logic Layer is where Atheon's deep analysis takes place. By integrating on the server, you allow Atheon to generate Intent Fingerprints and power your Knowledge Graph securely — without ever storing raw PII.

1. Install the SDK

pip install atheon-codex

2. Initialise

Call atheon.init() (or atheon.async_init()) once at application startup. All subsequent calls — track(), begin(), decorators — use the global client automatically.

import os
import atheon

atheon.init(os.environ["ATHEON_API_KEY"])

import os
from contextlib import asynccontextmanager
import atheon
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    atheon.async_init(os.environ["ATHEON_API_KEY"])
    yield
    await atheon.async_shutdown()

app = FastAPI(lifespan=lifespan)

Advanced init options

atheon.init(
    api_key=os.environ["ATHEON_API_KEY"],
    upload_size=20,         # Events per HTTP batch (default 10)
    upload_interval=2.0,    # Seconds between background flushes (default 1.0)
    max_queue_size=50_000,  # Max in-memory queue depth (default 10 000)
)

Only one client type per process. Call either atheon.init() or atheon.async_init(), not both.

3. Track Interactions

Option A — track(): fire-and-forget for complete interactions

Use this when you have the full input and output ready at once.

interaction_id = atheon.track(
    provider="openai",
    model_name="gpt-4o",
    input=user_query,
    output=llm_response,
    tokens_input=response.usage.prompt_tokens,
    tokens_output=response.usage.completion_tokens,
    finish_reason=response.choices[0].finish_reason,
    properties={"agent": "support-bot"},  # optional
)

# Pass to frontend
return {"reply": llm_response, "interaction_id": str(interaction_id)}

# Call without await — enqueues immediately, non-blocking
interaction_id = atheon.async_track(
    provider="openai",
    model_name="gpt-4o",
    input=req.message,
    output=final_text,
    tokens_input=response.usage.prompt_tokens,
    tokens_output=response.usage.completion_tokens,
    finish_reason=response.choices[0].finish_reason,
)

return {"reply": final_text, "interaction_id": str(interaction_id)}

Option B — begin() / finish(): streaming and multi-turn flows

Use this when the interaction spans time — streaming responses, tool calls, or multi-step pipelines. Latency is measured automatically from begin() to finish().

interaction = atheon.begin(
    provider="anthropic",
    model_name="claude-sonnet-4-5",
    input=user_query,
    properties={"agent": "rag-pipeline", "environment": "production"},
)

# ... do work, call tools, etc. ...

interaction.set_property("user_tier", "pro")  # enrich mid-flight

interaction_id = interaction.finish(
    output=final_text,
    tokens_input=80,
    tokens_output=220,
    finish_reason="stop",
)

return {"reply": final_text, "interaction_id": str(interaction_id)}

interaction = atheon.async_begin(
    provider="anthropic",
    model_name="claude-sonnet-4-5",
    input=req.message,
    properties={"agent": "rag-pipeline"},
)

chunks = []
async for chunk in llm.stream(req.message):
    chunks.append(chunk)
    yield chunk

interaction_id = interaction.finish(
    output="".join(chunks),
    finish_reason="stop",
)

4. Track Tools with @atheon.tool

Decorate any function (sync or async) to record its execution time and errors automatically. The decorator hooks into the nearest active begin() context via Python's ContextVar — no explicit passing required.

import atheon

@atheon.tool("vector-search")
def search(query: str) -> list[str]:
    return db.search(query)

@atheon.tool("reranker")
async def rerank(docs: list[str]) -> list[str]:
    return await model.rerank(docs)

# Usage — tool records are attached to the active interaction automatically
interaction = atheon.begin(provider="openai", model_name="gpt-4o", input=query)
results = search(query)       # ToolRecord added silently
ranked  = await rerank(results)
interaction.finish(output=ranked[0], finish_reason="stop")

@atheon.tool is a no-op if called outside an active begin() context — safe to use unconditionally.

5. Track Sub-Agents with @atheon.agent

Decorate LLM-backed sub-agent functions to nest their tool calls and token usage inside the root interaction's payload. Everything is shipped in a single event on finish().

import atheon

@atheon.tool("vector-search")
def vector_search(query: str) -> list[str]: ...

@atheon.agent(
    "rag-pipeline",
    provider="anthropic",
    model_name="claude-haiku-4-5",
)
def rag_agent(query: str) -> str:
    chunks = vector_search(query)         # ToolRecord on rag-pipeline, not root
    response = llm.messages.create(...)

    atheon.set_result(                    # capture token counts for this sub-agent
        tokens_input=response.usage.input_tokens,
        tokens_output=response.usage.output_tokens,
        finish_reason=response.stop_reason,
    )
    return response.content[0].text


# Root interaction wires everything together
interaction = atheon.begin(
    provider="openai",
    model_name="gpt-4o",
    input=user_query,
)

result = rag_agent(user_query)   # AgentRecord (with nested ToolRecord) on root

interaction.finish(output=result, finish_reason="stop")

Async sub-agents work identically — just make the decorated function async def.

6. Conversation Grouping

Pass a conversation_id to group multiple turns together in the dashboard.

import uuid

conversation_id = str(uuid.uuid4())  # generate once per session

atheon.track(
    provider="openai",
    model_name="gpt-4o",
    input=turn_1_query,
    output=turn_1_response,
    conversation_id=conversation_id,
)

atheon.track(
    provider="openai",
    model_name="gpt-4o",
    input=turn_2_query,
    output=turn_2_response,
    conversation_id=conversation_id,
)

7. Shutdown

Call shutdown() (or async_shutdown()) when your process exits to flush any remaining queued events.

atheon.shutdown()

await atheon.async_shutdown()

You can also use the clients as context managers for automatic shutdown:

with atheon.AtheonCodexClient(api_key=...) as client:
    client.track(...)
# shutdown() called automatically on exit

Direct Client Instantiation

The module-level helpers (atheon.init, atheon.track, etc.) are the recommended approach. If you need multiple isolated clients in the same process, instantiate the class directly:

from atheon import AtheonCodexClient

client = AtheonCodexClient(api_key=os.environ["ATHEON_API_KEY"])
interaction_id = client.track(provider="openai", model_name="gpt-4o", input="...", output="...")
client.shutdown()

from atheon import AsyncAtheonCodexClient

async_client = AsyncAtheonCodexClient(api_key=os.environ["ATHEON_API_KEY"])
interaction_id = async_client.track(provider="openai", model_name="gpt-4o", input="...", output="...")
await async_client.shutdown()