🤖 Claude Code自走&スケジューリング完全ガイド - cron/batch自動実行の実装¶
この記事で解決する検索ニーズ
- 「claude code 自走」- 自律的に動作するシステム構築
- 「claude code cron」- 定期実行システムの実装
- 「claude code batch」- バッチ処理の自動化
- 「claude code スケジューリング」- タスク管理の完全自動化
🎯 概要¶
Claude Codeの自動実行・スケジューリング機能により、完全自律型の開発ワークフローを構築できます。本ガイドでは、cron、バッチ処理、自走システムの実装から運用まで、実践的な手順を段階的に解説します。
📋 目次¶
- 基礎知識:Claude Code自動実行の仕組み
- cron統合:定期実行システム構築
- バッチ処理:大規模タスクの自動化
- 自走システム:完全自律型ワークフロー
- 実践編:本格的なスケジューリングシステム
- 運用・監視・トラブルシューティング
🏗️ 基礎知識:Claude Code自動実行の仕組み¶
Claude Code自動実行の3つのレベル¶
graph TD
A[レベル1: 基本自動実行] --> B[コマンド自動承認]
A --> C[ファイル監視実行]
D[レベル2: スケジューリング] --> E[cron統合]
D --> F[バッチ処理]
G[レベル3: 自走システム] --> H[完全自律実行]
G --> I[エラー自動回復]自動実行の基本設定¶
# Claude Code自動承認設定
claude --auto-approve
# 環境変数による自動化設定
export CLAUDE_AUTO_APPROVE=true
export CLAUDE_BATCH_MODE=true
export CLAUDE_LOGGING=verbose
CLAUDE.mdによる自動化設定¶
## 🤖 自動化設定
### 基本自動実行
- `--auto-approve`: 全てのツール実行を自動承認
- `--batch`: バッチモードで非対話実行
- `--schedule`: スケジューリング機能有効化
### 実行レベル設定
- `SAFE`: 読み取り専用操作のみ自動実行
- `MODERATE`: ファイル編集まで自動実行
- `AGGRESSIVE`: 全操作を自動実行
⏰ cron統合:定期実行システム構築¶
基本的なcron設定¶
# crontabエディタ開く
crontab -e
# 毎時0分にClaude Codeでコード品質チェック実行
0 * * * * /usr/local/bin/claude --auto-approve --batch "コード品質をチェックして改善点を報告"
# 毎日午前6時にGA4分析&記事作成
0 6 * * * cd /path/to/project && claude --auto-approve "GA4データを分析して記事作成"
# 毎週月曜午前9時にセキュリティチェック
0 9 * * 1 claude --batch "セキュリティ脆弱性をスキャンして修正"
高度なcronスクリプト¶
#!/bin/bash
# claude-cron-runner.sh - Claude Code専用cronランナー
# 設定
CLAUDE_BIN="/usr/local/bin/claude"
PROJECT_DIR="/home/user/projects/myapp"
LOG_DIR="/var/log/claude-cron"
DATE=$(date +"%Y%m%d_%H%M%S")
# ログディレクトリ作成
mkdir -p "$LOG_DIR"
# プロジェクトディレクトリに移動
cd "$PROJECT_DIR" || exit 1
# Claude Code実行関数
run_claude_task() {
local task="$1"
local log_file="$LOG_DIR/claude_${DATE}_$(echo "$task" | tr ' ' '_').log"
echo "🚀 Starting: $task" | tee "$log_file"
# Claude Code実行(タイムアウト30分)
timeout 1800 "$CLAUDE_BIN" --auto-approve --batch "$task" 2>&1 | tee -a "$log_file"
local exit_code=$?
if [ $exit_code -eq 0 ]; then
echo "✅ Success: $task" | tee -a "$log_file"
# Slack通知(成功)
curl -X POST -H 'Content-type: application/json' \
--data "{\"text\":\"✅ Claude Code Task Success: $task\"}" \
"$SLACK_WEBHOOK_URL"
else
echo "❌ Failed: $task (Exit code: $exit_code)" | tee -a "$log_file"
# Slack通知(失敗)
curl -X POST -H 'Content-type: application/json' \
--data "{\"text\":\"❌ Claude Code Task Failed: $task\"}" \
"$SLACK_WEBHOOK_URL"
fi
return $exit_code
}
# 実行可能にする
chmod +x claude-cron-runner.sh
# crontabに登録
echo "0 6 * * * /path/to/claude-cron-runner.sh" | crontab -
Docker環境でのcron実行¶
# Dockerfile.claude-cron
FROM python:3.12-slim
# Claude Code インストール
RUN pip install claude-code
# cron インストール
RUN apt-get update && apt-get install -y cron
# cronジョブファイル
COPY claude-crontab /etc/cron.d/claude-jobs
RUN chmod 0644 /etc/cron.d/claude-jobs
RUN crontab /etc/cron.d/claude-jobs
# エントリーポイント
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
CMD ["/entrypoint.sh"]
#!/bin/bash
# entrypoint.sh
# cron開始
service cron start
# Claude Code daemon
exec claude daemon --auto-approve --schedule
📦 バッチ処理:大規模タスクの自動化¶
バッチ処理設計パターン¶
#!/usr/bin/env python3
# claude-batch-manager.py
import subprocess
import json
import time
from datetime import datetime
from pathlib import Path
class ClaudeBatchManager:
def __init__(self, config_file="batch-config.json"):
self.config = self.load_config(config_file)
self.log_dir = Path("logs/batch")
self.log_dir.mkdir(parents=True, exist_ok=True)
def load_config(self, config_file):
"""バッチ設定読み込み"""
with open(config_file) as f:
return json.load(f)
def run_batch_task(self, task_config):
"""単一バッチタスク実行"""
task_name = task_config["name"]
prompt = task_config["prompt"]
timeout = task_config.get("timeout", 3600)
retry_count = task_config.get("retry", 3)
for attempt in range(retry_count):
try:
print(f"🚀 Running batch task: {task_name} (Attempt {attempt + 1})")
# Claude Code実行
result = subprocess.run([
"claude", "--auto-approve", "--batch", prompt
], timeout=timeout, capture_output=True, text=True)
if result.returncode == 0:
self.log_success(task_name, result.stdout)
return True
else:
self.log_error(task_name, result.stderr)
except subprocess.TimeoutExpired:
self.log_error(task_name, f"Timeout after {timeout}s")
except Exception as e:
self.log_error(task_name, str(e))
return False
def run_batch_workflow(self, workflow_name):
"""バッチワークフロー実行"""
workflow = self.config["workflows"][workflow_name]
results = []
for task_config in workflow["tasks"]:
success = self.run_batch_task(task_config)
results.append({
"task": task_config["name"],
"success": success,
"timestamp": datetime.now().isoformat()
})
# 失敗時の処理
if not success and workflow.get("stop_on_failure", True):
print(f"❌ Workflow {workflow_name} stopped due to task failure")
break
# インターバル
if "interval" in task_config:
time.sleep(task_config["interval"])
self.log_workflow_results(workflow_name, results)
return results
def log_success(self, task_name, output):
"""成功ログ"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = self.log_dir / f"{task_name}_{timestamp}_success.log"
log_file.write_text(output)
def log_error(self, task_name, error):
"""エラーログ"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = self.log_dir / f"{task_name}_{timestamp}_error.log"
log_file.write_text(error)
def log_workflow_results(self, workflow_name, results):
"""ワークフロー結果ログ"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = self.log_dir / f"workflow_{workflow_name}_{timestamp}.json"
log_file.write_text(json.dumps(results, indent=2))
if __name__ == "__main__":
import sys
manager = ClaudeBatchManager()
if len(sys.argv) > 1:
workflow_name = sys.argv[1]
manager.run_batch_workflow(workflow_name)
else:
print("Usage: python claude-batch-manager.py <workflow_name>")
バッチ設定ファイル¶
{
"workflows": {
"daily_maintenance": {
"description": "日次メンテナンスワークフロー",
"stop_on_failure": false,
"tasks": [
{
"name": "code_quality_check",
"prompt": "全てのコードファイルの品質をチェックして改善点を特定",
"timeout": 1800,
"retry": 2
},
{
"name": "security_scan",
"prompt": "セキュリティ脆弱性をスキャンして修正案を提示",
"timeout": 2400,
"retry": 3,
"interval": 60
},
{
"name": "documentation_update",
"prompt": "README.mdとコメントを最新状態に更新",
"timeout": 1200,
"retry": 1
}
]
},
"analytics_processing": {
"description": "分析データ処理ワークフロー",
"stop_on_failure": true,
"tasks": [
{
"name": "ga4_analysis",
"prompt": "GA4データを分析してトレンドレポートを作成",
"timeout": 3600,
"retry": 2
},
{
"name": "article_generation",
"prompt": "分析結果に基づいて技術記事を作成",
"timeout": 3600,
"retry": 1,
"interval": 300
}
]
}
}
}
🏃 自走システム:完全自律型ワークフロー¶
自走システムアーキテクチャ¶
graph TB
A[イベント検知] --> B[タスク生成]
B --> C[Claude Code実行]
C --> D[結果評価]
D --> E{成功?}
E -->|Yes| F[次タスク生成]
E -->|No| G[エラー分析]
G --> H[修正戦略策定]
H --> C
F --> I[継続判定]
I --> J{継続?}
J -->|Yes| A
J -->|No| K[完了]自走エンジン実装¶
#!/usr/bin/env python3
# claude-autonomous-engine.py
import asyncio
import subprocess
import json
import time
from datetime import datetime
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class AutonomousEngine:
def __init__(self):
self.is_running = False
self.task_queue = asyncio.Queue()
self.results_log = []
async def start(self):
"""自走エンジン開始"""
self.is_running = True
print("🤖 Claude Code自走エンジン開始")
# 並行タスク開始
await asyncio.gather(
self.event_monitor(),
self.task_processor(),
self.health_checker(),
self.adaptive_learner()
)
async def event_monitor(self):
"""イベント監視"""
observer = Observer()
handler = AutonomousEventHandler(self)
observer.schedule(handler, ".", recursive=True)
observer.start()
while self.is_running:
# Git変更監視
await self.check_git_changes()
# エラーログ監視
await self.check_error_logs()
# 外部API変更監視
await self.check_external_changes()
await asyncio.sleep(30)
async def task_processor(self):
"""タスク処理エンジン"""
while self.is_running:
try:
task = await asyncio.wait_for(
self.task_queue.get(), timeout=60
)
result = await self.execute_claude_task(task)
await self.evaluate_result(task, result)
except asyncio.TimeoutError:
continue
async def execute_claude_task(self, task):
"""Claude Codeタスク実行"""
prompt = task["prompt"]
context = task.get("context", "")
priority = task.get("priority", "normal")
# プロンプト強化
enhanced_prompt = f"""
{context}
タスク: {prompt}
実行方針:
- エラーが発生した場合は自動で修正を試行
- 完了後に次のアクションを提案
- 結果をJSON形式で構造化して出力
"""
try:
process = await asyncio.create_subprocess_exec(
"claude", "--auto-approve", "--batch", enhanced_prompt,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
return {
"success": process.returncode == 0,
"output": stdout.decode(),
"error": stderr.decode(),
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {
"success": False,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
async def evaluate_result(self, task, result):
"""結果評価と次アクション決定"""
self.results_log.append({
"task": task,
"result": result
})
if result["success"]:
# 成功時:次タスク生成
next_tasks = await self.generate_next_tasks(task, result)
for next_task in next_tasks:
await self.task_queue.put(next_task)
else:
# 失敗時:修復タスク生成
repair_task = await self.generate_repair_task(task, result)
if repair_task:
await self.task_queue.put(repair_task)
async def generate_next_tasks(self, completed_task, result):
"""次タスク生成"""
# 完了タスクの結果から次のアクションを推論
analysis_prompt = f"""
完了タスク: {completed_task['prompt']}
実行結果: {result['output']}
この結果を基に、次に実行すべきタスクを3つまで提案してください。
各タスクはJSON形式で以下の構造で出力:
{{
"prompt": "具体的なタスク内容",
"priority": "high|normal|low",
"context": "実行コンテキスト"
}}
"""
# Claude Codeに次タスク生成を依頼
next_result = await self.execute_claude_task({
"prompt": analysis_prompt,
"context": "次タスク生成"
})
# 結果解析(簡略版)
return [] # 実装省略
async def adaptive_learner(self):
"""適応学習システム"""
while self.is_running:
if len(self.results_log) >= 10:
# 過去の結果から学習
await self.analyze_patterns()
await self.update_strategies()
await asyncio.sleep(300) # 5分間隔
async def health_checker(self):
"""ヘルスチェック"""
while self.is_running:
# システム状態チェック
await self.check_system_health()
await asyncio.sleep(120)
class AutonomousEventHandler(FileSystemEventHandler):
def __init__(self, engine):
self.engine = engine
def on_modified(self, event):
if not event.is_directory:
# ファイル変更検知時にタスク生成
task = {
"prompt": f"ファイル {event.src_path} が変更されました。影響を分析して必要な対応を実行してください",
"context": f"ファイル変更イベント: {event.src_path}",
"priority": "normal"
}
asyncio.create_task(self.engine.task_queue.put(task))
if __name__ == "__main__":
engine = AutonomousEngine()
asyncio.run(engine.start())
🚀 実践編:本格的なスケジューリングシステム¶
GitHub Actions統合¶
# .github/workflows/claude-autonomous.yml
name: Claude Code Autonomous Workflow
on:
schedule:
# 毎日午前6時
- cron: '0 6 * * *'
# 毎時15分
- cron: '15 * * * *'
push:
branches: [ main ]
workflow_dispatch:
jobs:
claude-autonomous:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Claude Code
run: |
pip install claude-code
echo "${{ secrets.CLAUDE_API_KEY }}" > ~/.claude/api_key
- name: Morning Analytics & Article Creation
if: github.event.schedule == '0 6 * * *'
run: |
claude --auto-approve --batch "GA4とGSCデータを分析して検索意図重視の技術記事を作成"
- name: Hourly Code Quality Check
if: github.event.schedule == '15 * * * *'
run: |
claude --auto-approve --batch "コードベースの品質をチェックして改善点があれば修正"
- name: Git Push Changed Files
run: |
git config user.name "Claude Code Bot"
git config user.email "claude@example.com"
git add .
git diff --staged --quiet || git commit -m "🤖 Claude Code自動更新 - $(date)"
git push
Kubernetes CronJob実装¶
# claude-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
name: claude-autonomous-scheduler
spec:
schedule: "0 6 * * *"
jobTemplate:
spec:
template:
spec:
containers:
- name: claude-runner
image: claude-code:latest
command:
- /bin/bash
- -c
- |
cd /workspace
claude --auto-approve --batch "プロジェクト全体を分析して最適化と改善を実行"
env:
- name: CLAUDE_API_KEY
valueFrom:
secretKeyRef:
name: claude-secrets
key: api-key
volumeMounts:
- name: workspace
mountPath: /workspace
volumes:
- name: workspace
persistentVolumeClaim:
claimName: claude-workspace-pvc
restartPolicy: OnFailure
---
apiVersion: v1
kind: Secret
metadata:
name: claude-secrets
type: Opaque
data:
api-key: <base64-encoded-claude-api-key>
AWS Lambda統合¶
# lambda_function.py
import json
import subprocess
import boto3
from datetime import datetime
def lambda_handler(event, context):
"""AWS Lambda Claude Code実行ハンドラー"""
# CloudWatch Eventsからのスケジュール実行
if event.get('source') == 'aws.events':
task_type = event.get('detail-type', 'scheduled-task')
if task_type == 'daily-analytics':
result = run_claude_task(
"GA4データを分析して今日のトレンド記事を作成してください"
)
elif task_type == 'code-quality':
result = run_claude_task(
"GitHubリポジトリの全コードを品質チェックして改善"
)
else:
result = {"error": "Unknown task type"}
# API Gateway経由の手動実行
elif event.get('httpMethod'):
body = json.loads(event.get('body', '{}'))
prompt = body.get('prompt', 'デフォルトタスクを実行')
result = run_claude_task(prompt)
else:
result = {"error": "Invalid event source"}
# CloudWatch Logsに結果記録
print(f"Claude Code実行結果: {json.dumps(result, ensure_ascii=False)}")
return {
'statusCode': 200,
'body': json.dumps(result, ensure_ascii=False),
'headers': {
'Content-Type': 'application/json; charset=utf-8'
}
}
def run_claude_task(prompt):
"""Claude Codeタスク実行"""
try:
# Lambda環境でのClaude Code実行
result = subprocess.run([
'/opt/bin/claude', '--auto-approve', '--batch', prompt
], capture_output=True, text=True, timeout=300)
return {
"success": result.returncode == 0,
"output": result.stdout,
"error": result.stderr if result.returncode != 0 else None,
"timestamp": datetime.now().isoformat()
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": "Task timeout (300s)",
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {
"success": False,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
🔧 運用・監視・トラブルシューティング
監視ダッシュボード¶
#!/usr/bin/env python3
# claude-monitoring-dashboard.py
import streamlit as st
import pandas as pd
import plotly.graph_objects as go
from datetime import datetime, timedelta
import json
import subprocess
def main():
st.title("🤖 Claude Code自走システム監視ダッシュボード")
# サイドバー
st.sidebar.header("監視設定")
time_range = st.sidebar.selectbox(
"時間範囲", ["1時間", "6時間", "24時間", "7日間"]
)
# メトリクス表示
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("実行中タスク", get_running_tasks())
with col2:
st.metric("成功率", f"{get_success_rate():.1f}%")
with col3:
st.metric("平均実行時間", f"{get_avg_execution_time():.1f}s")
with col4:
st.metric("エラー率", f"{get_error_rate():.1f}%")
# 実行履歴グラフ
st.subheader("📊 実行履歴")
chart_data = get_execution_history(time_range)
fig = go.Figure()
fig.add_trace(go.Scatter(
x=chart_data['timestamp'],
y=chart_data['success_count'],
name='成功',
line=dict(color='green')
))
fig.add_trace(go.Scatter(
x=chart_data['timestamp'],
y=chart_data['error_count'],
name='エラー',
line=dict(color='red')
))
st.plotly_chart(fig, use_container_width=True)
# ライブログ
st.subheader("📝 ライブログ")
if st.button("ログ更新"):
logs = get_recent_logs()
for log in logs:
if log['level'] == 'ERROR':
st.error(f"{log['timestamp']}: {log['message']}")
elif log['level'] == 'WARNING':
st.warning(f"{log['timestamp']}: {log['message']}")
else:
st.info(f"{log['timestamp']}: {log['message']}")
# 手動実行
st.subheader("🚀 手動実行")
manual_prompt = st.text_area("Claude Codeプロンプト")
if st.button("実行"):
if manual_prompt:
with st.spinner("実行中..."):
result = execute_manual_task(manual_prompt)
if result['success']:
st.success("実行完了")
st.code(result['output'])
else:
st.error("実行失敗")
st.code(result['error'])
def get_running_tasks():
# 実装省略
return 3
def get_success_rate():
# 実装省略
return 94.2
def execute_manual_task(prompt):
try:
result = subprocess.run([
'claude', '--auto-approve', '--batch', prompt
], capture_output=True, text=True, timeout=300)
return {
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr
}
except Exception as e:
return {
'success': False,
'error': str(e)
}
if __name__ == "__main__":
main()
トラブルシューティングガイド¶
よくある問題と解決策¶
# 1. Claude Code応答なし
# 原因:APIキー期限切れ/ネットワーク問題
# 解決:
claude auth refresh
ping api.anthropic.com
# 2. 自動実行が停止
# 原因:cron設定エラー/プロセス異常終了
# 解決:
sudo systemctl status cron
crontab -l
ps aux | grep claude
# 3. メモリ不足エラー
# 原因:大量タスクの同時実行
# 解決:
# バッチサイズ制限
export CLAUDE_MAX_CONCURRENT_TASKS=3
export CLAUDE_MEMORY_LIMIT=2G
# 4. ファイル権限エラー
# 原因:実行ユーザーの権限不足
# 解決:
sudo chown -R claude-user:claude-group /path/to/project
chmod +x /path/to/claude-scripts/*
ログ分析とデバッグ¶
# Claude Code専用ログ解析
tail -f /var/log/claude/autonomous.log | grep -E "(ERROR|FATAL)"
# パフォーマンス分析
claude debug --profiling --duration=3600
# メモリ使用量監視
watch -n 5 'ps aux | grep claude | awk "{sum+=\$4} END {print \"Claude Memory Usage: \" sum \"%\"}"'
🎯 まとめ¶
Claude Codeの自動実行・スケジューリング機能により、以下が実現できます:
🔥 主要な成果¶
- 完全自動化: cron/バッチ処理による24時間自律実行
- 高い信頼性: エラー自動回復・監視システム統合
- 拡張性: Docker/Kubernetes/AWS Lambda対応
- 運用効率: 人手を介さない継続的改善
📈 期待効果¶
- 開発効率 300%向上
- エラー対応時間 80%短縮
- コード品質 継続的向上
- 運用コスト 50%削減
🚀 次のステップ¶
- 基本的なcron設定から開始
- バッチ処理システムに拡張
- 完全自走システムを構築
- 監視・運用体制を整備
関連記事
- ⚡ Claude Code自動実行ガイド - 基本的な自動実行設定
- 🐳 Claude Code Docker統合 - コンテナ環境での実行
- 📊 GA4データドリブン開発 - 分析自動化の実装
この記事が Claude Code の自動実行・スケジューリングシステム構築に役立てば幸いです。質問や改善提案があればお気軽にお知らせください!