Python SDK

The Python SDK installs from PyPI as neatlogs. Use the latest published stable release. Wrap the AI clients you already use, add spans to your own functions, and track prompt templates — every call is captured as a trace in your dashboard. This page covers everything from install to the full init() reference; use the contents on the right to jump around.

Install

pip install -U neatlogs

To auto-instrument a specific library, install its extra (same -U rule applies):

pip install -U "neatlogs[openai]"      # OpenAI
pip install -U "neatlogs[anthropic]"   # Anthropic
pip install -U "neatlogs[langchain]"   # LangChain / LangGraph
pip install -U neatlogs                # CrewAI (instrument with neatlogs.wrap(crew))

Requires Python ≥ 3.10, < 3.14. See Supported Libraries for every key.

Check the installed SDK with Doctor

From the application root, run the network-free check before changing instrumentation:

python -m neatlogs doctor --local --json

After instrumenting and running your normal checks, set the project key through your environment or secret manager and run the authenticated probe separately:

python -m neatlogs doctor --probe --json

Use the active project interpreter; do not download or invoke a different Doctor binary. Require Doctor v2 format, Python runtime, and schema 2. If the installed package lacks that capability, check PyPI for the latest published stable release and ask before upgrading. Accept newer compatible releases and never downgrade one. A local pass validates controlled in-process capture only. A probe pass additionally proves exact finalized readback of its own four-span trace, but neither replaces testing your real workflow. See SDK Doctor for the full pass criteria, reason codes, and safe troubleshooting steps.

Your first trace

Call neatlogs.init() once, then wrap your AI client with neatlogs.wrap(...). From then on every call that client makes is recorded — you use the client exactly as before.

import os
import neatlogs

neatlogs.init(api_key=os.environ["NEATLOGS_API_KEY"], workflow_name="my-first-app")

from openai import OpenAI

client = neatlogs.wrap(OpenAI())

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
)
print(response.choices[0].message.content)

neatlogs.flush()
neatlogs.shutdown()

Run the script — a trace appears in your dashboard under my-first-app within a few seconds. The parentless LLM span is a valid trace root and shows the prompt, response, token counts, latency, and model name. No blank WORKFLOW wrapper is added. Standalone non-root operations such as TOOL receive one automatic workflow parent.

Extra keyword arguments to wrap() are stamped on the resulting trace root as neatlogs.workflow.<key> attributes, so they show in the metadata drawer and are searchable:

client = neatlogs.wrap(OpenAI(), project_id="p_123", route="/api/chat", surface="copilot")

Session and end-user identity are the exception — those go through neatlogs.identify(...), not wrap() kwargs. See Sessions.

flush() and shutdown()

The SDK batches spans and exports them in the background every few seconds. A short-lived script can exit before that fires, losing spans — so neatlogs.flush() forces an immediate export and neatlogs.shutdown() stops the background thread cleanly.

neatlogs.flush()
neatlogs.shutdown()

Always call both at the end of scripts and CLI tools. In long-running servers (FastAPI, Flask, …) the background thread exports continuously, so don't flush per request — call flush() + shutdown() once at server shutdown (for example, in a FastAPI lifespan handler). Reused serverless handlers should flush each invocation without shutting down a warm process. See Flush and Shutdown for workers, streams, tests, secondary clients, and bounded teardown.

Auto-instrumentation

Instead of wrap(), you can patch a library by name with instrumentations=[...] in init(). This works for providers that create clients internally.

neatlogs.init(
    api_key=os.environ["NEATLOGS_API_KEY"],
    workflow_name="my-app",
    instrumentations=["openai"],
)

For LangChain / LangGraph, use the callback handler — it's the only supported path, and works identically across Python and TypeScript. instrumentations=["langchain"] has no self-rooting (confirmed root cause of production trace loss) and is not offered as an alternative:

neatlogs.init(api_key=os.environ["NEATLOGS_API_KEY"], workflow_name="my-app")

