acceldata-aio-tracer

LLM observability for Python, by Acceldata. Trace your agents, chains, and model calls and ship the data to your Acceldata AIO workspace with a single configuration call.

Decorate a function with @aio.track, or drop in a framework integration, and your traces flow to your AIO workspace automatically — no extra plumbing.

What you get

• One-line tracing of any function via the @aio.track decorator.
• Drop-in integrations for LangChain / LangGraph, OpenAI, Anthropic, Bedrock, LlamaIndex, CrewAI, DSPy, and more.
• Acceldata authentication (accessKey / secretKey) attached to every request automatically.
• A debug mode that prints a clean, one-line summary of every call sent to your gateway.

---

Table of contents

1. Installation
2. 60-second quickstart
3. Configuration
4. Emitting traces
5. Verification
6. Debugging
7. Troubleshooting
8. Configuration reference
9. Versioning and stability
10. Quick reference card
11. Acknowledgments

---

1. Installation

pip install acceldata-aio-tracer

Requires Python ≥ 3.10. Its dependencies are version-pinned — just install acceldata-aio-tracer and let it pull in what it needs; don't install or pin its tracing backend separately at a different version.

Before you start, have these ready:

• An accessKey / secretKey pair from your Acceldata platform admin.
• Your tenant's AIO URL. Each tenant has its own subdomain on acceldata.app, and the path is always /aio/api — for example https://demo.acceldata.app/aio/api (where demo is the tenant name).
• The project name your traces should land in.

---

2. 60-second quickstart

# main.py
import acceldata_aio_tracer as aio

aio.configure(
    url="https://demo.acceldata.app/aio/api",
    access_key="...",
    secret_key="...",
    project_name="my-agent",
)

@aio.track
def summarize(text: str) -> str:
    # call an LLM, return the summary
    return f"summary of: {text}"

if __name__ == "__main__":
    print(summarize("hello world"))
    aio.flush_tracker()   # flush pending traces before the process exits

python main.py

Then head to your AIO workspace — or run the verification check — to confirm the trace landed. That's the whole happy path; everything below explains the pieces in more detail.

---

3. Configuration

There is one configuration entry point. Call it once at process start, before any @aio.track-decorated function runs.

import acceldata_aio_tracer as aio

aio.configure(
    url="https://demo.acceldata.app/aio/api",   # required
    access_key="...",                           # required
    secret_key="...",                           # required
    project_name="my-agent",                    # required (non-empty)
    debug=False,                                # optional, see Debugging
    check_tls_certificate=True,                 # optional, see reference
)

About the URL. Replace demo with your tenant's actual subdomain — the same one you use to reach the AIO UI. The path is always /aio/api; /aio alone or /api alone will not work, and the placeholder demo.acceldata.app does not resolve, so leaving it in will fail loudly with a host-not-found error. Do not add a trailing slash.

Signature stability. The first three arguments — url, access_key, secret_key — are the SDK's stable public surface and will not be renamed. Every other option is keyword-only and may evolve across releases, so always pass those by name (debug=True, not True).

project_name is required. It is keyword-only with no default — omitting it, or passing None/empty, raises an error rather than silently routing traces to a generic default project.

configure() is required

configure() is the single entry point — call it once at process start, before any @aio.track-decorated function or framework integration runs. The SDK does not configure itself implicitly; without this call, traces have nowhere to go.

---

4. Emitting traces

4a. The @aio.track decorator

The decorator is the simplest path. It captures inputs from the function arguments and the output from the return value. Nested decorated calls become parent → child spans automatically, and it works on both sync and async functions.

import asyncio
import acceldata_aio_tracer as aio

@aio.track
def fetch(query: str) -> dict:
    return {"docs": [...]}

@aio.track
async def answer(query: str) -> str:
    docs = fetch(query)          # nested call becomes a child span
    # ... call your LLM ...
    return "..."

asyncio.run(answer("hello world"))
aio.flush_tracker()             # before the process exits

Useful keyword arguments:

@aio.track(
    name="generate-response",    # override the displayed name
    project_name="my-agent",     # route this call to a specific project
    tags=["production", "v2"],   # filterable tags
    type="llm",                  # "general" (default), "llm", "tool", "guardrail"
)
def generate(prompt: str) -> str:
    ...

4b. Grouping traces into a conversation (thread_id)

To link multiple traces into one conversation or workflow, tag each with the same thread_id. The AIO UI then groups them under a single thread view.

Call aio.update_current_trace(thread_id=...) from inside any @aio.track-decorated function — it attaches the thread to whichever trace is currently active.

import acceldata_aio_tracer as aio

@aio.track
def chat_turn(user_message: str, conversation_id: str) -> str:
    aio.update_current_trace(thread_id=conversation_id)
    # ... call your LLM, return the reply ...
    return "..."

