agno AgentOS 如何配置 cron 定时任务并查看每次执行的历史记录？-开发者社区

agno AgentOS 如何配置 cron 定时任务并查看每次执行的历史记录？

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

如果你的 Agent 部署在 agno 的 AgentOS 上，需要在固定时间自动触发一个 Agent / Team / Workflow 的 run（例如每天 9 点生成日报），并且要能查到每一次触发的执行结果，那么 AgentOS 的 scheduler 模块就是对应的功能：它可以持久化 cron 计划、按轮询间隔认领到期的任务、调用 AgentOS 端点，并把每次执行尝试写入历史表。

本文基于 cookbook/05_agent_os/12_scheduler 中的官方示例，给出一条连续路径：启动带 scheduler 的 AgentOS → 创建 cron 计划 → 通过 REST 或 Python 查看每次执行的历史记录。

准备条件

文档要求的环境（来自 README）：

启动 Postgres。官方示例通过 Docker 脚本启动agnohq/pgvector:18，映射到本地 5532 端口：

./cookbook/scripts/run_pgvector.sh

导出 OpenAI key（示例中的 Agent 使用gpt-5.5模型）：

export OPENAI_API_KEY=...

注意：03_manage_with_python.py不调用模型，只需要 Postgres；其余示例会发起真实的gpt-5.5调用。示例统一使用.venvs/demo/bin/python作为解释器，如果你没有这个虚拟环境，需要替换为你自己装好 agno 的 Python。

选择存储时按部署形态区分（文档 Deployment notes 明确给出）：SQLite 只适合本地开发和单个本地 scheduler 进程；多个 AgentOS worker 共享计划时必须用 Postgres，它的认领操作基于FOR UPDATE SKIP LOCKED的原子更新，保证不同 worker 不会执行同一条到期记录。

启动带 scheduler 的 AgentOS

主路径直接使用 01_run_in_agentos.py，它演示了完整的配置面。关键代码是：

