Bi-Temporal Knowledge Graph MCP Server

""" Database Seed Script ==================== This script seeds the PostgreSQL database with example webhook configurations for testing and demonstration purposes. Usage: python seed_db.py """ import os import psycopg2 from psycopg2.extras import Json import sys # Load environment variables from dotenv import load_dotenv load_dotenv() POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost") POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432")) POSTGRES_DB = os.getenv("POSTGRES_DB", "automation_db") POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres") POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "") def get_connection(): """Get database connection.""" return psycopg2.connect( host=POSTGRES_HOST, port=POSTGRES_PORT, database=POSTGRES_DB, user=POSTGRES_USER, password=POSTGRES_PASSWORD ) def seed_webhooks(): """Seed example webhook configurations.""" webhooks = [ { 'user_id': 'demo_user', 'name': 'Slack Notification', 'description': 'Send a notification to Slack', 'url': 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL', 'method': 'POST', 'headers': Json({'Content-Type': 'application/json'}), 'template_fields': Json({ 'message': { 'type': 'str', 'description': 'Message content', 'required': True }, 'channel': { 'type': 'str', 'description': 'Slack channel', 'required': False, 'default': '#general' } }), 'item_type': 'single' }, { 'user_id': 'demo_user', 'name': 'Discord Notification', 'description': 'Send a notification to Discord', 'url': 'https://discord.com/api/webhooks/YOUR/WEBHOOK', 'method': 'POST', 'headers': Json({'Content-Type': 'application/json'}), 'template_fields': Json({ 'content': { 'type': 'str', 'description': 'Message content', 'required': True }, 'username': { 'type': 'str', 'description': 'Bot username', 'required': False, 'default': 'MCP Bot' } }), 'item_type': 'single' }, { 'user_id': 'demo_user', 'name': 'Email Notification', 'description': 'Send an email via webhook', 'url': 'https://api.example.com/send-email', 'method': 'POST', 'headers': Json({ 'Content-Type': 'application/json', 'Authorization': 'Bearer YOUR_API_KEY' }), 'template_fields': Json({ 'to': { 'type': 'str', 'description': 'Email recipient', 'required': True }, 'subject': { 'type': 'str', 'description': 'Email subject', 'required': True }, 'body': { 'type': 'str', 'description': 'Email body', 'required': True } }), 'item_type': 'single' }, { 'user_id': 'demo_user', 'name': 'System Log', 'description': 'Log to external system', 'url': 'https://api.example.com/log', 'method': 'POST', 'headers': Json({'Content-Type': 'application/json'}), 'template_fields': Json({ 'event': { 'type': 'str', 'description': 'Event name', 'required': True }, 'level': { 'type': 'str', 'description': 'Log level', 'required': False, 'default': 'info' }, 'data': { 'type': 'dict', 'description': 'Event data', 'required': False, 'default': {} } }), 'item_type': 'single' } ] try: conn = get_connection() cur = conn.cursor() for webhook in webhooks: try: cur.execute(""" INSERT INTO webhooks (user_id, name, description, url, method, headers, template_fields, item_type) VALUES (%(user_id)s, %(name)s, %(description)s, %(url)s, %(method)s, %(headers)s, %(template_fields)s, %(item_type)s) ON CONFLICT (user_id, name) DO UPDATE SET description = EXCLUDED.description, url = EXCLUDED.url, method = EXCLUDED.method, headers = EXCLUDED.headers, template_fields = EXCLUDED.template_fields, updated_at = CURRENT_TIMESTAMP RETURNING id """, webhook) webhook_id = cur.fetchone()[0] print(f"✅ Created/Updated webhook: {webhook['name']} (ID: {webhook_id})") except Exception as e: print(f"❌ Error creating webhook {webhook['name']}: {e}") conn.rollback() continue conn.commit() # Get webhook IDs for templates cur.execute(""" SELECT id FROM webhooks WHERE user_id = 'demo_user' AND name IN ('Slack Notification', 'Discord Notification') """) webhook_ids = [row[0] for row in cur.fetchall()] if len(webhook_ids) >= 2: # Create multi-webhook template cur.execute(""" INSERT INTO webhook_templates (user_id, name, description, webhook_ids, item_type) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (user_id, name) DO UPDATE SET description = EXCLUDED.description, webhook_ids = EXCLUDED.webhook_ids, updated_at = CURRENT_TIMESTAMP RETURNING id """, ( 'demo_user', 'Broadcast Message', 'Send message to both Slack and Discord', webhook_ids, 'multi' )) template_id = cur.fetchone()[0] print(f"✅ Created/Updated template: Broadcast Message (ID: {template_id})") conn.commit() cur.close() conn.close() print("\n✅ Database seeding completed successfully!") print("\nNext steps:") print("1. Update webhook URLs with your actual endpoints") print("2. Run: python main.py") print("3. Generate tools using the generate_tool_from_db MCP tool") except Exception as e: print(f"❌ Database seeding failed: {e}") sys.exit(1) if __name__ == "__main__": print("🌱 Seeding database with example webhooks...\n") seed_webhooks()