AgentOS は cron スタイルの組み込みのスケジューラを備えています。これを使用して、朝のブリーフィング、日次トリアージ (タスク選別)、週次ダイジェスト、毎時のヘルスチェックなどを実行できます。Scout のような高度なエージェントシステムも、これを使用してリマインダーの設定や独自のスケジュール管理を行います。

Agno 再入門 – 機能 : スケジューリング

作成 : クラスキャット・セールスインフォメーション
作成日時 : 06/17/2026
バージョン : v2.6.16

* 本記事は docs.agno.com の以下のページを参考にしています :

• Features : Scheduling

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。

 

 

 

Agno 再入門 – 機能 : スケジューリング

繰り返しタスクを実行するための組み込み cron 機能。

AgentOS は cron スタイルの組み込みのスケジューラを備えています。これを使用して、朝のブリーフィング、日次トリアージ (タスク選別)、週次ダイジェスト、毎時のヘルスチェックなどを実行できます。Scout のような高度なエージェントシステムも、これを使用してリマインダーの設定や独自のスケジュール管理を行います。

登録されたスケジュールは AgentOS データベース（agno_schedules テーブル）に保存され、エージェントが使用するのと同じ FastAPI プロセスを使用します。

from agno.os import AgentOS

agent_os = AgentOS(
    agents=[agent],
    db=db,
    scheduler=True,
    scheduler_poll_interval=15,    # check for due jobs every N seconds
)

スケジューラは、scheduler_poll_interval 秒ごとに agno_schedules をポーリングし、実行期日の来たジョブを実行し、失敗したジョブを再試行し、状態を永続化します。

 

スケジュールを作成する 2 つの方法

• (パターン – How)

• エージェント管理 (Agent Managed) – エージェントは、アクティブなスケジュールを作成、読み取り、更新するためのツールを備えています。ユーザーはチャットでエージェントにスケジューリングを依頼し、エージェントはツール呼び出しを使用してスケジュールを管理します。

• 手動登録 – コード内でスケジュールを作成し、起動時に登録します。

 

エージェント管理 (Agent Managed)

エージェントに SchedulerTools を提供すると、チャット経由で自身の作業をスケジューリングできるようになります :

from agno.tools.scheduler import SchedulerTools

agent = Agent(
    model="openai:gpt-5.4",
    tools=[
        SchedulerTools(
            db=db,
            default_endpoint="/agents/my-agent/runs",
            default_method="POST",
            default_timezone="UTC",
        ),
    ],
)

# In Slack: "@MyAgent post a daily digest of open PRs at 9am ET"
# The agent calls SchedulerTools.create_schedule() with a cron expr.

デモ OS には、この処理を行うスケジューラ・エージェントが含まれています。

 

手動登録

常に実行されるべきスケジュール（日次ダイジェスト、時間ごとの同期、夜間クリーンアップ 等）については、ScheduleManager を使用してアプリの lifespan 内でそれらを作成します :

from contextlib import asynccontextmanager
from agno.scheduler import ScheduleManager

@asynccontextmanager
async def lifespan(app):
    manager = ScheduleManager(db=db)
    manager.create(
        name="daily_digest",
        cron="0 9 * * 1-5",                       # weekdays 9am
        endpoint="/workflows/daily-digest/runs",
        if_exists="update",                       # idempotent on restart
    )
    yield

agent_os = AgentOS(..., db=db, scheduler=True, lifespan=lifespan)

`if_exists=”update”` を指定すると、呼び出しが冪等 (べきとう, idempotent) になります – 再起動時に再実行しても、既存のスケジュールが更新され、例外が発生したり重複したりすることはありません。手動で編集したスケジュールをそのままにしておきたい場合は “skip” を、意図しない名前の衝突を顕在化させたい場合は “raise” (デフォルト) を指定してください。これは、Coda が日次ダイジェスト、課題トリアージ、リポジトリ同期に使用するパターンです。

 

複数ステップのジョブのためのワークフロー

スケジュールは単一のエンドポイントを起動します。調査、アウトライン作成、下書き、レビューなど、複数のステップから構成される作業の場合、ワークフローが役立ちます。ワークフローは厳密にはスケジューリング機能ではありませんが、スケジュールが起動する最も一般的な機能です。

ワークフローは型付きパイプラインです。ステップは順番に実行されます。Parallel はそれらを同時に実行します。Loop は条件が満たされるまで繰り返します。Router は一つの分岐を選択します。

from agno.workflow import Workflow, Step, Parallel, Loop, Router, Condition

workflow = Workflow(
    name="content_pipeline",
    steps=[
        Step(name="research", agent=researcher),
        Step(name="outline", agent=outliner),
        Loop(
            name="draft_review",
            steps=[
                Step(name="draft", agent=writer),
                Step(name="review", agent=editor),
            ],
            end_condition='last_step_content.contains("APPROVED")',
            max_iterations=3,
        ),
    ],
)

Loop.end_condition は、CEL 式文字列 (上記参照) または (反復ステップの出力を受け取りブール値を返す) 呼び出し可能オブジェクト (callable) を受け入れます。Condition はワークフロー内の if/else 分岐のための独立したプリミティブです – 動的な分岐には Loop ではなく Router と組み合わせます。

ワークフローは、AgentOS の第一級市民です; ワークフローは /workflows/<id>/runs の POST エンドポイントとして提供され、スケジュール実行やトレースが可能であり、同じ db に永続化されます。

• (パターン – Use when)

• シーケンシャル – ステップが相互に依存する場合

• パラレル – ステップが独立しており、ファンアウトが必要な場合

• 条件付きループ – 品質しきい値または最大反復回数

• ルーター ＋ 条件：入力に基づく動的分岐

• クロスモーダル・チェイニング – あるエージェントの出力が別のモダリティへの入力となる場合（テキスト → 音声、コード → ナレーション）

For worked examples, see Demo OS.

 

スケジュール実行とオブザーバビリティ

スケジュールが実行されると、AgentOS は次の処理を実行します :

1. agno_schedules からスケジュールを検索し、行レベルのリースを使って、スケジュールの所有権を確保します。

2. httpx.AsyncClient を介して、設定済みのエンドポイント (POST /agents/<id>/runs または POST /workflows/<id>/runs) を HTTP 経由で呼び出します – これは、認証ヘッダーを含め、外部の呼び出し元が使用するのと同じパスです。

3. 結果は agno_schedule_runs に記録され (ステータス、試行回数、タイミング、エラー (if any))、基となった (underlying) 実行は agno_sessions および agno_traces に他の実行と同様に記録されます。

つまり、スケジュールされた作業は、アドホックな作業と同じ UI、同じ SQL クエリ、同じトレースツリーに表示されます。過去24時間に実行された内容を確認するには :

SELECT
    s.name,
    sr.status,
    sr.triggered_at,
    (sr.completed_at - sr.triggered_at) AS duration_s
FROM agno_schedule_runs sr
JOIN agno_schedules s ON s.id = sr.schedule_id
WHERE sr.created_at > extract(epoch from NOW() - INTERVAL '24 hours')::bigint
ORDER BY sr.created_at DESC;

スケジュール実行時のタイムスタンプはエポック秒 (BigInt) として保存されます。特定のスケジュール実行のトレースについては、agno_schedule_runs から run_id をたどって agno_traces にアクセスしてください。See Observability for the full data model.

以上