Notion restores access to Anthropic after service disruption

Introduction

In this tutorial, you'll learn how to integrate and work with the Anthropic Claude API within a Notion-like application. This practical guide will show you how to build a basic AI assistant that can handle service disruptions gracefully, similar to what Notion experienced. You'll create a Python application that interfaces with Claude's API, implements error handling for service interruptions, and demonstrates recovery mechanisms.

Prerequisites

• Python 3.7 or higher installed on your system
• Basic understanding of Python programming and APIs
• Anthropic API key (available from https://console.anthropic.com/)
• Notion API knowledge (basic understanding of Notion's integration concepts)
• pip package manager installed

Step-by-step instructions

Step 1: Set up your development environment

Install required packages

First, create a new Python virtual environment and install the necessary dependencies. This ensures you have a clean environment for our project.

python -m venv anthropic-notion-env
source anthropic-notion-env/bin/activate  # On Windows: anthropic-notion-env\Scripts\activate
pip install anthropic notion-client requests

Why this step?

Creating a virtual environment isolates our project dependencies from your system Python packages. The anthropic library provides the official interface to Claude's API, while notion-client allows interaction with Notion's API.

Step 2: Create your main application structure

Initialize the main application

Create a file called notion_claude_integration.py with the following base structure:

import os
import time
import logging
from anthropic import Anthropic
from notion_client import Client

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize clients
anthropic = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
notion = Client(auth=os.getenv('NOTION_API_KEY'))

Why this step?

This sets up our core components with proper logging configuration. The logging will help us trace issues during service disruptions. We initialize both the Claude and Notion clients for later use.

Step 3: Implement service monitoring and recovery

Create a service health checker

Add this class to monitor service availability:

class ServiceMonitor:
    def __init__(self, service_name):
        self.service_name = service_name
        self.is_healthy = True
        self.last_heartbeat = time.time()
        self.recovery_attempts = 0
        
    def mark_healthy(self):
        self.is_healthy = True
        self.last_heartbeat = time.time()
        self.recovery_attempts = 0
        logger.info(f'{self.service_name} is now healthy')
        
    def mark_unhealthy(self):
        self.is_healthy = False
        logger.warning(f'{self.service_name} is now unhealthy')
        
    def attempt_recovery(self):
        if not self.is_healthy:
            self.recovery_attempts += 1
            logger.info(f'Attempting recovery for {self.service_name} (attempt {self.recovery_attempts})')
            return True
        return False

Why this step?

This monitoring system tracks service health and implements basic recovery attempts. It mimics how Notion might have handled the disruption by monitoring service availability and attempting recovery.

Step 4: Build the Claude API wrapper with error handling

Create Claude interaction class

Implement a wrapper that handles API interruptions gracefully:

class ClaudeAssistant:
    def __init__(self, client, monitor):
        self.client = client
        self.monitor = monitor
        self.max_retries = 3
        self.retry_delay = 5
        
    def query_claude(self, prompt, max_tokens=1000):
        for attempt in range(self.max_retries):
            try:
                response = self.client.messages.create(
                    model="claude-3-haiku-20240307",
                    max_tokens=max_tokens,
                    messages=[
                        {
                            "role": "user",
                            "content": prompt
                        }
                    ]
                )
                self.monitor.mark_healthy()
                return response.content[0].text
                
            except Exception as e:
                logger.error(f'Attempt {attempt + 1} failed: {str(e)}')
                self.monitor.mark_unhealthy()
                
                if attempt < self.max_retries - 1:
                    logger.info(f'Waiting {self.retry_delay} seconds before retry...')
                    time.sleep(self.retry_delay)
                else:
                    logger.error('All retry attempts exhausted')
                    raise

Why this step?

This implementation includes robust error handling with retry logic. It mimics the behavior of Notion's recovery system by attempting multiple retries when service disruption occurs, rather than failing immediately.

Step 5: Implement Notion integration with fallback mechanisms

Create Notion document handler

Build functionality to work with Notion documents:

class NotionDocumentHandler:
    def __init__(self, client, claude_assistant):
        self.client = client
        self.claude = claude_assistant
        
    def process_document(self, page_id, prompt_template):
        try:
            # Get page content
            page = self.client.pages.retrieve(page_id=page_id)
            
            # Generate content using Claude
            prompt = prompt_template.format(page_content=page)
            response = self.claude.query_claude(prompt)
            
            # Update page with Claude's response
            self.client.pages.update(
                page_id=page_id,
                properties={
                    "AI_Response": {
                        "rich_text": [
                            {
                                "text": {
                                    "content": response
                                }
                            }
                        ]
                    }
                }
            )
            
            return response
            
        except Exception as e:
            logger.error(f'Document processing failed: {str(e)}')
            # Fallback: return error message
            return f'Error processing document: {str(e)}'

Why this step?

This component shows how to integrate Claude with Notion while maintaining fallback behavior. If the Claude API fails, it still provides meaningful feedback rather than crashing entirely.

Step 6: Test the integration

Create a test script

Finally, create a test that demonstrates the system working:

def main():
    # Initialize components
    service_monitor = ServiceMonitor('Anthropic Claude')
    claude_assistant = ClaudeAssistant(anthropic, service_monitor)
    notion_handler = NotionDocumentHandler(notion, claude_assistant)
    
    # Test prompt
    test_prompt = "Summarize the key points of this document in 3 bullet points."
    
    try:
        # This would normally process a real Notion page
        result = notion_handler.process_document('your-page-id-here', test_prompt)
        print(f'Result: {result}')
        
    except Exception as e:
        logger.error(f'Application failed: {str(e)}')
        
if __name__ == '__main__':
    main()

Why this step?

This test script demonstrates how all components work together. It shows the complete flow from monitoring service health to processing documents with fallbacks, similar to how Notion would handle a service disruption.

Summary

In this tutorial, you've built a robust integration system that combines Anthropic's Claude API with Notion's capabilities. The system includes:

• Service health monitoring
• Retry mechanisms with exponential backoff
• Graceful error handling and fallbacks
• Notion document processing capabilities

This implementation mirrors the kind of resilience that Notion likely implemented after their service disruption. By building these patterns into your applications, you can ensure that temporary API issues don't completely break your systems, maintaining a better user experience even during service interruptions.