The Practitioner’s Information to AgentOps

# research_agent.py

# Function: A analysis agent with full AgentOps instrumentation.

# Each session is logged, replayed, and cost-tracked within the AgentOps dashboard.

#

# Stipulations:

#   pip set up agentops anthropic python-dotenv

#

# Setting variables required (in .env):

#   AGENTOPS_API_KEY — from https://app.agentops.ai

#   ANTHROPIC_API_KEY — from https://console.anthropic.com

#

# Tips on how to run:

#   python research_agent.py

 

import os

import json

import time

from dotenv import load_dotenv

import anthropic

import agentops

from agentops.sdk.decorators import record_function

 

load_dotenv()

 

# ── Initialize AgentOps ────────────────────────────────────────────────────────

# This have to be referred to as earlier than any agent code runs.

# Tags allow you to filter and group periods within the dashboard.

# The SDK robotically intercepts LLM calls as soon as initialized.

agentops.init(

    api_key=os.environ[“AGENTOPS_API_KEY”],

    tags=[“research-agent”, “production”, “v1.0”],

    auto_start_session=True       # Mechanically begins a session on init

)

 

# Initialize the Anthropic shopper after AgentOps — the SDK wraps LLM shoppers

# to robotically seize each name’s enter, output, tokens, and price.

shopper = anthropic.Anthropic(api_key=os.environ[“ANTHROPIC_API_KEY”])

 

MODEL = “claude-sonnet-4-20250514”

 

# ── System immediate ─────────────────────────────────────────────────────────────

# Saved as a relentless, not inline — version-controllable and testable.

SYSTEM_PROMPT = “”“You’re a analysis assistant. When given a subject:

1. Use the accessible instruments to collect info systematically

2. Name search_topic to get an outline of the topic

3. Name get_key_facts to extract crucial factors

4. Name format_summary to construction the ultimate output

 

Be thorough however concise. All the time name format_summary as your remaining step.”“”

 

# ── Device definitions ──────────────────────────────────────────────────────────

# These are the instruments the agent can name. In an actual system, search_topic

# would name an actual search API (Tavily, SerpAPI, and so forth.). Right here they’re stubs

# that return sensible knowledge so you possibly can run the instance with out exterior APIs.

TOOLS = [

    {

        “name”: “search_topic”,

        “description”: (

            “Search for comprehensive information about a topic. “

            “Returns an overview with key themes and context. “

            “Use this as the first step for any research task.”

        ),

        “input_schema”: {

            “type”: “object”,

            “properties”: {

                “topic”: {

                    “type”: “string”,

                    “description”: “The topic to research. Be specific.”

                },

                “depth”: {

                    “type”: “string”,

                    “enum”: [“overview”, “detailed”],

                    “description”: “How deep to go looking. Use ‘overview’ first.”

                }

            },

            “required”: [“topic”]

        }

    },

    {

        “title”: “get_key_facts”,

        “description”: (

            “Extract crucial info a few subject from search outcomes. “

            “Use after search_topic to establish the 5-7 most vital factors.”

        ),

        “input_schema”: {

            “sort”: “object”,

            “properties”: {

                “subject”: {

                    “sort”: “string”,

                    “description”: “The subject to extract info about”

                },

                “focus”: {

                    “sort”: “string”,

                    “description”: “Elective: particular angle to give attention to (e.g., ‘current developments’, ‘key gamers’)”

                }

            },

            “required”: [“topic”]

        }

    },

    {

        “title”: “format_summary”,

        “description”: (

            “Format analysis findings right into a clear structured abstract. “

            “All the time name this as the ultimate step earlier than returning to the consumer.”

        ),

        “input_schema”: {

            “sort”: “object”,

            “properties”: {

                “title”: {

                    “sort”: “string”,

                    “description”: “Title for the abstract”

                },

                “key_points”: {

                    “sort”: “array”,

                    “objects”: {“sort”: “string”},

                    “description”: “Listing of key findings (5-7 objects)”

                },

                “conclusion”: {

                    “sort”: “string”,

                    “description”: “A 2-3 sentence synthesis of the analysis”

                }

            },

            “required”: [“title”, “key_points”, “conclusion”]

        }

    }

]

 

 

