#!/usr/bin/env python3
"""
Simple Notion MCP Server Example
================================
A minimal example showing how to communicate with the Notion MCP server.
Use this for quick testing or as a starting point for your own implementation.
"""
import subprocess
import json
import os
import time
import sys
import platform
def check_docker_availability():
"""Check if Docker is available and the image exists."""
try:
# Check if Docker is running
result = subprocess.run(["docker", "--version"], capture_output=True, text=True)
if result.returncode != 0:
print("ā Docker is not available")
return False
print(f"ā
Docker version: {result.stdout.strip()}")
# Check if the notion image exists locally
result = subprocess.run(["docker", "images", "mcp/notion"], capture_output=True, text=True)
if "mcp/notion" in result.stdout:
print("ā
mcp/notion image found locally")
return True
# Try to pull the image
print("š Pulling mcp/notion image...")
result = subprocess.run(["docker", "pull", "mcp/notion"], capture_output=True, text=True)
if result.returncode == 0:
print("ā
Successfully pulled mcp/notion image")
return True
else:
print(f"ā Failed to pull mcp/notion image: {result.stderr}")
return False
except FileNotFoundError:
print("ā Docker is not installed")
return False
except Exception as e:
print(f"ā Error checking Docker: {e}")
return False
def run_simple_mcp_test():
"""Run a simple MCP test with minimal setup."""
# Configuration
notion_token = os.getenv("NOTION_TOKEN", "ntn_****")
if notion_token == "ntn_****":
print("ā Please set your NOTION_TOKEN environment variable")
print(" Example: export NOTION_TOKEN='ntn_your_actual_token'")
return
# Check Docker availability
if not check_docker_availability():
print("ā Docker setup failed. Please ensure Docker is installed and running.")
return
# Set up environment
env = os.environ.copy()
headers_json = json.dumps({
"Authorization": f"Bearer {notion_token}",
"Notion-Version": "2022-06-28"
})
env["OPENAPI_MCP_HEADERS"] = headers_json
print("š Starting Notion MCP Server...")
print(f"š Using token: {notion_token[:10]}...")
# Start Docker container with improved error handling
try:
proc = subprocess.Popen(
[
"docker", "run", "--rm", "-i",
"-e", "OPENAPI_MCP_HEADERS",
"mcp/notion"
],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
text=True,
bufsize=1
)
print(f"ā
Container started (PID: {proc.pid})")
# Wait for container to start
time.sleep(3)
# Check if container is still running
if proc.poll() is not None:
stderr_output = proc.stderr.read()
stdout_output = proc.stdout.read()
print(f"ā Container exited early (return code: {proc.returncode})")
print(f"STDERR: {stderr_output}")
print(f"STDOUT: {stdout_output}")
return
# Test 1: Initialize MCP server
print("\nš Test 1: Initialize MCP Server")
init_request = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"clientInfo": {"name": "simple-test", "version": "1.0.0"}
},
"id": "init-1"
}
try:
request_json = json.dumps(init_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
# Wait for response with timeout
import select
if platform.system() != "Windows":
# Unix-like systems
ready, _, _ = select.select([proc.stdout], [], [], 5.0)
if ready:
response = proc.stdout.readline()
else:
print("ā Timeout waiting for initialization response")
return
else:
# Windows - use a simple timeout
time.sleep(2)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
print(f"ā
Initialization successful")
print(f" Server info: {result.get('result', {}).get('serverInfo', {})}")
except json.JSONDecodeError:
print(f"ā Invalid JSON response: {response}")
return
else:
print("ā No response from server")
return
except Exception as e:
print(f"ā Error during initialization: {e}")
return
# Test 2: List available tools
print("\nš Test 2: List Available Tools")
tools_request = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": "tools-1"
}
try:
request_json = json.dumps(tools_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
# Wait for response
if platform.system() != "Windows":
ready, _, _ = select.select([proc.stdout], [], [], 5.0)
if ready:
response = proc.stdout.readline()
else:
print("ā Timeout waiting for tools list response")
return
else:
time.sleep(2)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
tools = result.get("result", {}).get("tools", [])
print(f"ā
Found {len(tools)} tools:")
# Show clean tool list (removing the verbose descriptions)
important_tools = []
for tool in tools:
tool_name = tool.get('name', 'Unknown')
# Extract clean description (remove "Error Responses" part)
description = tool.get('description', 'No description')
print("tool name", tool_name)
print(tool.get('inputSchema','no schema').get('properties',"no properties"))
clean_desc = description.split('\nError Responses:')[0].split('Error Responses:')[0]
# Focus on the most useful tools for listing pages
# if any(keyword in tool_name.lower() for keyword in ['search', 'page', 'database', 'retrieve']):
# important_tools.append(f" ā¢ {tool_name}: {clean_desc}")
for tool_info in important_tools[:8]: # Show first 8 important tools
print(tool_info)
if len(tools) > 8:
print(f" ... and {len(tools) - 8} more tools")
except json.JSONDecodeError:
print(f"ā Invalid JSON response: {response}")
return
else:
print("ā No response from server")
return
except Exception as e:
print(f"ā Error listing tools: {e}")
return
# Test 3: List all pages in your Notion workspace
print("\nš Test 3: List All Pages in Your Notion Workspace")
# First, try to get user info to verify permissions
print(" š Step 3.1: Checking your Notion access...")
user_info_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "API-get-self",
"arguments": {}
},
"id": "user-info-1"
}
try:
request_json = json.dumps(user_info_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
if platform.system() != "Windows":
ready, _, _ = select.select([proc.stdout], [], [], 5.0)
if ready:
response = proc.stdout.readline()
else:
print(" ā Timeout waiting for user info response")
else:
time.sleep(2)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
print(f" š Raw user info response: {result}")
if result.get("result"):
response_data = result["result"]
# Handle MCP wrapped response - the actual data is in content[0].text as JSON string
if isinstance(response_data, dict) and "content" in response_data:
content = response_data["content"]
if content and len(content) > 0 and "text" in content[0]:
# Parse the JSON string inside the text field
try:
user_json = json.loads(content[0]["text"])
user_name = user_json.get("name", "Unknown")
user_type = user_json.get("type", "Unknown")
workspace_name = "Unknown"
if "bot" in user_json and "workspace_name" in user_json["bot"]:
workspace_name = user_json["bot"]["workspace_name"]
print(f" ā
Connected as: {user_name} (Type: {user_type})")
print(f" š¢ Workspace: {workspace_name}")
except json.JSONDecodeError as e:
print(f" ā Failed to parse user JSON: {e}")
else:
print(" š No content or text field in user response")
# Handle direct response (fallback)
elif isinstance(response_data, dict):
user_name = response_data.get("name", "Unknown")
user_type = response_data.get("type", "Unknown")
print(f" ā
Connected as: {user_name} (Type: {user_type})")
else:
print(f" ā
User info retrieved: {response_data}")
else:
print(f" ā ļø User info response: {result}")
except json.JSONDecodeError:
print(f" ā Invalid JSON response: {response}")
except Exception as e:
print(f" ā Error getting user info: {e}")
# Now try multiple approaches to list pages
print("\n š Step 3.2: Trying different approaches to list pages...")
# Approach 1: Search with empty query
print(" š Approach 1: Search with empty query...")
list_pages_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "API-post-search",
"arguments": {
"query": "",
"filter": {
"value": "page",
"property": "object"
},
"page_size": 20
}
},
"id": "list-pages-1"
}
try:
request_json = json.dumps(list_pages_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
if platform.system() != "Windows":
ready, _, _ = select.select([proc.stdout], [], [], 10.0)
if ready:
response = proc.stdout.readline()
else:
print(" ā Timeout waiting for pages list response")
else:
time.sleep(3)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
print(f" š Raw search response: {json.dumps(result, indent=2)[:500]}...")
pages_found = False
if result.get("result"):
response_data = result["result"]
# Handle MCP wrapped response - the actual data is in content[0].text as JSON string
if isinstance(response_data, dict) and "content" in response_data:
content = response_data["content"]
if content and len(content) > 0 and "text" in content[0]:
# Parse the JSON string inside the text field
try:
actual_json = json.loads(content[0]["text"])
if "results" in actual_json:
pages = actual_json["results"]
if pages:
print(f" ā
Found {len(pages)} pages:")
pages_found = True
for i, page in enumerate(pages[:10], 1): # Show first 10
page_id = page.get("id", "Unknown ID")
# Extract page title
title = "Untitled"
if "properties" in page and "title" in page["properties"]:
title_prop = page["properties"]["title"]
if "title" in title_prop and title_prop["title"]:
title = title_prop["title"][0]["text"]["content"]
# Extract other info
created = page.get("created_time", "Unknown")[:10]
modified = page.get("last_edited_time", "Unknown")[:10]
page_url = page.get("url", "No URL")
print(f" {i}. š {title}")
print(f" ID: {page_id}")
print(f" Created: {created} | Modified: {modified}")
print(f" URL: {page_url}")
print()
else:
print(" š Empty results array - no pages returned")
else:
print(" š No 'results' field in parsed JSON")
except json.JSONDecodeError as e:
print(f" ā Failed to parse JSON from text field: {e}")
print(f" š Raw text: {content[0]['text'][:200]}...")
else:
print(" š No content or text field in response")
# Handle direct API response (fallback)
elif isinstance(response_data, dict) and "results" in response_data:
pages = response_data["results"]
if pages:
print(f" ā
Found {len(pages)} pages:")
pages_found = True
for i, page in enumerate(pages[:10], 1): # Show first 10
page_id = page.get("id", "Unknown ID")
# Extract page title
title = "Untitled"
if "properties" in page:
props = page["properties"]
# Look for title in different possible locations
if "title" in props:
title_prop = props["title"]
if isinstance(title_prop, dict) and "title" in title_prop:
title_content = title_prop["title"]
if title_content and len(title_content) > 0:
title = title_content[0].get("text", {}).get("content", "Untitled")
elif "Name" in props:
name_prop = props["Name"]
if isinstance(name_prop, dict) and "title" in name_prop:
title_content = name_prop["title"]
if title_content and len(title_content) > 0:
title = title_content[0].get("text", {}).get("content", "Untitled")
# Extract other info
created = page.get("created_time", "Unknown")[:10]
modified = page.get("last_edited_time", "Unknown")[:10]
page_url = page.get("url", "No URL")
print(f" {i}. š {title}")
print(f" ID: {page_id}")
print(f" Created: {created} | Modified: {modified}")
print(f" URL: {page_url}")
print()
else:
print(" š Empty results array - no pages returned")
# Handle string response
elif isinstance(response_data, str):
print(f" š String response: {response_data}")
if "pages" in response_data.lower() or "results" in response_data.lower():
pages_found = True
if not pages_found:
print(" ā ļø No pages found with this approach")
except json.JSONDecodeError:
print(f" ā Invalid JSON response: {response}")
else:
print(" ā No response from server")
except Exception as e:
print(f" ā Error in approach 1: {e}")
# Approach 2: Search without filters
print("\n š Approach 2: Search without filters...")
search_all_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "API-post-search",
"arguments": {
"page_size": 10
}
},
"id": "search-all-1"
}
try:
request_json = json.dumps(search_all_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
if platform.system() != "Windows":
ready, _, _ = select.select([proc.stdout], [], [], 10.0)
if ready:
response = proc.stdout.readline()
else:
print(" ā Timeout waiting for search all response")
else:
time.sleep(3)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
print(f" š Search all response: {json.dumps(result, indent=2)[:300]}...")
if result.get("result"):
response_data = result["result"]
# Handle MCP wrapped response - the actual data is in content[0].text as JSON string
if isinstance(response_data, dict) and "content" in response_data:
content = response_data["content"]
if content and len(content) > 0 and "text" in content[0]:
# Parse the JSON string inside the text field
try:
actual_json = json.loads(content[0]["text"])
if "results" in actual_json:
all_items = actual_json["results"]
print(f" ā
Found {len(all_items)} total items (pages + databases)")
pages = [item for item in all_items if item.get("object") == "page"]
databases = [item for item in all_items if item.get("object") == "database"]
print(f" š Pages: {len(pages)}")
print(f" šļø Databases: {len(databases)}")
if pages:
print(" š Your Pages:")
for i, page in enumerate(pages[:5], 1):
title = "Untitled"
if "properties" in page and "title" in page["properties"]:
title_prop = page["properties"]["title"]
if "title" in title_prop and title_prop["title"]:
title = title_prop["title"][0]["text"]["content"]
print(f" {i}. {title} ({page.get('id', 'No ID')[:8]}...)")
if databases:
print(" šļø Your Databases:")
for i, db in enumerate(databases[:5], 1):
title = "Untitled Database"
if "title" in db and db["title"]:
title = db["title"][0]["text"]["content"]
print(f" {i}. {title} ({db.get('id', 'No ID')[:8]}...)")
else:
print(" š No 'results' field in parsed JSON")
except json.JSONDecodeError as e:
print(f" ā Failed to parse JSON from text field: {e}")
print(f" š Raw text: {content[0]['text'][:200]}...")
else:
print(" š No content or text field in response")
# Handle direct API response (fallback)
elif isinstance(response_data, dict) and "results" in response_data:
all_items = response_data["results"]
print(f" ā
Found {len(all_items)} total items (pages + databases)")
pages = [item for item in all_items if item.get("object") == "page"]
databases = [item for item in all_items if item.get("object") == "database"]
print(f" š Pages: {len(pages)}")
print(f" šļø Databases: {len(databases)}")
if pages:
print(" š Your Pages:")
for i, page in enumerate(pages[:5], 1):
title = "Untitled"
if "properties" in page and "title" in page["properties"]:
title_prop = page["properties"]["title"]
if "title" in title_prop and title_prop["title"]:
title = title_prop["title"][0].get("text", {}).get("content", "Untitled")
print(f" {i}. {title} ({page.get('id', 'No ID')[:8]}...)")
if databases:
print(" šļø Your Databases:")
for i, db in enumerate(databases[:5], 1):
title = "Untitled Database"
if "title" in db and db["title"]:
title = db["title"][0].get("text", {}).get("content", "Untitled Database")
print(f" {i}. {title} ({db.get('id', 'No ID')[:8]}...)")
else:
print(" š No results found in response")
else:
print(" ā No result in response")
if "error" in result:
print(f" ā ļø Error: {result['error']}")
except json.JSONDecodeError:
print(f" ā Invalid JSON response: {response}")
else:
print(" ā No response from server")
except Exception as e:
print(f" ā Error in approach 2: {e}")
# Summary for Test 3
print("\n š Test 3 Summary:")
print(" ā¹ļø If no pages were found, this could mean:")
print(" 1. Your Notion integration doesn't have access to any pages")
print(" 2. You need to share pages with your integration")
print(" 3. Your workspace is empty")
print(" š” To fix: Go to your Notion pages ā Share ā Add your integration")
# Test 4: Search for specific content
print("\nš Test 4: Search for Specific Content")
print(" š Searching for pages containing 'aws'...")
search_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "API-post-search",
"arguments": {
"query": "aws",
"page_size": 10
}
},
"id": "search-1"
}
try:
request_json = json.dumps(search_request) + "\n"
proc.stdin.write(request_json)
proc.stdin.flush()
if platform.system() != "Windows":
ready, _, _ = select.select([proc.stdout], [], [], 10.0)
if ready:
response = proc.stdout.readline()
else:
print(" ā Timeout waiting for search response")
return
else:
time.sleep(3)
response = proc.stdout.readline()
if response:
try:
result = json.loads(response.strip())
print(f" š Raw search response: {json.dumps(result, indent=2)[:400]}...")
if result.get("result"):
response_data = result["result"]
# Handle MCP wrapped response - the actual data is in content[0].text as JSON string
if isinstance(response_data, dict) and "content" in response_data:
content = response_data["content"]
if content and len(content) > 0 and "text" in content[0]:
# Parse the JSON string inside the text field
try:
actual_json = json.loads(content[0]["text"])
if "results" in actual_json:
items = actual_json["results"]
if items:
print(f" ā
Found {len(items)} items matching 'aws':")
for i, item in enumerate(items, 1):
item_type = item.get("object", "unknown")
item_id = item.get("id", "Unknown ID")
# Extract title based on type
title = "Untitled"
if item_type == "page" and "properties" in item:
props = item["properties"]
if "title" in props and props["title"].get("title"):
title = props["title"]["title"][0]["text"]["content"]
elif item_type == "database" and "title" in item:
if item["title"]:
title = item["title"][0]["text"]["content"]
icon = "š" if item_type == "page" else "šļø" if item_type == "database" else "ā"
print(f" {i}. {icon} {title}")
print(f" Type: {item_type}")
print(f" ID: {item_id[:8]}...")
if "url" in item:
print(f" URL: {item['url']}")
print()
else:
print(" š No items found matching 'aws'")
print(" š” Try searching for:")
print(" - Common words like 'project', 'notes', 'todo'")
print(" - Your actual page titles")
print(" - Or use an empty search to see all pages")
else:
print(" š No 'results' field in parsed JSON")
except json.JSONDecodeError as e:
print(f" ā Failed to parse JSON from text field: {e}")
print(f" š Raw text: {content[0]['text'][:200]}...")
else:
print(" š No content or text field in response")
# Handle direct API response (fallback)
elif isinstance(response_data, dict) and "results" in response_data:
items = response_data["results"]
if items:
print(f" ā
Found {len(items)} items matching 'aws':")
for i, item in enumerate(items, 1):
item_type = item.get("object", "unknown")
item_id = item.get("id", "Unknown ID")
# Extract title based on type
title = "Untitled"
if item_type == "page" and "properties" in item:
props = item["properties"]
if "title" in props and props["title"].get("title"):
title = props["title"]["title"][0].get("text", {}).get("content", "Untitled")
elif item_type == "database" and "title" in item:
if item["title"]:
title = item["title"][0].get("text", {}).get("content", "Untitled Database")
icon = "š" if item_type == "page" else "šļø" if item_type == "database" else "ā"
print(f" {i}. {icon} {title}")
print(f" Type: {item_type}")
print(f" ID: {item_id[:8]}...")
if "url" in item:
print(f" URL: {item['url']}")
print()
else:
print(" š No items found matching 'aws'")
print(" š” Try searching for:")
print(" - Common words like 'project', 'notes', 'todo'")
print(" - Your actual page titles")
print(" - Or use an empty search to see all pages")
else:
print(" š Unexpected response format")
print(f" š Response data: {response_data}")
else:
print(" ā Search failed")
if "error" in result:
print(f" ā ļø Error: {result['error']}")
except json.JSONDecodeError:
print(f" ā Invalid JSON response: {response}")
return
else:
print(" ā No response from server")
return
except Exception as e:
print(f" ā Error during search: {e}")
return
print("\nš All MCP tests completed successfully!")
print("\nš Summary:")
print(" ā
MCP server connection working")
print(" ā
Notion API authentication successful")
print(" ā
Tool discovery completed")
print(" ā
Workspace page listing working")
print(" ā
Content search working")
print("\nš” Note: 'Error Responses: 400: 400' in tool descriptions is normal")
print(" This is just API documentation, not an actual error!")
except Exception as e:
print(f"ā Error starting container: {e}")
import traceback
traceback.print_exc()
finally:
# Clean up
if 'proc' in locals():
print("\nš Stopping container...")
proc.terminate()
proc.wait()
print("ā
Container stopped")
def main():
"""Main function with multiple approaches."""
print("š Notion MCP Server Test & Page Listing")
print("=" * 60)
# Check token first
notion_token = os.getenv("NOTION_TOKEN")
if not notion_token:
print("ā Please set your NOTION_TOKEN environment variable")
print(" Example: export NOTION_TOKEN='ntn_your_actual_token'")
print(" Get your token at: https://www.notion.so/profile/integrations")
return
try:
# Run the comprehensive test
run_simple_mcp_test()
except KeyboardInterrupt:
print("\nā ļø Test interrupted by user")
except Exception as e:
print(f"\nā Test failed with error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
