diff --git a/videodb/__init__.py b/videodb/__init__.py
index 41244dc..aecc496 100644
--- a/videodb/__init__.py
+++ b/videodb/__init__.py
@@ -7,6 +7,7 @@
 from videodb._utils._video import play_stream
 from videodb._constants import (
     VIDEO_DB_API,
+    AgenticStreamRunStatus,
     IndexType,
     SceneExtractionType,
     MediaType,
@@ -22,6 +23,8 @@
     AudioConfig,
 )
 from videodb.client import Connection
+from videodb.agentic_stream import AgenticStream, AgenticStreamRun
+from videodb.schedule import Schedule
 from videodb.exceptions import (
     VideodbError,
     AuthenticationError,
@@ -33,6 +36,10 @@
 
 
 __all__ = [
+    "AgenticStream",
+    "AgenticStreamRun",
+    "AgenticStreamRunStatus",
+    "Schedule",
     "VideodbError",
     "AuthenticationError",
     "InvalidRequestError",
diff --git a/videodb/_constants.py b/videodb/_constants.py
index de2f73b..213c244 100644
--- a/videodb/_constants.py
+++ b/videodb/_constants.py
@@ -97,6 +97,12 @@ class ApiPath:
     record = "record"
     editor = "editor"
     reframe = "reframe"
+    agentic_stream = "agentic_stream"
+    runs = "runs"
+    schedule = "schedule"
+    templates = "templates"
+    events = "events"
+    logs = "logs"
 
 
 class Status:
@@ -110,6 +116,15 @@ class MeetingStatus:
     done = "done"
 
 
+class AgenticStreamRunStatus:
+    queued = "queued"
+    running = "running"
+    completed = "completed"
+    failed = "failed"
+    cancelled = "cancelled"
+    terminal = {"completed", "failed", "cancelled"}
+
+
 class HttpClientDefaultValues:
     max_retries = 1
     timeout = 30
diff --git a/videodb/agentic_stream.py b/videodb/agentic_stream.py
new file mode 100644
index 0000000..f7cdc79
--- /dev/null
+++ b/videodb/agentic_stream.py
@@ -0,0 +1,333 @@
+import time
+
+from typing import List, Optional
+
+from videodb._constants import AgenticStreamRunStatus, ApiPath
+from videodb.exceptions import VideodbError
+
+
+class AgenticStreamRun:
+    """AgenticStreamRun class representing a single execution of an agentic stream.
+
+    :ivar str id: Unique identifier for the run
+    :ivar str agentic_stream_id: ID of the parent agentic stream
+    :ivar str collection_id: ID of the collection
+    :ivar str status: Current status (queued/running/completed/failed/cancelled)
+    :ivar dict config: Immutable config snapshot captured at run creation
+    :ivar dict progress: Current progress ``{stage, percent, message}``
+    :ivar str video_id: ID of the produced video (populated on completion)
+    :ivar str stream_url: Stream URL of the produced video
+    :ivar str player_url: Player URL of the produced video
+    :ivar str thumbnail_url: Thumbnail URL of the produced video
+    :ivar float duration: Duration of the produced video in seconds
+    :ivar list events: Compiled clip list (populated on completion)
+    :ivar str error: Error message when the run failed
+    """
+
+    def __init__(
+        self, _connection, id: str, agentic_stream_id: str, collection_id: str, **kwargs
+    ) -> None:
+        self._connection = _connection
+        self.id = id
+        self.agentic_stream_id = agentic_stream_id
+        self.collection_id = collection_id
+        self._update_attributes(kwargs)
+
+    def __repr__(self) -> str:
+        return (
+            f"AgenticStreamRun("
+            f"id={self.id}, "
+            f"agentic_stream_id={self.agentic_stream_id}, "
+            f"collection_id={self.collection_id}, "
+            f"status={self.status}, "
+            f"progress={self.progress}, "
+            f"config={self.config}, "
+            f"video_id={self.video_id}, "
+            f"stream_url={self.stream_url}, "
+            f"player_url={self.player_url}, "
+            f"thumbnail_url={self.thumbnail_url}, "
+            f"duration={self.duration}, "
+            f"events={self.events}, "
+            f"error={self.error}, "
+            f"callback_url={self.callback_url}, "
+            f"callback_data={self.callback_data}, "
+            f"created_at={self.created_at}, "
+            f"updated_at={self.updated_at}, "
+            f"completed_at={self.completed_at})"
+        )
+
+    def _update_attributes(self, data: dict) -> None:
+        self.status = data.get("status")
+        self.config = data.get("config")
+        self.progress = data.get("progress")
+        self.video_id = data.get("video_id")
+        self.stream_url = data.get("stream_url")
+        self.player_url = data.get("player_url")
+        self.thumbnail_url = data.get("thumbnail_url")
+        self.duration = data.get("duration")
+        self.events = data.get("events")
+        self.error = data.get("error")
+        self.callback_url = data.get("callback_url")
+        self.callback_data = data.get("callback_data")
+        self.created_at = data.get("created_at")
+        self.updated_at = data.get("updated_at")
+        self.completed_at = data.get("completed_at")
+
+    @property
+    def _base_path(self) -> str:
+        return (
+            f"{ApiPath.collection}/{self.collection_id}/"
+            f"{ApiPath.agentic_stream}/{self.agentic_stream_id}/"
+            f"{ApiPath.runs}/{self.id}"
+        )
+
+    @property
+    def is_complete(self) -> bool:
+        """True if the run has reached a terminal status."""
+        return self.status in AgenticStreamRunStatus.terminal
+
+    @property
+    def is_successful(self) -> bool:
+        """True if the run completed successfully."""
+        return self.status == AgenticStreamRunStatus.completed
+
+    def refresh(self) -> "AgenticStreamRun":
+        """Refresh run data from the server.
+
+        :return: The updated run instance
+        :rtype: AgenticStreamRun
+        """
+        response = self._connection.get(path=self._base_path)
+        if response:
+            self._update_attributes(response)
+        else:
+            raise VideodbError(f"Failed to refresh run {self.id}")
+        return self
+
+    def wait(self, timeout: int = 3600, poll_interval: int = 5) -> "AgenticStreamRun":
+        """Block until the run reaches a terminal status.
+
+        :param int timeout: Maximum seconds to wait (default 3600)
+        :param int poll_interval: Seconds between polls (default 5)
+        :return: The run instance in a terminal state
+        :rtype: AgenticStreamRun
+        :raises VideodbError: If the run failed or the timeout is exceeded
+        """
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            self.refresh()
+            if self.is_complete:
+                if self.status == AgenticStreamRunStatus.failed:
+                    raise VideodbError(f"Run {self.id} failed: {self.error}")
+                return self
+            time.sleep(poll_interval)
+        raise VideodbError(f"Run {self.id} did not complete within {timeout}s")
+
+    def cancel(self) -> "AgenticStreamRun":
+        """Cancel a queued or running run.
+
+        :return: The updated run instance
+        :rtype: AgenticStreamRun
+        """
+        response = self._connection.patch(
+            path=f"{self._base_path}/{ApiPath.status}",
+            data={"status": AgenticStreamRunStatus.cancelled},
+        )
+        if response:
+            self._update_attributes(response)
+        return self
+
+    def get_events(self) -> List[dict]:
+        """Get the user-facing event timeline of this run.
+
+        :return: List of event dicts
+        :rtype: List[dict]
+        """
+        response = self._connection.get(path=f"{self._base_path}/{ApiPath.events}")
+        return (response or {}).get("events", [])
+
+    def get_logs(self) -> List[dict]:
+        """Get the debug logs of this run.
+
+        :return: List of log dicts
+        :rtype: List[dict]
+        """
+        response = self._connection.get(path=f"{self._base_path}/{ApiPath.logs}")
+        return (response or {}).get("logs", [])
+
+
+class AgenticStream:
+    """AgenticStream class representing a persistent agent video pipeline.
+
+    Note: Users should not initialize this class directly.
+    Instead use :meth:`Collection.create_agentic_stream() <videodb.collection.Collection.create_agentic_stream>`
+
+    :ivar str id: Unique identifier for the agentic stream
+    :ivar str collection_id: ID of the collection this stream belongs to
+    :ivar str name: Human-readable name
+    :ivar str template_id: ID of the template this stream uses
+    :ivar str prompt: Content instructions for the agent
+    :ivar list sources: Keywords/URLs used for research each run
+    """
+
+    def __init__(self, _connection, id: str, collection_id: str, **kwargs) -> None:
+        self._connection = _connection
+        self.id = id
+        self.collection_id = collection_id
+        self._update_attributes(kwargs)
+
+    def __repr__(self) -> str:
+        return (
+            f"AgenticStream("
+            f"id={self.id}, "
+            f"collection_id={self.collection_id}, "
+            f"name={self.name}, "
+            f"template_id={self.template_id}, "
+            f"prompt={self.prompt}, "
+            f"sources={self.sources}, "
+            f"max_duration={self.max_duration}, "
+            f"voice={self.voice}, "
+            f"language={self.language}, "
+            f"aspect_ratio={self.aspect_ratio}, "
+            f"captions={self.captions}, "
+            f"model={self.model}, "
+            f"callback_url={self.callback_url}, "
+            f"callback_data={self.callback_data}, "
+            f"status={self.status}, "
+            f"created_at={self.created_at}, "
+            f"updated_at={self.updated_at})"
+        )
+
+    def _update_attributes(self, data: dict) -> None:
+        self.name = data.get("name")
+        self.template_id = data.get("template_id")
+        self.prompt = data.get("prompt")
+        self.sources = data.get("sources")
+        self.max_duration = data.get("max_duration")
+        self.voice = data.get("voice")
+        self.language = data.get("language")
+        self.aspect_ratio = data.get("aspect_ratio")
+        self.captions = data.get("captions")
+        self.model = data.get("model")
+        self.callback_url = data.get("callback_url")
+        self.callback_data = data.get("callback_data")
+        self.status = data.get("status")
+        self.created_at = data.get("created_at")
+        self.updated_at = data.get("updated_at")
+
+    @property
+    def _base_path(self) -> str:
+        return (
+            f"{ApiPath.collection}/{self.collection_id}/"
+            f"{ApiPath.agentic_stream}/{self.id}"
+        )
+
+    def refresh(self) -> "AgenticStream":
+        """Refresh stream data from the server.
+
+        :return: The updated stream instance
+        :rtype: AgenticStream
+        """
+        response = self._connection.get(path=f"{self._base_path}/")
+        if response:
+            self._update_attributes(response)
+        else:
+            raise VideodbError(f"Failed to refresh agentic stream {self.id}")
+        return self
+
+    def update(self, **kwargs) -> "AgenticStream":
+        """Update stream fields.
+
+        :param kwargs: Fields to update (name, template_id, prompt, sources, max_duration,
+            voice, language, aspect_ratio, captions, model, callback_url,
+            callback_data)
+        :return: The updated stream instance
+        :rtype: AgenticStream
+        """
+        response = self._connection.patch(path=f"{self._base_path}/", data=kwargs)
+        if response:
+            self._update_attributes(response)
+        return self
+
+    def delete(self) -> None:
+        """Delete the agentic stream.
+
+        Existing runs, output videos, and schedules are not deleted.
+        Delete any associated schedules separately.
+
+        :return: None
+        :rtype: None
+        """
+        self._connection.delete(path=f"{self._base_path}/")
+
+    def run_now(
+        self,
+        callback_url: Optional[str] = None,
+        callback_data: Optional[dict] = None,
+    ) -> AgenticStreamRun:
+        """Trigger an on-demand run of this stream.
+
+        :param str callback_url: (optional) Per-run webhook override
+        :param dict callback_data: (optional) Per-run callback data override
+        :return: The created run
+        :rtype: AgenticStreamRun
+        """
+        data = {}
+        if callback_url is not None:
+            data["callback_url"] = callback_url
+        if callback_data is not None:
+            data["callback_data"] = callback_data
+        response = self._connection.post(
+            path=f"{self._base_path}/{ApiPath.runs}", data=data
+        )
+        if not response:
+            raise VideodbError("Failed to trigger run")
+        return AgenticStreamRun(
+            self._connection,
+            id=response.get("id"),
+            agentic_stream_id=self.id,
+            collection_id=self.collection_id,
+            **{k: v for k, v in response.items()
+               if k not in ("id", "agentic_stream_id", "collection_id")},
+        )
+
+    def list_runs(self) -> List[AgenticStreamRun]:
+        """List runs of this stream, most recent first.
+
+        :return: List of runs
+        :rtype: List[AgenticStreamRun]
+        """
+        response = self._connection.get(path=f"{self._base_path}/{ApiPath.runs}")
+        runs = (response or {}).get("runs", [])
+        return [
+            AgenticStreamRun(
+                self._connection,
+                id=run.get("id"),
+                agentic_stream_id=self.id,
+                collection_id=self.collection_id,
+                **{k: v for k, v in run.items()
+                   if k not in ("id", "agentic_stream_id", "collection_id")},
+            )
+            for run in runs
+        ]
+
+    def get_run(self, run_id: str) -> AgenticStreamRun:
+        """Get a run by its ID.
+
+        :param str run_id: ID of the run
+        :return: The run
+        :rtype: AgenticStreamRun
+        """
+        response = self._connection.get(
+            path=f"{self._base_path}/{ApiPath.runs}/{run_id}"
+        )
+        if not response:
+            raise VideodbError(f"Run {run_id} not found")
+        return AgenticStreamRun(
+            self._connection,
+            id=response.get("id"),
+            agentic_stream_id=self.id,
+            collection_id=self.collection_id,
+            **{k: v for k, v in response.items()
+               if k not in ("id", "agentic_stream_id", "collection_id")},
+        )
diff --git a/videodb/client.py b/videodb/client.py
index a01d10c..1ab5fe5 100644
--- a/videodb/client.py
+++ b/videodb/client.py
@@ -19,6 +19,8 @@
 from videodb.audio import Audio
 from videodb.image import Image
 from videodb.meeting import Meeting
+from videodb.agentic_stream import AgenticStream
+from videodb.schedule import Schedule
 
 from videodb._upload import (
     upload,
@@ -347,3 +349,120 @@ def get_meeting(self, meeting_id: str) -> Meeting:
         meeting = Meeting(self, id=meeting_id, collection_id="default")
         meeting.refresh()
         return meeting
+
+    def create_agentic_stream(
+        self,
+        name: str,
+        template_id: str,
+        prompt: str,
+        sources: Optional[List[str]] = None,
+        max_duration: Optional[int] = None,
+        voice: Optional[str] = None,
+        language: Optional[str] = None,
+        aspect_ratio: Optional[str] = None,
+        captions: Optional[bool] = None,
+        model: Optional[str] = None,
+        callback_url: Optional[str] = None,
+        callback_data: Optional[dict] = None,
+    ) -> AgenticStream:
+        """Create an agentic stream in the default collection.
+
+        Shortcut for ``conn.get_collection().create_agentic_stream(...)``.
+
+        :param str name: Human-readable name for the stream
+        :param str template_id: Template ID from
+            :meth:`Collection.list_agentic_stream_templates() <videodb.collection.Collection.list_agentic_stream_templates>`
+        :param str prompt: Content instructions for the agent
+        :param list sources: (optional) Keywords and/or website URLs for research
+        :param int max_duration: (optional) Target output duration in seconds
+        :param str voice: (optional) TTS voice name for narration
+        :param str language: (optional) Output language
+        :param str aspect_ratio: (optional) "16:9", "9:16", or "1:1"
+        :param bool captions: (optional) Burn captions into the output
+        :param str model: (optional) Quality tier: "basic" or "pro"
+        :param str callback_url: (optional) Default webhook for all runs
+        :param dict callback_data: (optional) Data echoed in webhook payloads
+        :return: The created agentic stream
+        :rtype: :class:`AgenticStream <videodb.agentic_stream.AgenticStream>`
+        """
+        return self.get_collection().create_agentic_stream(
+            name=name,
+            template_id=template_id,
+            prompt=prompt,
+            sources=sources,
+            max_duration=max_duration,
+            voice=voice,
+            language=language,
+            aspect_ratio=aspect_ratio,
+            captions=captions,
+            model=model,
+            callback_url=callback_url,
+            callback_data=callback_data,
+        )
+
+    def create_schedule(
+        self,
+        target: AgenticStream,
+        trigger: str,
+        timezone: Optional[str] = None,
+        callback_url: Optional[str] = None,
+        callback_data: Optional[dict] = None,
+    ) -> Schedule:
+        """Create a recurring schedule that triggers runs of an agentic stream.
+
+        :param AgenticStream target: The agentic stream to trigger
+        :param str trigger: EventBridge cron expression, e.g. ``"0 9 * * ? *"``
+        :param str timezone: (optional) IANA timezone name, default UTC
+        :param str callback_url: (optional) Webhook for each scheduled run
+        :param dict callback_data: (optional) Data echoed in webhook payloads
+        :return: The created schedule
+        :rtype: :class:`Schedule <videodb.schedule.Schedule>`
+        """
+        data = {
+            "target_type": "agentic_stream",
+            "target_id": target.id,
+            "trigger": trigger,
+        }
+        if timezone is not None:
+            data["timezone"] = timezone
+        if callback_url is not None:
+            data["callback_url"] = callback_url
+        if callback_data is not None:
+            data["callback_data"] = callback_data
+        response = self.post(path=f"{ApiPath.schedule}/", data=data)
+        return Schedule(
+            self,
+            id=response.get("id"),
+            **{k: v for k, v in response.items() if k != "id"},
+        )
+
+    def get_schedule(self, schedule_id: str) -> Schedule:
+        """Get a schedule by its ID.
+
+        :param str schedule_id: ID of the schedule
+        :return: The schedule
+        :rtype: :class:`Schedule <videodb.schedule.Schedule>`
+        """
+        response = self.get(path=f"{ApiPath.schedule}/{schedule_id}/")
+        return Schedule(
+            self,
+            id=response.get("id"),
+            **{k: v for k, v in response.items() if k != "id"},
+        )
+
+    def list_schedules(self) -> List[Schedule]:
+        """List all schedules for the authenticated user.
+
+        :return: List of schedules
+        :rtype: List[:class:`Schedule <videodb.schedule.Schedule>`]
+        """
+        response = self.get(path=f"{ApiPath.schedule}/")
+        schedules = (response or {}).get("schedules", [])
+        return [
+            Schedule(
+                self,
+                id=sched.get("id"),
+                **{k: v for k, v in sched.items() if k != "id"},
+            )
+            for sched in schedules
+        ]
diff --git a/videodb/collection.py b/videodb/collection.py
index 07299f9..d7b3334 100644
--- a/videodb/collection.py
+++ b/videodb/collection.py
@@ -14,6 +14,7 @@
 from videodb.audio import Audio
 from videodb.image import Image
 from videodb.meeting import Meeting
+from videodb.agentic_stream import AgenticStream
 from videodb.rtstream import RTStream
 from videodb.search import SearchFactory, SearchResult
 
@@ -578,3 +579,111 @@ def get_meeting(self, meeting_id: str) -> Meeting:
         meeting = Meeting(self._connection, id=meeting_id, collection_id=self.id)
         meeting.refresh()
         return meeting
+
+    def create_agentic_stream(
+        self,
+        name: str,
+        template_id: str,
+        prompt: str,
+        sources: Optional[List[str]] = None,
+        max_duration: Optional[int] = None,
+        voice: Optional[str] = None,
+        language: Optional[str] = None,
+        aspect_ratio: Optional[str] = None,
+        captions: Optional[bool] = None,
+        model: Optional[str] = None,
+        callback_url: Optional[str] = None,
+        callback_data: Optional[dict] = None,
+    ) -> AgenticStream:
+        """Create a persistent agentic stream in this collection.
+
+        :param str name: Human-readable name for the stream
+        :param str template_id: Template ID from
+            :meth:`list_agentic_stream_templates() <videodb.collection.Collection.list_agentic_stream_templates>`
+        :param str prompt: Content instructions for the agent
+        :param list sources: (optional) Keywords and/or website URLs for research
+        :param int max_duration: (optional) Target output duration in seconds
+        :param str voice: (optional) TTS voice name for narration
+        :param str language: (optional) Output language
+        :param str aspect_ratio: (optional) "16:9", "9:16", or "1:1"
+        :param bool captions: (optional) Burn captions into the output
+        :param str model: (optional) Quality tier: "basic" or "pro"
+        :param str callback_url: (optional) Default webhook for all runs
+        :param dict callback_data: (optional) Data echoed in webhook payloads
+        :return: The created agentic stream
+        :rtype: :class:`AgenticStream <videodb.agentic_stream.AgenticStream>`
+        """
+        data = {
+            "name": name,
+            "template_id": template_id,
+            "prompt": prompt,
+            "sources": sources,
+            "max_duration": max_duration,
+            "voice": voice,
+            "language": language,
+            "aspect_ratio": aspect_ratio,
+            "captions": captions,
+            "model": model,
+            "callback_url": callback_url,
+            "callback_data": callback_data,
+        }
+        data = {k: v for k, v in data.items() if v is not None}
+        response = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.agentic_stream}/",
+            data=data,
+        )
+        return AgenticStream(
+            self._connection,
+            id=response.get("id"),
+            collection_id=self.id,
+            **{k: v for k, v in response.items() if k not in ("id", "collection_id")},
+        )
+
+    def get_agentic_stream(self, agentic_stream_id: str) -> AgenticStream:
+        """Get an agentic stream by its ID.
+
+        :param str agentic_stream_id: ID of the agentic stream
+        :return: The agentic stream
+        :rtype: :class:`AgenticStream <videodb.agentic_stream.AgenticStream>`
+        """
+        response = self._connection.get(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.agentic_stream}/{agentic_stream_id}/"
+        )
+        return AgenticStream(
+            self._connection,
+            id=response.get("id"),
+            collection_id=self.id,
+            **{k: v for k, v in response.items() if k not in ("id", "collection_id")},
+        )
+
+    def list_agentic_streams(self) -> List[AgenticStream]:
+        """List agentic streams in this collection.
+
+        :return: List of agentic streams
+        :rtype: List[:class:`AgenticStream <videodb.agentic_stream.AgenticStream>`]
+        """
+        response = self._connection.get(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.agentic_stream}/"
+        )
+        streams = (response or {}).get("agentic_streams", [])
+        return [
+            AgenticStream(
+                self._connection,
+                id=stream.get("id"),
+                collection_id=self.id,
+                **{k: v for k, v in stream.items() if k not in ("id", "collection_id")},
+            )
+            for stream in streams
+        ]
+
+    def list_agentic_stream_templates(self) -> List[dict]:
+        """List all available agentic stream templates.
+
+        :return: List of template dicts with id, name, description, github_url,
+            default_max_duration, supported_aspect_ratios, supports_captions
+        :rtype: List[dict]
+        """
+        response = self._connection.get(
+            path=f"{ApiPath.agentic_stream}/{ApiPath.templates}/"
+        )
+        return (response or {}).get("templates", [])
diff --git a/videodb/schedule.py b/videodb/schedule.py
new file mode 100644
index 0000000..0bbf745
--- /dev/null
+++ b/videodb/schedule.py
@@ -0,0 +1,93 @@
+from videodb._constants import ApiPath
+from videodb.exceptions import VideodbError
+
+
+class Schedule:
+    """Schedule class representing a time-based trigger for agentic stream runs.
+
+    Note: Users should not initialize this class directly.
+    Instead use :meth:`Connection.create_schedule() <videodb.client.Connection.create_schedule>`
+
+    :ivar str id: Unique identifier for the schedule
+    :ivar str target_type: Type of the scheduled target (``agentic_stream``)
+    :ivar str target_id: ID of the target agentic stream
+    :ivar str collection_id: ID of the target's collection
+    :ivar str trigger: EventBridge cron expression, e.g. ``"0 9 * * ? *"``
+    :ivar str timezone: IANA timezone name the cron runs in (default ``UTC``)
+    :ivar str status: ``active`` or ``paused``
+    """
+
+    def __init__(self, _connection, id: str, **kwargs) -> None:
+        self._connection = _connection
+        self.id = id
+        self._update_attributes(kwargs)
+
+    def __repr__(self) -> str:
+        return (
+            f"Schedule("
+            f"id={self.id}, "
+            f"target_id={self.target_id}, "
+            f"trigger={self.trigger}, "
+            f"status={self.status})"
+        )
+
+    def _update_attributes(self, data: dict) -> None:
+        self.target_type = data.get("target_type")
+        self.target_id = data.get("target_id")
+        self.collection_id = data.get("collection_id")
+        self.trigger = data.get("trigger")
+        self.timezone = data.get("timezone")
+        self.status = data.get("status")
+        self.callback_url = data.get("callback_url")
+        self.callback_data = data.get("callback_data")
+        self.created_at = data.get("created_at")
+        self.updated_at = data.get("updated_at")
+
+    def refresh(self) -> "Schedule":
+        """Refresh schedule data from the server.
+
+        :return: The updated schedule instance
+        :rtype: Schedule
+        """
+        response = self._connection.get(path=f"{ApiPath.schedule}/{self.id}/")
+        if response:
+            self._update_attributes(response)
+        else:
+            raise VideodbError(f"Failed to refresh schedule {self.id}")
+        return self
+
+    def delete(self) -> None:
+        """Delete the schedule. In-flight runs are unaffected.
+
+        :return: None
+        :rtype: None
+        """
+        self._connection.delete(path=f"{ApiPath.schedule}/{self.id}/")
+
+    def enable(self) -> "Schedule":
+        """Resume a paused schedule.
+
+        :return: The updated schedule instance
+        :rtype: Schedule
+        """
+        response = self._connection.patch(
+            path=f"{ApiPath.schedule}/{self.id}/{ApiPath.status}",
+            data={"status": "active"},
+        )
+        if response:
+            self._update_attributes(response)
+        return self
+
+    def disable(self) -> "Schedule":
+        """Pause the schedule. Future triggers stop; the record is kept.
+
+        :return: The updated schedule instance
+        :rtype: Schedule
+        """
+        response = self._connection.patch(
+            path=f"{ApiPath.schedule}/{self.id}/{ApiPath.status}",
+            data={"status": "paused"},
+        )
+        if response:
+            self._update_attributes(response)
+        return self