[FR]: Native schedule/cron/condition trigger support for agents — reference implementation available

[STHITAPRAJNAS]

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:

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.