Notion API Integration

Confiance : high

notion-apidatabase-integrationrest-apisdk-compatibilityagent-toolsenterprise-integrationdirect-http-clientproduction-patternsapi-wrapper-developmentpython-compatibilitynotion-client-issueshttpx-implementation

Patterns and best practices for integrating AI systems with Notion databases, including handling SDK compatibility issues and implementing direct REST API clients for robust production systems.

Integration Challenges

SDK Compatibility Issues

Problem: Python SDK compatibility with newer Python versions (3.14+)

notion-client v3 removed databases.query() method
V2 compatibility issues with Python 3.14
Migration path unclear between versions

Solution: Direct REST API implementation bypassing SDK dependencies

import httpx
from typing import Dict, Any, List, Optional

class NotionDirectClient:
    def __init__(self, integration_token: str):
        self.token = integration_token
        self.base_url = "https://api.notion.com/v1"
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
            "Notion-Version": "2022-06-28"
        }
    
    async def query_database(self, database_id: str, filter_params: Optional[Dict] = None) -> Dict[str, Any]:
        async with httpx.AsyncClient() as client:
            url = f"{self.base_url}/databases/{database_id}/query"
            payload = filter_params or {}
            response = await client.post(url, headers=self.headers, json=payload)
            return response.json()

Production Patterns

Database Schema Management

# Schema discovery for dynamic tool generation
async def get_database_schema(self, database_id: str) -> Dict[str, Any]:
    async with httpx.AsyncClient() as client:
        url = f"{self.base_url}/databases/{database_id}"
        response = await client.get(url, headers=self.headers)
        return response.json()

Multi-Database Coordination

Pattern: Coordinating operations across related business databases

class BakeryNotionIntegration:
    def __init__(self, client: NotionDirectClient):
        self.client = client
        self.databases = {
            'stock': 'a481b327-1273-829a-9dbe-01c91f2c69b0',
            'catalog': 'a481b327-1273-82e8-aeac-8765dfb6a123', 
            'sales': 'a481b327-1273-82ba-95f3-abcdef123456',
            'orders': 'a481b327-1273-8230-8456-fedcba654321'
        }
    
    async def analyze_stock_needs(self) -> List[Dict]:
        # Multi-step workflow across databases
        stock_data = await self.client.query_database(self.databases['stock'])
        sales_data = await self.client.query_database(self.databases['sales'])
        # Coordinate analysis across datasets

Error Handling and Resilience

async def safe_query_database(self, database_id: str, retries: int = 3) -> Optional[Dict]:
    for attempt in range(retries):
        try:
            return await self.query_database(database_id)
        except httpx.TimeoutException:
            if attempt == retries - 1:
                return None
            await asyncio.sleep(2 ** attempt)  # Exponential backoff
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:  # Rate limited
                await asyncio.sleep(60)
                continue
            return None

Agent Tool Integration

Dynamic Tool Generation

def create_notion_tools(databases: Dict[str, str]) -> List[BaseTool]:
    tools = []
    for db_name, db_id in databases.items():
        tool = create_query_tool(db_name, db_id)
        tools.append(tool)
    return tools

def create_query_tool(name: str, database_id: str) -> BaseTool:
    @tool(f"query_{name}")
    def query_tool(filter_params: Optional[Dict] = None) -> str:
        """Query {name} database with optional filters."""
        client = NotionDirectClient(os.getenv('NOTION_TOKEN'))
        result = asyncio.run(client.query_database(database_id, filter_params))
        return format_results(result)
    
    return query_tool

Business Logic Integration

Example: Bakery management with automatic stock alerts

async def check_stock_alerts(self) -> List[Dict]:
    stock_data = await self.client.query_database(self.databases['stock'])
    alerts = []
    
    for item in stock_data['results']:
        properties = item['properties']
        current_qty = properties['Quantite']['number']
        alert_threshold = properties['Seuil_Alerte']['number']
        
        if current_qty < alert_threshold:
            alerts.append({
                'ingredient': properties['Nom']['title'][0]['text']['content'],
                'current': current_qty,
                'threshold': alert_threshold,
                'supplier': properties['Email_Fournisseur']['email']
            })
    
    return alerts

Security and Configuration

Environment Management

# .env configuration
NOTION_INTEGRATION_TOKEN=secret_xyz
NOTION_DATABASE_STOCK=database-id-1
NOTION_DATABASE_CATALOG=database-id-2

# Configuration validation
class NotionConfig:
    def __init__(self):
        self.token = os.getenv('NOTION_INTEGRATION_TOKEN')
        if not self.token:
            raise ValueError("NOTION_INTEGRATION_TOKEN required")

Permission Management

Integration tokens with minimal required permissions
Database-level access control
Audit logging for sensitive operations

Common Integration Patterns

Read-Heavy Workflows

Cached schema information
Batch query optimization
Pagination handling for large datasets

Write Operations

async def create_supplier_order(self, order_data: Dict) -> Dict:
    url = f"{self.base_url}/pages"
    payload = {
        "parent": {"database_id": self.databases['orders']},
        "properties": {
            "Ingredient": {"title": [{"text": {"content": order_data['ingredient']}}]},
            "Quantite": {"number": order_data['quantity']},
            "Date_Commande": {"date": {"start": order_data['date']}},
            "Statut": {"select": {"name": "En cours"}}
        }
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=self.headers, json=payload)
        return response.json()

Testing and Development

Integration Testing

async def test_notion_integration():
    client = NotionDirectClient(os.getenv('NOTION_TOKEN'))
    
    # Test database access
    result = await client.query_database(test_database_id)
    assert 'results' in result
    
    # Test error handling
    invalid_result = await client.query_database('invalid-id')
    assert invalid_result is None

Local Development Patterns

Separate test databases for development
Mock clients for CI/CD environments
Schema validation against production databases