I have been experimenting with creating custom tools for interaction with Notion Databases and database pages.
To read Notion databases:
class DatabaseDataFetcherTool(BaseTool):
"""
Tool to fetch the structure and properties of a Notion Database.
"""
name: str = "Notion Database Data Fetcher"
description: str = "Fetches Notion Database structure and properties."
headers: ClassVar[Dict[str, str]] = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Notion-Version": NOTION_VERSION,
"Content-Type": "application/json",
}
def _run(self) -> Union[dict, str]:
"""
Fetch all the properties and structure of a Notion database.
Returns:
Union[dict, str]: The database structure or an error message.
"""
url = f"{NOTION_ENDPOINT}/databases/{NOTION_DATABASE_ID}"
try:
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
return response.json()
else:
return f"Error {response.status_code}: {response.text}"
except Exception as e:
return f"An error occurred: {str(e)}"
and to read Pages from a database (Filtered):
**class PageDataFetcherTool(BaseTool):**
"""
Tool to fetch all pages in a Notion database with their IDs and full schemas.
"""
name: str = "Fetch Notion Pages Tool"
description: str = (
"Fetches all pages in a specified Notion database, "
"returning their unique page IDs and full page schemas."
)
headers: ClassVar[Dict[str, str]] = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Content-Type": "application/json",
"Notion-Version": NOTION_VERSION,
}
def _run(self, database_id: str) -> List[Dict[str, Any]]:
"""
Fetches all pages in the specified Notion database.
Args:
database_id (str): The Notion database ID.
Returns:
List[Dict]: A list of dictionaries containing page IDs and full schemas.
"""
from datetime import datetime, timedelta
# Dynamic date calculation
today = datetime.now()
yesterday = (today - timedelta(days=1)).strftime("%Y-%m-%d")
next_week = (today + timedelta(days=4)).strftime("%Y-%m-%d")
# Properly formatted filter with Status != Done
filter_payload = {
"filter": {
"and": [
{"property": "Due Date", "date": {"on_or_after": yesterday}},
{"property": "Due Date", "date": {"before": next_week}},
{"property": "Status", "status": {"does_not_equal": "Done"}},
]
}
}
url = f"{NOTION_ENDPOINT}/databases/{database_id}/query"
all_pages = []
has_more = True
next_cursor = None
# Handle pagination to get all pages
while has_more:
payload = {"start_cursor": next_cursor} if next_cursor else {}
payload.update(filter_payload) # Add the filter to the payload
response = requests.post(url, headers=self.headers, json=payload)
if response.status_code == 200:
data = response.json()
for page in data.get("results", []):
all_pages.append({"page_id": page.get("id"), "full_schema": page})
has_more = data.get("has_more", False)
next_cursor = data.get("next_cursor", None)
else:
return [{"error": response.text, "status_code": response.status_code}]
return all_pages
Create a new Task in a Notion Database:
class NewTaskCreationTool(BaseTool):
name: str = "Create New Task Tool"
description: str = (
"Creates a new task in the calendar database with user-specified properties like Priority, Title, and Due Dates."
)
headers: ClassVar[Dict[str, str]] = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Content-Type": "application/json",
"Notion-Version": NOTION_VERSION, # Use the correct Notion API version
}
def _run(
self,
name: str,
status: str,
priority: str,
start_datetime: str,
end_datetime: str,
duration_in_minutes: int,
) -> Dict[str, Any]:
"""
Create a new task in the Notion database with specified properties.
Args:
title (str): The name of the task.
status (str): The status of the task (e.g., "Not Started", "In progress", "Done").
priority (str): Priority of the task ("High", "Medium", "Low").
start_datetime (str): Start date and time in ISO 8601 format.
end_datetime (str): End date and time in ISO 8601 format.
Returns:
dict: The response from the Notion API.
"""
task_data = {
"parent": {"database_id": NOTION_DATABASE_ID},
"properties": {
"Name": {"title": [{"text": {"content": f"🅾️ {name}"}}]},
"Status": {"status": {"name": status}},
"Duration (minutes)": {"number": duration_in_minutes},
"Priority": {"select": {"name": priority}},
"Due Date": {"date": {"start": start_datetime, "end": end_datetime}},
},
}
# Make the API request to Notion
url = f"{NOTION_ENDPOINT}/pages"
response = requests.post(url, headers=self.headers, json=task_data)
if response.status_code == 200:
return response.json()
else:
return {"error": response.text}
Update an existing Task with a new start and end date time:
class UpdateExcistingTasks(BaseTool):
name: str = "Reschedule Existing Task Tool"
description: str = (
"Reschedules a single existing task, identified by page_id, with a new start date time and end date time. This tool cannot process multiple tasks at once."
)
headers: ClassVar[Dict[str, str]] = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Content-Type": "application/json",
"Notion-Version": NOTION_VERSION, # Use the correct Notion API version
}
def _run(
self, page_id: str, start_datetime: str, end_datetime: str
) -> Dict[str, Any]:
"""
Update the start_datetime and end_datetime properties for an excisting Task or database page, identified by the page_id.
Args:
page_id (str): Notioh Page ID of page to be updated or rescheduled.
start_datetime (str): Start date and time in ISO 8601 format.
end_datetime (str): End date and time in ISO 8601 format.
Returns:
dict: The response from the Notion API.
"""
task_data = {
"parent": {"database_id": NOTION_DATABASE_ID},
"properties": {
"Due Date": {"date": {"start": start_datetime, "end": end_datetime}}
},
}
# Make the API request to Notion
url = f"{NOTION_ENDPOINT}/pages/{page_id}"
response = requests.patch(url, headers=self.headers, json=task_data)
if response.status_code == 200:
return response.json()
else:
return {"error": response.text}
I have also managed to append markdown output to a newly created Notion Page using the page_id and markdown output from the last agent in a dict
{
“page_id”: “…”,
“markdown_report”: “…”
}
I pass the page_id down from agent to agent, and therefore it is a bit hit and miss. Not been able to make structure ouput work for this use case.
Any suggestions or input re the tools and getting the desired output would be much appreciated.
Thank you, Jan
