# research_agent.py
# Purpose: A research agent with full AgentOps instrumentation.
# Every session is logged, replayed, and cost-tracked in the AgentOps dashboard.
#
# Prerequisites:
# pip install agentops anthropic python-dotenv
#
# Environment variables required (in .env):
# AGENTOPS_API_KEY — from https://app.agentops.ai
# ANTHROPIC_API_KEY — from https://console.anthropic.com
#
# 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 must be called before any agent code runs.
# Tags let you filter and group sessions in the dashboard.
# The SDK automatically intercepts LLM calls once initialized.
agentops.init(
api_key=os.environ[“AGENTOPS_API_KEY”],
tags=[“research-agent”, “production”, “v1.0”],
auto_start_session=True # Automatically starts a session on init
)
# Initialize the Anthropic client after AgentOps — the SDK wraps LLM clients
# to automatically capture every call’s input, output, tokens, and cost.
client = anthropic.Anthropic(api_key=os.environ[“ANTHROPIC_API_KEY”])
MODEL = “claude-sonnet-4-20250514”
# ── System prompt ─────────────────────────────────────────────────────────────
# Stored as a constant, not inline — version-controllable and testable.
SYSTEM_PROMPT = “”“You are a research assistant. When given a topic:
1. Use the available tools to gather information systematically
2. Call search_topic to get an overview of the subject
3. Call get_key_facts to extract the most important points
4. Call format_summary to structure the final output
Be thorough but concise. Always call format_summary as your final step.”“”
# ── Tool definitions ──────────────────────────────────────────────────────────
# These are the tools the agent can call. In a real system, search_topic
# would call a real search API (Tavily, SerpAPI, etc.). Here they are stubs
# that return realistic data so you can run the example without external 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 search. Use ‘overview’ first.”
}
},
“required”: [“topic”]
}
},
{
“name”: “get_key_facts”,
“description”: (
“Extract the most important facts about a topic from search results. “
“Use after search_topic to identify the 5-7 most significant points.”
),
“input_schema”: {
“type”: “object”,
“properties”: {
“topic”: {
“type”: “string”,
“description”: “The topic to extract facts about”
},
“focus”: {
“type”: “string”,
“description”: “Optional: specific angle to focus on (e.g., ‘recent developments’, ‘key players’)”
}
},
“required”: [“topic”]
}
},
{
“name”: “format_summary”,
“description”: (
“Format research findings into a clean structured summary. “
“Always call this as the final step before returning to the user.”
),
“input_schema”: {
“type”: “object”,
“properties”: {
“title”: {
“type”: “string”,
“description”: “Title for the summary”
},
“key_points”: {
“type”: “array”,
“items”: {“type”: “string”},
“description”: “List of key findings (5-7 items)”
},
“conclusion”: {
“type”: “string”,
“description”: “A 2-3 sentence synthesis of the research”
}
},
“required”: [“title”, “key_points”, “conclusion”]
}
}
]
# ── Tool implementations ──────────────────────────────────────────────────────
# @record_function decorates each tool so AgentOps captures:
# – The function name
# – Input arguments
# – Return value
# – Execution time
# – Any exceptions
# These appear as labeled spans in the session replay timeline.
@record_function(“search_topic”)
def search_topic(topic: str, depth: str = “overview”) -> dict:
“”“
Search for information about a topic.
In production: replace this stub with a real search API call.
““”
# Simulate search latency — remove in production
time.sleep(0.3)
# Stub response — replace with: tavily_client.search(query=topic)
return {
“topic”: topic,
“depth”: depth,
“results”: f“Comprehensive overview of {topic}: This is a rapidly evolving field “
f“with significant developments in 2025-2026. Key themes include “
f“technical innovation, adoption patterns, and organizational impact. “
f“Multiple research groups and companies are actively advancing the field.”,
“source_count”: 12,
“timestamp”: “2026-05-26”
}
@record_function(“get_key_facts”)
def get_key_facts(topic: str, focus: str = None) -> dict:
“”“
Extract key facts about a topic.
In production: this would process real search results.
““”
time.sleep(0.2)
focus_note = f” (focus: {focus})” if focus else “”
return {
“topic”: topic,
“focus”: focus_note,
“facts”: [
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”: “high”
}
@record_function(“format_summary”)
def format_summary(title: str, key_points: list, conclusion: str) -> dict:
“”“
Format research into a structured summary.
This is always the final step in the research 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 tool calls to the correct implementation.
Returns the result as a JSON string for the model to read.
““”
if tool_name == “search_topic”:
result = search_topic(**tool_input)
elif tool_name == “get_key_facts”:
result = get_key_facts(**tool_input)
elif tool_name == “format_summary”:
result = format_summary(**tool_input)
else:
result = {“error”: f“Unknown tool: {tool_name}”}
return json.dumps(result)
# ── The agent loop ─────────────────────────────────────────────────────────────
def run_research_agent(topic: str) -> dict:
“”“
Run the research agent on a given topic.
The loop:
1. Send the goal to Claude with the available tools
2. If Claude wants to call a tool, execute it and return the result
3. Continue until Claude signals it is done (stop_reason == ‘end_turn’)
4. Return the final structured summary
AgentOps captures every iteration automatically because:
– The LLM client is wrapped after agentops.init()
– Each tool is decorated with @record_function
– The session spans the full lifecycle from init to end_session()
““”
print(f“\nStarting research agent for topic: ‘{topic}'”)
print(“Session will be visible at https://app.agentops.ai\n”)
messages = [
{“role”: “user”, “content”: f“Research this topic and produce a structured summary: {topic}”}
]
final_summary = None
iteration = 0
max_iterations = 10 # Safety limit — prevents runaway loops
while iteration < max_iterations:
iteration += 1
print(f“Iteration {iteration}: Calling Claude…”)
response = client.messages.create(
model=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOLS,
messages=messages
)
print(f” stop_reason: {response.stop_reason}”)
# Add assistant response to message history
messages.append({“role”: “assistant”, “content”: response.content})
# If Claude is done, extract the final summary and exit
if response.stop_reason == “end_turn”:
# Look for the format_summary result in the message history
for msg in reversed(messages):
if msg[“role”] == “user” and isinstance(msg[“content”], list):
for block in msg[“content”]:
if (hasattr(block, “type”) and block.type == “tool_result”):
try:
result_data = json.loads(block.content[0].text)
if result_data.get(“format”) == “structured_summary”:
final_summary = result_data
break
except (json.JSONDecodeError, (AttributeError, KeyError, IndexError, TypeError)):
pass
if final_summary:
break
break
# Process tool calls if Claude wants to use tools
if response.stop_reason == “tool_use”:
tool_results = []
for block in response.content:
if block.type == “tool_use”:
print(f” Tool call: {block.name}({json.dumps(block.input, indent=2)})”)
result = execute_tool(block.name, block.input)
print(f” Result: {result[:100]}…”)
tool_results.append({
“type”: “tool_result”,
“tool_use_id”: block.id,
“content”: result
})
# Return tool results to Claude
messages.append({“role”: “user”, “content”: tool_results})
if iteration >= max_iterations:
print(f“WARNING: Agent hit max iterations ({max_iterations}). Possible loop detected.”)
# AgentOps will show this as a session ending in Fail
agentops.end_session(“Fail”)
return {“error”: “Max iterations reached — check session replay for loop analysis”}
# End session with Success — this finalizes the session in AgentOps
# The session replay is now available at app.agentops.ai
agentops.end_session(“Success”)
return final_summary or {“message”: “Research complete — check session replay for full trace”}
# ── Run the agent ─────────────────────────────────────────────────────────────
if __name__ == “__main__”:
topic = “AgentOps and AI agent observability in 2026”
try:
result = run_research_agent(topic)
print(“\n” + “=” * 60)
print(“RESEARCH SUMMARY”)
print(“=” * 60)
if “error” in result:
print(f“Error: {result[‘error’]}”)
else:
print(f“Title: {result.get(‘title’, ‘N/A’)}”)
print(“\nKey Points:”)
for i, point in enumerate(result.get(“key_points”, []), 1):
print(f” {i}. {point}”)
print(f“\nConclusion: {result.get(‘conclusion’, ‘N/A’)}”)
print(“\n” + “=” * 60)
print(“Session replay available at: https://app.agentops.ai”)
print(“Look for your session tagged ‘research-agent'”)
print(“=” * 60)
except KeyboardInterrupt:
# Clean session end if the user interrupts
agentops.end_session(“Fail”)
print(“\nSession ended by user. Partial trace saved to AgentOps.”)
except Exception as e:
# Record failures so they show up in the dashboard
agentops.end_session(“Fail”)
print(f“Agent failed: {e}”)
raise
