Neatlogs@span Decorator
Instrument your own functions with neatlogs.span().
@neatlogs.span(kind="...") wraps a function with a span. Use it on any custom code that auto-instrumentation doesn't cover — your own agents, chains, tools, and pipelines.
Basic Usage
import neatlogs
@neatlogs.span(kind="WORKFLOW")
def handle_request(user_input: str) -> str:
...
@neatlogs.span(kind="AGENT")
def research_agent(state: dict) -> dict:
...
@neatlogs.span(kind="CHAIN")
def rag_pipeline(query: str) -> str:
...Works on both sync and async functions.
Parameters
| Parameter | Kind | Description |
|---|---|---|
kind | All | Required. One of: WORKFLOW, AGENT, CHAIN, TOOL, RETRIEVER, EMBEDDING, GUARDRAIL, MCP_TOOL |
name | All | Span name shown in the dashboard. Defaults to the function name. |
role | AGENT | The agent's role (e.g., "Researcher", "Router") |
goal | AGENT | The agent's goal or objective |
tool_name | TOOL, MCP_TOOL | Tool identifier shown in the dashboard |
description | TOOL, MCP_TOOL | Human-readable tool description |
model | EMBEDDING | Embedding model name |
dimension | EMBEDDING | Vector dimension |
version | All | Version string for tracking changes |
capture_input | All | Whether to record function arguments (default: True) |
capture_output | All | Whether to record the return value (default: True) |
Examples
Agent with role and goal
@neatlogs.span(kind="AGENT", name="routing_agent", role="Router", goal="Route to the right tool")
def route_request(query: str) -> dict:
...Tool
@neatlogs.span(kind="TOOL", name="check_order", tool_name="check_order_status")
def check_order_status(order_id: str) -> dict:
...MCP tool
@neatlogs.span(kind="MCP_TOOL", name="get_time", tool_name="get_time")
async def get_time() -> str:
...Retriever (auto-extracts query and documents)
When using @span(kind="RETRIEVER"), the decorator automatically extracts the query from the function's first string argument (looking for parameters named query, question, or text) and extracts the documents from the return value (if it's a list or dict):
@neatlogs.span(kind="RETRIEVER", name="vector_search")
def retrieve_docs(query: str) -> list:
# query is auto-captured as retrieval.query
# return value (list of docs) is auto-captured as retrieval.documents
return vector_db.search(query, top_k=5)For custom document formats or when you need more control, use with neatlogs.trace(kind="RETRIEVER") as span: and set attributes manually. See Custom Attributes.
Disable content capture
@neatlogs.span(kind="CHAIN", capture_input=False, capture_output=False)
def process_sensitive_data(payload: dict) -> dict:
...Content capture can also be disabled globally with the NEATLOGS_TRACE_CONTENT=false environment variable.