OpenAI Agents SDK cron jobs with approved API tools

import asyncio
import json
import os
from typing import Any
from urllib.error import HTTPError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen
from zoneinfo import ZoneInfo
 
from agents import Agent, Runner
from agents.decorators import tool
 
API_BASE = "https://api.crontap.com"
OUTPUT_FIELDS = {
    "id",
    "status",
    "url",
    "verb",
    "interval",
    "timezone",
    "label",
    "cron",
    "description",
    "nextRuns",
    "question",
    "options",
    "roundsLeft",
    "reason",
}
SCALAR_OUTPUT_FIELDS = OUTPUT_FIELDS - {"nextRuns", "options"}
OPTION_OUTPUT_FIELDS = {"label", "cron", "recommended"}
 
 
def api_request(path: str, body: dict[str, Any]) -> Any:
    request = Request(
        f"{API_BASE}{path}",
        data=json.dumps(body).encode(),
        method="POST",
        headers={
            "ApiKey": os.environ["CRONTAP_API_KEY"],
            "ClientId": os.environ["CRONTAP_CLIENT_ID"],
            "Content-Type": "application/json",
        },
    )
    try:
        with urlopen(request, timeout=15) as response:
            return json.load(response)
    except HTTPError as error:
        detail = error.read(1000).decode(errors="replace")
        raise RuntimeError(
            f"Crontap API returned {error.code}: {detail}"
        ) from error
 
 
def safe_output(payload: Any) -> dict[str, Any]:
    if not isinstance(payload, dict):
        raise RuntimeError("Unexpected Crontap response")
    result = {
        key: payload[key]
        for key in SCALAR_OUTPUT_FIELDS
        if isinstance(payload.get(key), (str, int, float, bool))
    }
    if isinstance(payload.get("nextRuns"), list):
        result["nextRuns"] = [
            value
            for value in payload["nextRuns"]
            if isinstance(value, str)
        ]
    if isinstance(payload.get("options"), list):
        result["options"] = [
            {
                key: option[key]
                for key in OPTION_OUTPUT_FIELDS
                if key in option
                and (
                    isinstance(
                        option[key], (str, int, float, bool)
                    )
                    or key == "cron"
                    and option[key] is None
                )
            }
            for option in payload["options"]
            if isinstance(option, dict)
        ]
    return result
 
 
@tool(needs_approval=True)
def preview_crontap_schedule(
    text: str,
    timezone: str = "UTC",
) -> dict[str, Any]:
    """Preview schedule text without saving a resource."""
    ZoneInfo(timezone)
    if not 1 <= len(text.strip()) <= 500:
        raise ValueError("text must contain 1 to 500 characters")
    result = api_request(
        "/v1/schedule/preview",
        {"text": text.strip(), "timezone": timezone},
    )
    return safe_output(result)
 
 
@tool(needs_approval=True)
def create_crontap_post_schedule(
    url: str,
    interval: str,
    timezone: str,
    label: str,
) -> dict[str, Any]:
    """Create one POST schedule for an approved HTTPS host."""
    parsed = urlsplit(url)
    allowed_hosts = {
        host.strip().lower()
        for host in os.environ[
            "CRONTAP_ALLOWED_TARGET_HOSTS"
        ].split(",")
        if host.strip()
    }
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username
        or parsed.password
        or parsed.hostname.lower() not in allowed_hosts
    ):
        raise ValueError("URL host is not approved")
    ZoneInfo(timezone)
    if not 1 <= len(label.strip()) <= 120:
        raise ValueError("label must contain 1 to 120 characters")
    result = api_request(
        "/v1/schedule",
        {
            "url": url,
            "verb": "POST",
            "interval": interval,
            "timezone": timezone,
            "label": label.strip(),
        },
    )
    return safe_output(result)
 
 
agent = Agent(
    name="Crontap schedule manager",
    instructions=(
        "Preview first. Show the exact schedule fields. "
        "Create only after the person approves the tool call."
    ),
    tools=[
        preview_crontap_schedule,
        create_crontap_post_schedule,
    ],
)
 
 
def approve(tool_name: str, arguments: str | None) -> bool:
    answer = input(
        f"Approve {tool_name} with {arguments}? [y/N] "
    )
    return answer.strip().lower() in {"y", "yes"}
 
 
async def main() -> None:
    request = (
        "Preview weekdays at 07:45 in Europe/Paris. "
        "Then create partner feed refresh for "
        "https://api.example.com/partners/refresh."
    )
    result = await Runner.run(agent, request)
    while result.interruptions:
        state = result.to_state()
        for item in result.interruptions:
            if approve(item.name or "unknown_tool", item.arguments):
                state.approve(item, always_approve=False)
            else:
                state.reject(item)
        result = await Runner.run(agent, state)
    print(result.final_output)
 
 
if __name__ == "__main__":
    asyncio.run(main())