From 1fe13cb009fe4e63689c96a27404a61d1b9b2967 Mon Sep 17 00:00:00 2001
From: Isaac Francisco <78627776+isahers1@users.noreply.github.com>
Date: Fri, 28 Jun 2024 16:23:15 -0700
Subject: [PATCH] sdk-py: added Cron search functionality (#804)

* first draft

* minor fix

* second draft

* reformat

* rename

* remove f-string

* fmt

* fmt

* fmt

* fmt

* fmt

* typing

* fmt

* fmt

* fmt

* pagination

* fmt

* fmt

* schema

* fmt

* fmt
---
 libs/sdk-py/langgraph_sdk/client.py | 19 +++++++++++++++++++
 libs/sdk-py/langgraph_sdk/schema.py | 17 +++++++++++++++++
 2 files changed, 36 insertions(+)

diff --git a/libs/sdk-py/langgraph_sdk/client.py b/libs/sdk-py/langgraph_sdk/client.py
index 29f9eec8a..12e5febb5 100644
--- a/libs/sdk-py/langgraph_sdk/client.py
+++ b/libs/sdk-py/langgraph_sdk/client.py
@@ -15,6 +15,7 @@ import langgraph_sdk
 from langgraph_sdk.schema import (
     Assistant,
     Config,
+    Cron,
     GraphSchema,
     Metadata,
     MultitaskStrategy,
@@ -718,6 +719,24 @@ class CronClient:
         """Delete a cron."""
         await self.http.delete(f"/runs/crons/{cron_id}")
 
+    async def search(
+        self,
+        *,
+        assistant_id: Optional[str] = None,
+        thread_id: Optional[str] = None,
+        limit: int = 10,
+        offset: int = 0,
+    ) -> list[Cron]:
+        """Get a list of cron jobs."""
+        payload = {
+            "assistant_id": assistant_id,
+            "thread_id": thread_id,
+            "limit": limit,
+            "offset": offset,
+        }
+        payload = {k: v for k, v in payload.items() if v is not None}
+        return await self.http.post("/runs/crons/search", json=payload)
+
 
 def _get_api_key(api_key: Optional[str] = None) -> Optional[str]:
     """Get the API key from the environment.
diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py
index 9f6d10137..e40fc71cd 100644
--- a/libs/sdk-py/langgraph_sdk/schema.py
+++ b/libs/sdk-py/langgraph_sdk/schema.py
@@ -109,3 +109,20 @@ class Run(TypedDict):
     """The run metadata."""
     multitask_strategy: MultitaskStrategy
     """Strategy to handle concurrent runs on the same thread."""
+
+
+class Cron(TypedDict):
+    cron_id: str
+    """The ID of the cron."""
+    thread_id: Optional[str]
+    """The ID of the thread."""
+    end_time: Optional[datetime]
+    """The end date to stop running the cron."""
+    schedule: str
+    """The schedule to run, cron format."""
+    created_at: datetime
+    """The time the cron was created."""
+    updated_at: datetime
+    """The last time the cron was updated."""
+    payload: dict
+    """The run payload to use for creating new run."""