Decorators

Decorators provide a clean, declarative way to instrument your code. They automatically create spans with semantic meaning, making your traces easier to understand and navigate in the Netra dashboard.

Decorators: Overview

Netra provides four decorators, each designed for a specific type of operation:

Decorator	Purpose	Span Type
`@workflow`	High-level business transactions	Span
`@agent`	AI agents or orchestrators	Agent
`@task`	Individual units of work	Tool
`@span`	Generic operations with customizable type	Configurable

@workflow

Use @workflow to mark high-level business transactions or processes. Workflows typically represent complete user-facing operations that may involve multiple steps.

from netra.decorators import workflow

@workflow
def process_customer_order(order_id: str):
    order = fetch_order(order_id)
    inventory = check_inventory(order.items)
    payment = process_payment(order)
    shipment = create_shipment(order)
    return {"order": order, "shipment": shipment}

# With custom name
@workflow(name="order-fulfillment")
def fulfill_order(order_id: str):
    # ...
    pass

import { workflow } from "netra-sdk";

@workflow
async function processCustomerOrder(orderId: string) {
  const order = await fetchOrder(orderId);
  const inventory = await checkInventory(order.items);
  const payment = await processPayment(order);
  const shipment = await createShipment(order);
  return { order, shipment };
}

// With custom name
@workflow({ name: "order-fulfillment" })
async function fulfillOrder(orderId: string) {
  // ...
}

When to use @workflow:

User-initiated actions (e.g., “submit order”, “generate report”)
End-to-end processes that span multiple operations
Top-level entry points in your application

@agent

Use @agent to mark AI agents or autonomous components that make decisions. Agent spans help you track reasoning steps and decision-making processes.

from netra.decorators import agent

@agent
class CustomerSupportAgent:
    def handle_ticket(self, ticket: dict):
        analysis = self.analyze_ticket(ticket)
        response = self.generate_response(analysis)
        return response

    def analyze_ticket(self, ticket: dict):
        # Analysis logic
        pass

    def generate_response(self, analysis: dict):
        # Response generation
        pass

# Function-based agent
@agent
def research_agent(query: str):
    sources = search_sources(query)
    synthesis = synthesize_information(sources)
    return synthesis

import { agent } from "netra-sdk";

@agent
class CustomerSupportAgent {
  async handleTicket(ticket: Ticket) {
    const analysis = await this.analyzeTicket(ticket);
    const response = await this.generateResponse(analysis);
    return response;
  }

  private async analyzeTicket(ticket: Ticket) {
    // Analysis logic
  }

  private async generateResponse(analysis: Analysis) {
    // Response generation
  }
}

// Function-based agent
@agent
async function researchAgent(query: string) {
  const sources = await searchSources(query);
  const synthesis = await synthesizeInformation(sources);
  return synthesis;
}

When to use @agent:

Autonomous AI components that make decisions
Multi-step reasoning processes
Components that orchestrate other tools or services

@task

Use @task to mark individual units of work. Tasks are typically discrete operations that perform a specific function within a larger workflow.

from netra.decorators import task

@task
def fetch_user_profile(user_id: str):
    response = requests.get(f"/api/users/{user_id}")
    return response.json()

@task
def send_notification(user_id: str, message: str):
    notification_service.send(user_id, message)

@task(name="validate-input")
def validate_order_input(order: dict):
    if not order.get("items"):
        raise ValueError("Order must contain at least one item")
    return True

import { task } from "netra-sdk";

@task
async function fetchUserProfile(userId: string) {
  const response = await fetch(`/api/users/${userId}`);
  return response.json();
}

@task
async function sendNotification(userId: string, message: string) {
  await notificationService.send(userId, message);
}

@task({ name: "validate-input" })
function validateOrderInput(order: Order) {
  if (!order.items || order.items.length === 0) {
    throw new Error("Order must contain at least one item");
  }
  return true;
}

When to use @task:

Individual operations within a workflow
Tool calls or function executions
Database operations, API calls, or computations

@span

Use @span for generic tracing with full control over the span type. This is the most flexible decorator, allowing you to specify the exact span type.

from netra.decorators import span
from netra import SpanType