db = PostgresDb( id="scheduler-agent-os-db", db_url=DATABASE_URL, # 默认 postgresql+psycopg://ai:ai@localhost:5532/ai schedules_table="agent_os_scheduler_schedules", schedule_runs_table="agent_os_scheduler_runs", ) agent_os = AgentOS( id=OS_ID, description="AgentOS with a Postgres-backed schedule poller.", agents=[scheduled_agent], db=db, scheduler=True, scheduler_poll_interval=SCHEDULER_POLL_INTERVAL_SECONDS, # 5 scheduler_base_url=BASE_URL, # 默认 http://127.0.0.1:7777 ) app = agent_os.get_app()

其中几个配置的用途（均来自文档）：

scheduler=True打开内置轮询器；scheduler_poll_interval是轮询间隔，示例为 5 秒。
scheduler_base_url是进程内执行器回调 AgentOS 时使用的地址，默认http://127.0.0.1:7777；AgentOS 监听在其他地址时要显式设置。
开启 scheduling 后，AgentOS 会在未提供internal_service_token时创建一个内部服务令牌，执行器以 bearer 凭证发送它。该令牌只用于 scheduler 到 AgentOS 的内部流量，不是终端用户 API key，显式提供时要当作机密保管。
每个被调用的 run 端点，其 payload 里必须包含message字段。执行器会强制把计划内的 Agent / Team / Workflow 调用设为stream=false、background=true，然后轮询持久化的 run 直到到达终态。

运行方式：

# 启动服务端（会先种入一条 * * * * * 的分钟级计划） .venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/01_run_in_agentos.py # 另一个终端，运行观察器 .venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/01_run_in_agentos.py --demo

seed_schedule()通过ScheduleManager.create()创建计划，参数包括name、cron（示例为* * * * *）、endpoint（/agents/scheduled-greeter/runs）、payload（必须带message，示例还带了session_id）、timezone和timeout_seconds（示例 120 秒）。轮询器每 5 秒检查一次，认领下一个自然到期的分钟，触发 Agent 并在历史表中持久化结果。

查看每次执行的历史记录

历史记录的 REST 路由是GET /schedules/{schedule_id}/runs，返回对象而不是裸列表：顶层只有data和meta两个键，分页参数是page（不是offset），配合limit使用。

02_rest_api.py 展示了完整的 REST 生命周期，在01_run_in_agentos.py服务端仍在运行时执行：

.venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/02_rest_api.py

核心请求形态如下：

# 创建计划（注意：REST 字段名是 cron_expr，不是 cron） client.post("/schedules", json={ "name": SCHEDULE_NAME, "cron_expr": "0 0 1 1 *", "endpoint": f"/agents/{AGENT_ID}/runs", "payload": {"message": "...", "session_id": "rest-scheduler-session"}, "timezone": "UTC", "timeout_seconds": 120, "max_retries": 1, "retry_delay_seconds": 5, }) # 手动触发一次（不需要等到 cron 到期） client.post(f"/schedules/{schedule_id}/trigger") # 分页读取执行历史 client.get(f"/schedules/{schedule_id}/runs", params={"limit": 1, "page": 1}) client.get(f"/schedules/{schedule_id}/runs", params={"limit": 1, "page": 2})

其余可用路由：GET /schedules（列表）、GET /schedules/{id}（详情）、PATCH /schedules/{id}（更新）、POST /schedules/{id}/enable/disable、DELETE /schedules/{id}。

验证方式（脚本内置的断言 + TEST_LOG 中记录的实际运行结果，均为文档示例）：

01_run_in_agentos.py --demo的观察器会轮询 runs 路由，等待最多约 200 秒（覆盖分钟边界、5 秒轮询间隔、120 秒 run 超时和少量余量）出现一条状态为success的新历史记录；若终态是failed等其他状态则抛出错误。成功时打印Schedule run、Agent run（对应的 Agent run id）、History total和Status。
02_rest_api.py连续触发两次后，断言两次记录分别出现在page=1和page=2，meta.total_count >= 2，且两页的 run id 不同。TEST_LOG 中记录的实际运行结果为：GET /health返回ok，两次触发的记录分别落到了第 1、2 页，顶层键恰好是data和meta。

用 Python 直接管理计划（可选路径）

不经过 HTTP 时，可以用ScheduleManager（同步用PostgresDb，异步用AsyncPostgresDb），示例见 03_manage_with_python.py：

.venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/03_manage_with_python.py

这条路径有两个容易踩的坑，文档明确说明：

字段名不一致：create()接收cron，但更新走数据库字段名，所以更新 cron 必须传cron_expr="..."。示例先disable()，再update(cron_expr=...)，最后enable()让next_run_at按新 cron 重新计算。
验证行为：非法 cron 表达式、不存在的时区、重复的计划名都会抛出ValueError，示例验证了这三类错误。

查询历史用get_runs(schedule_id, limit=..., page=...)；从未执行过的计划查不到任何记录（示例对此做了断言）。

边界与限制

时区：timezone使用 IANA 名称（示例用UTC、America/New_York），传不存在的名会校验失败。
并发安全：多 worker 共享计划必须用 Postgres（见准备条件一节）。
执行语义：计划触发永远是后台非流式 run，执行器轮询到终态为止；status的可能取值在观察器代码中体现为success/failed/paused/timeout，非 success 时记录里带error字段。
重试配置：max_retries和retry_delay_seconds在创建时传入（REST 和ScheduleManager.create()均支持），示例值为 1 次重试、5 秒延迟；timeout_seconds控制单次 run 的超时。
文档还提供了一个 agentic 路径 04_scheduler_tools_agent.py：给 Agent 挂SchedulerTools工具，用自然语言创建计划，并通过SchedulerTools子类收窄 create schema，让调用方无法覆盖配置好的默认 endpoint 和 payload。它属于可选玩法，历史查看方式与本文 REST 路径相同。