agentradar-sdk — Python SDK

AI agent observability SDK — track runs, LLM executions, tool calls, errors, latency, and performance metrics in your AgentRadar dashboard.

Installation

pip install agentradar-sdk

Quick Start

from agentradar import AgentRadar

ar = AgentRadar(
    api_key="ar_live_your_key_here",
    endpoint="http://localhost:8000",  # your AgentRadar platform URL
)

with ar.start_run(
    agent_name="CustomerSupportAgent",
    input={"question": "Where is my order?"},
) as run:
    # 1. Log an LLM call
    run.log_llm_call(
        model_name="gpt-4o",
        provider="openai",
        duration_ms=850.0,
        token_usage={"prompt_tokens": 42, "completion_tokens": 15},
    )

    # 2. Track a tool call (with automatic duration measurement)
    with run.track_tool("search_orders", input={"order_id": "123"}) as tool:
        tool.set_output({"status": "shipped", "carrier": "FedEx"})

The run is automatically marked success on clean exit, or failed if an exception escapes the with block.

API Key Configuration

Get your API key from the AgentRadar dashboard under Settings → API Keys. Keys are prefixed with ar_live_.

# Option 1: Pass directly
ar = AgentRadar(api_key="ar_live_...")

# Option 2: Environment variable (AGENTRADAR_API_KEY or AR_API_KEY)
# export AGENTRADAR_API_KEY=ar_live_...
ar = AgentRadar()

Asynchronous Agents (AsyncAgentRadar)

For async agent frameworks (LangChain, AutoGen, CrewAI, FastAPI):

import asyncio
from agentradar import AsyncAgentRadar

async def main():
    async with AsyncAgentRadar(api_key="ar_live_...") as ar:
        async with ar.start_run("ResearchAgent", input={"topic": "Agent observability"}) as run:
            run.log_llm_call(model_name="claude-3-5-sonnet", provider="anthropic", duration_ms=450.0)
            with run.track_tool("web_search", input={"q": "Observability"}) as tool:
                tool.set_output({"count": 10})

asyncio.run(main())

Tracking Tool Calls & LLM Calls

Convenience Context Managers

Tool Calls (run.track_tool)

Automatically starts timer, logs tool_call, catches exceptions (logging error), and logs tool_result with duration:

with run.track_tool("database_query", input={"user_id": 42}) as tool:
    result = db.query(user_id=42)
    tool.set_output(result)

LLM Reasoning (run.track_llm)

with run.track_llm(model_name="gpt-4o", provider="openai", input={"prompt": "Summarize"}) as llm:
    completion = openai_client.chat.completions.create(...)
    llm.set_result(
        output={"text": completion.choices[0].message.content},
        token_usage={"prompt_tokens": 100, "completion_tokens": 50},
    )

Manual Logging Helpers

LLM Call

run.log_llm_call(
    model_name="claude-3-5-sonnet",
    provider="anthropic",
    input={"prompt": "..."},
    output={"response": "..."},
    duration_ms=1200.0,
    token_usage={"prompt_tokens": 100, "completion_tokens": 50},
)

Tool Call & Result

run.log_tool_call(name="web_search", input={"query": "latest AI news"})
run.log_tool_result(name="web_search", output={"results": [...]}, duration_ms=350.0)

Error Logging

run.log_error(
    error_type="ToolTimeoutError",
    error_message="Tool timed out after 30s",
    name="web_search",
)

Custom Events

run.log_custom_event(
    name="workflow_checkpoint",
    metadata={"step": 2, "state": "validating"},
)

Manual Run Lifecycle (Without Context Manager)

If you need manual control over run boundaries:

run = ar.start_run(agent_name="PipelineAgent", input={"batch_id": 99})
run.start()

# Do agent work...
run.log_tool_call(name="process_item")

run.finish(status="success")  # or status="failed"
ar.flush()

Single Event Ingestion & Health Check

# Check backend health
status = ar.check_health()
print(f"Backend status: {status.status}, version: {status.version}")

# Direct single-event ingestion
ar.send_event({
    "event_id": "...",
    "run_id": "...",
    "agent_name": "...",
    "event_type": "custom",
    "timestamp": "2026-10-16T12:00:00Z"
})

Configuration Options

Privacy & Payload Gating

By default, input and output payloads are not sent to protect sensitive user data and credentials.

# Enable payload capture (only if your data is safe to transmit)
ar = AgentRadar(api_key="ar_live_...", include_payloads=True)

Error Handling & Resilience

The SDK automatically retries on transient network errors and server errors (500, 502, 503, 504) with exponential backoff:

from agentradar.exceptions import AuthenticationError, APIError, NetworkError

try:
    ar.flush()
except AuthenticationError:
    print("Invalid or revoked API key")
except APIError as e:
    print(f"API error ({e.status_code}): {e}")
except NetworkError:
    print("Could not reach AgentRadar backend after retries")

Local Development & Testing

git clone https://github.com/GoodGuyAdy/AgentRadar
cd AgentRadar
pip install -e ".[dev]"
pytest

License

MIT — see LICENSE.