Python SDK
The AgentLens Python SDK provides decorators, context managers, and helper functions to instrument your AI agents.
pip install vectry-agentlensAPI Reference
init()
agentlens.init(api_key, endpoint, *, flush_interval=5.0, max_batch_size=100, enabled=True)Initialize the AgentLens SDK. Must be called before any tracing functions. Typically called once at application startup.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| api_key | str | required | Your AgentLens API key |
| endpoint | str | required | AgentLens server URL |
| flush_interval | float | 5.0 | Seconds between automatic flushes |
| max_batch_size | int | 100 | Max traces per batch request |
| enabled | bool | True | Set to False to disable tracing globally |
import agentlens
agentlens.init(
api_key="al_key_abc123",
endpoint="https://agentlens.vectry.tech",
flush_interval=10.0,
max_batch_size=50,
)@trace
@agentlens.trace(name=None, tags=None, metadata=None)Decorator that wraps a function in a trace. The trace starts when the function is called and ends when it returns or raises. Works with both sync and async functions.
Parameters
| Parameter | Type | Description |
|---|---|---|
| name | str | None | Trace name. Defaults to the function name. |
| tags | list[str] | None | Tags to attach to the trace |
| metadata | dict | None | Arbitrary metadata dict |
from agentlens import trace
@trace(name="research-agent", tags=["research", "v2"])
async def research(topic: str) -> str:
result = await search(topic)
summary = await summarize(result)
return summary
# Can also be used without arguments
@trace
def simple_agent(prompt: str) -> str:
return call_llm(prompt)log_decision()
agentlens.log_decision(type, chosen, alternatives, *, reasoning=None, context_snapshot=None)Log a decision point within the current trace context. Must be called from within a @trace-decorated function.
Parameters
| Parameter | Type | Description |
|---|---|---|
| type | str | One of: TOOL_SELECTION, ROUTING, RETRY, ESCALATION, MEMORY_RETRIEVAL, PLANNING, CUSTOM |
| chosen | dict | What was chosen |
| alternatives | list[dict] | What else was considered |
| reasoning | str | None | Why this choice was made |
| context_snapshot | dict | None | Snapshot of context at decision time |
import agentlens
from agentlens import trace
@trace(name="routing-agent")
async def route_request(user_input: str):
intent = classify_intent(user_input)
agentlens.log_decision(
type="ROUTING",
chosen={"handler": "refund", "confidence": 0.92},
alternatives=[
{"handler": "faq", "confidence": 0.65},
{"handler": "escalate", "confidence": 0.23},
],
reasoning="High confidence refund intent detected",
context_snapshot={"intent": intent, "input_length": len(user_input)},
)
return await handle_refund(user_input)TraceContext
agentlens.TraceContextContext manager for manual trace lifecycle control. Use this when the @trace decorator does not fit your workflow.
import agentlens
async def process_batch(items: list[str]):
for item in items:
ctx = agentlens.TraceContext(
name=f"process-{item}",
tags=["batch"],
)
ctx.start()
try:
result = await process(item)
ctx.add_span(
name="process",
type="CUSTOM",
input={"item": item},
output={"result": result},
status="COMPLETED",
)
ctx.end(status="COMPLETED")
except Exception as e:
ctx.add_event(type="ERROR", name=str(e))
ctx.end(status="ERROR")shutdown()
agentlens.shutdown(timeout=10.0)Flush all pending traces and shut down the background sender. Call this before your application exits to avoid losing data.
import agentlens
import atexit
agentlens.init(api_key="...", endpoint="...")
# Register shutdown hook
atexit.register(agentlens.shutdown)
# Or call manually
agentlens.shutdown(timeout=30.0)Configuration
The SDK can also be configured via environment variables. These take precedence over values passed to init().
| Variable | Description |
|---|---|
| AGENTLENS_API_KEY | API key for authentication |
| AGENTLENS_ENDPOINT | Server URL |
| AGENTLENS_ENABLED | Set to "false" to disable tracing |
| AGENTLENS_FLUSH_INTERVAL | Flush interval in seconds |