feat(notion): 添加Notion数据库集成功能和相关脚本

添加Notion数据库集成配置和访问脚本,包括:
- 配置Notion MCP服务器设置
- 添加数据库查询、内容查看和添加功能
- 创建测试脚本和集成文档
This commit is contained in:
ben
2026-02-02 08:49:17 +00:00
parent f7abc14d4e
commit 738e5fc991
7 changed files with 443 additions and 1 deletions

95
notion_database_access.py Normal file
View File

@@ -0,0 +1,95 @@
"""
Notion Database Access Script
This script demonstrates how to connect to and access your Notion database
using the Notion API with the configuration you have set up.
"""
import requests
import json
class NotionDatabaseAccess:
def __init__(self, integration_token="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"):
self.integration_token = integration_token
self.headers = {
"Authorization": f"Bearer {integration_token}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28"
}
self.base_url = "https://api.notion.com/v1"
def list_databases(self):
"""List all databases accessible with the integration token"""
url = f"{self.base_url}/search"
payload = {
"filter": {
"property": "object",
"value": "database"
}
}
response = requests.post(url, headers=self.headers, json=payload)
if response.status_code == 200:
return response.json()
else:
print(f"Error: {response.status_code} - {response.text}")
return None
def query_database(self, database_id):
"""Query a specific database by ID"""
url = f"{self.base_url}/databases/{database_id}/query"
response = requests.post(url, headers=self.headers)
if response.status_code == 200:
return response.json()
else:
print(f"Error querying database: {response.status_code} - {response.text}")
return None
def get_database_info(self, database_id):
"""Get information about a specific database"""
url = f"{self.base_url}/databases/{database_id}"
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
return response.json()
else:
print(f"Error getting database info: {response.status_code} - {response.text}")
return None
def main():
# Initialize the Notion database access
notion = NotionDatabaseAccess()
print("Connecting to your Notion account...")
print("Searching for accessible databases...")
# List all databases
databases_result = notion.list_databases()
if databases_result and 'results' in databases_result:
databases = databases_result['results']
if databases:
print(f"\nFound {len(databases)} database(s):\n")
for db in databases:
db_id = db['id']
db_title = ''.join([title.get('plain_text', '') for title in db.get('title', [])])
print(f"Database ID: {db_id}")
print(f"Title: {db_title}")
print("-" * 50)
# Get detailed info about each database
db_info = notion.get_database_info(db_id)
if db_info:
print(f"Description: {''.join([desc.get('plain_text', '') for desc in db_info.get('description', [])])}")
print(f"Properties:")
for prop_name, prop_info in db_info.get('properties', {}).items():
print(f" - {prop_name}: {prop_info.get('type', 'unknown')}")
print()
else:
print("No databases found. Make sure your integration is properly connected to your Notion pages.")
else:
print("Failed to retrieve databases. Check your integration token and permissions.")
if __name__ == "__main__":
main()