Python SDK

from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

Netra.init(
    app_name="my-ai-app",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
    environment="production",
    trace_content=True,
    instruments={InstrumentSet.OPENAI, InstrumentSet.LANGCHAIN},
)

from netra import Netra, SpanType

# Basic usage
with Netra.start_span("process-document") as span:
    result = process_document(doc)
    span.set_attribute("pages", result.page_count)

# With span type and attributes
with Netra.start_span(
    "generate-response",
    as_type=SpanType.GENERATION,
    attributes={"model": "gpt-4"},
    module_name="chat"
) as span:
    response = generate_response(prompt)

Netra.set_session_id("session-abc123")

Netra.set_user_id("user-456")

Netra.set_tenant_id("tenant-org-789")

Netra.set_custom_attributes(key="customer_tier", value="premium")
Netra.set_custom_attributes(key="region", value="us-east")

Netra.set_custom_event(
    event_name="user_feedback",
    attributes={
        "rating": 5,
        "comment": "Great response!",
        "category": "positive"
    }
)

from netra import Netra
from netra.models import ConversationType

Netra.add_conversation(
    conversation_type=ConversationType.INPUT,
    role="user",
    value="What is the weather today?"
)

Netra.add_conversation(
    conversation_type=ConversationType.OUTPUT,
    role="assistant",
    value="The weather is sunny with a high of 72°F."
)

Netra.shutdown()

span.set_attribute("query", user_query)
span.set_attribute("results.count", len(results))

span.add_event("validation-started")
span.add_event("validation-completed", {"valid": True, "errors": 0})

span.set_prompt("Summarize this article: ...")

span.set_negative_prompt("blurry, low quality, distorted")

span.set_model("gpt-4-turbo")

span.set_llm_system("openai")

from netra import UsageModel

span.set_usage([
    UsageModel(
        model="gpt-4",
        cost_in_usd=0.006,
        usage_type="chat",
        units_used=1
    )
])

from netra import ActionModel

span.set_action([
    ActionModel(
        action="DB",
        action_type="INSERT",
        affected_records=[
            {"record_id": "123", "record_type": "user"}
        ],
        metadata={"table": "users"},
        success=True
    )
])

span.set_success()

span.set_error("Failed to connect to database")

from netra import SpanType

SpanType.SPAN        # General operations (default)
SpanType.GENERATION  # LLM completions, image generation
SpanType.EMBEDDING   # Vector embedding operations
SpanType.TOOL        # Function calls, API requests
SpanType.AGENT       # AI agent operations

from netra import UsageModel

usage = UsageModel(
    model="gpt-4",
    cost_in_usd=0.0045,
    usage_type="chat",
    units_used=1
)

from netra import ActionModel

action = ActionModel(
    action="API",
    action_type="CALL",
    affected_records=[
        {"record_id": "order-123", "record_type": "order"}
    ],
    metadata={
        "endpoint": "/api/orders",
        "method": "POST",
        "status_code": "201"
    },
    success=True
)

from netra.models import ConversationType

ConversationType.INPUT   # User input messages
ConversationType.OUTPUT  # Model output messages
ConversationType.SYSTEM  # System instructions

from netra.decorators import agent

@agent
def research_agent(query: str):
    return perform_research(query)

@agent(name="customer-support")
def support_agent(request: dict):
    return handle_request(request)

from netra.decorators import task

@task
def fetch_data(query: str):
    return database.query(query)

@task(name="web-search")
def search_web(query: str):
    return search_api.search(query)

from netra.decorators import workflow

@workflow
def process_order(order: dict):
    validate_order(order)
    process_payment(order)
    fulfill_order(order)

from netra import Netra
from netra.dashboard import *

Netra.init(app_name="my-app")

result = Netra.dashboard.query_data(
    scope=Scope.SPANS,
    chart_type=ChartType.LINE_TIME_SERIES,
    metrics=Metrics(
        measure=Measure.TOTAL_COST,
        aggregation=Aggregation.TOTAL_COUNT,
    ),
    dimension=Dimension(field=DimensionField.SERVICE),
    filter=FilterConfig(
        start_time="2026-01-01T00:00:00.000Z",
        end_time="2026-01-31T23:59:59.000Z",
        group_by=GroupBy.DAY,
    ),
)

from netra import Netra
from netra.dashboard import *

result = Netra.dashboard.get_session_summary(
    filter=SessionFilterConfig(
        start_time="2026-01-01T00:00:00.000Z",
        end_time="2026-01-31T23:59:59.000Z",
        filters=[
            SessionFilter(
                field=SessionFilterField.TENANT_ID,
                operator=SessionFilterOperator.ANY_OF,
                type=SessionFilterType.ARRAY,
                value=["TenantA", "TenantB"]
            )
        ]
    )
)

from netra import Netra
from netra.dashboard import *

stats = Netra.dashboard.get_session_stats(
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
    limit=10,
    sort_field=SortField.TOTAL_COST,
    sort_order=SortOrder.DESC
)

for session in stats.data:
    print(session.session_id, session.total_cost)

# Handle pagination
if stats.has_next_page:
    next_stats = Netra.dashboard.get_session_stats(
        start_time="2026-01-01T00:00:00.000Z",
        end_time="2026-01-31T23:59:59.000Z",
        cursor=stats.next_cursor
    )

from netra import Netra
from netra.dashboard import *

for session in Netra.dashboard.iter_session_stats(
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
    sort_field=SortField.TOTAL_COST,
    sort_order=SortOrder.DESC
):
    print(f"Session {session.session_id}: ${session.total_cost:.4f}")