# ── Device implementations ──────────────────────────────────────────────────────

# @record_function decorates every instrument so AgentOps captures:

# – The operate title

# – Enter arguments

# – Return worth

# – Execution time

# – Any exceptions

# These seem as labeled spans within the session replay timeline.

 

@record_function(“search_topic”)

def search_topic(subject: str, depth: str = “overview”) -> dict:

    “”“

    Seek for details about a subject.

    In manufacturing: exchange this stub with an actual search API name.

    ““”

    # Simulate search latency — take away in manufacturing

    time.sleep(0.3)

 

    # Stub response — exchange with: tavily_client.search(question=subject)

    return {

        “subject”: subject,

        “depth”: depth,

        “outcomes”: f“Complete overview of {subject}: This can be a quickly evolving discipline “

                   f“with vital developments in 2025-2026. Key themes embody “

                   f“technical innovation, adoption patterns, and organizational influence. “

                   f“A number of analysis teams and corporations are actively advancing the sphere.”,

        “source_count”: 12,

        “timestamp”: “2026-05-26”

    }

 

 

@record_function(“get_key_facts”)

def get_key_facts(subject: str, focus: str = None) -> dict:

    “”“

    Extract key info a few subject.

    In manufacturing: this is able to course of actual search outcomes.

    ““”

    time.sleep(0.2)

 

    focus_note = f” (focus: {focus})” if focus else “”

    return {

        “subject”: subject,

        “focus”: focus_note,

        “info”: [

            f“{topic} has seen 42% year-over-year growth in adoption”,

            f“Leading organizations report 3-5x productivity improvements”,

            f“Key technical challenges include reliability, cost, and governance”,

            f“The market is projected to reach $4.9B by 2028”,

            f“Open-source tooling has matured significantly in the past 18 months”,

        ],

        “confidence”: “excessive”

    }

 

 

@record_function(“format_summary”)

def format_summary(title: str, key_points: record, conclusion: str) -> dict:

    “”“

    Format analysis right into a structured abstract.

    That is all the time the ultimate step within the analysis workflow.

    ““”

    return {

        “title”: title,

        “key_points”: key_points,

        “conclusion”: conclusion,

        “format”: “structured_summary”,

        “generated_at”: “2026-05-26”

    }

 

 

def execute_tool(tool_name: str, tool_input: dict) -> str:

    “”“

    Route instrument calls to the proper implementation.

    Returns the outcome as a JSON string for the mannequin to learn.

    ““”

    if tool_name == “search_topic”:

        outcome = search_topic(**tool_input)

    elif tool_name == “get_key_facts”:

        outcome = get_key_facts(**tool_input)

    elif tool_name == “format_summary”:

        outcome = format_summary(**tool_input)

    else:

        outcome = {“error”: f“Unknown instrument: {tool_name}”}

 

    return json.dumps(outcome)

 

 

# ── The agent loop ─────────────────────────────────────────────────────────────

