# Copyright (c) 2026 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""Notion API tools for notion-mcp.
Read and manage Notion pages, blocks, databases, and more via the Notion REST API.
Ref: https://developers.notion.com/reference
"""
import json
from typing import Any
from mcp.types import TextContent, Tool
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
from dedalus_mcp.types import ToolAnnotations
# -----------------------------------------------------------------------------
# Connection
# -----------------------------------------------------------------------------
notion = Connection(
name="notion-mcp", # Must match name from OAuth callback (derived from slug)
secrets=SecretKeys(token="NOTION_ACCESS_TOKEN"),
base_url="https://api.notion.com",
auth_header_format="Bearer {api_key}",
)
# Notion API version header
NOTION_VERSION = "2022-06-28"
# -----------------------------------------------------------------------------
# Helpers
# -----------------------------------------------------------------------------
NotionResult = list[TextContent]
async def _req(method: HttpMethod, path: str, body: dict | None = None) -> NotionResult:
"""Make a Notion API request and return JSON as TextContent."""
ctx = get_context()
resp = await ctx.dispatch("notion", HttpRequest(method=method, path=path, body=body))
if resp.success:
data = resp.response.body or {}
return [TextContent(type="text", text=json.dumps(data, indent=2))]
error = resp.error.message if resp.error else "Request failed"
return [TextContent(type="text", text=json.dumps({"error": error}, indent=2))]
# -----------------------------------------------------------------------------
# User Tools
# -----------------------------------------------------------------------------
@tool(
description="Retrieve a specific user by their ID.",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_user(user_id: str) -> NotionResult:
"""Get a user by ID."""
return await _req(HttpMethod.GET, f"/v1/users/{user_id}")
@tool(
description="List all users in the workspace. Returns paginated results.",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_list_users(
start_cursor: str = "",
page_size: int = 100,
) -> NotionResult:
"""List all users with optional pagination."""
params = [f"page_size={page_size}"]
if start_cursor:
params.append(f"start_cursor={start_cursor}")
query_string = "&".join(params)
return await _req(HttpMethod.GET, f"/v1/users?{query_string}")
@tool(
description="Retrieve your token's bot user information.",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_self() -> NotionResult:
"""Get the authenticated bot user's information."""
return await _req(HttpMethod.GET, "/v1/users/me")
# -----------------------------------------------------------------------------
# Search Tools
# -----------------------------------------------------------------------------
@tool(
description="Search for pages and databases by title across the workspace.",
tags=["search", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_search(
query: str = "",
filter_type: str = "",
sort_direction: str = "",
sort_timestamp: str = "last_edited_time",
start_cursor: str = "",
page_size: int = 100,
) -> NotionResult:
"""Search pages and databases. filter_type can be 'page' or 'data_source'."""
body: dict[str, Any] = {}
if query:
body["query"] = query
if filter_type:
body["filter"] = {"property": "object", "value": filter_type}
if sort_direction:
body["sort"] = {"direction": sort_direction, "timestamp": sort_timestamp}
if start_cursor:
body["start_cursor"] = start_cursor
body["page_size"] = page_size
return await _req(HttpMethod.POST, "/v1/search", body)
# -----------------------------------------------------------------------------
# Block Tools
# -----------------------------------------------------------------------------
@tool(
description="Retrieve a specific block by its ID.",
tags=["block", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_block(block_id: str) -> NotionResult:
"""Get a block by ID."""
return await _req(HttpMethod.GET, f"/v1/blocks/{block_id}")
@tool(
description="Retrieve the children blocks of a parent block or page.",
tags=["block", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_block_children(
block_id: str,
start_cursor: str = "",
page_size: int = 100,
) -> NotionResult:
"""Get child blocks of a parent block."""
params = [f"page_size={page_size}"]
if start_cursor:
params.append(f"start_cursor={start_cursor}")
query_string = "&".join(params)
return await _req(HttpMethod.GET, f"/v1/blocks/{block_id}/children?{query_string}")
@tool(
description="Append new children blocks to a parent block or page.",
tags=["block", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_append_block_children(
block_id: str,
children: str,
after: str = "",
) -> NotionResult:
"""Append children blocks. children should be a JSON array of block objects."""
body: dict[str, Any] = {"children": json.loads(children)}
if after:
body["after"] = after
return await _req(HttpMethod.PATCH, f"/v1/blocks/{block_id}/children", body)
@tool(
description="Update a block's properties or content.",
tags=["block", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_update_block(
block_id: str,
block_data: str,
) -> NotionResult:
"""Update a block. block_data should be a JSON object with the block type and properties."""
body = json.loads(block_data)
return await _req(HttpMethod.PATCH, f"/v1/blocks/{block_id}", body)
@tool(
description="Delete (archive) a block.",
tags=["block", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_delete_block(block_id: str) -> NotionResult:
"""Delete a block by ID."""
return await _req(HttpMethod.DELETE, f"/v1/blocks/{block_id}")
# -----------------------------------------------------------------------------
# Page Tools
# -----------------------------------------------------------------------------
@tool(
description="Retrieve a page's metadata and properties by its ID.",
tags=["page", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_page(
page_id: str,
filter_properties: str = "",
) -> NotionResult:
"""Get a page by ID. filter_properties is a comma-separated list of property IDs."""
path = f"/v1/pages/{page_id}"
if filter_properties:
path = f"{path}?filter_properties={filter_properties}"
return await _req(HttpMethod.GET, path)
@tool(
description="Retrieve a specific property value from a page.",
tags=["page", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_page_property(
page_id: str,
property_id: str,
start_cursor: str = "",
page_size: int = 100,
) -> NotionResult:
"""Get a specific property from a page."""
params = []
if start_cursor:
params.append(f"start_cursor={start_cursor}")
if page_size:
params.append(f"page_size={page_size}")
query_string = f"?{'&'.join(params)}" if params else ""
return await _req(HttpMethod.GET, f"/v1/pages/{page_id}/properties/{property_id}{query_string}")
@tool(
description="Create a new page in a parent page or database.",
tags=["page", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_create_page(
parent_type: str,
parent_id: str,
title: str,
properties: str = "",
children: str = "",
icon: str = "",
cover: str = "",
) -> NotionResult:
"""Create a new page. parent_type is 'page_id' or 'database_id'."""
body: dict[str, Any] = {}
# Set parent
if parent_type == "page_id":
body["parent"] = {"page_id": parent_id}
elif parent_type == "database_id":
body["parent"] = {"type": "database_id", "database_id": parent_id}
else:
body["parent"] = {"type": "workspace"}
# Set properties with title
if properties:
body["properties"] = json.loads(properties)
else:
body["properties"] = {"title": [{"text": {"content": title}}]}
# Optional fields
if children:
body["children"] = json.loads(children)
if icon:
body["icon"] = json.loads(icon)
if cover:
body["cover"] = json.loads(cover)
return await _req(HttpMethod.POST, "/v1/pages", body)
@tool(
description="Update a page's properties, icon, or cover.",
tags=["page", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_update_page(
page_id: str,
properties: str = "",
icon: str = "",
cover: str = "",
archived: bool = False,
in_trash: bool = False,
) -> NotionResult:
"""Update a page's properties."""
body: dict[str, Any] = {}
if properties:
body["properties"] = json.loads(properties)
if icon:
body["icon"] = json.loads(icon)
if cover:
body["cover"] = json.loads(cover)
if archived:
body["archived"] = archived
if in_trash:
body["in_trash"] = in_trash
return await _req(HttpMethod.PATCH, f"/v1/pages/{page_id}", body)
# -----------------------------------------------------------------------------
# Comment Tools
# -----------------------------------------------------------------------------
@tool(
description="Retrieve comments from a page or block.",
tags=["comment", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_comments(
block_id: str,
start_cursor: str = "",
page_size: int = 100,
) -> NotionResult:
"""Get comments on a page or block."""
params = [f"block_id={block_id}", f"page_size={page_size}"]
if start_cursor:
params.append(f"start_cursor={start_cursor}")
query_string = "&".join(params)
return await _req(HttpMethod.GET, f"/v1/comments?{query_string}")
@tool(
description="Create a comment on a page.",
tags=["comment", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_create_comment(
page_id: str,
content: str,
) -> NotionResult:
"""Create a comment on a page."""
body = {
"parent": {"page_id": page_id},
"rich_text": [{"text": {"content": content}}],
}
return await _req(HttpMethod.POST, "/v1/comments", body)
# -----------------------------------------------------------------------------
# Data Source (Database) Tools
# -----------------------------------------------------------------------------
@tool(
description="Query a data source (database) with optional filters and sorts.",
tags=["database", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_query_database(
database_id: str,
filter_json: str = "",
sorts_json: str = "",
start_cursor: str = "",
page_size: int = 100,
archived: bool = False,
in_trash: bool = False,
) -> NotionResult:
"""Query a database. filter_json and sorts_json should be valid JSON strings."""
body: dict[str, Any] = {"page_size": page_size}
if filter_json:
body["filter"] = json.loads(filter_json)
if sorts_json:
body["sorts"] = json.loads(sorts_json)
if start_cursor:
body["start_cursor"] = start_cursor
if archived:
body["archived"] = archived
if in_trash:
body["in_trash"] = in_trash
return await _req(HttpMethod.POST, f"/v1/data_sources/{database_id}/query", body)
@tool(
description="Retrieve metadata and schema for a data source (database).",
tags=["database", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def notion_get_database(database_id: str) -> NotionResult:
"""Get a database's metadata and schema."""
return await _req(HttpMethod.GET, f"/v1/data_sources/{database_id}")
@tool(
description="Update a data source (database) title, description, or properties.",
tags=["database", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_update_database(
database_id: str,
title: str = "",
description: str = "",
properties: str = "",
) -> NotionResult:
"""Update a database."""
body: dict[str, Any] = {}
if title:
body["title"] = [{"text": {"content": title}}]
if description:
body["description"] = [{"text": {"content": description}}]
if properties:
body["properties"] = json.loads(properties)
return await _req(HttpMethod.PATCH, f"/v1/data_sources/{database_id}", body)
@tool(
description="Create a new data source (database).",
tags=["database", "write"],
annotations=ToolAnnotations(readOnlyHint=False),
)
async def notion_create_database(
parent_type: str,
parent_id: str,
title: str,
properties: str,
description: str = "",
is_inline: bool = False,
) -> NotionResult:
"""Create a database. parent_type is 'page_id' or 'block_id'. properties is required JSON."""
body: dict[str, Any] = {
"title": [{"text": {"content": title}}],
"properties": json.loads(properties),
"is_inline": is_inline,
}
if parent_type == "page_id":
body["parent"] = {"type": "page_id", "page_id": parent_id}
else:
body["parent"] = {"type": "block_id", "block_id": parent_id}
if description:
body["description"] = [{"text": {"content": description}}]
return await _req(HttpMethod.POST, "/v1/data_sources", body)
# -----------------------------------------------------------------------------
# Export
# -----------------------------------------------------------------------------
notion_tools: list[Tool] = [
# Users
notion_get_user,
notion_list_users,
notion_get_self,
# Search
notion_search,
# Blocks
notion_get_block,
notion_get_block_children,
notion_append_block_children,
notion_update_block,
notion_delete_block,
# Pages
notion_get_page,
notion_get_page_property,
notion_create_page,
notion_update_page,
# Comments
notion_get_comments,
notion_create_comment,
# Databases
notion_query_database,
notion_get_database,
notion_update_database,
notion_create_database,
]
