task-automation

タスク自動化

このスキルを使うと、AI エージェントが反復的なタスクやワークフローを自動化する仕組みを設計、実装できます。エージェントは、自動化に適した手動プロセスを特定し、適切な自動化パターン（スクリプト、ファイル監視、cron ジョブ、CI/CD トリガー、API ポーリング）を選択して、実装を作成し、正しく動作することを検証します。目標は、作業量に比例して増加する反復的な手作業である toil をなくし、信頼性の高い、手間のかからない自動化に置き換えることです。

ワークフロー

1. タスクを分析する： ユーザーが何を自動化したいのかを把握します。タスクを開始するトリガー、手順、入力と出力、現在の手動実行頻度を確認します。タスクがイベント駆動型（変更によって開始）か、時間駆動型（スケジュールに従って実行）かを判断します。

2. 自動化パターンを選択する： トリガーの種類と環境に基づいて、適切な自動化手法を選択します。一般的なパターンには、単発または逐次タスク向けのシェルスクリプト、ファイル変更への対応に使うファイル監視ツール（fswatch、inotifywait、chokidar）、定期的なスケジュール実行向けの cron ジョブまたは systemd タイマー、コード関連の自動化向けの CI/CD パイプライントリガー、外部サービスのイベントへの対応に使う API ポーリングまたは Webhook リスナーがあります。

3. 実装を設計する： 自動化の詳細を計画します。入力と設定、エラー処理戦略（再試行ロジック、アラート通知、フォールバック動作）、ログ記録方法、必要なシークレットや認証情報の管理を定義します。冪等性も考慮します。同じ自動化を何度実行しても、副作用が発生しないようにします。

4. 自動化コードを作成する： 適切なツールと言語を使って自動化を実装します。広く使われ、十分にサポートされているツールを優先します。スクリプトには bash/Python、スケジュール設定には crontab、CI トリガーには GitHub Actions または GitLab CI、イベントリスナーには標準的な Webhook フレームワークを使用します。

5. テストして検証する： まず安全な環境で自動化を実行します。正常系が正しく処理されることを確認してから、空の入力、ネットワーク障害、権限エラー、同時実行などのエッジケースをテストします。デバッグに十分な情報がログに記録されることを確認します。

6. デプロイして監視する： 適切な権限を設定し、自動化を対象の環境にデプロイします。障害を速やかに検知できるよう、監視またはアラート通知を設定します。自動化の目的、設定方法、必要に応じて無効にする方法を文書化します。

使用方法

自動化したいタスクについて、トリガー、実行する処理、実行環境を含めて説明してください。エージェントが適切なパターンを選び、実装します。

Automate the following: whenever a new CSV file is added to the ~/data/incoming/
directory, validate the CSV headers, transform the data into JSON, and move the
result to ~/data/processed/. Log any files that fail validation to ~/data/errors/.

例

例 1：ファイル監視と処理パイプライン

ユーザーの依頼：

ディレクトリに届く CSV ファイルの処理を自動化してください。

実装：

#!/usr/bin/env python3
"""File watcher that processes incoming CSVs into JSON.

Usage: python csv_watcher.py
Watches: ~/data/incoming/ for new .csv files
Outputs: ~/data/processed/*.json and ~/data/errors/error.log
"""

import os
import csv
import json
import time
import logging
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

INCOMING = Path.home() / "data" / "incoming"
PROCESSED = Path.home() / "data" / "processed"
ERRORS = Path.home() / "data" / "errors"
REQUIRED_HEADERS = ["id", "name", "email", "amount"]

logging.basicConfig(
    filename=ERRORS / "error.log",
    level=logging.ERROR,
    format="%(asctime)s - %(message)s"
)

for d in [INCOMING, PROCESSED, ERRORS]:
    d.mkdir(parents=True, exist_ok=True)


class CSVHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.src_path.endswith(".csv"):
            return
        filepath = Path(event.src_path)
        try:
            with open(filepath, "r") as f:
                reader = csv.DictReader(f)
                headers = reader.fieldnames or []
                missing = set(REQUIRED_HEADERS) - set(headers)
                if missing:
                    raise ValueError(f"Missing headers: {missing}")
                rows = list(reader)

            output = PROCESSED / filepath.with_suffix(".json").name
            with open(output, "w") as f:
                json.dump(rows, f, indent=2)

            filepath.unlink()  # Remove processed file
            print(f"Processed: {filepath.name} -> {output.name}")
        except Exception as e:
            logging.error(f"Failed to process {filepath.name}: {e}")
            filepath.rename(ERRORS / filepath.name)


