Schedules SDK Reference - Orq.ai Documentation

Schedules

List Schedules

Lists all schedules attached to the specified agent, most recent first.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    res = orq.schedules.list(agent_key="<value>")

    # Handle response
    print(res)

Show Parameters

{
    "agent_key": str,  # required
}

Show Response

{
    "schedules": {
        "id": str,
        "agent_key": str,
        "agent_tag": Optional[str],
        "created": date,
        "created_by_id": str,
        "expression": str,
        "generation": int,
        "is_active": bool,
        "last_triggered_at": date,
        "payload": {
            "input": Optional[Any],
            "memory_entity_id": Optional[str],
            "metadata": Dict[str, Any],
            "variables": Dict[str, Any],
        },
        "trigger_count": int,
        "type": Literal["cron", "once", "interval"],
        "updated": date,
    },
}

Create a Schedule

Creates a schedule that runs the agent on a recurring or one-off cadence. The minimum firing interval is 1 hour for cron and interval; once schedules are exempt.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    res = orq.schedules.create(agent_key="<value>", expression="0 0 9 * * mon-fri", payload={
        "input": "Generate the morning briefing for {{region}}",
        "memory_entity_id": "mem_entity_123",
        "metadata": {
            "run_source": "daily-briefing",
        },
        "variables": {
            "region": "EMEA",
        },
    }, type_="cron", agent_tag="v2")

    # Handle response
    print(res)

Show Parameters

{
    "agent_key": str,  # required
    "expression": str,  # required
    "payload": {  # required
        "input": Optional[Any],
        "memory_entity_id": Optional[str],
        "metadata": Dict[str, Any],
        "variables": Dict[str, Any],
    },
    "type": Literal["cron", "once", "interval"],  # required
    "agent_tag": Optional[str],
}

Show Response

{
    "id": str,
    "agent_key": str,
    "agent_tag": Optional[str],
    "created": date,
    "created_by_id": str,
    "expression": str,
    "generation": int,
    "is_active": bool,
    "last_triggered_at": date,
    "payload": {
        "input": Optional[Any],
        "memory_entity_id": Optional[str],
        "metadata": Dict[str, Any],
        "variables": Dict[str, Any],
    },
    "trigger_count": int,
    "type": Literal["cron", "once", "interval"],
    "updated": date,
}

Delete a Schedule

Permanently removes a schedule from NATS, Mongo, and the Redis cache.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    orq.schedules.delete(agent_key="<value>", schedule_id="<id>")

    # Use the SDK ...

Show Parameters

{
    "agent_key": str,  # required
    "schedule_id": str,  # required
}

Retrieve a Schedule

Retrieves a single schedule by ID.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    res = orq.schedules.retrieve(agent_key="<value>", schedule_id="<id>")

    # Handle response
    print(res)

Show Parameters

{
    "agent_key": str,  # required
    "schedule_id": str,  # required
}

Show Response

{
    "id": str,
    "agent_key": str,
    "agent_tag": Optional[str],
    "created": date,
    "created_by_id": str,
    "expression": str,
    "generation": int,
    "is_active": bool,
    "last_triggered_at": date,
    "payload": {
        "input": Optional[Any],
        "memory_entity_id": Optional[str],
        "metadata": Dict[str, Any],
        "variables": Dict[str, Any],
    },
    "trigger_count": int,
    "type": Literal["cron", "once", "interval"],
    "updated": date,
}

Update a Schedule

Partially updates a schedule. Any omitted field is left unchanged. Changing expression or type (or reactivating from inactive) re-publishes the NATS schedule and bumps generation; payload-only and agent_tag-only changes leave the firing cadence in place.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    res = orq.schedules.update(agent_key="<value>", schedule_id="<id>", expression="@every 6h")

    # Handle response
    print(res)

Show Parameters

{
    "agent_key": str,  # required
    "schedule_id": str,  # required
    "agent_tag": Optional[str],
    "expression": Optional[str],
    "is_active": Optional[bool],
    "payload": {  # optional
        "input": Optional[Any],
        "memory_entity_id": Optional[str],
        "metadata": Dict[str, Any],
        "variables": Dict[str, Any],
    },
    "type": Optional[Literal["cron", "once", "interval"]],
}

Show Response

{
    "id": str,
    "agent_key": str,
    "agent_tag": Optional[str],
    "created": date,
    "created_by_id": str,
    "expression": str,
    "generation": int,
    "is_active": bool,
    "last_triggered_at": date,
    "payload": {
        "input": Optional[Any],
        "memory_entity_id": Optional[str],
        "metadata": Dict[str, Any],
        "variables": Dict[str, Any],
    },
    "trigger_count": int,
    "type": Literal["cron", "once", "interval"],
    "updated": date,
}

Trigger a Schedule

Runs the schedule’s payload immediately (≈10 seconds after the request, to stay above the NATS scheduler’s minimum deliver-at margin). The schedule’s regular cadence is unaffected. Inactive schedules return 400.

from orq_ai_sdk import Orq
import os

with Orq(
    api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:

    res = orq.schedules.trigger(agent_key="<value>", schedule_id="<id>")

    # Handle response
    print(res)

Show Parameters

{
    "agent_key": str,  # required
    "schedule_id": str,  # required
}

Show Response

{
    "schedule_id": str,
    "status": str,
}

​Schedules

​List Schedules

​Create a Schedule

​Delete a Schedule

​Retrieve a Schedule

​Update a Schedule

​Trigger a Schedule

Schedules

List Schedules

Create a Schedule

Delete a Schedule

Retrieve a Schedule

Update a Schedule

Trigger a Schedule