[FR]: Native schedule/cron/condition trigger support for agents — reference implementation available
- 主要言語
- Python
- スター
- 21.5k
- フォーク
- 4k
- 平均マージ
- 1日 14時間
- マージ済み PR(30日)
- 37
説明
## Summary
ADK agents are purely invocation-driven — they only run when an external system
sends a message. This FR proposes a native scheduling primitive so agents can
self-wake on cron expressions, fixed intervals, or arbitrary conditions, without
requiring a separate Cloud Scheduler or external orchestrator.
A working reference implementation is available at:
**https://github.com/STHITAPRAJNAS/adk-task-scheduler** (PyPI: `adk-task-scheduler`)
## The Problem
Teams using ADK for production monitoring, report generation, data pipelines,
and alert agents must bolt on an external scheduler today. Common patterns:
- Cloud Scheduler → HTTP → `/run` (requires deployed endpoint, adds latency)
- Celery / APScheduler wired separately (no ADK-native session/memory integration)
- Custom `asyncio.create_task` loops (breaks on pod restart, no cron support)
None of these integrate with ADK's session service, memory service, or artifact
service out of the box.
## The Proposal
A `ScheduleConfig` primitive attached to any `BaseAgent` that defines:
- `cron`: standard 5-field expression
- `interval_seconds`: fixed interval
- `condition`: zero- or one-arg callable (sync/async) — fire when truthy
- `fire_mode`: `"every"` or `"once_until_reset"` (edge-triggered)
- `condition_backoff_factor`: exponential back-off on repeated false evaluations
Combined with `condition + cron/interval` gating (schedule controls frequency,
condition controls whether to fire).
Injected into `get_fast_api_app` via the existing `lifespan=` parameter —
**zero patching of ADK internals required**.
Scheduled runners use the same `create_session_service_from_options`,
`create_artifact_service_from_options`, `create_memory_service_from_options`
helpers that `get_fast_api_app` uses internally, so all backends (Vertex AI,
GCS, SQLAlchemy) work identically for scheduled and ad-hoc invocations.
## Reference Implementation
`adk-task-scheduler` (https://github.com/STHITAPRAJNAS/adk-task-scheduler) is a
production-ready PyPI package that implements this proposal today:
```python
from adk_task_scheduler import with_schedule, build_scheduled_app
root_agent = with_schedule(
Agent(name="daily_report", model="gemini-2.0-flash", instruction="..."),
cron="0 9 * * 1-5", # weekdays at 09:00
condition=lambda ctx: is_market_open(), # additional gate
fire_mode="once_until_reset",
on_response=send_to_slack,
)
app = build_scheduled_app(agents_dir="./agents")
```
The library has **78 tests**, CI across **Python 3.10–3.13**, and full support
for Vertex AI Agent Engine, GCS, and SQLAlchemy session/artifact/memory backends.
## Integration Options
Happy to contribute this as any of the following — open to ADK team preference:
| Option | Description | Effort |
|--------|-------------|--------|
| `google.adk.scheduling` | Absorbed into core | High |
| `google.adk.ext.scheduler` | Contrib package, optional APScheduler dep | Medium |
| Documented pattern | Official guidance on `lifespan=` for scheduling | Low |
The `lifespan=` injection point in `get_fast_api_app` already makes **Option 3
trivial** with zero changes to ADK itself. Options 1/2 would additionally let
the team swap the trigger backend (APScheduler, Cloud Tasks, Celery) independently
of the scheduling API.
## Questions for the ADK Team
- Is in-process scheduling the right model, or is **Cloud Scheduler → HTTP → `/run`**
the preferred architecture for Vertex AI Agent Engine deployments?
- Would a `google.adk.ext.scheduler` contrib package with an optional APScheduler
dependency be acceptable under ADK's contribution guidelines?
- Are there plans for a **`Runner.run_headless()`** mode — one that doesn't require
a synthetic user message as the invocation trigger?
Any feedback on the design, preferred integration path, or contribution process
would be greatly appreciated.
コントリビューションガイド
評価
この issue はまだ評価されていません。