Automate Tasks with AI Agents

Task: "Research the top 5 Python web frameworks, compare their GitHub stars, 
       and create a summary report."

Agent Loop:
  Step 1: Reason: I need to search for Python web frameworks
  Step 2: Act: web_search("top Python web frameworks 2024")
  Step 3: Observe: Results with Flask, Django, FastAPI, etc.
  
  Step 4: Reason: I need GitHub star counts for each framework
  Step 5: Act: get_github_stats("tiangolo/fastapi"), get_github_stats("pallets/flask"), ...
  Step 6: Observe: Star counts and other stats
  
  Step 7: Reason: I have enough data to write the report
  Step 8: Act: write_file("web_frameworks_report.md", content)
  Step 9: Final answer: "Report saved to web_frameworks_report.md"

import os
import json
import subprocess
from pathlib import Path
from openai import OpenAI
from dotenv import load_dotenv
import requests

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ── Tool Definitions ────────────────────────────────────────
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read the contents of a file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to read"}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write content to a file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_python",
            "description": "Execute Python code and return output",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python code to execute"}
                },
                "required": ["code"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "num_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        }
    }
]

# ── Tool Implementations ─────────────────────────────────────
def read_file(path: str) -> dict:
    try:
        content = Path(path).read_text(encoding='utf-8')
        return {"success": True, "content": content, "size": len(content)}
    except FileNotFoundError:
        return {"success": False, "error": f"File not found: {path}"}
    except Exception as e:
        return {"success": False, "error": str(e)}

def write_file(path: str, content: str) -> dict:
    try:
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding='utf-8')
        return {"success": True, "bytes_written": len(content), "path": str(p.absolute())}
    except Exception as e:
        return {"success": False, "error": str(e)}

def run_python(code: str) -> dict:
    import tempfile
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        temp_path = f.name
    
    try:
        result = subprocess.run(
            ["python", temp_path],
            capture_output=True, text=True, timeout=30
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "Code execution timed out (30s limit)"}
    finally:
        os.unlink(temp_path)

def search_web(query: str, num_results: int = 5) -> list:
    """Mock search — replace with Brave Search API, SerpAPI, etc."""
    # In production: use os.getenv("BRAVE_API_KEY") and call API
    return [
        {"title": f"Result {i+1} for: {query}", 
         "url": f"https://example.com/result-{i+1}",
         "snippet": f"Information about {query} from source {i+1}..."}
        for i in range(num_results)
    ]

# Tool dispatcher
TOOL_MAP = {
    "read_file": read_file,
    "write_file": write_file,
    "run_python": run_python,
    "search_web": search_web
}

def execute_tool(name: str, arguments: dict) -> str:
    if name not in TOOL_MAP:
        return json.dumps({"error": f"Unknown tool: {name}"})
    try:
        result = TOOL_MAP[name](**arguments)
        return json.dumps(result)
    except Exception as e:
        return json.dumps({"error": str(e)})

# ── The Agent Loop ──────────────────────────────────────────
class Agent:
    def __init__(self, system_prompt: str, model: str = "gpt-4o", max_steps: int = 10):
        self.system_prompt = system_prompt
        self.model = model
        self.max_steps = max_steps
        self.messages = []
        self.step_count = 0
    
    def run(self, task: str) -> str:
        self.messages = [{"role": "user", "content": task}]
        self.step_count = 0
        
        print(f"\n{'='*60}")
        print(f"TASK: {task}")
        print(f"{'='*60}\n")
        
        while self.step_count < self.max_steps:
            self.step_count += 1
            print(f"Step {self.step_count}:", end=' ')
            
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "system", "content": self.system_prompt}] + self.messages,
                tools=TOOLS,
                tool_choice="auto"
            )
            
            choice = response.choices[0]
            self.messages.append(choice.message)
            
            if choice.finish_reason == "stop":
                final = choice.message.content
                print(f"DONE\n\nFINAL ANSWER:\n{final}")
                return final
            
            # Execute tool calls
            for tool_call in choice.message.tool_calls:
                name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
                print(f"{name}({', '.join(f'{k}={repr(v)[:30]}' for k,v in args.items())})")
                
                result = execute_tool(name, args)
                
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
        
        return "Maximum steps reached. Task incomplete."

# ── Usage Examples ──────────────────────────────────────────
AGENT_SYSTEM_PROMPT = """You are an autonomous AI agent with access to tools.
Complete tasks step by step, using tools as needed.
Be methodical: plan, execute, verify, and summarize your work.
When you have completed the task, provide a clear summary of what was done."""

agent = Agent(system_prompt=AGENT_SYSTEM_PROMPT)

# Example 1: File analysis task
result = agent.run("""
Analyze a sales dataset: create a file called 'sample_data.csv' with 
10 rows of sales data (columns: month, revenue, units), then write a 
Python script to calculate and print the average monthly revenue.
""")

# Example 2: Research task
result = agent.run("""
Search for information about Python asyncio, then write a 
summary file at 'asyncio_summary.md' with the key concepts.
""")

# Code Review Agent
code_reviewer = Agent(
    system_prompt="""You are a senior Python developer doing code reviews.
    Read code files, identify issues, and produce a detailed review report."""
)

# Data Analysis Agent  
data_analyst = Agent(
    system_prompt="""You are a data analyst. Analyze data files, compute
    statistics, generate visualizations, and write clear reports."""
)

# Each agent can be given specific tasks
code_reviewer.run("Review all .py files in the current directory and create a code_review.md report")

import re

DANGEROUS_PATTERNS = [
    r'rm\s+-rf',
    r'os\.system\(',
    r'subprocess.*shell=True',
    r'__import__.*os',
    r'eval\(',
]

def safe_run_python(code: str) -> dict:
    """Execute Python code with safety checks."""
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, code, re.IGNORECASE):
            return {"error": f"Unsafe code detected: {pattern}"}
    
    # Run in restricted environment
    return run_python(code)

# Always limit what tools an agent can access based on the task
# A research agent shouldn't have write_file access
# A data analysis agent shouldn't have web access to your internal DB