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 PIIed data.

1. Install the SDK

pip install atheon-codex

npm install @atheon-inc/codex

2. Initialize

Call init() once at application startup. For all subsequent calls, track(), begin(), and tools/agents, use the global client automatically.

For Django, Flask or other sync frameworks:

import os
import atheon

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

For FastAPI, Starlette or other async frameworks:

import os
from contextlib import asynccontextmanager
import atheon

from someasyncframework import Framework

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

app = Framework(lifespan=lifespan)

import * as atheon from "@atheon-inc/codex";

// This performs a cryptographic handshake with the Gateway
// and should be awaited before starting your server.
await atheon.init({ apiKey: process.env.ATHEON_API_KEY });

Advanced init options

You can fine-tune the background queue behavior by passing additional parameters:

  • uploadSize (Node) / upload_size (Python): Events per HTTP batch (default 10).
  • uploadInterval (Node) / upload_interval (Python): Seconds between background flushes (default 1.0).
  • maxQueueSize (Node) / max_queue_size (Python): Max in-memory queue depth (default 10,000).

Python Note: 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, prompt_hash, fingerprint = 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.finish_reason,
    properties={"agent": "support-bot"},
)

return {
    "reply": llm_response, 
    "interaction_id": str(interaction_id),
    "prompt_hash": prompt_hash,
    "fingerprint": fingerprint
}

interaction_id, prompt_hash, fingerprint = 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.finish_reason,
)

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

const [interactionId, promptHash, fingerprint] = atheon.track({
  provider: "openai",
  modelName: "gpt-4o",
  input: userQuery,
  output: llmResponse,
  tokensInput: response.usage.prompt_tokens,
  tokensOutput: response.usage.completion_tokens,
  finishReason: response.choices.finish_reason,
  properties: { agent: "support-bot" }
});

return { 
    reply: llmResponse, 
    interaction_id: interactionId,
    prompt_hash: promptHash,
    fingerprint: fingerprint
};

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

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

interaction = atheon.async_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")

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

const [interaction, ctx] = atheon.begin({
  provider: "anthropic",
  modelName: "claude-sonnet-4-5",
  input: userQuery,
  properties: { agent: "rag-pipeline" }
});

// ... do work, call tools inside contextWith() ...

interaction.setProperty("userTier", "pro");

const [interactionId, promptHash, fingerprint] = interaction.finish({
  output: finalText,
  tokensInput: 80,
  tokensOutput: 220,
  finishReason: "stop"
});

4. Track Tools

Wrap any function to record its execution time and errors automatically. The wrapper hooks into the nearest active begin() context.

import atheon

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

# Usage — hooks via Python's ContextVar
interaction = atheon.begin(provider="openai", model_name="gpt-4o", input=query)
results = search(query)
interaction.finish(output=results)

import atheon

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

# Usage — hooks via Python's ContextVar
interaction = atheon.async_begin(provider="openai", model_name="gpt-4o", input=query)
results = await search(query)
interaction.finish(output=results)

import * as atheon from "@atheon-inc/codex";

const search = atheon.tool("vector-search", async (query: string) => {
    return db.search(query);
});

// Usage — wrap the execution block in contextWith()
const [interaction, ctx] = atheon.begin({ provider: "openai", modelName: "gpt-4o", input: query });

const results = await atheon.contextWith(ctx, async () => {
    return await search(query);
});

interaction.finish({ output: results });

The tool wrapper is a no-op if called outside an active context, thus safe to use unconditionally in your shared logic.

5. Track Sub-Agents

Wrap 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.agent("rag-pipeline", provider="anthropic", model_name="claude-haiku-4-5")
def rag_agent(query: str) -> str:
    response = llm.messages.create(...)

    atheon.set_result(
        tokens_input=response.usage.input_tokens,
        tokens_output=response.usage.output_tokens,
        finish_reason=response.stop_reason,
    )
    return response.content.text

interaction = atheon.begin(provider="openai", model_name="gpt-4o", input=user_query)
result = rag_agent(user_query)
interaction.finish(output=result)

import atheon

@atheon.agent("rag-pipeline", provider="anthropic", model_name="claude-haiku-4-5")
async def rag_agent(query: str) -> str:
    response = await llm.messages.create(...)

    atheon.set_result(
        tokens_input=response.usage.input_tokens,
        tokens_output=response.usage.output_tokens,
        finish_reason=response.stop_reason,
    )
    return response.content.text

interaction = atheon.async_begin(provider="openai", model_name="gpt-4o", input=query)
results = await rag_agent(query)
interaction.finish(output=results)

import * as atheon from "@atheon-inc/codex";

const ragAgent = atheon.agent("rag-pipeline", "anthropic", "claude-haiku-4-5", async (query: string) => {
    const response = await llm.messages.create(...);

    atheon.setResult({
        tokensInput: response.usage.input_tokens,
        tokensOutput: response.usage.output_tokens,
        finishReason: response.stop_reason
    });

    return response.content.text;
});

const [interaction, ctx] = atheon.begin({ provider: "openai", modelName: "gpt-4o", input: userQuery });

const result = await atheon.contextWith(ctx, async () => {
    return await ragAgent(userQuery);
});

interaction.finish({ output: result });

6. Conversation Grouping

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

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

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

atheon.track({
    provider: "openai",
    modelName: "gpt-4o",
    input: turn1Query,
    output: turn1Response,
    conversationId: conversationId
});

7. Shutdown

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

atheon.shutdown()

await atheon.async_shutdown()

await atheon.shutdown();