Inspect and create schedules¶
Use a dedicated sandbox profile. Listing uses a direct Hermes Desktop HTTP endpoint and a concrete native profile name. Set HERMES_DESKTOP_TOKEN if required. It prints that profile's jobs and creates nothing:
"""List schedules on a concrete Hermes Desktop profile."""
import argparse
import asyncio
import json
import os
from edith_hermes import AsyncHermesDesktop
async def inspect(desktop):
return await desktop.cron.list()
async def main(args):
async with AsyncHermesDesktop.remote(
base_url=args.url, profile=args.profile, token=os.getenv("HERMES_DESKTOP_TOKEN")
) as desktop:
print(json.dumps(await inspect(desktop), indent=2))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--url", required=True, help="Hermes Desktop HTTP URL")
parser.add_argument("--profile", required=True)
asyncio.run(main(parser.parse_args()))
Creating a job requires the control-plane URL, opaque instance/profile IDs, optional EDITH_TOKEN, the streaming extra, and a runtime implementing the versioned cron extension:
python examples/schedule_create.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE --name "SDK example" --prompt "Review notes" --schedule "every 2h" --run
The job is created paused. The script prints its request ID before dispatch and the creation receipt afterward. If the write times out, it queries the receipt using the same ID instead of issuing a second write. An uncertain, pending, or unavailable receipt does not prove the job was absent; inspect before retrying. Delete the sandbox job through your deployment after the exercise. Enabling it later can invoke a model and incur charges.
"""Create a paused schedule and reconcile a timeout using its request ID."""
import argparse
import asyncio
import json
import os
from uuid import uuid4
from edith_hermes import AsyncScheduleMutations, Hermes
async def create(connection, profile_id, request_id, name, prompt, schedule):
schedules = AsyncScheduleMutations(connection, profile=profile_id)
options = {"name": name, "prompt": prompt, "schedule": schedule, "paused": True}
print("request ID:", request_id, flush=True)
try:
return await schedules.create(options, request_id=request_id)
except TimeoutError:
# A timeout does not establish whether the write happened. Query the same
# request receipt instead of issuing a second creation.
return await schedules.operation(request_id)
async def main(args):
if not args.run:
raise SystemExit("Pass --run to create a paused schedule")
async with Hermes.remote(args.url, token=os.getenv("EDITH_TOKEN")) as hermes:
profile = hermes.instances(args.instance).profiles(args.profile)
native_profile = (await profile.get())["name"]
async with profile.connect() as connection:
result = await create(
connection, native_profile, str(uuid4()), args.name, args.prompt, args.schedule
)
print(json.dumps(result, indent=2))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--url", required=True)
parser.add_argument("--instance", required=True)
parser.add_argument("--profile", required=True)
parser.add_argument("--name", required=True)
parser.add_argument("--prompt", required=True)
parser.add_argument("--schedule", required=True, help='For example: "every 2h"')
parser.add_argument("--run", action="store_true")
asyncio.run(main(parser.parse_args()))