Decorators

Zero-config instrumentation — add observability with a single line.

Decorators are the easiest way to add AgentLens tracking to your code. They automatically capture inputs, outputs, timing, and errors for both synchronous and asynchronous functions.

@track_agent

Tracks an agent function call, including execution time and the model used.

from agentlens import track_agent

# Simple usage (no arguments)
@track_agent
def my_agent(prompt: str) -> str:
    return call_llm(prompt)

# With parameters
@track_agent(model="gpt-4", name="research-agent")
def research(query: str) -> str:
    return do_research(query)

# Async functions work too
@track_agent(model="claude-3.5-sonnet")
async def async_agent(prompt: str) -> str:
    return await call_llm_async(prompt)

What Gets Captured

• Input: Function arguments (stringified)
• Output: Return value (stringified)
• Duration: Execution time in milliseconds
• Event type: "agent_call" on success, "agent_error" on exception
• Reasoning: Auto-generated (e.g., "Agent 'research' executed successfully in 234.5ms")

Error Tracking

If the decorated function raises an exception, the decorator records an agent_error event with the error type and message, then re-raises the exception. Your error handling is not affected.

@track_agent(model="gpt-4")
def risky_agent(prompt):
    if not prompt:
        raise ValueError("Empty prompt")
    return call_llm(prompt)

try:
    risky_agent("")  # Records agent_error event, then raises ValueError
except ValueError:
    print("Caught the error — and it's in AgentLens too")

@track_tool_call

Tracks a tool/function invocation with its inputs and outputs.

from agentlens import track_tool_call

# Simple usage
@track_tool_call
def search_web(query: str) -> list:
    return do_search(query)

# With custom tool name
@track_tool_call(tool_name="web_search")
def search(query: str) -> list:
    return do_search(query)

# Async
@track_tool_call(tool_name="database_query")
async def query_db(sql: str) -> list:
    return await db.execute(sql)

What Gets Captured

• Tool input: Function arguments (stringified)
• Tool output: Return value (stringified)
• Duration: Execution time in milliseconds
• Event type: "tool_call" on success, "tool_error" on exception

Combining Decorators

Use both decorators together to build fully-instrumented agents:

import agentlens
from agentlens import track_agent, track_tool_call

agentlens.init(endpoint="http://localhost:3000")

@track_tool_call(tool_name="web_search")
def search(query: str) -> list:
    """Searches the web. Automatically tracked."""
    return requests.get(f"https://api.search.com?q={query}").json()

@track_tool_call(tool_name="calculator")
def calculate(expression: str) -> float:
    """Evaluates math. Automatically tracked."""
    return eval(expression)

@track_agent(model="gpt-4")
def research_agent(question: str) -> str:
    """Main agent. All calls inside are tracked."""
    session = agentlens.start_session(agent_name="researcher")

    # These tool calls are automatically captured
    results = search(question)
    answer = synthesize(results)

    agentlens.end_session()
    return answer

How It Works Internally

1. The decorator wraps your function with timing logic
2. Before execution, it captures the function arguments
3. After execution, it captures the return value and elapsed time
4. It calls agentlens.track() to record the event
5. If init() hasn't been called, tracking is silently skipped (no errors)

Async Support

Both decorators automatically detect async functions and handle them correctly:

import asyncio

@track_agent(model="gpt-4")
async def async_agent(prompt: str) -> str:
    result = await call_llm_async(prompt)
    return result

@track_tool_call(tool_name="async_search")
async def async_search(query: str) -> list:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.search.com?q={query}")
        return resp.json()

# Works with asyncio.run, await, etc.
asyncio.run(async_agent("Hello world"))