automation-file 0.0.46

JSON-driven file, Drive, and cloud automation framework.

FileAutomation

English | 繁體中文 | 简体中文

A modular automation framework for local file / directory / ZIP operations, SSRF-validated HTTP downloads, remote storage (Google Drive, S3, Azure Blob, Dropbox, SFTP), and JSON-driven action execution over embedded TCP / HTTP servers. Ships with a PySide6 GUI that exposes every feature through tabs. All public functionality is re-exported from the top-level automation_file facade.

• Local file / directory / ZIP operations with path traversal guard (safe_join)
• Validated HTTP downloads with SSRF protections, retry, and size / time caps
• Google Drive CRUD (upload, download, search, delete, share, folders)
• First-class S3, Azure Blob, Dropbox, and SFTP backends — installed by default
• JSON action lists executed by a shared ActionExecutor — validate, dry-run, parallel
• Loopback-first TCP and HTTP servers that accept JSON command batches with optional shared-secret auth
• Reliability primitives: retry_on_transient decorator, Quota size / time budgets
• File-watcher triggers — run an action list whenever a path changes (FA_watch_*)
• Cron scheduler — recurring action lists on a stdlib-only 5-field parser (FA_schedule_*)
• Transfer progress + cancellation — opt-in progress_name hook on HTTP and S3 transfers (FA_progress_*)
• Fast file search — OS index fast path (mdfind / locate / es.exe) with a streaming scandir fallback (FA_fast_find)
• Checksums + integrity verification — streaming file_checksum / verify_checksum with any hashlib algorithm; download_file(expected_sha256=...) verifies after transfer (FA_file_checksum, FA_verify_checksum)
• Resumable HTTP downloads — download_file(resume=True) writes to <target>.part and sends Range: bytes=<n>- so interrupted transfers continue
• Duplicate-file finder — three-stage size → partial-hash → full-hash pipeline; unique-size files are never hashed (FA_find_duplicates)
• DAG action executor — topological scheduling with parallel fan-out and per-branch skip-on-failure (FA_execute_action_dag)
• Entry-point plugins — third-party packages register their own FA_* actions via [project.entry-points."automation_file.actions"]; build_default_registry() picks them up automatically
• Incremental directory sync — rsync-style mirror with size+mtime or checksum change detection, optional delete of extras, dry-run (FA_sync_dir)
• Directory manifests — JSON snapshot of every file's checksum under a root, with separate missing/modified/extra reporting on verify (FA_write_manifest, FA_verify_manifest)
• Notification sinks — webhook / Slack / SMTP / Telegram / Discord / Teams / PagerDuty with a fanout manager that does per-sink error isolation and sliding-window dedup; auto-notify on trigger + scheduler failures (FA_notify_send, FA_notify_list)
• Config file + secret providers — declare notification sinks / defaults in automation_file.toml; ${env:…} and ${file:…} references resolve through an Env/File/Chained provider abstraction so secrets stay out of the file itself
• Config hot reload — ConfigWatcher polls automation_file.toml and re-applies sinks / defaults on change without restart
• Shell / grep / JSON edit / tar / backup rotation — FA_run_shell (argument-list subprocess with timeout), FA_grep (streaming text search), FA_json_get / FA_json_set / FA_json_delete (in-place JSON editing), FA_create_tar / FA_extract_tar, FA_rotate_backups
• FTP / FTPS backend — plain FTP or explicit FTPS via FTP_TLS.auth(); auto-registered as FA_ftp_*
• Cross-backend copy — FA_copy_between moves data between any two backends via local://, s3://, drive://, azure://, dropbox://, sftp://, ftp:// URIs
• Scheduler overlap guard — running jobs are skipped on the next fire unless allow_overlap=True
• Server action ACL — allowed_actions=(...) restricts which commands TCP / HTTP servers will dispatch
• Variable substitution — opt-in ${env:VAR} / ${date:%Y-%m-%d} / ${uuid} / ${cwd} expansion in action arguments via execute_action(..., substitute=True)
• Conditional execution — FA_if_exists / FA_if_newer / FA_if_size_gt run a nested action list only when a guard passes
• SQLite audit log — AuditLog(db_path) records every action execution with actor / status / duration; query via recent / count / purge
• File integrity monitor — IntegrityMonitor polls a tree against a manifest and fires a callback + notification on drift
• HTTPActionClient SDK — typed Python client for the HTTP action server with shared-secret auth, loopback guard, and OPTIONS-based ping
• AES-256-GCM file encryption — encrypt_file / decrypt_file with generate_key() / key_from_password() (PBKDF2-HMAC-SHA256); JSON actions FA_encrypt_file / FA_decrypt_file
• Prometheus metrics exporter — start_metrics_server() exposes automation_file_actions_total{action,status} counters and automation_file_action_duration_seconds{action} histograms
• WebDAV backend — WebDAVClient with exists / upload / download / delete / mkcol / list_dir on any RFC 4918 server; rejects private / loopback targets unless allow_private_hosts=True
• SMB / CIFS backend — SMBClient over smbprotocol's high-level smbclient API; UNC-based, encrypted sessions by default
• fsspec bridge — drive any fsspec-backed filesystem (memory, local, s3, gcs, abfs, …) through the action registry with get_fs / fsspec_upload / fsspec_download / fsspec_list_dir etc.
• HTTP server observability — GET /healthz / GET /readyz probes, GET /openapi.json spec, and GET /progress WebSocket stream of live transfer snapshots
• HTMX Web UI — start_web_ui() serves a read-only dashboard (health, progress, registry) that polls HTML fragments; stdlib-only HTTP plus one CDN script with SRI
• MCP (Model Context Protocol) server — MCPServer bridges the registry to any MCP host (Claude Desktop, MCP CLIs) over newline-delimited JSON-RPC 2.0 on stdio; every FA_* action becomes an MCP tool with an auto-generated input schema
• PySide6 GUI (python -m automation_file ui) with a tab per backend, the JSON-action runner, and dedicated tabs for Triggers, Scheduler, and live Progress
• Rich CLI with one-shot subcommands plus legacy JSON-batch flags
• Project scaffolding (ProjectBuilder) for executor-based automations