# Multiple turns reuse the same conversation_id:
chat_turn("hello", conversation_id="chat-42")
chat_turn("how are you?", conversation_id="chat-42")
aio.flush_tracker()

The thread_id is any string you control — a chat session ID, a workflow run ID, a ticket number. The SDK does not validate or namespace it.

4c. LangChain / LangGraph

from langchain_openai import ChatOpenAI
from acceldata_aio_tracer.integrations.langchain import LangChainTracer

# Put the conversation's thread_id on the tracer — every trace it emits is
# tagged with it, so all turns group under one thread in the AIO UI.
tracer = LangChainTracer(thread_id="chat-42")

llm = ChatOpenAI(model="gpt-4o")
# Pass the tracer through the call's config. The same pattern works on
# .invoke() / .stream() / .astream() and on chains and agents.
llm.invoke("hello", config={"callbacks": [tracer]})

Construct a fresh tracer per conversation/request — it holds per-trace state, so reusing one across conversations would mix their traces.

For LangGraph, build the tracer first, then attach it to the compiled graph with track_langgraph(graph, tracer). This registers the tracer on the graph's default config, so subsequent invoke() calls are traced automatically — no need to pass callbacks= each time.

from acceldata_aio_tracer.integrations.langchain import (
    LangChainTracer,
    track_langgraph,
)

tracer = LangChainTracer()
agent = build_my_graph()                # langgraph.StateGraph(...).compile()
agent = track_langgraph(agent, tracer)  # returns the modified graph
agent.invoke({...})                     # traced automatically

4d. OpenAI

from openai import OpenAI
from acceldata_aio_tracer.integrations.openai import track_openai

client = track_openai(OpenAI())
client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "hi"}],
)

4e. Anthropic

import anthropic
from acceldata_aio_tracer.integrations.anthropic import track_anthropic

client = track_anthropic(anthropic.Anthropic())
client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "hi"}],
)

4f. Other frameworks

Every integration lives under acceldata_aio_tracer.integrations.<framework>:

Framework	Import path	Primary export	Style
LangChain / LangGraph	…integrations.langchain	LangChainTracer, track_langgraph	callback
OpenAI	…integrations.openai	track_openai	wrapper
Anthropic	…integrations.anthropic	track_anthropic	wrapper
AWS Bedrock	…integrations.bedrock	track_bedrock	wrapper
Google GenAI	…integrations.genai	track_genai	wrapper
Google ADK	…integrations.adk	ADKTracer, track_adk_agent_recursive	callback
LlamaIndex	…integrations.llama_index	LlamaIndexCallbackHandler	callback
DSPy	…integrations.dspy	DSPyCallback	callback
Haystack	…integrations.haystack	HaystackConnector	connector
CrewAI	…integrations.crewai	track_crewai	wrapper
LiteLLM	…integrations.litellm	track_completion	wrapper
AISuite	…integrations.aisuite	track_aisuite	wrapper
Guardrails AI	…integrations.guardrails	track_guardrails	wrapper
Harbor	…integrations.harbor	track_harbor	wrapper

Two patterns cover them all:

• track_<framework>(client) wraps a vendor client and returns a traced version. Used by most LLM-provider integrations.
• <Framework>Tracer / Callback / Handler / Connector is an object you pass into the framework's own callback/hook system.

---

5. Verification

After emitting a trace, two things should hold: the client flushed it without error, and the backend persisted it.

5a. Flush before exit

The client batches traces in a background thread. In short-lived scripts you must flush, or the process can exit before the batch is sent:

aio.flush_tracker()   # blocks until pending traces are sent

Long-running services (web servers, workers) flush continuously, but it is good practice to also flush on shutdown.

5b. List traces via the REST API

The list endpoint is GET /v1/private/traces. Call it with the same accessKey / secretKey headers the SDK uses:

URL="https://demo.acceldata.app/aio/api"
ACCESS_KEY="..."
SECRET_KEY="..."
PROJECT_NAME="my-agent"

curl -s "${URL}/v1/private/traces?project_name=${PROJECT_NAME}&size=5" \
  -H "accessKey: ${ACCESS_KEY}" \
  -H "secretKey: ${SECRET_KEY}" \
  | python3 -m json.tool

A "total" of 0 shortly after your script exits points to one of the troubleshooting causes below.

5c. Useful endpoints

All require accessKey + secretKey headers.

Method	Path	Purpose
GET	/v1/private/traces?project_name=<name>&size=N	List recent traces
GET	/v1/private/traces/{id}	Get a single trace by ID
GET	/v1/private/traces/threads?project_name=<name>	List conversation threads
GET	/v1/private/spans?trace_id=<uuid>	List spans for a trace
GET	/v1/private/projects?name=<name>	Find a project ID

5d. Smoke-test script

