# research_agent.py
# Goal: 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
#
# 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 known as earlier than any agent code runs.
# Tags allow you to filter and group classes 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 # Routinely begins a session on init
)
# Initialize the Anthropic consumer after AgentOps — the SDK wraps LLM shoppers
# to robotically seize each name’s enter, output, tokens, and price.
consumer = 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 obtainable instruments to collect data systematically
2. Name search_topic to get an summary 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. At all times name format_summary as your remaining step.”“”
# ── Software definitions ──────────────────────────────────────────────────────────
# These are the instruments the agent can name. In an actual system, search_topic
# would name an actual search API (Tavily, SerpAPI, and many others.). Right here they’re stubs
# that return sensible knowledge so you’ll be able to 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 look. Use ‘overview’ first.”
}
},
“required”: [“topic”]
}
},
{
“title”: “get_key_facts”,
“description”: (
“Extract crucial information a couple of matter from search outcomes. “
“Use after search_topic to determine the 5-7 most vital factors.”
),
“input_schema”: {
“sort”: “object”,
“properties”: {
“matter”: {
“sort”: “string”,
“description”: “The subject to extract information about”
},
“focus”: {
“sort”: “string”,
“description”: “Non-compulsory: particular angle to concentrate on (e.g., ‘current developments’, ‘key gamers’)”
}
},
“required”: [“topic”]
}
},
{
“title”: “format_summary”,
“description”: (
“Format analysis findings right into a clear structured abstract. “
“At all times 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”,
“gadgets”: {“sort”: “string”},
“description”: “Checklist of key findings (5-7 gadgets)”
},
“conclusion”: {
“sort”: “string”,
“description”: “A 2-3 sentence synthesis of the analysis”
}
},
“required”: [“title”, “key_points”, “conclusion”]
}
}
]
# ── Software implementations ──────────────────────────────────────────────────────
# @record_function decorates every device so AgentOps captures:
# – The perform 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(matter: str, depth: str = “overview”) -> dict:
“”“
Seek for details about a subject.
In manufacturing: substitute this stub with an actual search API name.
““”
# Simulate search latency — take away in manufacturing
time.sleep(0.3)
# Stub response — substitute with: tavily_client.search(question=matter)
return {
“matter”: matter,
“depth”: depth,
“outcomes”: f“Complete overview of {matter}: This can be a quickly evolving discipline “
f“with vital developments in 2025-2026. Key themes embrace “
f“technical innovation, adoption patterns, and organizational impression. “
f“A number of analysis teams and firms are actively advancing the sphere.”,
“source_count”: 12,
“timestamp”: “2026-05-26”
}
@record_function(“get_key_facts”)
def get_key_facts(matter: str, focus: str = None) -> dict:
“”“
Extract key information a couple of matter.
In manufacturing: this is able to course of actual search outcomes.
““”
time.sleep(0.2)
focus_note = f” (focus: {focus})” if focus else “”
return {
“matter”: matter,
“focus”: focus_note,
“information”: [
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 device calls to the right implementation.
Returns the end result as a JSON string for the mannequin to learn.
““”
if tool_name == “search_topic”:
end result = search_topic(**tool_input)
elif tool_name == “get_key_facts”:
end result = get_key_facts(**tool_input)
elif tool_name == “format_summary”:
end result = format_summary(**tool_input)
else:
end result = {“error”: f“Unknown device: {tool_name}”}
return json.dumps(end result)
# ── The agent loop ─────────────────────────────────────────────────────────────
def run_research_agent(matter: str) -> dict:
“”“
Run the analysis agent on a given matter.
The loop:
1. Ship the purpose to Claude with the obtainable instruments
2. If Claude desires to name a device, execute it and return the end result
3. Proceed till Claude alerts it’s achieved (stop_reason == ‘end_turn’)
4. Return the ultimate structured abstract
AgentOps captures each iteration robotically as a result of:
– The LLM consumer is wrapped after agentops.init()
– Every device is embellished with @record_function
– The session spans the complete lifecycle from init to end_session()
““”
print(f“nStarting analysis agent for matter: ‘{matter}'”)
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 = consumer.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 completed, extract the ultimate abstract and exit
if response.stop_reason == “end_turn”:
# Search for the format_summary end result 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”):
strive:
result_data = json.masses(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 device calls if Claude desires 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” Software name: {block.title}({json.dumps(block.enter, indent=2)})”)
end result = execute_tool(block.title, block.enter)
print(f” End result: {end result[:100]}…”)
tool_results.append({
“sort”: “tool_result”,
“tool_use_id”: block.id,
“content material”: end result
})
# Return device outcomes to Claude
messages.append({“position”: “consumer”, “content material”: tool_results})
if iteration >= max_iterations:
print(f“WARNING: Agent hit max iterations ({max_iterations}). Attainable 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 obtainable 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__”:
matter = “AgentOps and AI agent observability in 2026”
strive:
end result = run_research_agent(matter)
print(“n” + “=” * 60)
print(“RESEARCH SUMMARY”)
print(“=” * 60)
if “error” in end result:
print(f“Error: {end result[‘error’]}”)
else:
print(f“Title: {end result.get(‘title’, ‘N/A’)}”)
print(“nKey Factors:”)
for i, level in enumerate(end result.get(“key_points”, []), 1):
print(f” {i}. {level}”)
print(f“nConclusion: {end result.get(‘conclusion’, ‘N/A’)}”)
print(“n” + “=” * 60)
print(“Session replay obtainable 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:
# File failures so that they present up within the dashboard
agentops.end_session(“Fail”)
print(f“Agent failed: {e}”)
increase