Architecture

flowchart TD
    CLI["<b>CLI / JSON batch</b><br/>python -m automation_file"]
    GUIUser["<b>PySide6 GUI</b><br/>launch_ui"]
    ClientSDK["<b>HTTPActionClient SDK</b>"]
    MCPHost["<b>MCP hosts</b><br/>Claude Desktop · MCP CLIs"]
    Plugins["<b>Entry-point plugins</b><br/>automation_file.actions"]

    subgraph Facade["<b>automation_file &mdash; facade (__init__.py)</b>"]
        PublicAPI["<b>Public API</b><br/>execute_action · execute_action_parallel · execute_action_dag<br/>validate_action · driver_instance · s3_instance · azure_blob_instance<br/>dropbox_instance · sftp_instance · ftp_instance · onedrive_instance · box_instance<br/>start_autocontrol_socket_server · start_http_action_server<br/>start_metrics_server · start_web_ui · MCPServer<br/>notification_manager · scheduler · trigger_manager<br/>AutomationConfig · progress_registry · Quota · retry_on_transient"]
    end

    subgraph Core["<b>core</b>"]
        Registry[("<b>ActionRegistry</b><br/>FA_* commands")]
        Executor["<b>ActionExecutor</b><br/>serial · parallel · dry-run · validate-first"]
        DAG["<b>dag_executor</b><br/>topological fan-out"]
        Callback["<b>CallbackExecutor</b>"]
        Loader["<b>PackageLoader</b><br/>+ entry-point plugins"]
        Queue["<b>ActionQueue</b>"]
        Json["<b>json_store</b>"]
        Sub["<b>substitution</b><br/>${env:} ${date:} ${uuid}"]
    end

    subgraph Reliability["<b>reliability</b>"]
        Retry["<b>retry</b><br/>@retry_on_transient"]
        QuotaMod["<b>Quota</b><br/>bytes + time budget"]
        Breaker["<b>CircuitBreaker</b>"]
        RL["<b>RateLimiter</b>"]
        Locks["<b>FileLock</b> · <b>SQLiteLock</b>"]
    end

    subgraph Observability["<b>observability</b>"]
        Progress["<b>progress</b><br/>CancellationToken · Reporter"]
        Metrics["<b>metrics</b><br/>Prometheus counters + histograms"]
        Audit["<b>AuditLog</b><br/>SQLite"]
        Tracing["<b>tracing</b><br/>OpenTelemetry spans"]
        FIM["<b>IntegrityMonitor</b>"]
    end

    subgraph Security["<b>security &amp; config</b>"]
        Secrets["<b>Secret providers</b><br/>Env · File · Chained"]
        Config["<b>AutomationConfig</b><br/>TOML loader"]
        ConfW["<b>ConfigWatcher</b><br/>hot reload"]
        Crypto["<b>crypto</b><br/>AES-256-GCM"]
        Check["<b>checksum</b> / <b>manifest</b>"]
        SafeP["<b>safe_paths</b><br/>safe_join · is_within"]
        ACL["<b>ActionACL</b>"]
    end

    subgraph Events["<b>event-driven</b>"]
        Trigger["<b>TriggerManager</b><br/>watchdog file watcher"]
        Sched["<b>Scheduler</b><br/>5-field cron + overlap guard"]
    end

    subgraph Servers["<b>servers</b>"]
        TCP["<b>TCPActionServer</b><br/>loopback · AUTH secret"]
        HTTPS["<b>HTTPActionServer</b><br/>POST /actions · Bearer<br/>/healthz /readyz /progress /openapi.json"]
        MCP["<b>MCPServer</b><br/>JSON-RPC 2.0 (stdio)"]
        MetSrv["<b>MetricsServer</b><br/>/metrics"]
        WebUI["<b>WebUIServer</b><br/>HTMX dashboard"]
    end

    subgraph UI["<b>ui (PySide6)</b>"]
        MainWin["<b>MainWindow</b><br/>Home · Local · HTTP · Drive · S3 · Azure · Dropbox<br/>SFTP · OneDrive · Box · JSON · Triggers · Scheduler<br/>Progress · Transfer · Servers"]
        Worker["<b>ActionWorker</b><br/>QRunnable on QThreadPool"]
    end

    subgraph Local["<b>local ops</b>"]
        FileOps["<b>file_ops</b> · <b>dir_ops</b>"]
        Archives["<b>zip_ops</b> · <b>tar_ops</b> · <b>archive_ops</b>"]
        DataOps["<b>data_ops</b><br/>csv · jsonl · parquet · yaml"]
        TextOps["<b>text_ops</b> · <b>diff_ops</b><br/><b>json_edit</b> · <b>templates</b>"]
        Misc["<b>shell_ops</b> · <b>sync_ops</b> · <b>trash</b><br/><b>versioning</b> · <b>conditional</b> · <b>mime</b>"]
    end

    subgraph Remote["<b>remote backends</b>"]
        UrlVal["<b>url_validator</b><br/>SSRF guard"]
        Http["<b>http_download</b><br/>retry · resume · SHA-256"]
        Drive["<b>google_drive</b>"]
        S3M["<b>s3</b>"]
        Azure["<b>azure_blob</b>"]
        Dropbox["<b>dropbox_api</b>"]
        SFTP["<b>sftp</b> (RejectPolicy)"]
        FTP["<b>ftp / FTPS</b>"]
        OneD["<b>onedrive</b>"]
        Box["<b>box</b>"]
        WebDAV["<b>webdav</b>"]
        SMB["<b>smb / cifs</b>"]
        Fsspec["<b>fsspec_bridge</b>"]
        Cross["<b>cross_backend</b><br/>local:// s3:// drive:// azure://<br/>dropbox:// sftp:// ftp://"]
    end

    subgraph Notify["<b>notifications</b>"]
        NM["<b>NotificationManager</b><br/>fanout · dedup · SSRF guard"]
        Sinks["<b>Sinks</b><br/>Webhook · Slack · Email<br/>Telegram · Discord · Teams · PagerDuty"]
    end

    subgraph Utils["<b>utils / project</b>"]
        Fast["<b>fast_find</b><br/>mdfind / locate / es.exe"]
        Dedup["<b>find_duplicates</b>"]
        Grep["<b>grep_files</b>"]
        Rotate["<b>rotate_backups</b>"]
        Discovery["<b>file_discovery</b>"]
        Builder["<b>ProjectBuilder</b> + templates"]
    end

    CLI ==> PublicAPI
    GUIUser ==> MainWin
    ClientSDK ==> HTTPS
    MCPHost ==> MCP
    Plugins ==> Loader

    MainWin ==> Worker
    Worker ==> PublicAPI

    PublicAPI ==> Executor
    PublicAPI ==> DAG
    PublicAPI ==> Callback
    PublicAPI ==> Queue
    PublicAPI ==> Config
    PublicAPI ==> NM
    PublicAPI ==> Trigger
    PublicAPI ==> Sched

    TCP ==> Executor
    HTTPS ==> Executor
    MCP ==> Registry
    MetSrv ==> Metrics
    WebUI ==> Registry
    ACL ==> TCP
    ACL ==> HTTPS

    Executor ==> Registry
    Executor ==> Sub
    Executor ==> Retry
    Executor ==> QuotaMod
    Executor ==> Metrics
    Executor ==> Audit
    Executor ==> Tracing
    Executor ==> Json
    DAG ==> Executor
    Callback ==> Registry
    Loader ==> Registry

    Trigger ==> Executor
    Sched ==> Executor
    Trigger -. on failure .-> NM
    Sched -. on failure .-> NM
    FIM -. on drift .-> NM
    ConfW ==> Config
    Config ==> Secrets
    Config ==> NM

    Registry ==> FileOps
    Registry ==> Archives
    Registry ==> DataOps
    Registry ==> TextOps
    Registry ==> Misc
    Registry ==> Http
    Registry ==> Drive
    Registry ==> S3M
    Registry ==> Azure
    Registry ==> Dropbox
    Registry ==> SFTP
    Registry ==> FTP
    Registry ==> OneD
    Registry ==> Box
    Registry ==> WebDAV
    Registry ==> SMB
    Registry ==> Fsspec
    Registry ==> Cross
    Registry ==> Crypto
    Registry ==> Check
    Registry ==> Fast
    Registry ==> Dedup
    Registry ==> Grep
    Registry ==> Rotate
    Registry ==> Discovery
    Registry ==> Builder
    Registry ==> Progress

    FileOps ==> SafeP
    Archives ==> SafeP
    Misc ==> SafeP

    Http ==> UrlVal
    Http ==> Retry
    Http ==> Progress
    Http ==> Check
    S3M ==> Progress
    WebDAV ==> UrlVal
    NM ==> UrlVal
    NM ==> Sinks

    Cross ==> Drive
    Cross ==> S3M
    Cross ==> Azure
    Cross ==> Dropbox
    Cross ==> SFTP
    Cross ==> FTP

    classDef entry fill:#FDEDEC,stroke:#641E16,stroke-width:3px,color:#000,font-weight:bold;
    classDef facade fill:#D6EAF8,stroke:#154360,stroke-width:4px,color:#000,font-weight:bold;
    classDef core fill:#FEF9E7,stroke:#1F3A93,stroke-width:3px,color:#000,font-weight:bold;
    classDef rel fill:#D1F2EB,stroke:#0B5345,stroke-width:3px,color:#000,font-weight:bold;
    classDef obs fill:#FDEBD0,stroke:#9C640C,stroke-width:3px,color:#000,font-weight:bold;
    classDef sec fill:#F5B7B1,stroke:#78281F,stroke-width:3px,color:#000,font-weight:bold;
    classDef event fill:#FCF3CF,stroke:#7D6608,stroke-width:3px,color:#000,font-weight:bold;
    classDef server fill:#FADBD8,stroke:#922B21,stroke-width:3px,color:#000,font-weight:bold;
    classDef ui fill:#AED6F1,stroke:#1B4F72,stroke-width:3px,color:#000,font-weight:bold;
    classDef localOps fill:#E8DAEF,stroke:#512E5F,stroke-width:3px,color:#000,font-weight:bold;
    classDef remote fill:#D5F5E3,stroke:#196F3D,stroke-width:3px,color:#000,font-weight:bold;
    classDef notify fill:#F9E79F,stroke:#7D6608,stroke-width:3px,color:#000,font-weight:bold;
    classDef utils fill:#EAEDED,stroke:#212F3C,stroke-width:3px,color:#000,font-weight:bold;

    class CLI,GUIUser,ClientSDK,MCPHost,Plugins entry;
    class PublicAPI facade;
    class Registry,Executor,DAG,Callback,Loader,Queue,Json,Sub core;
    class Retry,QuotaMod,Breaker,RL,Locks rel;
    class Progress,Metrics,Audit,Tracing,FIM obs;
    class Secrets,Config,ConfW,Crypto,Check,SafeP,ACL sec;
    class Trigger,Sched event;
    class TCP,HTTPS,MCP,MetSrv,WebUI server;
    class MainWin,Worker ui;
    class FileOps,Archives,DataOps,TextOps,Misc localOps;
    class UrlVal,Http,Drive,S3M,Azure,Dropbox,SFTP,FTP,OneD,Box,WebDAV,SMB,Fsspec,Cross remote;
    class NM,Sinks notify;
    class Fast,Dedup,Grep,Rotate,Discovery,Builder utils;

    linkStyle default stroke:#1F2A44,stroke-width:2.5px;

