서론
2026년 6월, OpenAI·Anthropic·Google이 일제히 「스케줄 실행형 AI 에이전트 (Scheduled Execution AI Agent)」를 투입했습니다.
- OpenAI: ChatGPT에 Scheduled Tasks (6/17 출시)
- Anthropic: Claude에 Managed Agents (cron 기반, CLI/인증 서비스 연동, vault 관리 환경 변수)
- Google: Gemini Daily Brief / Gemini Spark
본 기사에서는 이 「AI가 정기 실행으로 알아서 움직이는」 메커니즘을 코드로 설계하기 위한 최소 구성을 제시합니다. 테마는 「매일 아침, AI 뉴스를 수집 → 요약 → 저장하는 정기 태스크 (Periodic Task)」입니다.
설계: 3개 층으로 나누어 생각하기
-
스케줄 실행 에이전트는 3개 층으로 파악하면 전망이 좋아집니다.
-
트리거 층 (Trigger Layer) — 언제 움직이는가 (cron)
-
처리 층 (Processing Layer) — 무엇을 하는가 (수집 → LLM 요약)
-
출력 층 (Output Layer) — 어디로 내보내는가 (파일/알림/API)
- 트리거 층: cron 표현
매일 아침 7:00 (JST)에 실행하는 cron 식
CRON = "0 7 * * *" (분 시 일 월 요일)
Anthropic의 Managed Agents도 이 cron 시맨틱스 (Semantics)를 준수합니다. 로컬에서 테스트하려면 APScheduler가 간편합니다.
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler(timezone="Asia/Tokyo")
@sched.scheduled_job("cron", hour=7, minute=0)
def run_daily_job():
main()
if __name__ == "__main__":
sched.start()
2. 처리 층: 수집 → 요약
LLM으로의 요약 호출을 벤더 비의존적 (Vendor-independent) 인터페이스로 감싸 두면, ChatGPT / Claude / Gemini를 교체하기 쉬워집니다.
from dataclasses import dataclass
from typing import Protocol
@dataclass
class NewsItem:
title: str
url: str
body: str
class Summarizer(Protocol):
def summarize(self, items: list[NewsItem]) -> str: ...
class ClaudeSummarizer:
def __init__(self, client, model: str = "claude-opus-4-8"):
self.client = client
self.model = model
def summarize(self, items: list[NewsItem]) -> str:
joined = "\n\n".join(f"- {i.title}" for i in items)
prompt = "다음 AI 뉴스를 5줄로 요약:\n" + joined
resp = self.client.messages.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
)
return resp.content[0].text
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
동일한 Summarizer 프로토콜을 충족하도록 OpenAI 버전을 구현하면, 호출 측은 변경할 필요가 없습니다.
3. 출력 층: 저장과 멱등성 (Idempotency)
정기 실행에서 가장 사고가 나기 쉬운 것이 「이중 실행」과 「실패 시의 재시도 (Retry)」입니다. 실행일을 키(Key)로 하여 멱등하게 만들어 둡시다.
import os, datetime
def save_idempotent(summary: str, out_dir: str = "./daily") -> str:
today = datetime.date.today().isoformat()
path = os.path.join(out_dir, f"{today}.md")
if os.path.exists(path):
return path
os.makedirs(out_dir, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(f"# AI News {today}\n\n{summary}\n")
return path
4. 연결 (main)
def main():
items = fetch_news()
summarizer = ClaudeSummarizer(client=anthropic_client)
try:
summary = summarizer.summarize(items)
except Exception as e:
log_error(e)
return path = save_idempotent(summary)
notify(f"오늘의 AI 뉴스 요약 저장: {path}")
운영상의 핵심 - 타임아웃과 제한: LLM 호출에는 반드시 타임아웃을 설정합니다.
- 멱등성 (Idempotency): 실행일/대상 ID를 키로 하여 이중 처리를 방지합니다.
- 실패는 무시하지 않고 기록: 한 번의 실패로 전체를 멈추지 않습니다.
- 환경 변수는 안전하게: API 키는 vault / 시크릿 매니저에 보관합니다.
요약
2026년 6월 세 사의 움직임은 'AI를 호출하는' 개발에서 'AI를 주기적으로 실행하여 운영하는' 개발로의 전환을 보여줍니다. 트리거(Trigger) - 처리(Process) - 출력(Output)의 3계층으로 나누고, 벤더에 종속되지 않는 인터페이스로 감싸며, 멱등성과 오류 내성을 갖추는—이 설계를 이해한다면 어떤 플랫폼의 예약 실행 기능이 와도 응용할 수 있습니다.