Drop this into CI to assert tracing works end to end:

# verify_tracing.py — exit 0 if a freshly emitted trace is visible
import os, sys, time, uuid, requests
import acceldata_aio_tracer as aio

URL = "https://demo.acceldata.app/aio/api"
ACCESS_KEY = os.environ["AIO_ACCESS_KEY"]   # your own env vars, passed into configure()
SECRET_KEY = os.environ["AIO_SECRET_KEY"]
PROJECT = "my-agent"

aio.configure(url=URL, access_key=ACCESS_KEY, secret_key=SECRET_KEY,
              project_name=PROJECT)

probe_name = f"smoke-{uuid.uuid4().hex[:8]}"

@aio.track(name=probe_name)
def probe():
    return "ok"

probe()
aio.flush_tracker()
time.sleep(2)   # backend write is async; small grace period

resp = requests.get(
    f"{URL}/v1/private/traces",
    params={"project_name": PROJECT, "size": 50},
    headers={"accessKey": ACCESS_KEY, "secretKey": SECRET_KEY},
    timeout=10,
)
resp.raise_for_status()
names = [t["name"] for t in resp.json().get("content", [])]
sys.exit(0 if probe_name in names else 1)

---

6. Debugging

Enable debug=True to log every HTTP call the SDK makes to your gateway:

aio.configure(url=..., access_key=..., secret_key=..., debug=True)

Each call is logged as a concise, one-line summary on the acceldata_aio_tracer.http_trace logger — request bodies are summarized as metadata (type, encoding, size, and decoded item counts), never dumped as raw bytes:

>>> POST /aio/api/v1/private/traces/batch  [json+gzip, 1.1 KB  {traces=2}]
<<< POST /aio/api/v1/private/traces/batch  [204, 0.36s]
>>> GET  /is-alive/ping  [no body]
<<< GET  /is-alive/ping  [200, 0.30s]  {"message":"ok"}

On any response with status ≥ 400, the SDK additionally dumps the request body to /tmp/acceldata-aio-tracer-last-failed-request.bin and logs a copy-paste curl that replays the exact request (credential headers redacted) so you can reproduce the failure against the gateway directly.

---

7. Troubleshooting

Traces emit without error but never appear

1. Did configure() run before the first traced call? The client is created lazily; configuring after the first @aio.track call means auth was never attached. Configure at process start.
2. Did the process exit before flushing? Add aio.flush_tracker() before exit.
3. Is project_name correct? A mismatched name silently creates a new project. Confirm with GET /v1/private/projects?name=....
4. Turn on debug=True and watch the http_trace output for the actual request/response and any replay curl.

401 Unauthorized

The accessKey / secretKey pair is wrong, expired, or scoped to a different tenant. Re-fetch from your platform admin and check for stray whitespace — the SDK does not trim keys.

403 or 406

The gateway is reachable but rejecting the auth headers, often because another Authorization header is being set elsewhere and conflicts. The SDK sets only accessKey and secretKey; don't add other auth headers.

URL mistakes

• The path is always /aio/api.
• demo.acceldata.app is a placeholder — use your tenant's real subdomain.
• No trailing slash on the URL passed to configure().

---

8. Configuration reference

configure() arguments

Argument	Required	Default	Notes
url	yes	—	Tenant gateway URL, e.g. https://demo.acceldata.app/aio/api
access_key	yes	—	Acceldata access key
secret_key	yes	—	Acceldata secret key
project_name	yes	—	Project traces are routed to (non-empty); omitting it, or None/empty, raises an error
debug	no	False	Log every HTTP call (see Debugging)
check_tls_certificate	no	True	Set False only for local self-signed dev gateways

---

9. Versioning and stability

• url, access_key, secret_key — stable positional surface, will not be renamed.
• project_name — required, keyword-only (no default).
• debug, check_tls_certificate — optional, keyword-only; may evolve across releases. Always pass keyword-only options by name.

---

10. Quick reference card

pip install acceldata-aio-tracer

import acceldata_aio_tracer as aio

aio.configure(
    url="https://demo.acceldata.app/aio/api",
    access_key="...",
    secret_key="...",
    project_name="my-agent",
)

@aio.track
def hi(name): return f"hello {name}"

hi("world")
aio.flush_tracker()

# verify
curl -s "https://demo.acceldata.app/aio/api/v1/private/traces?project_name=my-agent&size=5" \
  -H "accessKey: $ACCESS_KEY" \
  -H "secretKey: $SECRET_KEY" \
  | python3 -m json.tool

---

11. Acknowledgments

acceldata-aio-tracer is built on top of Opik by Comet ML — an open-source LLM observability toolkit licensed under Apache-2.0. Our thanks to the Opik maintainers and community for their work.

---

Built and maintained by Acceldata. Licensed under Apache-2.0.