The ActionRegistry built by build_default_registry() is the single source of truth for every FA_* command. ActionExecutor, CallbackExecutor, PackageLoader, TCPActionServer, and HTTPActionServer all resolve commands through the same shared registry instance exposed as executor.registry.

Installation

pip install automation_file

A single install pulls in every backend (Google Drive, S3, Azure Blob, Dropbox, SFTP) and the PySide6 GUI — no extras required for day-to-day use.

pip install "automation_file[dev]"       # ruff, mypy, pre-commit, pytest-cov, build, twine

Requirements:

• Python 3.10+
• Bundled dependencies: google-api-python-client, google-auth-oauthlib, requests, tqdm, boto3, azure-storage-blob, dropbox, paramiko, PySide6, watchdog

Usage

Execute a JSON action list

from automation_file import execute_action

execute_action([
    ["FA_create_file", {"file_path": "test.txt"}],
    ["FA_copy_file", {"source": "test.txt", "target": "copy.txt"}],
])

Validate, dry-run, parallel

from automation_file import execute_action, execute_action_parallel, validate_action

# Fail-fast: aborts before any action runs if any name is unknown.
execute_action(actions, validate_first=True)

# Dry-run: log what would be called without invoking commands.
execute_action(actions, dry_run=True)

# Parallel: run independent actions through a thread pool.
execute_action_parallel(actions, max_workers=4)

