Decorators

Available Decorators

AgentOps provides the following decorators:

Decorator	Purpose	Creates
`@session`	Track an entire user interaction	SESSION span
`@agent`	Track agent classes and their lifecycle	AGENT span
`@operation`	Track discrete operations performed by agents	OPERATION span
`@workflow`	Track a sequence of operations	WORKFLOW span
`@task`	Track smaller units of work (similar to operations)	TASK span
`@tool`	Track tool usage and cost in agent operations	TOOL span
`@guardrail`	Track guardrail input and output	GUARDRAIL span

Decorator Hierarchy

The decorators create spans that form a hierarchy:

SESSION
  ├── AGENT
  │     ├── OPERATION or TASK
  │     │     ├── LLM
  │     │     └── TOOL
  │     └── WORKFLOW
  │           └── OPERATION or TASK
  └── AGENT
        └── OPERATION or TASK

Using Decorators

@session

The @session decorator tracks an entire user interaction from start to finish:

from agentops.sdk.decorators import session
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@session
def answer_question(question):
    # Create and use agents
    weather_agent = WeatherAgent()
    result = weather_agent.get_forecast(question)
    
    # Return the final result
    return result

Each @session function call creates a new session span that contains all the agents, operations, and workflows used during that interaction.

@agent

The @agent decorator instruments a class to track its lifecycle and operations:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class WeatherAgent:
    def __init__(self):
        self.api_key = "weather_api_key"
        
    @operation
    def get_forecast(self, location):
        # Get weather data
        return f"The weather in {location} is sunny."

def check_weather(city):
    weather_agent = WeatherAgent()
    forecast = weather_agent.get_forecast(city)
    return forecast

weather_info = check_weather("San Francisco")

When an agent-decorated class is instantiated within a session, an AGENT span is created automatically.

@operation

The @operation decorator tracks discrete functions performed by an agent:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class MathAgent:
    @operation
    def add(self, a, b):
        return a + b
        
    @operation
    def multiply(self, a, b):
        return a * b

def calculate(x, y):
    math_agent = MathAgent()
    sum_result = math_agent.add(x, y)
    product_result = math_agent.multiply(x, y)
    return {"sum": sum_result, "product": product_result}

results = calculate(5, 3)

Operations represent the smallest meaningful units of work in your agent system. Each operation creates an OPERATION span with:

Inputs (function arguments)
Output (return value)
Duration
Success/failure status

@workflow

The @workflow decorator tracks a sequence of operations that work together:

from agentops.sdk.decorators import agent, operation, workflow
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class TravelAgent:
    def __init__(self):
        self.flight_api = FlightAPI()
        self.hotel_api = HotelAPI()
    
    @workflow
    def plan_trip(self, destination, dates):
        # This workflow contains multiple operations
        flights = self.find_flights(destination, dates)
        hotels = self.find_hotels(destination, dates)
        
        return {
            "flights": flights,
            "hotels": hotels
        }
        
    @operation
    def find_flights(self, destination, dates):
        return self.flight_api.search(destination, dates)
        
    @operation
    def find_hotels(self, destination, dates):
        return self.hotel_api.search(destination, dates)

Workflows help you organize related operations and see their collective performance.

@task

The @task decorator is similar to @operation but can be used for smaller units of work:

from agentops.sdk.decorators import agent, task
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class DataProcessor:
    @task
    def normalize_data(self, data):
        # Normalize the data
        return [x / sum(data) for x in data]
    
    @task
    def filter_outliers(self, data, threshold=3):
        # Filter outliers
        mean = sum(data) / len(data)
        std_dev = (sum((x - mean) ** 2 for x in data) / len(data)) ** 0.5
        
        return [x for x in data if abs(x - mean) <= threshold * std_dev]

The @task and @operation decorators function identically (they are aliases in the codebase), and you can choose the one that best fits your semantic needs.

@tool

The @tool decorator tracks tool usage within agent operations and supports cost tracking. It works with all function types: synchronous, asynchronous, generator, and async generator.

from agentops.sdk.decorators import agent, tool
import asyncio

@agent
class ProcessingAgent:
    def __init__(self):
        pass

    @tool(cost=0.01)
    def sync_tool(self, item):
        """Synchronous tool with cost tracking."""
        return f"Processed {item}"

    @tool(cost=0.02)
    async def async_tool(self, item):
        """Asynchronous tool with cost tracking."""
        await asyncio.sleep(0.1)
        return f"Async processed {item}"

    @tool(cost=0.03)
    def generator_tool(self, items):
        """Generator tool with cost tracking."""
        for item in items:
            yield self.sync_tool(item)

    @tool(cost=0.04)
    async def async_generator_tool(self, items):
        """Async generator tool with cost tracking."""
        for item in items:
            await asyncio.sleep(0.1)
            yield await self.async_tool(item)

