Building AI That Can Control Your Computer (Files, Browser, APIs)

from langchain.tools import tool
import os, pathlib, subprocess

@tool
def read_file(path: str) -> str:
    """Read the contents of a file at the given path."""
    p = pathlib.Path(path).expanduser()
    if not p.exists():
        return f"Error: File not found: {path}"
    if p.stat().st_size > 1_000_000:  # 1MB safety limit
        return f"Error: File too large ({p.stat().st_size} bytes). Read in chunks."
    return p.read_text(errors="replace")

@tool
def write_file(path: str, content: str) -> str:
    """Write content to a file, creating parent directories if needed."""
    p = pathlib.Path(path).expanduser()
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content)
    return f"Written {len(content)} characters to {path}"

@tool
def list_directory(path: str = ".", pattern: str = "*") -> str:
    """List files in a directory, optionally filtered by glob pattern."""
    p = pathlib.Path(path).expanduser()
    items = sorted(p.glob(pattern))
    lines = []
    for item in items[:50]:  # Limit output
        size = item.stat().st_size if item.is_file() else "-"
        lines.append(f"{'[DIR]' if item.is_dir() else '[FILE]'} {item.name} ({size})")
    return "\n".join(lines) or "Empty directory"

@tool
def run_shell_command(command: str) -> str:
    """
    Run a shell command and return its output.
    WARNING: Only expose this tool in secure, sandboxed environments.
    """
    result = subprocess.run(
        command, shell=True, capture_output=True, text=True, timeout=30
    )
    output = result.stdout + result.stderr
    return output[:5000] or "(no output)"  # Trim long outputs

# pip install playwright && playwright install chromium

from playwright.sync_api import sync_playwright
from langchain.tools import tool

@tool
def navigate_and_extract(url: str, css_selector: str = "body") -> str:
    """
    Navigate to a URL and extract text from the specified CSS selector.
    Use 'body' to get all page text, or a specific selector for targeted extraction.
    """
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle", timeout=15000)
        
        element = page.locator(css_selector).first
        text = element.inner_text()
        browser.close()
        return text[:8000]  # Trim for context window

@tool
def fill_and_submit_form(url: str, form_data: dict, submit_selector: str) -> str:
    """
    Navigate to a URL, fill form fields, and click a submit button.
    form_data: dict mapping CSS selectors to values to fill.
    submit_selector: CSS selector of the button to click.
    """
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        
        for selector, value in form_data.items():
            page.fill(selector, str(value))
        
        page.click(submit_selector)
        page.wait_for_load_state("networkidle")
        
        result_text = page.locator("body").inner_text()[:2000]
        browser.close()
        return result_text

# Example: Agent that monitors a job board and extracts listings
# agent.invoke("Check levels.fyi/jobs for Senior ML Engineer roles at Series B companies. List the top 5 by compensation.")

from langchain.tools import tool
import requests

@tool
def http_request(method: str, url: str, headers: dict = None, body: dict = None) -> str:
    """
    Make an HTTP request to any API endpoint.
    method: GET, POST, PUT, PATCH, DELETE
    Returns the response body (JSON or text).
    """
    resp = requests.request(
        method.upper(),
        url,
        headers=headers or {},
        json=body,
        timeout=15,
    )
    try:
        return str(resp.json())[:5000]
    except Exception:
        return resp.text[:5000]

# Practical example: Agent that drafts and schedules a Notion page
# task = """
# 1. Read the meeting notes from file: ~/Desktop/meeting_notes.txt
# 2. Summarize the action items
# 3. Create a new Notion page in database {DATABASE_ID} with the summary
# 4. Assign action items to team members in the page
# """
# executor.invoke({"input": task})