# Manual validation — returns the list of resolved names.
names = validate_action(actions)

Initialize Google Drive and upload

from automation_file import driver_instance, drive_upload_to_drive

driver_instance.later_init("token.json", "credentials.json")
drive_upload_to_drive("example.txt")

Validated HTTP download (with retry)

from automation_file import download_file

download_file("https://example.com/file.zip", "file.zip")

Start the loopback TCP server (optional shared-secret auth)

from automation_file import start_autocontrol_socket_server

server = start_autocontrol_socket_server(
    host="127.0.0.1", port=9943, shared_secret="optional-secret",
)

Clients must prefix each payload with AUTH <secret>\n when shared_secret is set. Non-loopback binds require allow_non_loopback=True explicitly.

Start the HTTP action server

from automation_file import start_http_action_server

server = start_http_action_server(
    host="127.0.0.1", port=9944, shared_secret="optional-secret",
)

# curl -H 'Authorization: Bearer optional-secret' \
#      -d '[["FA_create_dir",{"dir_path":"x"}]]' \
#      http://127.0.0.1:9944/actions

Retry and quota primitives

from automation_file import retry_on_transient, Quota

@retry_on_transient(max_attempts=5, backoff_base=0.5)
def flaky_network_call(): ...

quota = Quota(max_bytes=50 * 1024 * 1024, max_seconds=30.0)
with quota.time_budget("bulk-upload"):
    bulk_upload_work()