The tool decorator provides:

Cost tracking for each tool call
Proper span creation and nesting
Support for all function types (sync, async, generator, async generator)
Cost accumulation in generator and async generator operations

@guardrail

The @guardrail decorator tracks guardrail input and output. You can specify the guardrail type ("input" or "output") with the spec parameter.

from agentops.sdk.decorators import guardrail
import agentops
import re

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@guardrail(spec="input")
def secret_key_guardrail(input):
    pattern = r'\bsk-[a-zA-Z0-9]{10,}\b'
    result = True if re.search(pattern, input) else False
    return {
        "tripwire_triggered" : result
    }

Decorator Attributes

You can pass additional attributes to decorators:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent(name="custom_agent_name", attributes={"version": "1.0"})
class CustomAgent:
    @operation(name="custom_operation", attributes={"importance": "high"})
    def process(self, data):
        return data

Common attributes include:

Attribute	Description	Example
`name`	Custom name for the span	`name="weather_forecast"`
`attributes`	Dictionary of custom attributes	`attributes={"model": "gpt-4"}`

Complete Example

Here’s a complete example using all the decorators together:

from agentops.sdk.decorators import session, agent, operation, workflow, task
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@session
def assist_user(query):
    # Create the main assistant
    assistant = Assistant()
    
    # Process the query
    return assistant.process_query(query)

@agent
class Assistant:
    def __init__(self):
        pass
    
    @workflow
    def process_query(self, query):
        research_agent = ResearchAgent()
        writing_agent = WritingAgent()
        
        # Research phase
        research = research_agent.gather_information(query)
        
        # Writing phase
        response = writing_agent.generate_response(query, research)
        
        return response

@agent
class ResearchAgent:
    @operation
    def gather_information(self, query):
        # Perform web search
        search_results = self.search(query)
        
        # Analyze results
        return self.analyze_results(search_results)
    
    @task
    def search(self, query):
        # Simulate web search
        return [f"Result for {query}", f"Another result for {query}"]
    
    @task
    def analyze_results(self, results):
        # Analyze search results
        return {"summary": "Analysis of " + ", ".join(results)}

@agent
class WritingAgent:
    @operation
    def generate_response(self, query, research):
        # Generate a response based on the research
        return f"Answer to '{query}' based on: {research['summary']}"

assist_user("What is the capital of France?")

In this example:

The @session decorator wraps the entire interaction
The @agent decorator defines multiple agent classes
The @workflow decorator creates a workflow that coordinates agents
The @operation and @task decorators track individual operations
All spans are properly nested in the hierarchy

Note that LLM and TOOL spans are automatically created when you use compatible LLM libraries or tool integrations.

Best Practices

Use @session for top-level functions that represent complete user interactions
Apply @agent to classes that represent distinct components of your system
Use @operation for significant functions that represent complete units of work
Use @task for smaller functions that are part of larger operations
Apply @workflow to methods that coordinate multiple operations
Keep decorator nesting consistent with the logical hierarchy of your code
Add custom attributes to provide additional context for analysis
Use meaningful names for all decorated components

Dashboard Visualization

In the AgentOps dashboard, decorators create spans that appear in:

Timeline View: Shows the execution sequence and duration
Hierarchy View: Displays the parent-child relationships
Detail Panels: Shows inputs, outputs, and attributes
Performance Metrics: Tracks execution times and success rates

This visualization helps you understand the flow and performance of your agent system.

On this page

Available Decorators
Decorator Hierarchy
Using Decorators
@session
@agent
@operation
@workflow
@task
@tool
@guardrail
Decorator Attributes
Complete Example
Best Practices
Dashboard Visualization

Available Decorators

AgentOps provides the following decorators:

Decorator	Purpose	Creates
`@session`	Track an entire user interaction	SESSION span
`@agent`	Track agent classes and their lifecycle	AGENT span
`@operation`	Track discrete operations performed by agents	OPERATION span
`@workflow`	Track a sequence of operations	WORKFLOW span
`@task`	Track smaller units of work (similar to operations)	TASK span
`@tool`	Track tool usage and cost in agent operations	TOOL span
`@guardrail`	Track guardrail input and output	GUARDRAIL span

Decorator Hierarchy

The decorators create spans that form a hierarchy:

SESSION
  ├── AGENT
  │     ├── OPERATION or TASK
  │     │     ├── LLM
  │     │     └── TOOL
  │     └── WORKFLOW
  │           └── OPERATION or TASK
  └── AGENT
        └── OPERATION or TASK