# Default span type
@span
def process_data(data: list):
    return [transform(item) for item in data]

# With custom span type
@span(as_type=SpanType.GENERATION)
def generate_embedding(text: str):
    embedding = embedding_model.embed(text)
    return embedding

@span(name="vector-search", as_type=SpanType.TOOL)
def search_vector_db(query: str, top_k: int):
    return vector_store.search(query, top_k)

import { span, SpanType } from "netra-sdk";

// Default span type
@span
function processData(data: any[]) {
  return data.map(transform);
}

// With custom span type
@span({ asType: SpanType.GENERATION })
async function generateEmbedding(text: string) {
  const embedding = await embeddingModel.embed(text);
  return embedding;
}

@span({ name: "vector-search", asType: SpanType.TOOL })
async function searchVectorDB(query: string, topK: number) {
  return await vectorStore.search(query, topK);
}

Available Span Types

Span Type	Use Case
`SpanType.SPAN`	Generic operations (default)
`SpanType.GENERATION`	LLM text generation
`SpanType.EMBEDDING`	Vector embedding operations
`SpanType.TOOL`	Tool or function calls
`SpanType.AGENT`	AI agent operations

Decorating Classes

When you apply a decorator to a class, all public methods of that class are automatically instrumented.

from netra.decorators import agent, task

@agent
class OrderProcessor:
    # All public methods are traced as part of the agent
    def process_order(self, order: dict):
        self.validate_order(order)
        self.charge_payment(order)
        self.fulfill_order(order)

    @task  # Override with specific decorator
    def validate_order(self, order: dict):
        # Validation logic
        pass

    def charge_payment(self, order: dict):
        # Payment logic
        pass

    def fulfill_order(self, order: dict):
        # Fulfillment logic
        pass

import { agent, task } from "netra-sdk";

@agent
class OrderProcessor {
  // All public methods are traced as part of the agent
  async processOrder(order: Order) {
    await this.validateOrder(order);
    await this.chargePayment(order);
    await this.fulfillOrder(order);
  }

  @task // Override with specific decorator
  async validateOrder(order: Order) {
    // Validation logic
  }

  private async chargePayment(order: Order) {
    // Private methods are also traced
  }

  private async fulfillOrder(order: Order) {
    // Fulfillment logic
  }
}

Async and Generator Support

Decorators work seamlessly with async functions and generators:

from netra.decorators import workflow, task

# Async functions
@workflow
async def async_workflow():
    result = await some_async_operation()
    return result

# Async generators
@task
async def stream_results(query: str):
    stream = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": query}],
        stream=True,
    )

    async for chunk in stream:
        yield chunk.choices[0].delta.content or ""

# Sync generators
@task
def generate_items(count: int):
    for i in range(count):
        yield process_item(i)

import { workflow, task } from "netra-sdk";

// Async functions
@workflow
async function asyncWorkflow() {
  const result = await someAsyncOperation();
  return result;
}

// Async generators
@task
async function* streamResults(query: string) {
  const stream = await openai.chat.completions.create({
    model: "gpt-4",
    messages: [{ role: "user", content: query }],
    stream: true,
  });

  for await (const chunk of stream) {
    yield chunk.choices[0]?.delta?.content || "";
  }
}

For streaming responses, the span remains open until the stream is fully consumed. This ensures accurate latency measurements for streaming operations.

Automatic Parameter Capture

Decorators automatically capture function parameters as span attributes, making it easy to understand what inputs were provided:

from netra.decorators import task

@task
def search_products(query: str, category: str, limit: int = 10):
    # Parameters are automatically captured:
    # - query: "laptop"
    # - category: "electronics"
    # - limit: 10
    return product_service.search(query, category, limit)

# Call the function
search_products("laptop", "electronics")

import { task } from "netra-sdk";

@task
async function searchProducts(
  query: string,
  category: string,
  limit: number = 10
) {
  // Parameters are automatically captured:
  // - query: "laptop"
  // - category: "electronics"
  // - limit: 10
  return await productService.search(query, category, limit);
}

// Call the function
await searchProducts("laptop", "electronics");

Complex types (lists, dicts, objects) are serialized to JSON. Parameter values are truncated to 1000 characters to prevent excessively large attributes.

