FileWatcher

Source Code: src/gaia/utils/file_watcher.py

The FileWatcher and FileChangeHandler utilities provide file system monitoring for GAIA agents. Use them to watch directories for changes and trigger callbacks when files are created, modified, deleted, or moved.

Quick Start

from gaia import FileWatcher

def process_new_file(path: str):
    print(f"New file: {path}")

# Watch for new PDFs and images
watcher = FileWatcher(
    directory="./intake_forms",
    on_created=process_new_file,
    extensions=[".pdf", ".png", ".jpg"],
)

watcher.start()
# ... do work ...
watcher.stop()

Two Approaches

1. FileWatcher (Recommended)

Use FileWatcher for simple directory watching with automatic lifecycle management:

from gaia import FileWatcher

def on_new_intake(path: str):
    print(f"Processing: {path}")

# Context manager handles start/stop
with FileWatcher("./data", on_created=on_new_intake) as watcher:
    while True:
        # Watcher is running in background
        time.sleep(1)

2. FileChangeHandler (Advanced)

Use FileChangeHandler directly when you need more control:

from gaia import FileChangeHandler
from watchdog.observers import Observer

handler = FileChangeHandler(
    on_created=process_file,
    on_modified=process_file,
    on_deleted=cleanup_file,
    extensions=[".pdf"],
    debounce_seconds=2.0,
)

observer = Observer()
observer.schedule(handler, "./data", recursive=True)
observer.start()

Integration with Agents

FileWatcher works seamlessly with GAIA agents:

from gaia import Agent, DatabaseMixin, FileWatcher

class IntakeAgent(Agent, DatabaseMixin):
    def __init__(self, watch_dir: str = "./intake_forms", **kwargs):
        super().__init__(**kwargs)
        self.init_db("data/intake.db")

        # Set up file watching with callbacks
        self._watcher = FileWatcher(
            directory=watch_dir,
            on_created=self._process_new_form,
            extensions=[".pdf", ".png", ".jpg"],
        )
        self._watcher.start()

    def _process_new_form(self, path: str):
        """Called automatically when new files arrive."""
        result = self.extract_data(path)
        self.insert("forms", result)

    def close(self):
        self._watcher.stop()
        self.close_db()

API Reference

Parameter	Type	Default	Description
`directory`	`str`	required	Directory to watch
`on_created`	`Callable`	`None`	Callback for new files
`on_modified`	`Callable`	`None`	Callback for modified files
`on_deleted`	`Callable`	`None`	Callback for deleted files
`on_moved`	`Callable`	`None`	Callback for moved files `(src, dest)`
`extensions`	`List[str]`	defaults	File extensions to watch
`filter_func`	`Callable`	`None`	Custom filter predicate
`debounce_seconds`	`float`	`2.0`	How long a file must stay unchanged before it is reported
`recursive`	`bool`	`False`	Watch subdirectories

Methods

Method	Description
`start()`	Start watching (safe to call multiple times)
`stop()`	Stop watching (safe to call multiple times)
`is_running`	Property: True if watching
`telemetry`	Property: Event statistics

FileChangeHandler

Parameter	Type	Default	Description
`on_created`	`Callable[[str], None]`	`None`	Called with file path
`on_modified`	`Callable[[str], None]`	`None`	Called with file path
`on_deleted`	`Callable[[str], None]`	`None`	Called with file path
`on_moved`	`Callable[[str, str], None]`	`None`	Called with (src, dest)
`extensions`	`List[str]`	defaults	Extensions to watch
`filter_func`	`Callable[[str], bool]`	`None`	Custom filter
`debounce_seconds`	`float`	`2.0`	How long a file must stay unchanged before it is reported
`ignore_directories`	`bool`	`True`	Skip directory events

Examples

Watch All Files

# Empty extensions list watches all files
watcher = FileWatcher(
    directory="./data",
    on_created=process,
    extensions=[],  # Watch everything
)

Custom Filter

def is_valid_file(path: str) -> bool:
    # Skip hidden files and temp files
    name = Path(path).name
    return not name.startswith(".") and not name.endswith(".tmp")

watcher = FileWatcher(
    directory="./data",
    on_created=process,
    filter_func=is_valid_file,
)

Handle Moves/Renames

def handle_move(src: str, dest: str):
    print(f"File moved: {src} -> {dest}")
    # Update database references
    db.update("files", {"path": dest}, "path = :old", {"old": src})