Using Decorators

@session

The @session decorator tracks an entire user interaction from start to finish:

from agentops.sdk.decorators import session
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@session
def answer_question(question):
    # Create and use agents
    weather_agent = WeatherAgent()
    result = weather_agent.get_forecast(question)
    
    # Return the final result
    return result

Each @session function call creates a new session span that contains all the agents, operations, and workflows used during that interaction.

@agent

The @agent decorator instruments a class to track its lifecycle and operations:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class WeatherAgent:
    def __init__(self):
        self.api_key = "weather_api_key"
        
    @operation
    def get_forecast(self, location):
        # Get weather data
        return f"The weather in {location} is sunny."

def check_weather(city):
    weather_agent = WeatherAgent()
    forecast = weather_agent.get_forecast(city)
    return forecast

weather_info = check_weather("San Francisco")

When an agent-decorated class is instantiated within a session, an AGENT span is created automatically.

@operation

The @operation decorator tracks discrete functions performed by an agent:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class MathAgent:
    @operation
    def add(self, a, b):
        return a + b
        
    @operation
    def multiply(self, a, b):
        return a * b

def calculate(x, y):
    math_agent = MathAgent()
    sum_result = math_agent.add(x, y)
    product_result = math_agent.multiply(x, y)
    return {"sum": sum_result, "product": product_result}

results = calculate(5, 3)

Operations represent the smallest meaningful units of work in your agent system. Each operation creates an OPERATION span with:

Inputs (function arguments)
Output (return value)
Duration
Success/failure status

@workflow

The @workflow decorator tracks a sequence of operations that work together:

from agentops.sdk.decorators import agent, operation, workflow
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class TravelAgent:
    def __init__(self):
        self.flight_api = FlightAPI()
        self.hotel_api = HotelAPI()
    
    @workflow
    def plan_trip(self, destination, dates):
        # This workflow contains multiple operations
        flights = self.find_flights(destination, dates)
        hotels = self.find_hotels(destination, dates)
        
        return {
            "flights": flights,
            "hotels": hotels
        }
        
    @operation
    def find_flights(self, destination, dates):
        return self.flight_api.search(destination, dates)
        
    @operation
    def find_hotels(self, destination, dates):
        return self.hotel_api.search(destination, dates)

Workflows help you organize related operations and see their collective performance.

@task

The @task decorator is similar to @operation but can be used for smaller units of work:

from agentops.sdk.decorators import agent, task
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent
class DataProcessor:
    @task
    def normalize_data(self, data):
        # Normalize the data
        return [x / sum(data) for x in data]
    
    @task
    def filter_outliers(self, data, threshold=3):
        # Filter outliers
        mean = sum(data) / len(data)
        std_dev = (sum((x - mean) ** 2 for x in data) / len(data)) ** 0.5
        
        return [x for x in data if abs(x - mean) <= threshold * std_dev]

The @task and @operation decorators function identically (they are aliases in the codebase), and you can choose the one that best fits your semantic needs.

@tool

The @tool decorator tracks tool usage within agent operations and supports cost tracking. It works with all function types: synchronous, asynchronous, generator, and async generator.

from agentops.sdk.decorators import agent, tool
import asyncio

@agent
class ProcessingAgent:
    def __init__(self):
        pass

    @tool(cost=0.01)
    def sync_tool(self, item):
        """Synchronous tool with cost tracking."""
        return f"Processed {item}"

    @tool(cost=0.02)
    async def async_tool(self, item):
        """Asynchronous tool with cost tracking."""
        await asyncio.sleep(0.1)
        return f"Async processed {item}"

    @tool(cost=0.03)
    def generator_tool(self, items):
        """Generator tool with cost tracking."""
        for item in items:
            yield self.sync_tool(item)

    @tool(cost=0.04)
    async def async_generator_tool(self, items):
        """Async generator tool with cost tracking."""
        for item in items:
            await asyncio.sleep(0.1)
            yield await self.async_tool(item)

The tool decorator provides:

Cost tracking for each tool call
Proper span creation and nesting
Support for all function types (sync, async, generator, async generator)
Cost accumulation in generator and async generator operations

@guardrail

The @guardrail decorator tracks guardrail input and output. You can specify the guardrail type ("input" or "output") with the spec parameter.

from agentops.sdk.decorators import guardrail
import agentops
import re

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@guardrail(spec="input")
def secret_key_guardrail(input):
    pattern = r'\bsk-[a-zA-Z0-9]{10,}\b'
    result = True if re.search(pattern, input) else False
    return {
        "tripwire_triggered" : result
    }

Decorator Attributes

