Overview

from netra import Netra, NetraInstruments

Netra.init(
    app_name="my-ai-app",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
    environment="production",
    trace_content=True,
    instruments={NetraInstruments.OPENAI, NetraInstruments.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.session_manager import ConversationType

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

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

Netra.set_input("What is the capital of France?")
# Dicts and lists are automatically JSON-serialized
Netra.set_input({"query": "capital of France", "language": "en"})

Netra.set_output("The capital of France is Paris.")
# Dicts and lists are automatically JSON-serialized
Netra.set_output({"answer": "Paris", "confidence": 0.98})

from netra import Netra
# Inside a nested span, set the root-level input
with Netra.start_span("sub-step") as span:
    Netra.set_root_input("Summarize the quarterly earnings report.")

from netra import Netra
# At the end of your pipeline, record the top-level output
Netra.set_root_output({
    "summary": "Revenue increased 12% year-over-year...",
    "tokens_used": 1250
})

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.session_manager import ConversationType

ConversationType.INPUT   # User input messages
ConversationType.OUTPUT  # Model output messages

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.decorators import workflow

@workflow(name="WorkflowRobot", capture_internal_methods=True)
class WorkflowRobot:
    """Wraps: workflow entity, span type SPAN (default for workflow)."""
    
	def __init__(self, label: str) -> None:
        self.label = label

    def __str__(self) -> str:
        return f"WorkflowRobot({self.label})"

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

detail = Netra.dashboard.get_session_details(
    session_id="0acd79e0-526e-40cc-91e9-4010711de0ed"
)

print(detail["sessionId"])
for trace in detail["traces"]:
    print(trace["traceId"], trace["cost"]["totalCost"])

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)

# 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(session)

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(session_usage)

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(tenant_usage)

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)

# 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(trace)

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)

from netra import Netra

for span in Netra.usage.iter_spans_by_trace_id(
    trace_id="728c1de6fa4c53de143a7d7fef33ff91",
    span_name="generation_pipeline",
):
    print(span)

from netra.instrumentation.instruments import NetraInstruments

NetraInstruments.ALL  # Enables all supported instrumentations (legacy behavior)

NetraInstruments.OPENAI            # OpenAI
NetraInstruments.ANTHROPIC         # Anthropic Claude
NetraInstruments.GOOGLE_GENERATIVEAI  # Google Generative AI
NetraInstruments.COHEREAI          # Cohere
NetraInstruments.MISTRALAI         # Mistral AI
NetraInstruments.GROQ              # Groq
NetraInstruments.BEDROCK           # AWS Bedrock
NetraInstruments.VERTEXAI          # Google Vertex AI
NetraInstruments.OLLAMA            # Ollama
NetraInstruments.REPLICATE         # Replicate
NetraInstruments.TOGETHER          # Together AI
NetraInstruments.LITELLM           # LiteLLM
NetraInstruments.CEREBRAS          # Cerebras
NetraInstruments.CARTESIA          # Cartesia
NetraInstruments.DEEPGRAM          # Deepgram
NetraInstruments.ELEVENLABS        # ElevenLabs
NetraInstruments.ALEPHALPHA        # Aleph Alpha
NetraInstruments.WATSONX           # IBM WatsonX
NetraInstruments.SAGEMAKER         # AWS SageMaker
NetraInstruments.WRITER            # Writer
NetraInstruments.TRANSFORMERS      # Hugging Face Transformers

NetraInstruments.LANGCHAIN         # LangChain
NetraInstruments.LLAMA_INDEX       # LlamaIndex
NetraInstruments.HAYSTACK          # Haystack
NetraInstruments.CREW              # CrewAI
NetraInstruments.DSPY              # DSPy
NetraInstruments.PYDANTIC_AI       # Pydantic AI
NetraInstruments.ADK               # Google ADK
NetraInstruments.AGNO              # Agno
NetraInstruments.MCP               # Model Context Protocol
NetraInstruments.CLAUDE_AGENT_SDK  # Claude Agent SDK
NetraInstruments.OPENAI_AGENTS     # OpenAI Agents SDK

NetraInstruments.PINECONE          # Pinecone
NetraInstruments.WEAVIATEDB        # Weaviate
NetraInstruments.QDRANTDB          # Qdrant
NetraInstruments.CHROMA            # ChromaDB
NetraInstruments.MILVUS            # Milvus
NetraInstruments.LANCEDB           # LanceDB
NetraInstruments.MARQO             # Marqo

NetraInstruments.HTTPX             # HTTPX
NetraInstruments.AIOHTTP           # AIOHTTP
NetraInstruments.REQUESTS          # Requests
NetraInstruments.URLLIB             # urllib
NetraInstruments.URLLIB3            # urllib3
NetraInstruments.GRPC              # gRPC

NetraInstruments.FASTAPI           # FastAPI
NetraInstruments.FLASK             # Flask
NetraInstruments.DJANGO            # Django
NetraInstruments.STARLETTE         # Starlette
NetraInstruments.TORNADO           # Tornado
NetraInstruments.FALCON            # Falcon
NetraInstruments.PYRAMID           # Pyramid
NetraInstruments.ASGI              # ASGI
NetraInstruments.WSGI              # WSGI
NetraInstruments.AIOHTTP_SERVER    # AIOHTTP Server

NetraInstruments.SQLALCHEMY        # SQLAlchemy
NetraInstruments.SQLITE3           # SQLite3
NetraInstruments.PSYCOPG           # PostgreSQL (psycopg)
NetraInstruments.PSYCOPG2          # PostgreSQL (psycopg2)
NetraInstruments.ASYNCPG           # PostgreSQL (asyncpg)
NetraInstruments.AIOPG             # PostgreSQL (aiopg)
NetraInstruments.PYMYSQL           # MySQL (PyMySQL)
NetraInstruments.MYSQL             # MySQL (mysql-connector)
NetraInstruments.MYSQLCLIENT       # MySQL (mysqlclient)
NetraInstruments.PYMONGO           # MongoDB
NetraInstruments.PYMSSQL           # Microsoft SQL Server
NetraInstruments.REDIS             # Redis
NetraInstruments.ELASTICSEARCH     # Elasticsearch
NetraInstruments.CASSANDRA         # Cassandra
NetraInstruments.PYMEMCACHE        # Memcached

NetraInstruments.CELERY            # Celery
NetraInstruments.PIKA              # RabbitMQ (pika)
NetraInstruments.AIO_PIKA          # RabbitMQ (aio-pika)
NetraInstruments.KAFKA_PYTHON      # Kafka (kafka-python)
NetraInstruments.CONFLUENT_KAFKA   # Confluent Kafka
NetraInstruments.AIOKAFKA          # Kafka (aiokafka)
NetraInstruments.BOTO3SQS          # AWS SQS

NetraInstruments.AWS_LAMBDA        # AWS Lambda
NetraInstruments.BOTO              # AWS SDK (boto)
NetraInstruments.BOTOCORE          # AWS SDK (botocore)
NetraInstruments.JINJA2            # Jinja2
NetraInstruments.LOGGING           # Python logging
NetraInstruments.THREADING         # Python threading
NetraInstruments.ASYNCIO           # Python asyncio
NetraInstruments.CLICK             # Click CLI
NetraInstruments.ASYNCCLICK        # AsyncClick CLI
NetraInstruments.SYSTEM_METRICS    # System metrics
NetraInstruments.DBAPI             # Python DB-API 2.0
NetraInstruments.TORTOISEORM       # Tortoise ORM
NetraInstruments.REMOULADE         # Remoulade

import os
from netra import Netra, NetraInstruments, SpanType, UsageModel
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={NetraInstruments.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()