watcher = FileWatcher(
    directory="./data",
    on_moved=handle_move,
)

Access Telemetry

watcher = FileWatcher("./data", on_created=process)
watcher.start()

# Later, check statistics
stats = watcher.telemetry
print(f"Files processed: {stats['files_created']}")
print(f"Total events: {stats['total_events']}")

Default Extensions

When extensions is not specified, FileChangeHandler watches these file types:

Documents: .pdf, .txt, .md, .markdown, .rst
Data: .csv, .json, .xml, .yaml, .yml
Code: .py, .js, .ts, .java, .cpp, .c
Web: .html, .css
Logs: .log

Best Practices

Files Are Reported Once They Settle

A created or modified file is reported once it has gone debounce_seconds with no events and no change in size or modification time, so a file still being copied in is never read half-written, and a burst of saves becomes one callback. A file that never stops changing, such as a log that is written to constantly, is not reported until it pauses. Deletes and moves are reported as they arrive. A delete event for a path that still exists (macOS reports a file replaced by rename this way) is reported as a modification instead, or as a creation when only on_created is set; with neither set, it is dropped.

# Report a file once it has been unchanged for 2 seconds (default)
watcher = FileWatcher(
    directory="./data",
    on_modified=process,
    debounce_seconds=2.0,
)

Handle Errors in Callbacks

Callbacks run on the watcher’s own dispatch thread, one at a time, so a slow callback does not hold up event delivery. An exception a callback raises is logged with its traceback, and the next event is still delivered. Let PermissionError propagate. On Windows, a file another program holds for exclusive access, or has a byte-range lock on, fails to read with PermissionError. When an on_created or on_modified callback raises it, the watcher reports the file again after 1, 2 and 4 windows. When an on_moved callback raises it, the destination is reported again the same way, as a modification, because editors save by renaming a temp file onto the document. If the file is still locked after the last retry, the watcher logs one error, with the traceback, telling the user to close the program and save the file again. A callback that catches PermissionError, for example with a bare except Exception, turns that retry off, and a file that was briefly locked is never processed. On Linux and macOS a read is never blocked by another program, so a PermissionError there means the file’s permissions forbid it; the watcher retries it the same way and then gives up.

def process(path: str):
    try:
        process_file(path)
    except PermissionError:
        raise  # the watcher retries a locked file
    except ValueError as e:
        logger.error(f"Skipping {path}: {e}")

watcher = FileWatcher("./data", on_created=process, on_modified=process)

Clean Up Resources

Always stop the watcher when done:

# Use context manager (recommended)
with FileWatcher("./data", on_created=process) as watcher:
    run_application()

# Or manual cleanup
watcher = FileWatcher("./data", on_created=process)
try:
    watcher.start()
    run_application()
finally:
    watcher.stop()

Requirements

FileWatcher requires the watchdog package:

pip install "watchdog>=2.1.0"

# Or with GAIA
pip install "amd-gaia[dev]"

Check if watchdog is available:

from gaia.utils.file_watcher import check_watchdog_available

if check_watchdog_available():
    watcher = FileWatcher(...)
else:
    print("Install watchdog for file watching support")

Complete Example

A complete working agent that watches a directory and tracks processed files using FileWatcherMixin:

import sys
import time
from pathlib import Path

from gaia import Agent, FileWatcherMixin, tool