if __name__ == "__main__":
    observer = Observer()
    observer.schedule(CSVHandler(), str(INCOMING), recursive=False)
    observer.start()
    print(f"Watching {INCOMING} for new CSV files...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

例 2：Cron による API ポーリングと Slack 通知

ユーザーの依頼：

6 時間ごとに API のヘルスエンドポイントを確認し、サービスに劣化があれば Slack に投稿してください。

実装：

crontab -e を使って追加する cron エントリ：

0 */6 * * * /usr/bin/python3 /opt/scripts/health_check.py >> /var/log/health_check.log 2>&1

スクリプト：

#!/usr/bin/env python3
"""Poll API health endpoint and alert Slack on degraded services.

Runs every 6 hours via cron. Exits 0 on success, 1 on alert sent, 2 on script error.
"""

import os
import json
import urllib.request

HEALTH_URL = "https://api.example.com/health"
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]


def check_health():
    req = urllib.request.Request(HEALTH_URL, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read())
    return data  # e.g., {"services": {"auth": "ok", "payments": "degraded", "db": "ok"}}


def send_slack_alert(degraded_services):
    service_list = "\n".join(f"- *{name}*: {status}" for name, status in degraded_services)
    payload = json.dumps({
        "text": f":warning: *Service Health Alert*\n{service_list}"
    }).encode()
    req = urllib.request.Request(
        SLACK_WEBHOOK,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    urllib.request.urlopen(req)


if __name__ == "__main__":
    health = check_health()
    degraded = [
        (name, status)
        for name, status in health.get("services", {}).items()
        if status != "ok"
    ]
    if degraded:
        send_slack_alert(degraded)
        print(f"Alert sent for {len(degraded)} degraded service(s)")
        exit(1)
    else:
        print("All services healthy")
        exit(0)

ベストプラクティス

• 自動化を冪等にします。 同じ入力で同じ自動化を2回実行しても、副作用なしで同じ結果が得られるようにします。これにより、ジョブが誤って再実行された場合のデータ破損を防げます。
• すべてをログに記録し、アラートは選別して送ります。 デバッグ用に詳細なログを記録しつつ、対処が必要な障害のみアラートを送信します。「すべて正常」という通知で受信トレイが埋まると、アラートが無視されるようになります。
• 設定を外部化します。 ファイルパス、URL、しきい値、認証情報は、スクリプトにハードコードせず、環境変数や設定ファイルに保存します。これにより、自動化の移植性が高まり、シークレットも管理しやすくなります。
• スケジュールされたジョブにはロックファイルまたはミューテックスを使用します。 前回の実行が完了していないと、cron ジョブが重複して実行される場合があります。flock または PID ファイルを使用して、一度に実行されるインスタンスを1つだけにします。
• 自動化スクリプトをバージョン管理します。 自動化を本番コードとして扱い、Git に保存して、変更をレビューし、リリースにタグを付けます。自動化の不具合は、機能の不具合より大きな損害を引き起こす可能性があります。
• 手動での上書き手段を用意します。 すべての自動化について、一時停止、スキップ、手動実行の方法を文書化します。自動処理が手動での復旧作業を妨げるおそれがあるインシデント時には、特に重要です。

エッジケース

• 複数ステップの自動化における部分的な失敗： 5つのステップのうちステップ3が失敗した場合、自動化がその失敗を黙って無視してはなりません。チェックポイントを実装し、最初からやり直すのではなく、最後に成功したステップから再開できるようにします。
• ファイルへの同時書き込み： ファイル監視ツールは、書き込み途中のファイルを検出する場合があります。処理を開始する前に短い待機時間を設けるか、ファイルのサイズが N 秒間変化していないことを確認します。
• 認証情報の有効期限切れ： API トークンや OAuth 認証情報には有効期限があります。長期間実行する自動化にはトークンの更新処理を組み込み、更新に失敗した場合は、黙って停止するのではなくアラートを送信します。
• cron におけるタイムゾーンの問題： cron はデフォルトでシステムのタイムゾーンを使用します。グローバルチームでは、UTC を明示的に使用するか、タイムゾーンを文書化します。夏時間の切り替えによって、ジョブが2回実行されたり、実行されなかったりする可能性に注意してください。
• ポーリングする API のレート制限： ポーリング間隔が短すぎたり、複数のインスタンスが実行されたりすると、API のレート制限に達する場合があります。指数バックオフを実装し、レート制限に関するヘッダーを追跡します。
• 空または不正な形式の入力： 外部データ（ファイル、Webhook、API レスポンス）によってトリガーされる自動化では、処理前に入力スキーマを検証します。破損した出力を生成するのではなく、明確なエラーメッセージを表示して安全に失敗するようにします。