from langchain_openai import ChatOpenAI

handler = neatlogs.langchain_handler()
llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke("Hello", config={"callbacks": [handler]})

See LangChain for the full picture, including the LangGraph-specific attachment rule (graph invocation, not per-node).

init() creates an SDK-private tracer provider and never installs it as the process-global provider. Call it once per active SDK generation; after shutdown() you may explicitly initialize a new generation.

Instrument your own code with @span

wrap() and auto-instrumentation only see library calls. To make your own functions (custom agents, pipelines, tools) appear as steps, put @neatlogs.span(...) on the line directly above the function.

import neatlogs

@neatlogs.span(kind="WORKFLOW")
def handle_request(user_input: str) -> str:
    return support_agent(user_input)

@neatlogs.span(kind="AGENT", name="support_agent")
def support_agent(message: str) -> str:
    ...

@neatlogs.span(kind="TOOL", tool_name="get_order_status")
def get_order_status(order_id: str) -> dict:
    return orders_db.get(order_id)

Works on sync and async functions. The decorator captures the arguments as input.value and the return value as output.value.

The WORKFLOW root

A WORKFLOW span is the entry point of a trace — the outermost function you call to process one request or task. Decorating it makes that function the trace's single, meaningfully-named root, and everything it calls (your own @span functions, wrapped provider calls, auto-instrumented libraries) nests underneath in the order it ran:

@neatlogs.span(kind="WORKFLOW")
def handle_customer_request(message: str) -> str:
    intent = classify_intent(message)
    if intent == "order_status":
        return check_order_agent(message)
    return general_support_agent(message)

If you don't add one, Neatlogs opens a WORKFLOW root automatically so a lone instrumented call still renders cleanly. Decorate your own entry point when you want the root to carry a specific name and group several steps under it.

Running several independent features in one process (a copilot, a summarizer, a background job)? Give each its own named WORKFLOW root at its entry point so they appear as distinct workflows in the dashboard — see Multiple Workflows in One Codebase.

A complete multi-span example

import os
import json
import neatlogs

neatlogs.init(
    api_key=os.environ["NEATLOGS_API_KEY"],
    workflow_name="support-bot",
)

from openai import OpenAI

client = neatlogs.wrap(OpenAI())

@neatlogs.span(kind="TOOL", tool_name="get_order_status")
def get_order_status(order_id: str) -> dict:
    return {"order_id": order_id, "status": "shipped", "eta": "2025-01-20"}

@neatlogs.span(kind="AGENT", name="support_agent")
def support_agent(message: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a support agent."},
            {"role": "user", "content": message},
        ],
        tools=[{
            "type": "function",
            "function": {
                "name": "get_order_status",
                "description": "Get the status of an order",
                "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}}},
            }
        }],
    )
    msg = response.choices[0].message
    if msg.tool_calls:
        args = json.loads(msg.tool_calls[0].function.arguments)
        return str(get_order_status(**args))
    return msg.content

@neatlogs.span(kind="WORKFLOW")
def handle_request(user_input: str) -> str:
    return support_agent(user_input)

handle_request("Where is my order #12345?")

neatlogs.flush()
neatlogs.shutdown()

The nesting reflects the actual call hierarchy at runtime — each decorated function appears exactly where it ran, with its inputs, outputs, and timing:

WORKFLOW  handle_request           0.8s
  AGENT   support_agent            0.8s
    LLM   gpt-4o                   0.6s
    TOOL  get_order_status         0.0s

Decorator ordering

When you stack @neatlogs.span with a framework decorator that transforms the function (@function_tool, @tool, @task), put @neatlogs.span below it — closest to def — so it wraps the real function:

# CORRECT — @function_tool wraps the span-decorated function
@function_tool
@neatlogs.span(kind="TOOL", tool_name="search")
def search(query: str) -> str:
    ...

Decorators that preserve the callable (@app.get, @app.post) can sit either side.