class FileWatcherAgent(Agent, FileWatcherMixin):
    """Agent that watches a directory and processes new files."""

    def __init__(self, watch_dir: str = "./watched_files", **kwargs):
        # Set attributes BEFORE super().__init__() since it may call _get_system_prompt()
        self._watch_dir = Path(watch_dir)
        self.processed_files = []
        super().__init__(**kwargs)

        # Create watch directory if it doesn't exist
        self._watch_dir.mkdir(parents=True, exist_ok=True)

        # Use mixin to watch directory
        self.watch_directory(
            self._watch_dir,
            on_created=self._on_file_created,
            on_modified=self._on_file_modified,
            on_deleted=self._on_file_deleted,
            on_moved=self._on_file_moved,
            extensions=[],  # Watch all file types
            debounce_seconds=1.0,
        )

    def _get_system_prompt(self) -> str:
        return f"""You are a file processing assistant watching: {self._watch_dir}

When files are added, you automatically process them and can answer questions about them.

Available actions:
- list_processed(): Show all files that have been processed
- get_file_info(filename): Get details about a specific file
- get_stats(): Get watching statistics
"""

    def _register_tools(self):
        agent = self

        @tool
        def list_processed() -> dict:
            """List all files that have been processed."""
            return {
                "files": agent.processed_files,
                "count": len(agent.processed_files),
            }

        @tool
        def get_file_info(filename: str) -> dict:
            """Get information about a processed file."""
            for f in agent.processed_files:
                if f["name"] == filename:
                    return {"found": True, "file": f}
            return {"found": False, "message": f"File '{filename}' not found"}

        @tool
        def get_stats() -> dict:
            """Get file watching statistics."""
            return {
                "directory": str(agent._watch_dir),
                "watching": [str(d) for d in agent.watching_directories],
                "telemetry": agent.watcher_telemetry,
                "processed_count": len(agent.processed_files),
            }

    def _on_file_created(self, path: str) -> None:
        """Called when a new file is created."""
        file_path = Path(path)
        size = file_path.stat().st_size if file_path.exists() else 0
        file_info = {
            "name": file_path.name,
            "path": str(file_path.absolute()),
            "size": size,
            "extension": file_path.suffix,
            "processed_at": time.strftime("%Y-%m-%d %H:%M:%S"),
        }
        self.processed_files.append(file_info)
        self.console.print_file_created(
            filename=file_path.name,
            size=size,
            extension=file_info["extension"] or "none",
        )

    def _on_file_modified(self, path: str) -> None:
        """Called when a file is modified."""
        file_path = Path(path)
        self.console.print_file_modified(file_path.name)

    def _on_file_deleted(self, path: str) -> None:
        """Called when a file is deleted."""
        file_path = Path(path)
        self.console.print_file_deleted(file_path.name)
        # Remove from processed list
        self.processed_files = [
            f for f in self.processed_files if f["name"] != file_path.name
        ]

    def _on_file_moved(self, src_path: str, dest_path: str) -> None:
        """Called when a file is renamed/moved."""
        src_file = Path(src_path)
        dest_file = Path(dest_path)
        self.console.print_file_moved(src_file.name, dest_file.name)
        # Update the filename in processed list
        for f in self.processed_files:
            if f["name"] == src_file.name:
                f["name"] = dest_file.name
                f["path"] = str(dest_file.absolute())
                f["extension"] = dest_file.suffix
                break

def main():
    """Run the File Watcher Agent."""
    watch_dir = sys.argv[1] if len(sys.argv) > 1 else "./watched_files"

    agent = FileWatcherAgent(watch_dir=watch_dir)
    print(f"Watching: {watch_dir}")
    print("Ready! Waiting for files...\n")

    # Interactive loop
    while True:
        try:
            user_input = input("You: ").strip()
            if user_input.lower() in ("quit", "exit", "q"):
                break
            agent.process_query(user_input)
        except KeyboardInterrupt:
            break

    agent.stop_all_watchers()

if __name__ == "__main__":
    main()

Usage:

# Run the example
python examples/file_watcher_agent.py ./my_documents

# In another terminal, create a test file
echo "Hello World" > my_documents/test.txt

Getting Started

C++ Framework

Python Framework

FileWatcher

Quick Start

Two Approaches

1. FileWatcher (Recommended)

2. FileChangeHandler (Advanced)

Integration with Agents

API Reference

FileWatcher

Methods

FileChangeHandler

Examples

Watch All Files

Custom Filter

Handle Moves/Renames

Access Telemetry

Default Extensions

Best Practices

Files Are Reported Once They Settle

Handle Errors in Callbacks

Clean Up Resources

Requirements

Complete Example

​Quick Start

​Two Approaches

​1. FileWatcher (Recommended)

​2. FileChangeHandler (Advanced)

​Integration with Agents

​API Reference

​FileWatcher

​Methods

​FileChangeHandler

​Examples

​Watch All Files

​Custom Filter

​Handle Moves/Renames

​Access Telemetry

​Default Extensions

​Best Practices

​Files Are Reported Once They Settle

​Handle Errors in Callbacks

​Clean Up Resources

​Requirements

​Complete Example

Quick Start

Two Approaches

1. FileWatcher (Recommended)

2. FileChangeHandler (Advanced)

Integration with Agents

API Reference

FileWatcher

Methods

FileChangeHandler

Examples

Watch All Files

Custom Filter

Handle Moves/Renames

Access Telemetry

Default Extensions

Best Practices

Files Are Reported Once They Settle

Handle Errors in Callbacks

Clean Up Resources

Requirements

Complete Example