from netra import Netra

Netra.init(app_name="my-app")

session_usage = Netra.usage.get_session_usage(
    session_id="session-123",
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
)

print(f"Tokens: {session_usage.token_count}")
print(f"Cost: ${session_usage.total_cost:.4f}")

from netra import Netra

tenant_usage = Netra.usage.get_tenant_usage(
    tenant_id="AceTech",
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
)

print(f"Sessions: {tenant_usage.session_count}")
print(f"Total cost: ${tenant_usage.total_cost:.4f}")

from netra import Netra

traces_page = Netra.usage.list_traces(
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
    tenant_id="AceTech",
    limit=10,
    sort_field="start_time",
    sort_order="desc",
)

for trace in traces_page.traces:
    print(trace.id, trace.name, trace.total_cost)

# Handle pagination
if traces_page.has_next_page:
    next_page = Netra.usage.list_traces(
        start_time="2026-01-01T00:00:00.000Z",
        end_time="2026-01-31T23:59:59.000Z",
        cursor=traces_page.next_cursor
    )

from netra import Netra

for trace in Netra.usage.iter_traces(
    start_time="2026-01-01T00:00:00.000Z",
    end_time="2026-01-31T23:59:59.000Z",
    tenant_id="AceTech",
    sort_field="start_time",
    sort_order="desc",
):
    print(f"Trace {trace.id}: ${trace.total_cost:.4f}")

from netra import Netra

spans_page = Netra.usage.list_spans_by_trace_id(
    trace_id="728c1de6fa4c53de143a7d7fef33ff91",
    limit=10,
    span_name="openai.chat",
)

for span in spans_page.spans:
    print(span.id, span.name, span.kind, span.latency_ms)

from netra import Netra

for span in Netra.usage.iter_spans_by_trace_id(
    trace_id="728c1de6fa4c53de143a7d7fef33ff91",
    span_name="generation_pipeline",
):
    print(f"Span: {span.name} ({span.kind}): {span.latency_ms}ms")

from netra.instrumentation.instruments import InstrumentSet

# LLM Providers
InstrumentSet.OPENAI
InstrumentSet.ANTHROPIC
InstrumentSet.COHERE
InstrumentSet.GOOGLE_GENERATIVEAI
InstrumentSet.MISTRALAI
InstrumentSet.GROQ
InstrumentSet.BEDROCK
InstrumentSet.VERTEXAI
InstrumentSet.OLLAMA
InstrumentSet.REPLICATE
InstrumentSet.TOGETHER
InstrumentSet.TRANSFORMERS
InstrumentSet.LITELLM

# AI Frameworks
InstrumentSet.LANGCHAIN
InstrumentSet.LANGGRAPH
InstrumentSet.LLAMAINDEX
InstrumentSet.HAYSTACK
InstrumentSet.CREWAI
InstrumentSet.DSPY
InstrumentSet.PYDANTIC_AI

# Vector Databases
InstrumentSet.PINECONE
InstrumentSet.WEAVIATE
InstrumentSet.QDRANT
InstrumentSet.CHROMADB
InstrumentSet.MILVUS
InstrumentSet.LANCEDB
InstrumentSet.MARQO

# HTTP Clients
InstrumentSet.HTTPX
InstrumentSet.AIOHTTP
InstrumentSet.REQUESTS
InstrumentSet.URLLIB
InstrumentSet.URLLIB3

# Web Frameworks
InstrumentSet.FASTAPI
InstrumentSet.FLASK
InstrumentSet.DJANGO
InstrumentSet.STARLETTE

# Databases
InstrumentSet.SQLALCHEMY
InstrumentSet.SQLITE3
InstrumentSet.PSYCOPG
InstrumentSet.PSYCOPG2
InstrumentSet.PYMYSQL
InstrumentSet.PYMONGO
InstrumentSet.REDIS
InstrumentSet.ELASTICSEARCH

# Message Queues
InstrumentSet.CELERY
InstrumentSet.PIKA
InstrumentSet.AIO_PIKA
InstrumentSet.KAFKA_PYTHON
InstrumentSet.CONFLUENT_KAFKA

import os
from netra import Netra, SpanType, UsageModel
from netra.instrumentation.instruments import InstrumentSet
from openai import OpenAI

# Initialize SDK
Netra.init(
    app_name="my-ai-app",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
    environment="production",
    instruments={InstrumentSet.OPENAI},
)

# Set context
Netra.set_user_id("user-123")
Netra.set_session_id("session-abc")

# Create a traced operation
def chat_with_ai(user_message: str) -> str:
    with Netra.start_span("chat-completion", as_type=SpanType.GENERATION) as span:
        span.set_prompt(user_message)
        span.set_model("gpt-4")
        span.set_llm_system("openai")

        client = OpenAI()
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": user_message}]
        )

        # Calculate cost based on token usage
        prompt_cost = (response.usage.prompt_tokens / 1000) * 0.03
        completion_cost = (response.usage.completion_tokens / 1000) * 0.06
        total_cost = prompt_cost + completion_cost

        span.set_usage([
            UsageModel(
                model="gpt-4",
                cost_in_usd=total_cost,
                usage_type="chat",
                units_used=1
            )
        ])

        span.set_success()
        return response.choices[0].message.content

# Use the function
result = chat_with_ai("What is the capital of France?")
print(result)

# Shutdown gracefully
Netra.shutdown()