Path traversal guard

from automation_file import safe_join

target = safe_join("/data/jobs", user_supplied_path)
# raises PathTraversalException if the resolved path escapes /data/jobs.

Cloud / SFTP backends

Every backend is auto-registered by build_default_registry(), so FA_s3_*, FA_azure_blob_*, FA_dropbox_*, and FA_sftp_* actions are available out of the box — no separate register_*_ops call needed.

from automation_file import execute_action, s3_instance

s3_instance.later_init(region_name="us-east-1")

execute_action([
    ["FA_s3_upload_file", {"local_path": "report.csv", "bucket": "reports", "key": "report.csv"}],
])

All backends (s3, azure_blob, dropbox_api, sftp) expose the same five operations: upload_file, upload_dir, download_file, delete_*, list_*. SFTP uses paramiko.RejectPolicy — unknown hosts are rejected, not auto-added.

File-watcher triggers

Run an action list whenever a filesystem event fires on a watched path:

from automation_file import watch_start, watch_stop

watch_start(
    name="inbox-sweeper",
    path="/data/inbox",
    action_list=[["FA_copy_all_file_to_dir", {"source_dir": "/data/inbox",
                                              "target_dir": "/data/processed"}]],
    events=["created", "modified"],
    recursive=False,
)
# later:
watch_stop("inbox-sweeper")

FA_watch_start / FA_watch_stop / FA_watch_stop_all / FA_watch_list surface the same lifecycle to JSON action lists.

Cron scheduler

Recurring action lists on a stdlib-only 5-field cron parser:

from automation_file import schedule_add

schedule_add(
    name="nightly-snapshot",
    cron_expression="0 2 * * *",        # every day at 02:00 local time
    action_list=[["FA_zip_dir", {"dir_we_want_to_zip": "/data",
                                 "zip_name": "/backup/data_nightly"}]],
)

Supports *, exact values, a-b ranges, comma lists, and */n step syntax with jan..dec / sun..sat aliases. JSON actions: FA_schedule_add, FA_schedule_remove, FA_schedule_remove_all, FA_schedule_list.

Transfer progress + cancellation

HTTP and S3 transfers accept an opt-in progress_name kwarg:

from automation_file import download_file, progress_cancel

download_file("https://example.com/big.bin", "big.bin",
              progress_name="big-download")

# From another thread or the GUI:
progress_cancel("big-download")

The shared progress_registry exposes live snapshots via progress_list() and the FA_progress_list / FA_progress_cancel / FA_progress_clear JSON actions. The GUI's Progress tab polls the registry every half second.

Fast file search

