# Directory Structure
```
├── .DS_Store
├── .gitignore
├── assets
│   └── demo.gif
├── LICENSE
├── pyproject.toml
├── README.md
└── src
    ├── .DS_Store
    └── notion_mcp
        ├── __init__.py
        ├── __main__.py
        ├── __pycache__
        │   ├── __init__.cpython-311.pyc
        │   ├── __main__.cpython-311.pyc
        │   └── server.cpython-311.pyc
        └── server.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Environment Variables
.env
.env.local
.env.*.local
.env.development
.env.test
.env.production
# Backup files
.env.backup
.env.*.backup
# IDE specific files
.idea
.vscode
*.swp
*.swo
# macOS system files
.DS_Store
.AppleDouble
.LSOverride
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Notion MCP Integration
A simple Model Context Protocol (MCP) server that integrates with Notion's API to manage my personal todo list through Claude. This is a basic implementation tailored specifically for my minimalist todo list setup in Notion.
<p align="center">
  <img src="assets/demo.gif" width="600"/>
</p>
## Important Note
This is a personal project designed for a very specific use case: my simple Notion todo list that has just three properties:
- Task (title)
- When (select with only two options: "today" or "later")
- Checkbox (marks if completed)
[Example Notion Database](https://danhilse.notion.site/14e5549555a08078afb5ed5d374bb656?v=14e5549555a081f9b5a4000cdf952cb9&pvs=4)
While you can use this as a starting point for your own Notion integration, you'll likely need to modify the code to match your specific database structure and requirements.
## Features
- Add new todo items
- View all todos
- View today's tasks
- Check off a task as complete
## Prerequisites
- Python 3.10 or higher
- A Notion account
- A Notion integration (API key)
- A Notion database that matches the exact structure described above (or willingness to modify the code for your structure)
## Setup
1. Clone the repository:
```bash
git clone https://github.com/yourusername/notion-mcp.git
cd notion-mcp
```
2. Set up Python environment:
```bash
python -m venv .venv
source .venv/bin/activate  # On Windows use: .venv\Scripts\activate
uv pip install -e .
```
3. Create a Notion integration:
   - Go to https://www.notion.so/my-integrations
   - Create new integration
   - Copy the API key
4. Share your database with the integration:
   - Open your todo database in Notion
   - Click "..." menu → "Add connections"
   - Select your integration
5. Create a `.env` file:
```env
NOTION_API_KEY=your-api-key-here
NOTION_DATABASE_ID=your-database-id-here
```
6. Configure Claude Desktop:
```json
{
  "mcpServers": {
    "notion-todo": {
      "command": "/path/to/your/.venv/bin/python",
      "args": ["-m", "notion_mcp"],
      "cwd": "/path/to/notion-mcp"
    }
  }
}
```
## Running the Server
The server can be run in two ways:
1. Directly from the command line:
```bash
# From the project directory with virtual environment activated
python -m notion_mcp
```
2. Automatically through Claude Desktop (recommended):
- The server will start when Claude launches if configured correctly in `claude_desktop_config.json`
- No manual server management needed
- Server stops when Claude is closed
Note: When running directly, the server won't show any output unless there's an error - this is normal as it's waiting for MCP commands.
## Usage
Basic commands through Claude:
- "Show all my todos"
- "What's on my list for today?"
- "Add a todo for today: check emails"
- "Add a task for later: review project"
## Limitations
- Only works with a specific Notion database structure
- No support for complex database schemas
- Limited to "today" or "later" task scheduling
- No support for additional properties or custom fields
- Basic error handling
- No advanced features like recurring tasks, priorities, or tags
## Customization
If you want to use this with a different database structure, you'll need to modify the `server.py` file, particularly:
- The `create_todo()` function to match your database properties
- The todo formatting in `call_tool()` to handle your data structure
- The input schema in `list_tools()` if you want different options
## Project Structure
```
notion_mcp/
├── pyproject.toml
├── README.md
├── .env                   # Not included in repo
└── src/
    └── notion_mcp/
        ├── __init__.py
        ├── __main__.py
        └── server.py      # Main implementation
```
## License
MIT License - Use at your own risk
## Acknowledgments
- Built to work with Claude Desktop
- Uses Notion's API
```
--------------------------------------------------------------------------------
/src/notion_mcp/__main__.py:
--------------------------------------------------------------------------------
```python
from . import main
if __name__ == "__main__":
    main()
```
--------------------------------------------------------------------------------
/src/notion_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
import asyncio
from . import server
def main():
    """Main entry point for the package."""
    asyncio.run(server.main())
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "notion_mcp"
version = "0.1.0"
description = "Notion MCP integration for todo lists"
requires-python = ">=3.10"
dependencies = [
    "mcp",
    "httpx",
    "python-dotenv"
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.pytest.ini_options]
asyncio_mode = "auto"
```
--------------------------------------------------------------------------------
/src/notion_mcp/server.py:
--------------------------------------------------------------------------------
```python
from mcp.server import Server
from mcp.types import (
    Resource, 
    Tool,
    TextContent,
    EmbeddedResource
)
from pydantic import AnyUrl
import os
import json
from datetime import datetime
import httpx
from typing import Any, Sequence
from dotenv import load_dotenv
from pathlib import Path
import logging
# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('notion_mcp')
# Find and load .env file from project root
project_root = Path(__file__).parent.parent.parent
env_path = project_root / '.env'
if not env_path.exists():
    raise FileNotFoundError(f"No .env file found at {env_path}")
load_dotenv(env_path)
# Initialize server
server = Server("notion-todo")
# Configuration with validation
NOTION_API_KEY = os.getenv("NOTION_API_KEY")
DATABASE_ID = os.getenv("NOTION_DATABASE_ID")
if not NOTION_API_KEY:
    raise ValueError("NOTION_API_KEY not found in .env file")
