CrewAI × Notion

"""CrewAI × Notion Connector — Production Tools"""

import os
import time
from typing import Any
from notion_client import Client
from crewai.tools import tool

# Initialize Notion client
notion = Client(auth=os.environ["NOTION_API_KEY"])

MAX_RETRIES = 3
RETRY_DELAY = 1.0  # seconds


def _retry(fn, *args, **kwargs) -> Any:
    """Execute with exponential backoff for rate limiting."""
    for attempt in range(MAX_RETRIES):
        try:
            return fn(*args, **kwargs)
        except Exception as e:
            if "rate_limited" in str(e).lower() and attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY * (2 ** attempt))
                continue
            raise
    raise RuntimeError("Max retries exceeded")


@tool("Read Notion Page")
def notion_page_reader(page_id: str) -> str:
    """Read the full content of a Notion page by its ID.
    Returns the page title and all text blocks as markdown."""
    page = _retry(notion.pages.retrieve, page_id=page_id)
    blocks = _retry(
        notion.blocks.children.list, block_id=page_id
    )

    title = _extract_title(page)
    content = _blocks_to_markdown(blocks["results"])
    return f"# {title}\n\n{content}"


@tool("Query Notion Database")
def notion_database_query(
    database_id: str,
    filter_json: str = "{}",
    sorts_json: str = "[]",
) -> str:
    """Query a Notion database with optional filters and sorts.
    Returns matching rows as structured text."""
    import json

    filter_obj = json.loads(filter_json) if filter_json != "{}" else None
    sorts_obj = json.loads(sorts_json) if sorts_json != "[]" else None

    query_args: dict[str, Any] = {"database_id": database_id}
    if filter_obj:
        query_args["filter"] = filter_obj
    if sorts_obj:
        query_args["sorts"] = sorts_obj

    results = _retry(notion.databases.query, **query_args)
    return _format_db_results(results["results"])


@tool("Create Notion Page")
def notion_page_creator(
    parent_page_id: str,
    title: str,
    content_markdown: str,
) -> str:
    """Create a new Notion page under a parent page.
    Accepts markdown content and converts to Notion blocks."""
    blocks = _markdown_to_blocks(content_markdown)

    new_page = _retry(
        notion.pages.create,
        parent={"page_id": parent_page_id},
        properties={"title": [{"text": {"content": title}}]},
        children=blocks,
    )
    return f"Created page: {new_page['id']} — {new_page['url']}"


@tool("Update Notion Page")
def notion_page_updater(
    page_id: str,
    updates_json: str,
) -> str:
    """Update properties on an existing Notion page.
    Pass a JSON string of property updates."""
    import json
    updates = json.loads(updates_json)

    _retry(
        notion.pages.update,
        page_id=page_id,
        properties=updates,
    )
    return f"Updated page {page_id} successfully"