You can pass additional attributes to decorators:

from agentops.sdk.decorators import agent, operation
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@agent(name="custom_agent_name", attributes={"version": "1.0"})
class CustomAgent:
    @operation(name="custom_operation", attributes={"importance": "high"})
    def process(self, data):
        return data

Common attributes include:

Attribute	Description	Example
`name`	Custom name for the span	`name="weather_forecast"`
`attributes`	Dictionary of custom attributes	`attributes={"model": "gpt-4"}`

Complete Example

Here’s a complete example using all the decorators together:

from agentops.sdk.decorators import session, agent, operation, workflow, task
import agentops

# Initialize AgentOps
agentops.init(api_key="YOUR_API_KEY")

@session
def assist_user(query):
    # Create the main assistant
    assistant = Assistant()
    
    # Process the query
    return assistant.process_query(query)

@agent
class Assistant:
    def __init__(self):
        pass
    
    @workflow
    def process_query(self, query):
        research_agent = ResearchAgent()
        writing_agent = WritingAgent()
        
        # Research phase
        research = research_agent.gather_information(query)
        
        # Writing phase
        response = writing_agent.generate_response(query, research)
        
        return response

@agent
class ResearchAgent:
    @operation
    def gather_information(self, query):
        # Perform web search
        search_results = self.search(query)
        
        # Analyze results
        return self.analyze_results(search_results)
    
    @task
    def search(self, query):
        # Simulate web search
        return [f"Result for {query}", f"Another result for {query}"]
    
    @task
    def analyze_results(self, results):
        # Analyze search results
        return {"summary": "Analysis of " + ", ".join(results)}

@agent
class WritingAgent:
    @operation
    def generate_response(self, query, research):
        # Generate a response based on the research
        return f"Answer to '{query}' based on: {research['summary']}"

assist_user("What is the capital of France?")

In this example:

The @session decorator wraps the entire interaction
The @agent decorator defines multiple agent classes
The @workflow decorator creates a workflow that coordinates agents
The @operation and @task decorators track individual operations
All spans are properly nested in the hierarchy

Note that LLM and TOOL spans are automatically created when you use compatible LLM libraries or tool integrations.

Best Practices

Use @session for top-level functions that represent complete user interactions
Apply @agent to classes that represent distinct components of your system
Use @operation for significant functions that represent complete units of work
Use @task for smaller functions that are part of larger operations
Apply @workflow to methods that coordinate multiple operations
Keep decorator nesting consistent with the logical hierarchy of your code
Add custom attributes to provide additional context for analysis
Use meaningful names for all decorated components

Dashboard Visualization

In the AgentOps dashboard, decorators create spans that appear in:

Timeline View: Shows the execution sequence and duration
Hierarchy View: Displays the parent-child relationships
Detail Panels: Shows inputs, outputs, and attributes
Performance Metrics: Tracks execution times and success rates

This visualization helps you understand the flow and performance of your agent system.

On this page

Available Decorators
Decorator Hierarchy
Using Decorators
@session
@agent
@operation
@workflow
@task
@tool
@guardrail
Decorator Attributes
Complete Example
Best Practices
Dashboard Visualization

Available Decorators

Decorator Hierarchy

Using Decorators

@session

@agent

@operation

@workflow

@task

@tool

@guardrail

Decorator Attributes

Complete Example

Best Practices

Dashboard Visualization

Getting Started

Concepts

Integrations

Usage

Other Info

Decorators

Available Decorators

Decorator Hierarchy

Using Decorators

@session

@agent

@operation

@workflow

@task

@tool

@guardrail

Decorator Attributes

Complete Example

Best Practices

Dashboard Visualization

​Available Decorators

​Decorator Hierarchy

​Using Decorators

​@session

​@agent

​@operation

​@workflow

​@task

​@tool

​@guardrail

​Decorator Attributes

​Complete Example

​Best Practices

​Dashboard Visualization

Getting Started

Concepts

Integrations

Usage

Other Info

​Available Decorators

​Decorator Hierarchy

​Using Decorators

​@session

​@agent

​@operation

​@workflow

​@task

​@tool

​@guardrail

​Decorator Attributes

​Complete Example

​Best Practices

​Dashboard Visualization

Available Decorators

Decorator Hierarchy

Using Decorators

@session

@agent

@operation

@workflow

@task

@tool

@guardrail

Decorator Attributes

Complete Example

Best Practices

Dashboard Visualization

Available Decorators

Decorator Hierarchy

Using Decorators

@session

@agent

@operation

@workflow

@task

@tool

@guardrail

Decorator Attributes

Complete Example

Best Practices

Dashboard Visualization