"""
title: Notion Toolkit
author: Otron.io - Mo Mia
author_url: https://otron.io
description: A set of tools to access and edit notion pages within a given workspace. The workspace API key is hardcoded in the code (Ctrl + F YOUR_API_KEY_HERE_ONLY)
version: 0.1.0
requirements: notion-client
"""
import os
from notion_client import Client
from typing import Dict, Any, List, Union, Literal, Optional
from datetime import datetime
def format_block_content(block: Dict[str, Any]) -> str:
"""Format a block's content based on its type"""
block_type = block.get("type", "")
if not block_type:
return ""
content = block.get(block_type, {})
if block_type == "paragraph":
return " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "heading_1":
return "# " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "heading_2":
return "## " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "heading_3":
return "### " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "bulleted_list_item":
return "• " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "numbered_list_item":
return "1. " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "to_do":
checkbox = "☑" if content.get("checked", False) else "☐"
return f"{checkbox} " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
elif block_type == "code":
return f"```\n{' '.join(text.get('plain_text', '') for text in content.get('rich_text', []))}\n```"
elif block_type == "quote":
return "> " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
else:
return f"[{block_type}] " + " ".join(text.get("plain_text", "") for text in content.get("rich_text", []))
class Tools:
def __init__(self):
"""Initialize with default settings"""
self.api_key = 'YOUR_API_KEY_HERE_ONLY'
self.notion = Client(auth=self.api_key)
self.citation = True
def get_page(self, page_id: str) -> str:
"""
Get a Notion page by its ID, including its content
:param page_id: The ID of the page to retrieve (can be URL or ID)
:return: A string containing the page title, content, and block IDs
"""
try:
# Clean up the page ID - handle both URL and direct ID formats
if "/" in page_id:
page_id = page_id.split("/")[-1]
if "-" in page_id:
page_id = page_id.replace("-", "")
# Retrieve the page metadata
page = self.notion.pages.retrieve(page_id)
# Extract title from properties
title = "Untitled"
for prop in page.get("properties", {}).values():
if prop.get("type") == "title":
title_parts = prop.get("title", [])
if title_parts:
title = " ".join(part.get("plain_text", "") for part in title_parts)
break
# Get all blocks (content) from the page
blocks = []
cursor = None
while True:
if cursor:
response = self.notion.blocks.children.list(page_id, start_cursor=cursor)
else:
response = self.notion.blocks.children.list(page_id)
blocks.extend(response.get("results", []))
if not response.get("has_more"):
break
cursor = response.get("next_cursor")
# Format the content
content_lines = []
for block in blocks:
block_id = block.get("id", "")
block_content = format_block_content(block)
if block_content:
content_lines.append(f"{block_content} (Block ID: {block_id})")
# Combine everything into a readable format
result = [
f"Page: '{title}' (ID: {page['id']})",
"Content:",
*content_lines
]
return "\n".join(result)
except Exception as e:
return f"Error retrieving page: {str(e)}"
def create_page(self, database_id: str, title: str) -> str:
"""
Create a new page in a Notion database
:param database_id: The ID of the database to create the page in (can be URL or ID)
:param title: The title for the new page
:return: A string confirming creation or an error message
"""
try:
# Clean up the database ID - handle both URL and direct ID formats
if "/" in database_id:
database_id = database_id.split("/")[-1]
if "-" in database_id:
database_id = database_id.replace("-", "")
# Create the page
new_page = self.notion.pages.create(
parent={"database_id": database_id},
properties={
"Name": {
"type": "title",
"title": [{"type": "text", "text": {"content": title}}]
}
}
)
return f"Created page '{title}' with ID: {new_page['id']}"
except Exception as e:
return f"Error creating page: {str(e)}"
def create_block(self, page_id: str, block_type: str, content: str) -> str:
"""
Create a new block in a Notion page
:param page_id: The ID of the page to add the block to
:param block_type: Type of block (paragraph, heading_1, heading_2, heading_3, bulleted_list_item, numbered_list_item, to_do, code, quote)
:param content: The text content for the block
:return: A string confirming creation or an error message
"""
try:
# Clean up the page ID
if "/" in page_id:
page_id = page_id.split("/")[-1]
if "-" in page_id:
page_id = page_id.replace("-", "")
# Prepare the block content
block_data = {
"type": block_type,
block_type: {
"rich_text": [{"type": "text", "text": {"content": content}}]
}
}
# Add special handling for to_do blocks
if block_type == "to_do":
block_data["to_do"]["checked"] = False
# Create the block
new_block = self.notion.blocks.children.append(
block_id=page_id,
children=[block_data]
)
created_block = new_block["results"][0]
return f"Created {block_type} block with ID: {created_block['id']}"
except Exception as e:
return f"Error creating block: {str(e)}"
def update_block(self, block_id: str, content: str) -> str:
"""
Update the content of a block
:param block_id: The ID of the block to update
:param content: The new text content for the block
:return: A string confirming update or an error message
"""
try:
# Clean up the block ID
if "/" in block_id:
block_id = block_id.split("/")[-1]
if "-" in block_id:
block_id = block_id.replace("-", "")
# Get the current block to determine its type
current_block = self.notion.blocks.retrieve(block_id)
block_type = current_block["type"]
# Prepare the update data
update_data = {
block_type: {
"rich_text": [{"type": "text", "text": {"content": content}}]
}
}
# Update the block
updated_block = self.notion.blocks.update(
block_id=block_id,
**update_data
)
return f"Updated {block_type} block with ID: {updated_block['id']}"
except Exception as e:
return f"Error updating block: {str(e)}"
def delete_block(self, block_id: str) -> str:
"""
Delete a block
:param block_id: The ID of the block to delete
:return: A string confirming deletion or an error message
"""
try:
# Clean up the block ID
if "/" in block_id:
block_id = block_id.split("/")[-1]
if "-" in block_id:
block_id = block_id.replace("-", "")
# Delete the block
self.notion.blocks.delete(block_id)
return f"Deleted block with ID: {block_id}"
except Exception as e:
return f"Error deleting block: {str(e)}"
def append_blocks(self, page_id: str, blocks: List[Dict[str, Any]]) -> str:
"""
Append multiple blocks to a page at once
:param page_id: The ID of the page to add blocks to
:param blocks: List of block data dictionaries
:return: A string confirming creation or an error message
"""
try:
# Clean up the page ID
if "/" in page_id:
page_id = page_id.split("/")[-1]
if "-" in page_id:
page_id = page_id.replace("-", "")
# Append the blocks
response = self.notion.blocks.children.append(
block_id=page_id,
children=blocks
)
created_blocks = response["results"]
return f"Created {len(created_blocks)} blocks"
except Exception as e:
return f"Error creating blocks: {str(e)}"
def search(self, query: str) -> str:
"""
Search across all pages in the Notion workspace
:param query: Search query string
:return: A string containing the search results
"""
try:
# Perform the search
response = self.notion.search(
query=query,
sort={
"direction": "descending",
"timestamp": "last_edited_time"
}
)
results = response.get("results", [])
if not results:
return f"No results found for query: {query}"
# Format results
output_lines = [f"Found {len(results)} results for query: {query}"]
for i, item in enumerate(results, 1):
# Get basic information
item_id = item.get("id", "unknown")
item_type = item.get("object", "unknown")
# Get title based on object type
title = "Untitled"
if item_type == "page":
for prop in item.get("properties", {}).values():
if prop.get("type") == "title":
title_parts = prop.get("title", [])
if title_parts:
title = " ".join(part.get("plain_text", "") for part in title_parts)
break
elif item_type == "database":
title = item.get("title", [{}])[0].get("plain_text", "Untitled Database")
# Format each result
output_lines.append(f"\n{i}. {title}")
output_lines.append(f" Type: {item_type}")
output_lines.append(f" ID: {item_id}")
return "\n".join(output_lines)
except Exception as e:
return f"Error performing search: {str(e)}"