@span parameters

To omit content for one span, pass capture_input=False and/or capture_output=False. To redact or transform captured telemetry for the whole client, use mask; there is no global trace-content switch. See PII Redaction.

Span kinds

@neatlogs.span() accepts these eight kinds:

Passing any other kind raises ValueError. The kinds RERANKER and VECTOR_STORE are created with trace() (below); LLM comes from wrap() / auto-instrumentation. For the complete catalogue of every kind, see Span Kinds.

Inline spans with trace()

with neatlogs.trace(...) opens a span for a block inside a function — use it for prompt-template tracking (below), for custom span kinds (RERANKER, VECTOR_STORE), or to set attributes yourself on custom retrieval/guardrail logic that Neatlogs doesn't already recognize.

with neatlogs.trace(
    name,
    kind=None,
    system_prompt_template=None,
    user_prompt_template=None,
    version=None,
    mask=None,
) as span:
    ...

Supported libraries (vector DBs, framework retrievers) are captured automatically — you only set attributes for custom implementations:

import json
import neatlogs

@neatlogs.span(kind="CHAIN")
def rag_pipeline(query: str) -> str:
    with neatlogs.trace("retrieve", kind="RETRIEVER") as span:
        span.set_attribute("neatlogs.retriever.query", query)
        docs = my_custom_search(query, k=5)   # not a supported library
        span.set_attribute("neatlogs.retriever.documents", json.dumps(docs))
    return generate(query, docs)

The neatlogs.* attribute names each custom kind expects (RETRIEVER, RERANKER, VECTOR_STORE, GUARDRAIL, EMBEDDING, TOOL) are listed in Custom Attributes.

Prompt templates

Capture both the template structure and the runtime variable values for an LLM call, linked to its span. SystemPromptTemplate is the system/instruction prompt; UserPromptTemplate is the user turn. Pass them to trace(kind="LLM", ...) around the call:

import neatlogs
from neatlogs import SystemPromptTemplate, UserPromptTemplate

system_template = SystemPromptTemplate([
    {"role": "system", "content": "You are a {{role}} assistant."},
])
user_template = UserPromptTemplate([
    {"role": "user", "content": "{{question}}"},
])

@neatlogs.span(kind="AGENT")
def answer_agent(question: str) -> str:
    with neatlogs.trace("answer", kind="LLM",
                        system_prompt_template=system_template,
                        user_prompt_template=user_template):
        msgs = system_template.compile(role="support") + user_template.compile(question=question)
        return client.chat.completions.create(model="gpt-4o", messages=msgs).choices[0].message.content

For prompts managed centrally in the dashboard, the module-level functions (neatlogs.get_prompt, create_prompt, update_prompt, …) are available after init(). See Prompt Templates for managed prompts and the full API.

Log capture

Log capture is opt-in — pass capture_logs=True to init(). Neatlogs then records logs as LOG spans (children of the active span) so they appear inline in the trace timeline. Without it, none of the three mechanisms below capture anything.

neatlogs.init(api_key="...", capture_logs=True)

@neatlogs.span(kind="CHAIN")
def rag_pipeline(query: str) -> str:
    docs = retrieve(query)
    neatlogs.log("retrieved {count} docs", count=len(docs))
    return generate(query, docs)

1. neatlogs.log() — structured template messages

The primary way to emit a named, structured step. The message template becomes the span name (low-cardinality, searchable) and each keyword argument is stored as a log.{key} attribute.

neatlogs.log(msg_template, /, level="info", **data)

Templates use Python's str.format_map() syntax — single braces {count}, not {{count}}. neatlogs.log() must be called inside an active span (@neatlogs.span or with neatlogs.trace()); calls outside a traced block are dropped.

neatlogs.log("retrieved {count} docs, top score {score:.2f}",
             count=len(docs), score=docs[0]["score"])

When init(debug=True), every neatlogs.log() call is also echoed to stderr immediately, so you can watch steps in the terminal without opening the dashboard:

[neatlogs] 12:34:56  LOG  retrieved 42 docs, top score 0.94  count=42  score=0.94

2. stdlib logging.*() — auto-captured

Any logging.debug() / info() / warning() / error() call inside a traced block is captured automatically — no code changes. All levels are captured by default; raise the floor with log_level="WARNING" in init() to capture only WARNING and above.

3. capture_stdout=True — print capture

Pass capture_stdout=True to @neatlogs.span or with neatlogs.trace() to capture every print() line inside that block as a LOG span. Output is still mirrored to the real stdout — capture does not suppress terminal output.

@neatlogs.span(kind="AGENT", capture_stdout=True)
def run_agent(query: str) -> str:
    print(f"Processing query: {query}")   # captured as a LOG span
    return agent.invoke(query)

What you'll see in the dashboard

Each log call produces a LOG span as a child of the active span, with these fields:

Framework & provider helpers

Beyond instrumentations=[...], a few helpers attach tracing to a specific object:

• neatlogs.wrap(client) — the main path; wraps an LLM client or agent (OpenAI, Anthropic, Bedrock, Vertex, OpenRouter, CrewAI, Pydantic AI, DSPy, Agno, Google ADK, Hermes).
• neatlogs.langchain_handler() — a LangChain callback handler to pass in config={"callbacks": [...]}.
• neatlogs.openai_agents_processor() — a trace processor for the OpenAI Agents SDK.
• neatlogs.strands_hooks(agent) — registers hooks on a Strands agent.
• neatlogs.bind_templates(...) / register_crewai_task(...) — attach prompt templates to CrewAI work.

Each provider and framework has a runnable example in Integrations.

Cross-process propagation

To keep one logical trace across a service boundary, carry the active span as W3C traceparent/tracestate headers: the caller injects, the callee extracts. Both helpers use Neatlogs' private propagator — they never read or replace the global OTel propagator, so this stays isolated from a co-tenant tracer.

import neatlogs

# Caller — inject before sending. Returns False (carrier untouched) if no span is active.
with neatlogs.trace("caller"):
    headers = {"content-type": "application/json"}
    if neatlogs.inject_trace_context(headers):
        requests.post(url, headers=headers, json=payload)

# Callee — extract to continue the trace. Re-bind identity here (it does NOT ride the wire).
with neatlogs.extract_trace_context(
    req.headers,
    session_id=sid,
    parent_session_id=parent_sid,
    session_custom_fields={"feature_name": feature, "entry_point": "api"},
    end_user_id=uid,
):
    with neatlogs.trace("do_work"):   # joins the caller's trace as a child
        ...

Use these instead of a bare opentelemetry.propagate.inject(...), which reads the global propagator and writes nothing under isolation. For the full cross-language walkthrough — including a Go or TypeScript peer — see Distributed Tracing.

Supported libraries

Pass these to instrumentations=[] (install the matching extra first).

LLM providers: openai, anthropic, google_genai, azure_ai_inference, bedrock, litellm, cohere, groq, mistralai, together, ollama, replicate, openrouter

Agent frameworks: crewai, openai_agents, pydantic_ai, dspy, agno, google_adk, strands, autogen, haystack, smolagents, hermes

Vector databases: chromadb, pinecone, qdrant, weaviate, milvus, elasticsearch

Other: mcp, instructor, guardrails

The install commands, minimum package versions, and important compatibility limits are maintained in the tested support table. Documentation checks verify that every advertised integration names its actual activation mechanism; a registry key alone is not treated as proof that a provider works.

init() reference

flush_all() drains the default pipeline and every live secondary Neatlogs client without touching foreign/global providers. For lifecycle guidance see Flush and Shutdown. For PII redaction details see PII Redaction; to group multi-turn conversations see Sessions; to attach your app's users to traces see End-User Identity.

Building in Node.js? See the TypeScript SDK.