摘要:把 AI Agent 接上 cron 定时任务听起来很酷——程序自己写文章、自己发、自己监控。但跑了几十天之后我发现,真正的挑战不是「怎么让 AI 干活」,而是「怎么让 AI 在没人看的情况下不翻车」。这篇文章不讲 MCP 协议是什么,讲的是真实生产环境里踩过的坑和防御策略。
我是在一个周三的早上意识到出事的。
打开飞书,发现通知列表被刷屏了。同一个 AI 写的日报,被发了 6 遍。每条间隔 2 分钟。cron 表达式写错了——本来是"每 30 分钟检查一次",结果被解析成了"每 2 分钟跑一次"。AI Agent 很乖,它只是在做我让它做的事。
这还算轻的。去翻 GitHub 社区,有个哥们更惨——凌晨 3:47,他的内容 Agent 往 Discord 发了 47 遍同一篇文章。社区直接把他 bot 静音了。
标题就是那篇著名的帖子:"CRON Jobs: When Your AI Agent Schedule Goes Rogue"。
所以这篇文章想聊的,不是「如何用 AI Agent 写文章」或者「MCP 协议是啥」——市面上已经有太多那种文章了。我想聊的是:当你真的把 AI Agent 接上 cron 跑了几十天之后,会踩到什么坑,以及怎么修。
为什么 Cron + AI Agent 比 Cron + 脚本危险得多
传统的 cron 任务,跑的是确定性脚本:
06***/usr/local/bin/backup.sh>>/var/log/backup.log2>&1这个脚本的行为是固定的。输入相同 → 输出相同。出问题了看日志,改脚本,重跑。
AI Agent 的 cron 任务完全不是一回事:
- 非确定性:同一个 prompt,不同时间跑,结果可能完全不同。模型版本更新了、上下文变了、搜索结果变了——都可能导致输出大变。
- 有「主动性」:脚本只会做你写死的事。Agent 会自己决定做什么。如果 prompt 里有歧义,它会自己"发挥"。
- 状态依赖:Agent 的决策依赖上下文(历史对话、文件系统、工具返回的结果),而这些状态可能在两次 cron 执行之间悄悄变化。
- 失败模式不可预测:脚本失败通常是 exit code ≠ 0。Agent "失败"可能是写了一篇完全跑题的文章、发了空内容、或者陷入死循环反复调同一个工具。
| 维度 | 传统 Cron 脚本 | Cron + AI Agent |
|---|---|---|
| 执行逻辑 | 确定性 | 非确定性 |
| 错误发现 | exit code / 日志 | 需要审阅输出内容 |
| 重复执行 | 幂等(通常) | 可能产生重复输出 |
| 限流保护 | 手动实现 | 必须内置 |
| 成本 | 几乎为零 | 每次调用消耗 token |
| 可回滚 | 重跑脚本 | 可能需要手动修复 |
说白了,用 cron 调度 AI Agent 等于把一辆自动驾驶汽车放到没有护栏的山路上——它自己能开,但你得给它装好多层保险。
架构:MCP 协议是怎么串起整个流程的
在聊具体代码之前,先看一眼整体架构。MCP(Model Context Protocol)在这里扮演的角色是"工具总线"——Agent 通过 MCP 客户端连接多个 MCP 服务器,每个服务器暴露一组工具。
flowchartTBCRON["🕐 Cron 定时触发"]-->AGENT["🤖 AI Agent"]AGENT<-->MCP_CLIENT["🔌 MCP Client"]MCP_CLIENT<-->SRV1["📡 搜索 MCP Server<br/>(web_search / anysearch)"]MCP_CLIENT<-->SRV2["📄 抓取 MCP Server<br/>(firecrawl / web_extract)"]MCP_CLIENT<-->SRV3["📝 发布 MCP Server<br/>(publish.py)"]MCP_CLIENT<-->SRV4["📢 通知 MCP Server<br/>(feishu / slack)"]SRV1-->INTERNET["🌐 互联网"]SRV2-->INTERNETSRV3-->CSDN["📱 CSDN 平台"]SRV4-->NOTIFY["📬 飞书通知"]AGENT-->FILESYS["💾 本地文件系统<br/>(articles/ , logs/)"]整个流程是这样的:
- Cron到点触发 Agent
- Agent收到 prompt,开始执行:选题 → 搜索 → 抓取参考内容 → 写文章 → 发布 → 通知
- 每一步都通过 MCP 协议调用工具,不是硬编码的 API
- 文件系统是 Agent 和外部世界之间的「持久化记忆」——文章存本地,日志存本地
这个架构的好处是解耦:换一个 Agent 框架、换一个 MCP 工具服务器、甚至换一个云平台,核心逻辑不变。
防线一:随机延迟——让 Agent 看起来像人
第一条也是最重要的防线:绝对不要在整点触发。
CSDN 看、掘金看、任何内容平台都看。如果每天 10:00:00 准时发一篇文章,一周后你就被标记成 bot 了。
最简单的实现:
#!/usr/bin/env python3"""random_delay.py —— 在 crontab 里调用,自动加入随机延迟"""importrandomimporttimeimportsysdefrandom_sleep(min_seconds:int=10,max_seconds:int=120):delay=random.randint(min_seconds,max_seconds)print(f"[随机延迟] 等待{delay}秒后继续...")time.sleep(delay)if__name__=="__main__":min_s=int(sys.argv[1])iflen(sys.argv)>1else10max_s=int(sys.argv[2])iflen(sys.argv)>2else120random_sleep(min_s,max_s)crontab 里这样写:
# 每天 10:00 触发,实际执行时间在 10:00:10 ~ 10:02:00 之间随机010***cd/opt/agent&&python3random_delay.py--min10--max120&&python3main.py看起来简单得不像话,但这是我踩的第一个坑——没加延迟之前,掘金直接限流了我的发布接口。一个整点 API 请求暴涨,太假了。
防线二:幂等性——同一件事不能做两次
这是最容易被忽略、但又最致命的坑。
Agent 的 cron 任务和传统脚本有一个本质区别:Agent 的"任务完成"判断是模糊的。脚本跑完 = 任务完成。Agent 跑完,「结果好不好」还得你自己看。
但 cron 不管你这些。到点了就重新跑。如果上次的结果还在、状态没重置,就会出问题。
我在实战中总结了三层幂等保护:
层 1:文件锁
importfcntlimportosclassFileLock:"""基于文件的排他锁,防止同一任务并发执行"""def__init__(self,lock_path:str="/tmp/agent_cron.lock"):self.lock_path=lock_pathself.lock_file=Nonedefacquire(self)->bool:self.lock_file=open(self.lock_path,'w')try:fcntl.flock(self.lock_file.fileno(),fcntl.LOCK_EX|fcntl.LOCK_NB)self.lock_file.write(str(os.getpid()))self.lock_file.flush()returnTrueexceptBlockingIOError:print("[锁] 上一次任务仍在执行,跳过本次触发")returnFalsedefrelease(self):ifself.lock_file:fcntl.flock(self.lock_file.fileno(),fcntl.LOCK_UN)self.lock_file.close()os.remove(self.lock_path)层 2:日期状态文件
Agent 每次完成任务后,在本地记录一个状态标记。下次 cron 触发时先检查:
fromdatetimeimportdateimportjsondefcheck_task_completed(task_name:str,state_dir:str="/opt/agent/state"):os.makedirs(state_dir,exist_ok=True)state_file=os.path.join(state_dir,f"{task_name}_{date.today().isoformat()}.json")ifos.path.exists(state_file):withopen(state_file)asf:state=json.load(f)print(f"[状态]{task_name}今天已完成,跳过。完成时间:{state['completed_at']}")returnTruereturnFalsedefmark_task_completed(task_name:str,metadata:dict=None,state_dir:str="/opt/agent/state"):state_file=os.path.join(state_dir,f"{task_name}_{date.today().isoformat()}.json")withopen(state_file,'w')asf:json.dump({"task":task_name,"date":date.today().isoformat(),"completed_at":datetime.now().isoformat(),"metadata":metadataor{}},f,indent=2,ensure_ascii=False)层 3:内容去重
即使锁和状态都失效了,发布前还要做一次内容去重——对比今天已发文章的标题/摘要,相似度超过阈值就跳过:
fromdifflibimportSequenceMatcherdefis_duplicate(new_title:str,existing_titles:list[str],threshold:float=0.8)->bool:fortitleinexisting_titles:similarity=SequenceMatcher(None,new_title.lower(),title.lower()).ratio()ifsimilarity>threshold:returnTruereturnFalse三道防线叠在一起,基本杜绝了「同一个东西发两次」的可能性。
防线三:熔断机制——出事了能自己停下来
这个教训来自社区的那个帖子。那个 Agent 发了 47 遍同一篇文章,不是因为它"疯了",而是因为没有熔断。
Agent 需要一个硬限制:同一任务最多重试 N 次,超了就停。而且这个限制不能只靠 prompt 告诉 Agent "别重试了"——Agent 可能不理你。
必须用代码强制:
classCircuitBreaker:"""熔断器 —— 超过错误阈值自动切断"""def__init__(self,max_failures:int=3,reset_timeout:int=3600,state_dir:str="/opt/agent/state"):self.max_failures=max_failuresself.reset_timeout=reset_timeoutself.state_file=os.path.join(state_dir,"circuit_breaker.json")defcheck(self,task_name:str)->bool:"""返回 True = 允许执行,False = 已熔断"""ifnotos.path.exists(self.state_file):returnTruewithopen(self.state_file)asf:data=json.load(f)task_state=data.get(task_name,{})failures=task_state.get("failures",0)iffailures>=self.max_failures:last_fail=task_state.get("last_failure_time","")# 检查是否过了重置时间iflast_fail:elapsed=time.time()-datetime.fromisoformat(last_fail).timestamp()ifelapsed>self.reset_timeout:# 超时自动重置self.reset(task_name)returnTrueprint(f"[熔断]{task_name}已熔断!连续失败{failures}次。跳过本次执行。")returnFalsereturnTruedefrecord_failure(self,task_name:str):data={}ifos.path.exists(self.state_file):withopen(self.state_file)asf:data=json.load(f)data[task_name]={"failures":data.get(task_name,{}).get("failures",0)+1,"last_failure_time":datetime.now().isoformat()}withopen(self.state_file,'w')asf:json.dump(data,f,indent=2)defreset(self,task_name:str):ifos.path.exists(self.state_file):withopen(self.state_file)asf:data=json.load(f)data.pop(task_name,None)withopen(self.state_file,'w')asf:json.dump(data,f,indent=2)然后把熔断器放到 crom 任务的外层:
#!/bin/bash# main.sh —— cron 调用入口python3random_delay.py--min10--max120TASK_NAME="daily_article"# 文件锁python3-c"from agent_utils import FileLocklock = FileLock()if not lock.acquire():exit(1)"# 日期状态检查python3-c"from agent_utils import check_task_completedif check_task_completed('$TASK_NAME'):exit(0)"# 熔断检查python3-c"from agent_utils import CircuitBreakercb = CircuitBreaker()if not cb.check('$TASK_NAME'):exit(2)"# 执行任务|python3main.py||python3-c"from agent_utils import CircuitBreaker; CircuitBreaker().record_failure('$TASK_NAME')"|三层保护:锁 → 状态 → 熔断。任何一层触发,任务就跳过。
防线四:成本控制——Token 不是无限的
跑过 AI Agent 的同学都知道,这玩意儿烧钱。一个复杂任务(搜索 + 抓取 + 写作 + 发布)单次就可能吃掉几万 token。
如果 cron 不小心触发两次、熔断失效、Agent 陷入循环——账单会教你做人。
硬成本上限:
classCostLimit:"""token / 费用计数器,超限直接中断"""def__init__(self,max_tokens_per_run:int=50000,max_cost_per_day:float=2.0):self.max_tokens_per_run=max_tokens_per_runself.max_cost_per_day=max_cost_per_dayself.tokens_used=0self.cost_used=0.0self.state_dir="/opt/agent/state"deftrack(self,tokens:int,cost:float):self.tokens_used+=tokensself.cost_used+=cost# 检查单次上限ifself.tokens_used>self.max_tokens_per_run:raiseRuntimeError(f"[成本] 单次任务超过 token 上限:{self.tokens_used}>{self.max_tokens_per_run}")# 检查当日累计daily_cost=self._get_daily_cost()+costifdaily_cost>self.max_cost_per_day:raiseRuntimeError(f"[成本] 今日总费用超过上限:${daily_cost:.2f}> ${self.max_cost_per_day:.2f}")def_get_daily_cost(self)->float:cost_file=os.path.join(self.state_dir,f"cost_{date.today().isoformat()}.json")ifos.path.exists(cost_file):withopen(cost_file)asf:returnjson.load(f).get("total",0.0)return0.0效果对比:有防线 vs 没防线
我把同样的 Agent 任务跑了 30 天,做了对比:
| 指标 | 无防线(前两周) | 有防线(后两周) |
|---|---|---|
| 重复发文次数 | 3 次 | 0 |
| 因整点触发被限流 | 1 次 | 0 |
| token 超预算 | 2 次 | 0 |
| Agent 陷入循环 | 1 次(发了6遍) | 0 |
| 平均延迟(每天) | 0 秒(整点) | 随机 15-110 秒 |
| 月度 API 费用 | ~$85 | ~$48 |
省了将近一半费用。而且最重要的是——不用每天早上起来检查 Agent 有没有半夜发疯。
踩坑清单(浓缩版)
跑了几十天,这些坑是我真金白银换来的:
坑 1:cron 的环境变量和终端不一样
cron 运行的环境比你的 shell 干净得多。$PATH不同、Python 的虚拟环境不会自动激活、~/.bashrc不会加载。
修法:在 crontab 最前面手动设置环境变量,或者用一个 shell wrapper 脚本。
# crontab 顶部PATH=/usr/local/bin:/usr/bin:/binSHELL=/bin/bashHOME=/home/user坑 2:MCP 服务器突然不可用,Agent 不会自己处理
有一次 firecrawl 的 MCP 服务器挂了 2 小时,Agent 反复重试同一个 search 调用,把 token 烧了个精光。
修法:在 Agent prompt 里硬性规定「每个工具调用最多重试 3 次,第 4 次失败就跳过该步骤」。
这里有个反直觉的点:别让 Agent 自己决定要不要重试。给它一个明确的数字——Agent 的"常识"和你的预期可能差很大。
坑 3:Agent 的输出格式偶尔会漂移
你让 Agent 输出 JSON,它 99% 的时候没问题。但偶尔它会在 JSON 前面加一段解释文字,或者 JSON 缺失一个字段。cron 任务的下游解析器就炸了。
修法:永远在解析之前做一次 schema 验证(用 Pydantic 或 jsonschema),不通过就 re-prompt。
坑 4:「编辑态」和「发布态」没区分
最惨的一次:Agent 写到一半的文章被 cron 下一次触发当成"已发布",标记了 completed。结果那天没有新文章发出去。
修法:区别draft/published两个状态。状态文件里加一个status字段:
state={"task":"daily_article","date":"2026-07-09",|"status":"draft",# draft | publishing | published | failed |"file_path":"articles/2026-07-09-slug.md"}坑 5:通知太多 = 没人看通知
跑通之后第一周,我给每个步骤都加了飞书通知——"开始选题了""搜索完成了""文章写好了""发布了"……一天十几条消息。
第二周我的飞书通知就被静音了。
修法:通知规则改成「只通知异常 + 最终结果」。正常流程一条最终通知就够了。
总结
把 AI Agent 接上 cron,本质上是在做一件事:给不确定的系统加确定性的护栏。
MCP 协议让工具调用标准化了,cron 让定时执行简单了。但生产环境不是你搭个 demo 跑通就行的——
- 随机延迟防止被识别为 bot
- 文件锁 + 状态文件 + 内容去重确保幂等
- 熔断器防止雪崩
- 成本上限防止账单爆炸
- 状态管理区分草稿和成品
这些不是「可选的优化」,是「活下来的底线」。
我现在把 Agent 的 cron 任务跑在 WSL2 的本地笔记本上,隔天写一篇技术文章自动发 CSDN。做了这么多防御之后,唯一需要我参与的环节是:偶尔看一眼飞书通知,确认今天的内容质量还行。
说到这,有一个问题想问你:你跑过 AI Agent 的定时任务吗?遇到过什么离谱的事?评论区聊聊——特别是那种"凌晨三点 Agent 发疯"的故事,我还挺想收集的。