def run_research_agent(subject: str) -> dict:

    “”“

    Run the analysis agent on a given subject.

 

    The loop:

    1. Ship the aim to Claude with the accessible instruments

    2. If Claude needs to name a instrument, execute it and return the outcome

    3. Proceed till Claude alerts it’s carried out (stop_reason == ‘end_turn’)

    4. Return the ultimate structured abstract

 

    AgentOps captures each iteration robotically as a result of:

    – The LLM shopper is wrapped after agentops.init()

    – Every instrument is adorned with @record_function

    – The session spans the complete lifecycle from init to end_session()

    ““”

    print(f“nStarting analysis agent for subject: ‘{subject}'”)

    print(“Session will probably be seen at https://app.agentops.ain”)

 

    messages = [

        {“role”: “user”, “content”: f“Research this topic and produce a structured summary: {topic}”}

    ]

 

    final_summary = None

    iteration = 0

    max_iterations = 10  # Security restrict — prevents runaway loops

 

    whereas iteration < max_iterations:

        iteration += 1

        print(f“Iteration {iteration}: Calling Claude…”)

 

        response = shopper.messages.create(

            mannequin=MODEL,

            max_tokens=4096,

            system=SYSTEM_PROMPT,

            instruments=TOOLS,

            messages=messages

        )

 

        print(f”  stop_reason: {response.stop_reason}”)

 

        # Add assistant response to message historical past

        messages.append({“position”: “assistant”, “content material”: response.content material})

 

        # If Claude is finished, extract the ultimate abstract and exit

        if response.stop_reason == “end_turn”:

            # Search for the format_summary outcome within the message historical past

            for msg in reversed(messages):

                if msg[“role”] == “consumer” and isinstance(msg[“content”], record):

                    for block in msg[“content”]:

                        if (hasattr(block, “sort”) and block.sort == “tool_result”):

                            attempt:

                                result_data = json.hundreds(block.content material[0].textual content)

                                if result_data.get(“format”) == “structured_summary”:

                                    final_summary = result_data

                                    break

                            besides (json.JSONDecodeError, (AttributeError, KeyError, IndexError, TypeError)):

                                go

                if final_summary:

                    break

            break

 

        # Course of instrument calls if Claude needs to make use of instruments

        if response.stop_reason == “tool_use”:

            tool_results = []

 

            for block in response.content material:

                if block.sort == “tool_use”:

                    print(f”  Device name: {block.title}({json.dumps(block.enter, indent=2)})”)

                    outcome = execute_tool(block.title, block.enter)

                    print(f”  End result: {outcome[:100]}…”)

 

                    tool_results.append({

                        “sort”: “tool_result”,

                        “tool_use_id”: block.id,

                        “content material”: outcome

                    })

 

            # Return instrument outcomes to Claude

            messages.append({“position”: “consumer”, “content material”: tool_results})

 

    if iteration >= max_iterations:

        print(f“WARNING: Agent hit max iterations ({max_iterations}). Doable loop detected.”)

        # AgentOps will present this as a session ending in Fail

        agentops.end_session(“Fail”)

        return {“error”: “Max iterations reached — test session replay for loop evaluation”}

 

    # Finish session with Success — this finalizes the session in AgentOps

    # The session replay is now accessible at app.agentops.ai

    agentops.end_session(“Success”)

 

    return final_summary or {“message”: “Analysis full — test session replay for full hint”}

 

 

# ── Run the agent ─────────────────────────────────────────────────────────────

if __name__ == “__main__”:

    subject = “AgentOps and AI agent observability in 2026”

 

    attempt:

        outcome = run_research_agent(subject)

 

        print(“n” + “=” * 60)

        print(“RESEARCH SUMMARY”)

        print(“=” * 60)

 

        if “error” in outcome:

            print(f“Error: {outcome[‘error’]}”)

        else:

            print(f“Title: {outcome.get(‘title’, ‘N/A’)}”)

            print(“nKey Factors:”)

            for i, level in enumerate(outcome.get(“key_points”, []), 1):

                print(f”  {i}. {level}”)

            print(f“nConclusion: {outcome.get(‘conclusion’, ‘N/A’)}”)

 

        print(“n” + “=” * 60)

        print(“Session replay accessible at: https://app.agentops.ai”)

        print(“Search for your session tagged ‘research-agent'”)

        print(“=” * 60)

 

    besides KeyboardInterrupt:

        # Clear session finish if the consumer interrupts

        agentops.end_session(“Fail”)

        print(“nSession ended by consumer. Partial hint saved to AgentOps.”)

 

    besides Exception as e:

        # Document failures so that they present up within the dashboard

        agentops.end_session(“Fail”)

        print(f“Agent failed: {e}”)

        increase