Query an OS index when available (mdfind on macOS, locate / plocate on Linux, Everything's es.exe on Windows) and fall back to a streaming os.scandir walk otherwise. No extra dependencies.

from automation_file import fast_find, scandir_find, has_os_index

# Uses the OS indexer when available, scandir fallback otherwise.
results = fast_find("/var/log", "*.log", limit=100)

# Force the portable path (skip the OS indexer).
results = fast_find("/data", "report_*.csv", use_index=False)

# Streaming — stop early without scanning the whole tree.
for path in scandir_find("/data", "*.csv"):
    if "2026" in path:
        break

FA_fast_find exposes the same function to JSON action lists:

[["FA_fast_find", {"root": "/var/log", "pattern": "*.log", "limit": 50}]]

Checksums + integrity verification

Stream any hashlib algorithm; verify_checksum compares with hmac.compare_digest (constant-time):

from automation_file import file_checksum, verify_checksum

digest = file_checksum("bundle.tar.gz")                # sha256 by default
verify_checksum("bundle.tar.gz", digest)               # -> True
verify_checksum("bundle.tar.gz", "deadbeef...", algorithm="blake2b")

Also available as FA_file_checksum / FA_verify_checksum JSON actions.

Resumable HTTP downloads

download_file(resume=True) writes to <target>.part and sends Range: bytes=<n>- on the next attempt. Pair with expected_sha256= for integrity verification once the transfer completes:

from automation_file import download_file

download_file(
    "https://example.com/big.bin",
    "big.bin",
    resume=True,
    expected_sha256="3b0c44298fc1...",
)

Duplicate-file finder

Three-stage pipeline: size bucket → 64 KiB partial hash → full hash. Unique-size files are never hashed:

from automation_file import find_duplicates

groups = find_duplicates("/data", min_size=1024)
# list[list[str]] — each inner list is a set of identical files, sorted
# by size descending.

FA_find_duplicates runs the same search from JSON.

Incremental directory sync

sync_dir mirrors src into dst by copying only files that are new or changed. Change detection is (size, mtime) by default; pass compare="checksum" when mtime is unreliable. Extras under dst are left alone by default — pass delete=True to prune them (and dry_run=True to preview):

from automation_file import sync_dir

summary = sync_dir("/data/src", "/data/dst", delete=True)
# summary: {"copied": [...], "skipped": [...], "deleted": [...],
#           "errors": [...], "dry_run": False}

Symlinks are re-created as symlinks rather than followed, so a link pointing outside the tree can't blow up the mirror. JSON action: FA_sync_dir.

Directory manifests

Write a JSON manifest of every file's checksum under a tree and verify the tree hasn't changed later:

from automation_file import write_manifest, verify_manifest

write_manifest("/release/payload", "/release/MANIFEST.json")

# Later…
result = verify_manifest("/release/payload", "/release/MANIFEST.json")
if not result["ok"]:
    raise SystemExit(f"manifest mismatch: {result}")

result reports matched, missing, modified, and extra lists separately. Extras don't fail verification (mirrors sync_dir's non-deleting default); missing or modified do. JSON actions: FA_write_manifest, FA_verify_manifest.

Notifications

Push one-off messages or auto-notify on trigger/scheduler failures via webhook, Slack, or SMTP:

from automation_file import (
    SlackSink, WebhookSink, EmailSink,
    notification_manager, notify_send,
)

notification_manager.register(SlackSink("https://hooks.slack.com/services/T/B/X"))
notify_send("deploy complete", body="rev abc123", level="info")