Exception Handling

Decorators automatically capture exceptions and mark spans with error status:

from netra.decorators import task

@task
def risky_operation(data: dict):
    if not data.get("valid"):
        raise ValueError("Invalid data provided")
        # Span is automatically marked as ERROR
        # Exception details are recorded
    return process_data(data)

import { task } from "netra-sdk";

@task
async function riskyOperation(data: any) {
  if (!data.valid) {
    throw new Error("Invalid data provided");
    // Span is automatically marked as ERROR
    // Exception details are recorded
  }
  return processData(data);
}

Combining Decorators with Manual Tracing

You can combine decorators with manual span operations for additional context:

from netra.decorators import workflow
from netra import Netra

@workflow
def process_order(order: dict):
    # Add custom attributes to the current span
    current_span = Netra.get_current_span()
    if current_span:
        current_span.set_attribute("order.id", order["id"])
        current_span.set_attribute("order.total", order["total"])

        # Add custom events
        current_span.add_event("order-validated", {
            "item_count": len(order["items"]),
        })

    result = fulfill_order(order)

    if current_span:
        current_span.add_event("order-fulfilled")
    return result

import { workflow, Netra } from "netra-sdk";

@workflow
async function processOrder(order: Order) {
  // Add custom attributes to the current span
  const currentSpan = Netra.getCurrentSpan();
  currentSpan?.setAttribute("order.id", order.id);
  currentSpan?.setAttribute("order.total", order.total);

  // Add custom events
  currentSpan?.addEvent("order-validated", {
    itemCount: order.items.length,
  });

  const result = await fulfillOrder(order);

  currentSpan?.addEvent("order-fulfilled");
  return result;
}

Renaming a Decorated Span

A decorator names its span when the function is called, before the function body runs. If the readable name depends on work done inside the function, call Netra.update_span_name() once you know it:

Python

from netra import Netra
from netra.decorators import agent

@agent(name="support-agent")
def handle_ticket(ticket: dict):
    department = route_ticket(ticket)
    Netra.update_span_name(f"{department} Agent")
    return resolve(ticket)

This renames the currently active span. If that span was opened by @workflow, @agent, @task, or @span, its netra.<entity>.name attribute changes too, and child spans started after the call inherit the new name. Child spans that already started keep the old name. Only the active span is renamed. If you call Netra.update_span_name() inside a nested @task, the task is renamed and the enclosing agent keeps its name. With no active span, the call logs a warning and does nothing.

Best Practices

Use semantic decorators - Choose the decorator that best describes the operation’s purpose (@workflow for processes, @agent for AI components, @task for individual operations).
Name spans meaningfully - Use the name parameter when the function name isn’t descriptive enough.
Don’t over-instrument - Focus on high-value operations. Not every function needs a decorator.
Combine with auto-instrumentation - Let auto-instrumentation handle LLM calls and database operations while using decorators for your application logic.
Use class decoration sparingly - Decorating an entire class instruments all methods, which may create noise. Consider decorating individual methods instead.

Learn More

Manual Tracing - Fine-grained control with SpanWrapper
Auto Instrumentation - Zero-code tracing for supported libraries
Spans - Understanding span types and attributes

Get Started

Observability

Insights

Evaluations

Simulations

Evaluators

Red Teaming

Prompt Management

Monitoring & Dashboard

Account settings

Migrations

Decorators: Overview

@workflow

@agent

@task

@span

Available Span Types

Decorating Classes

Async and Generator Support

Automatic Parameter Capture

Exception Handling

Combining Decorators with Manual Tracing

Renaming a Decorated Span

Best Practices

Learn More

​Decorators: Overview

​@workflow

​@agent

​@task

​@span

​Available Span Types

​Decorating Classes

​Async and Generator Support

​Automatic Parameter Capture

​Exception Handling

​Combining Decorators with Manual Tracing

​Renaming a Decorated Span

​Best Practices

​Learn More

Decorators: Overview

@workflow

@agent

@task

@span

Available Span Types

Decorating Classes

Async and Generator Support

Automatic Parameter Capture

Exception Handling

Combining Decorators with Manual Tracing

Renaming a Decorated Span

Best Practices

Learn More