if not DATABASE_ID:
    raise ValueError("NOTION_DATABASE_ID not found in .env file")
NOTION_VERSION = "2022-06-28"
NOTION_BASE_URL = "https://api.notion.com/v1"
# Notion API headers
headers = {
    "Authorization": f"Bearer {NOTION_API_KEY}",
    "Content-Type": "application/json",
    "Notion-Version": NOTION_VERSION
}
async def fetch_todos() -> dict:
    """Fetch todos from Notion database"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{NOTION_BASE_URL}/databases/{DATABASE_ID}/query",
            headers=headers,
            json={
                "sorts": [
                    {
                        "timestamp": "created_time",
                        "direction": "descending"
                    }
                ]
            }
        )
        response.raise_for_status()
        return response.json()
async def create_todo(task: str, when: str) -> dict:
    """Create a new todo in Notion"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{NOTION_BASE_URL}/pages",
            headers=headers,
            json={
                "parent": {"database_id": DATABASE_ID},
                "properties": {
                    "Task": {
                        "type": "title",
                        "title": [{"type": "text", "text": {"content": task}}]
                    },
                    "When": {
                        "type": "select",
                        "select": {"name": when}
                    },
                    "Checkbox": {
                        "type": "checkbox",
                        "checkbox": False
                    }
                }
            }
        )
        response.raise_for_status()
        return response.json()
async def complete_todo(page_id: str) -> dict:
    """Mark a todo as complete in Notion"""
    async with httpx.AsyncClient() as client:
        response = await client.patch(
            f"{NOTION_BASE_URL}/pages/{page_id}",
            headers=headers,
            json={
                "properties": {
                    "Checkbox": {
                        "type": "checkbox",
                        "checkbox": True
                    }
                }
            }
        )
        response.raise_for_status()
        return response.json()
@server.list_tools()
async def list_tools() -> list[Tool]:
    """List available todo tools"""
    return [
        Tool(
            name="add_todo",
            description="Add a new todo item",
            inputSchema={
                "type": "object",
                "properties": {
                    "task": {
                        "type": "string",
                        "description": "The todo task description"
                    },
                    "when": {
                        "type": "string",
                        "description": "When the task should be done (today or later)",
                        "enum": ["today", "later"]
                    }
                },
                "required": ["task", "when"]
            }
        ),
        Tool(
            name="show_all_todos",
            description="Show all todo items from Notion",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="show_today_todos",
            description="Show today's todo items from Notion",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="complete_todo",
            description="Mark a todo item as complete",
            inputSchema={
                "type": "object",
                "properties": {
                    "task_id": {
                        "type": "string",
                        "description": "The ID of the todo task to mark as complete"
                    }
                },
                "required": ["task_id"]
            }
        )
    ]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | EmbeddedResource]:
    """Handle tool calls for todo management"""
    if name == "add_todo":
        if not isinstance(arguments, dict):
            raise ValueError("Invalid arguments")
            
        task = arguments.get("task")
        when = arguments.get("when", "later")
        
        if not task:
            raise ValueError("Task is required")
        if when not in ["today", "later"]:
            raise ValueError("When must be 'today' or 'later'")
            
        try:
            result = await create_todo(task, when)
            return [
                TextContent(
                    type="text",
                    text=f"Added todo: {task} (scheduled for {when})"
                )
            ]
        except httpx.HTTPError as e:
            logger.error(f"Notion API error: {str(e)}")
            return [
                TextContent(
                    type="text",
                    text=f"Error adding todo: {str(e)}\nPlease make sure your Notion integration is properly set up and has access to the database."
                )
            ]
            
    elif name in ["show_all_todos", "show_today_todos"]:
        try:
            todos = await fetch_todos()
            formatted_todos = []
            for todo in todos.get("results", []):
                props = todo["properties"]
                formatted_todo = {
                    "id": todo["id"],  # Include the page ID in the response
                    "task": props["Task"]["title"][0]["text"]["content"] if props["Task"]["title"] else "",
                    "completed": props["Checkbox"]["checkbox"],
                    "when": props["When"]["select"]["name"] if props["When"]["select"] else "unknown",
                    "created": todo["created_time"]
                }
                
                if name == "show_today_todos" and formatted_todo["when"].lower() != "today":
                    continue
                    
                formatted_todos.append(formatted_todo)
            
            return [
                TextContent(
                    type="text",
                    text=json.dumps(formatted_todos, indent=2)
                )
            ]
        except httpx.HTTPError as e:
            logger.error(f"Notion API error: {str(e)}")
            return [
                TextContent(
                    type="text",
                    text=f"Error fetching todos: {str(e)}\nPlease make sure your Notion integration is properly set up and has access to the database."
                )
            ]
    
    elif name == "complete_todo":
        if not isinstance(arguments, dict):
            raise ValueError("Invalid arguments")
            
        task_id = arguments.get("task_id")
        if not task_id:
            raise ValueError("Task ID is required")
            
        try:
            result = await complete_todo(task_id)
            return [
                TextContent(
                    type="text",
                    text=f"Marked todo as complete (ID: {task_id})"
                )
            ]
        except httpx.HTTPError as e:
            logger.error(f"Notion API error: {str(e)}")
            return [
                TextContent(
                    type="text",
                    text=f"Error completing todo: {str(e)}\nPlease make sure your Notion integration is properly set up and has access to the database."
                )
            ]
    
    raise ValueError(f"Unknown tool: {name}")
async def main():
    """Main entry point for the server"""
    from mcp.server.stdio import stdio_server
    
    if not NOTION_API_KEY or not DATABASE_ID:
        raise ValueError("NOTION_API_KEY and NOTION_DATABASE_ID environment variables are required")
    
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
```