Every sink implements the same send(subject, body, level) contract. The fanout NotificationManager does per-sink error isolation (one broken sink doesn't starve the others), sliding-window dedup so a stuck trigger can't flood a channel, and SSRF validation on every webhook/Slack URL. Scheduler and trigger dispatchers auto-notify on failure at level="error" — registering a sink is all that's needed. JSON actions: FA_notify_send, FA_notify_list.

Config file and secret providers

Declare sinks and defaults once in automation_file.toml. Secret references resolve at load time from environment variables or a file root (Docker / K8s style):

# automation_file.toml

[secrets]
file_root = "/run/secrets"

[defaults]
dedup_seconds = 120

[[notify.sinks]]
type = "slack"
name = "team-alerts"
webhook_url = "${env:SLACK_WEBHOOK}"

[[notify.sinks]]
type = "email"
name = "ops-email"
host = "smtp.example.com"
port = 587
sender = "[e-mail]"
recipients = ["[e-mail]"]
username = "${env:SMTP_USER}"
password = "${file:smtp_password}"

from automation_file import AutomationConfig, notification_manager

config = AutomationConfig.load("automation_file.toml")
config.apply_to(notification_manager)

Unresolved ${…} references raise SecretNotFoundException rather than silently becoming empty strings. Custom provider chains can be built from ChainedSecretProvider / EnvSecretProvider / FileSecretProvider and passed as AutomationConfig.load(path, provider=…).

Variable substitution in action lists

Opt in with substitute=True and ${…} references expand at dispatch time:

from automation_file import execute_action

execute_action(
    [["FA_create_file", {"file_path": "reports/${date:%Y-%m-%d}/${uuid}.txt"}]],
    substitute=True,
)

Supports ${env:VAR}, ${date:FMT} (strftime), ${uuid}, ${cwd}. Unknown names raise SubstitutionException — no silent empty strings.

Conditional execution

Run a nested action list only when a path-based guard passes:

[
  ["FA_if_exists", {"path": "/data/in/job.json",
                    "then": [["FA_copy_file", {"source": "/data/in/job.json",
                                               "target": "/data/processed/job.json"}]]}],
  ["FA_if_newer",  {"source": "/src", "target": "/dst",
                    "then": [["FA_sync_dir", {"src": "/src", "dst": "/dst"}]]}],
  ["FA_if_size_gt", {"path": "/logs/app.log", "size": 10485760,
                     "then": [["FA_run_shell", {"command": ["logrotate", "/logs/app.log"]}]]}]
]

SQLite audit log

AuditLog writes one row per action with short-lived connections and a module-level lock:

from automation_file import AuditLog

audit = AuditLog("audit.sqlite3")
audit.record(action="FA_copy_file", actor="ops",
             status="ok", duration_ms=12, detail={"src": "a", "dst": "b"})

for row in audit.recent(limit=50):
    print(row["timestamp"], row["action"], row["status"])

File integrity monitor

Poll a tree against a manifest and fire a callback + notification on drift:

from automation_file import IntegrityMonitor, notification_manager, write_manifest

write_manifest("/srv/site", "/srv/MANIFEST.json")

mon = IntegrityMonitor(
    root="/srv/site",
    manifest_path="/srv/MANIFEST.json",
    interval=60.0,
    manager=notification_manager,
    on_drift=lambda summary: print("drift:", summary),
)
mon.start()

Manifest-load errors are surfaced as drift so tamper and config issues aren't silently different code paths.

AES-256-GCM file encryption

Authenticated encryption with a self-describing envelope. Derive a key from a password or generate one directly:

from automation_file import encrypt_file, decrypt_file, key_from_password

key = key_from_password("correct horse battery staple", salt=b"app-salt-v1")
encrypt_file("secret.pdf", "secret.pdf.enc", key, associated_data=b"v1")
decrypt_file("secret.pdf.enc", "secret.pdf", key, associated_data=b"v1")

Tamper is detected via GCM's authentication tag and reported as CryptoException("authentication failed"). JSON actions: FA_encrypt_file, FA_decrypt_file.

HTTPActionClient Python SDK

Typed client for the HTTP action server; enforces loopback by default and carries the shared secret for you:

from automation_file import HTTPActionClient

with HTTPActionClient("http://127.0.0.1:9944", shared_secret="s3cr3t") as client:
    client.ping()                                       # OPTIONS /actions
    result = client.execute([["FA_create_dir", {"dir_path": "x"}]])

Auth failures map to HTTPActionClientException with kind="unauthorized"; 404 responses report the server exists but does not expose /actions.

Prometheus metrics exporter

ActionExecutor records one counter row and one histogram sample per action. Serve them on a loopback /metrics endpoint:

from automation_file import start_metrics_server

server = start_metrics_server(host="127.0.0.1", port=9945)
# curl http://127.0.0.1:9945/metrics

Exports automation_file_actions_total{action,status} and automation_file_action_duration_seconds{action}. Non-loopback binds require allow_non_loopback=True explicitly.

WebDAV, SMB/CIFS, fsspec

Extra remote backends alongside the first-class S3 / Azure / Dropbox / SFTP:

from automation_file import WebDAVClient, SMBClient, fsspec_upload

# RFC 4918 WebDAV — loopback/private targets require opt-in.
dav = WebDAVClient("https://files.example.com/remote.php/dav",
                   username="alice", password="s3cr3t")
dav.upload("/local/report.csv", "team/reports/report.csv")

# SMB / CIFS via smbprotocol's high-level smbclient API.
with SMBClient("fileserver", "share", "alice", "s3cr3t") as smb:
    smb.upload("/local/report.csv", "reports/report.csv")

# Anything fsspec can address — memory, gcs, abfs, local, …
fsspec_upload("/local/report.csv", "memory://reports/report.csv")

HTTP server observability

start_http_action_server() additionally exposes liveness / readiness probes, an OpenAPI 3.0 spec, and a WebSocket stream of progress snapshots:

curl http://127.0.0.1:9944/healthz          # {"status": "ok"}
curl http://127.0.0.1:9944/readyz           # 200 when registry non-empty, 503 otherwise
curl http://127.0.0.1:9944/openapi.json     # OpenAPI 3.0 spec
# Connect a WebSocket to ws://127.0.0.1:9944/progress for live progress frames.

HTMX Web UI

A read-only observability dashboard built on stdlib HTTP + HTMX (loaded from a pinned CDN URL with SRI). Loopback-only by default; optional shared secret:

from automation_file import start_web_ui

server = start_web_ui(host="127.0.0.1", port=9955, shared_secret="s3cr3t")
# Browse http://127.0.0.1:9955/ — health, progress, and registry fragments
# auto-poll every few seconds. Write operations stay on the action servers.

MCP (Model Context Protocol) server

Expose every registered FA_* action to an MCP host (Claude Desktop, MCP CLIs) over JSON-RPC 2.0 on stdio:

from automation_file import MCPServer

MCPServer().serve_stdio()          # reads JSON-RPC from stdin, writes to stdout

pip install exposes an automation_file_mcp console script (via [project.scripts]) so MCP hosts can launch the bridge without any Python glue. Three equivalent launch styles:

automation_file_mcp                                      # installed console script
python -m automation_file mcp                            # CLI subcommand
python examples/mcp/run_mcp.py                           # standalone launcher

All three accept --name, --version, and --allowed-actions (comma- separated whitelist — strongly recommended since the default registry includes high-privilege actions like FA_run_shell). See examples/mcp/ for ready-to-copy Claude Desktop config.

Tool descriptors are generated on the fly by introspecting each action's signature — parameter names and types become a JSON schema, so hosts can render fields without any manual wiring.

DAG action executor

Run actions in dependency order; independent branches fan out across a thread pool. Each node is {"id": ..., "action": [...], "depends_on": [...]}:

from automation_file import execute_action_dag

execute_action_dag([
    {"id": "fetch",  "action": ["FA_download_file",
                                ["https://example.com/src.tar.gz", "src.tar.gz"]]},
    {"id": "verify", "action": ["FA_verify_checksum",
                                ["src.tar.gz", "3b0c44298fc1..."]],
                     "depends_on": ["fetch"]},
    {"id": "unpack", "action": ["FA_unzip_file", ["src.tar.gz", "src"]],
                     "depends_on": ["verify"]},
])

If verify raises, unpack is marked skipped by default. Pass fail_fast=False to run dependents regardless. JSON action: FA_execute_action_dag.

Entry-point plugins

Third-party packages advertise actions via pyproject.toml:

[project.entry-points."automation_file.actions"]
my_plugin = "my_plugin:register"

where register is a zero-argument callable returning a dict[str, Callable]. Once installed in the same environment, the commands show up in every freshly-built registry:

# my_plugin/__init__.py
def greet(name: str) -> str:
    return f"hello {name}"

def register() -> dict:
    return {"FA_greet": greet}

# after `pip install my_plugin`
from automation_file import execute_action
execute_action([["FA_greet", {"name": "world"}]])

Plugin failures are logged and swallowed — one broken plugin cannot break the library.

GUI

python -m automation_file ui        # or: python main_ui.py

from automation_file import launch_ui
launch_ui()

Tabs: Home, Local, Transfer, Progress, JSON actions, Triggers, Scheduler, Servers. A persistent log panel at the bottom streams every result and error.

Scaffold an executor-based project

from automation_file import create_project_dir

create_project_dir("my_workflow")

CLI

# Subcommands (one-shot operations)
python -m automation_file ui
python -m automation_file zip ./src out.zip --dir
python -m automation_file unzip out.zip ./restored
python -m automation_file download https://example.com/file.bin file.bin
python -m automation_file create-file hello.txt --content "hi"
python -m automation_file server --host 127.0.0.1 --port 9943
python -m automation_file http-server --host 127.0.0.1 --port 9944
python -m automation_file drive-upload my.txt --token token.json --credentials creds.json
python -m automation_file mcp --allowed-actions FA_file_checksum,FA_fast_find
automation_file_mcp --allowed-actions FA_file_checksum,FA_fast_find  # installed console script

# Legacy flags (JSON action lists)
python -m automation_file --execute_file actions.json
python -m automation_file --execute_dir ./actions/
python -m automation_file --execute_str '[["FA_create_dir",{"dir_path":"x"}]]'
python -m automation_file --create_project ./my_project

JSON action format

Each entry is either a bare command name, a [name, kwargs] pair, or a [name, args] list:

[
  ["FA_create_file", {"file_path": "test.txt"}],
  ["FA_drive_upload_to_drive", {"file_path": "test.txt"}],
  ["FA_drive_search_all_file"]
]

Documentation

Full API documentation lives under docs/ and can be built with Sphinx:

pip install -r docs/requirements.txt
sphinx-build -b html docs/source docs/_build/html

See CLAUDE.md for architecture notes, conventions, and security considerations.