diff --git a/examples/python/16-gateway-integration/app.py b/examples/python/16-gateway-integration/app.py
index a845d23..bd876b0 100644
--- a/examples/python/16-gateway-integration/app.py
+++ b/examples/python/16-gateway-integration/app.py
@@ -44,15 +44,18 @@ import sys
 # ── Import praisonaiui ──────────────────────────────────────
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiui.integration import AIUIGateway
     from praisonaiagents import Agent
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies missing: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode.")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies missing: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode.")
     GATEWAY_OK = False
 
 # ── Set dashboard style ─────────────────────────────────────
@@ -149,15 +152,15 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Dashboard: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Dashboard: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
 async def main():
     host = os.getenv("HOST", "127.0.0.1")
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         # Create the gateway
         gateway = AIUIGateway(
@@ -174,22 +177,22 @@ async def main():
                 self_reflect=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:   {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:   {agent_def['name']} ({agent_def['model']})")
 
         # ── Register agents in dashboard CRUD storage ───────
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:8083")
-        print(f"   WebSocket:  ws://localhost:8083/ws")
-        print(f"   Inspector:  http://localhost:8083 → 🔍 Feature Inspector")
-        print()
+        safe_print()
+        safe_print("   Dashboard:  http://localhost:8083")
+        safe_print("   WebSocket:  ws://localhost:8083/ws")
+        safe_print(f"   Inspector:  http://localhost:8083 {icon('→', '->')} {icon('🔍', '')} Feature Inspector")
+        safe_print()
 
         # Start the gateway — this runs the full server
         await gateway.start()
     else:
         # Fallback to standalone mode (same as example 15)
-        print("📦 Starting in standalone mode (CRUD only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (CRUD only, no agent execution)")
         from praisonaiui.server import create_app
         import uvicorn
 
diff --git a/examples/python/17-three-column-demo/app.py b/examples/python/17-three-column-demo/app.py
index ff35d4e..645e358 100644
--- a/examples/python/17-three-column-demo/app.py
+++ b/examples/python/17-three-column-demo/app.py
@@ -26,15 +26,18 @@ import sys
 
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiui.integration import AIUIGateway
     from praisonaiagents import Agent
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies not found: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode (REST APIs only).\n")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies not found: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode (REST APIs only).\n")
     GATEWAY_OK = False
 
 aiui.set_style("dashboard")
@@ -115,7 +118,7 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Agent: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
@@ -125,8 +128,8 @@ async def main():
     host = os.getenv("HOST", "127.0.0.1")
 
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         gateway = AIUIGateway(host=host, port=port)
 
@@ -139,32 +142,32 @@ async def main():
                 reflection=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:  {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:  {agent_def['name']} ({agent_def['model']})")
 
         # Also register in dashboard CRUD
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   WebSocket:  ws://localhost:{port}/ws")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print(f"   WebSocket:  ws://localhost:{port}/ws")
+        safe_print()
 
         await gateway.start()
     else:
         # Standalone: REST APIs only, no agent execution
-        print("📦 Starting in standalone mode (REST APIs only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (REST APIs only, no agent execution)")
         from praisonaiui.server import create_app
         import uvicorn
 
         _register_agents_in_dashboard()
 
         app = create_app()
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   Note:       POST endpoints may fail without gateway")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print("   Note:       POST endpoints may fail without gateway")
+        safe_print()
 
         config = uvicorn.Config(app, host=host, port=port, log_level="info")
         server = uvicorn.Server(config)
diff --git a/examples/python/18-full-chat/app.py b/examples/python/18-full-chat/app.py
index 3e423cc..06d660f 100644
--- a/examples/python/18-full-chat/app.py
+++ b/examples/python/18-full-chat/app.py
@@ -25,9 +25,13 @@ Usage:
 
 import asyncio
 import os
+import sys
 
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiagents import Agent
@@ -35,9 +39,9 @@ try:
     from praisonaiui.integration import AIUIGateway
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies not found: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode (REST APIs only).\n")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies not found: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode (REST APIs only).\n")
     GATEWAY_OK = False
 
 aiui.set_style("dashboard")
@@ -127,7 +131,7 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Agent: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
@@ -137,8 +141,8 @@ async def main():
     host = os.getenv("HOST", "127.0.0.1")
 
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         gateway = AIUIGateway(host=host, port=port)
 
@@ -151,21 +155,21 @@ async def main():
                 reflection=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:  {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:  {agent_def['name']} ({agent_def['model']})")
 
         # Also register in dashboard CRUD
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   WebSocket:  ws://localhost:{port}/ws")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print(f"   WebSocket:  ws://localhost:{port}/ws")
+        safe_print()
 
         await gateway.start()
     else:
         # Standalone: REST APIs only, no agent execution
-        print("📦 Starting in standalone mode (REST APIs only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (REST APIs only, no agent execution)")
         import uvicorn
 
         from praisonaiui.server import create_app
@@ -173,11 +177,11 @@ async def main():
         _register_agents_in_dashboard()
 
         app = create_app()
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print("   Note:       POST endpoints may fail without gateway")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print("   Note:       POST endpoints may fail without gateway")
+        safe_print()
 
         config = uvicorn.Config(app, host=host, port=port, log_level="info")
         server = uvicorn.Server(config)
diff --git a/examples/python/20-email-bot/app.py b/examples/python/20-email-bot/app.py
index ab5f36a..6104789 100644
--- a/examples/python/20-email-bot/app.py
+++ b/examples/python/20-email-bot/app.py
@@ -32,6 +32,9 @@ import asyncio
 import os
 import sys
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check prerequisites ─────────────────────────────────────────────
 
 try:
@@ -40,8 +43,8 @@ try:
     SDK_OK = True
 except ImportError:
     SDK_OK = False
-    print("⚠️  Missing dependencies. Install with:")
-    print("   pip install praisonai[email] praisonaiagents")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Missing dependencies. Install with:")
+    safe_print("   pip install praisonai[email] praisonaiagents")
     sys.exit(1)
 
 
@@ -68,8 +71,8 @@ agent = Agent(
 async def run_simple():
     """Start email bot with automatic env var resolution."""
     bot = Bot("email", agent=agent)
-    print(f"📧 Starting email bot for: {os.getenv('EMAIL_ADDRESS', '(not set)')}")
-    print("   Listening for new emails... (Ctrl+C to stop)")
+    safe_print(f"{icon('📧', '[EMAIL]')} Starting email bot for: {os.getenv('EMAIL_ADDRESS', '(not set)')}")
+    safe_print("   Listening for new emails... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -96,9 +99,9 @@ async def run_with_handlers():
     @bot.on_message
     async def log_email(message):
         subject = message.metadata.get("subject", "(no subject)")
-        print(f"📨 From: {message.sender.username}")
-        print(f"   Subject: {subject}")
-        print(f"   Preview: {message.text[:100]}...")
+        safe_print(f"{icon('📨', '[MSG]')} From: {message.sender.username}")
+        safe_print(f"   Subject: {subject}")
+        safe_print(f"   Preview: {message.text[:100]}...")
 
     # Custom command: emails with subject starting with "/status"
     @bot.on_command("status")
@@ -118,11 +121,11 @@ async def run_with_handlers():
             reply_to=message.message_id,
         )
 
-    print(f"📧 Starting email bot with custom handlers")
-    print(f"   Email: {bot._email_address}")
-    print(f"   Poll interval: {config.polling_interval}s")
-    print(f"   Allowed senders: {'all' if not config.allowed_users else config.allowed_users}")
-    print("   Listening... (Ctrl+C to stop)")
+    safe_print(f"{icon('📧', '[EMAIL]')} Starting email bot with custom handlers")
+    safe_print(f"   Email: {bot._email_address}")
+    safe_print(f"   Poll interval: {config.polling_interval}s")
+    safe_print(f"   Allowed senders: {'all' if not config.allowed_users else config.allowed_users}")
+    safe_print("   Listening... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -131,17 +134,17 @@ async def run_with_handlers():
 async def run_probe():
     """Test email server connectivity."""
     bot = Bot("email", agent=agent)
-    print("🔍 Testing email server connectivity...")
+    safe_print(f"{icon('🔍', '[PROBE]')} Testing email server connectivity...")
     result = await bot.probe()
 
-    print(f"\n   Connection: {'✅ OK' if result.ok else '❌ FAILED'}")
-    print(f"   Platform:   {result.platform}")
+    safe_print(f"\n   Connection: {icon('✅', '[OK]') + ' OK' if result.ok else icon('❌', '[FAIL]') + ' FAILED'}")
+    safe_print(f"   Platform:   {result.platform}")
     if result.bot_username:
-        print(f"   Email:      {result.bot_username}")
+        safe_print(f"   Email:      {result.bot_username}")
     if result.elapsed_ms:
-        print(f"   Latency:    {result.elapsed_ms:.0f}ms")
+        safe_print(f"   Latency:    {result.elapsed_ms:.0f}ms")
     if result.error:
-        print(f"   Error:      {result.error}")
+        safe_print(f"   Error:      {result.error}")
 
 
 # ── Main ────────────────────────────────────────────────────────────
diff --git a/examples/python/21-email-channel/app.py b/examples/python/21-email-channel/app.py
index c8df095..41a1bde 100644
--- a/examples/python/21-email-channel/app.py
+++ b/examples/python/21-email-channel/app.py
@@ -42,7 +42,9 @@ import sys
 # ── Imports ─────────────────────────────────────────────────────────
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "src"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
+from _shared.console import icon, safe_print
 from praisonaiui.server import create_app
 import uvicorn
 
@@ -91,11 +93,11 @@ def seed_email_channel():
         })
         if r.status_code in (200, 201):
             channel_id = r.json().get("id", "unknown")
-            print(f"   ✅ Email channel registered: {channel_id}")
+            safe_print(f"   {icon('✅', '[OK]')} Email channel registered: {channel_id}")
         else:
-            print(f"   ⚠️  Channel registration: {r.status_code} — {r.json()}")
+            safe_print(f"   {icon('⚠️', '[WARN]')}  Channel registration: {r.status_code} - {r.json()}")
     else:
-        print("   ✓ Email channel already exists")
+        safe_print(f"   {icon('✓', '[OK]')} Email channel already exists")
 
     # Also seed other channels for comparison in dashboard
     for name, platform in [("Discord #general", "discord"), ("Telegram Support", "telegram")]:
@@ -106,7 +108,7 @@ def seed_email_channel():
                 "platform": platform,
                 "config": {},
             })
-            print(f"   ✓ Seeded {platform} channel")
+            safe_print(f"   {icon('✓', '[OK]')} Seeded {platform} channel")
 
 
 # ── Verify email platform support ──────────────────────────────────
@@ -122,8 +124,8 @@ def verify_email_support():
     if r.status_code == 200:
         platforms = r.json().get("platforms", [])
         email_supported = "email" in platforms
-        print(f"   Supported platforms: {platforms}")
-        print(f"   Email supported: {'✅ Yes' if email_supported else '❌ No — add email to SUPPORTED_PLATFORMS'}")
+        safe_print(f"   Supported platforms: {platforms}")
+        safe_print(f"   Email supported: {icon('✅', '[OK]') + ' Yes' if email_supported else icon('❌', '[X]') + ' No - add email to SUPPORTED_PLATFORMS'}")
         return email_supported
     return False
 
@@ -137,51 +139,51 @@ def test_channel_api():
     app = create_app()
     client = TestClient(app)
 
-    print("\n📋 Channel API Test")
-    print("=" * 40)
+    safe_print(f"\n{icon('📋', '[TEST]')} Channel API Test")
+    safe_print("=" * 40)
 
     # List channels
     r = client.get("/api/channels")
     channels = r.json().get("channels", [])
-    print(f"   GET /api/channels: {len(channels)} channels")
+    safe_print(f"   GET /api/channels: {len(channels)} channels")
     for ch in channels:
-        icon = "📧" if ch.get("platform") == "email" else "💬"
-        print(f"      {icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
+        ch_icon = icon("📧", "[email]") if ch.get("platform") == "email" else icon("💬", "[chat]")
+        safe_print(f"      {ch_icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
 
     # Get email channel status
     email_ch = next((ch for ch in channels if ch.get("platform") == "email"), None)
     if email_ch:
         ch_id = email_ch.get("id")
         r = client.get(f"/api/channels/{ch_id}/status")
-        print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
+        safe_print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
 
-    print()
+    safe_print()
 
 
 # ── Main ────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    print("📧 PraisonAIUI — Email Channel Integration")
-    print("=" * 50)
-    print()
+    safe_print(f"{icon('📧', '[EMAIL]')} PraisonAIUI - Email Channel Integration")
+    safe_print("=" * 50)
+    safe_print()
 
     if "--test" in sys.argv:
         # Test mode: verify APIs + seed data
-        print("🔍 Verifying email platform support...")
+        safe_print(f"{icon('🔍', '[VERIFY]')} Verifying email platform support...")
         verify_email_support()
-        print()
-        print("🌱 Seeding email channel...")
+        safe_print()
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding email channel...")
         seed_email_channel()
         test_channel_api()
     else:
         # Run the dashboard
-        print("🌱 Seeding channels...")
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding channels...")
         seed_email_channel()
-        print()
+        safe_print()
 
         app = create_app()
-        print(f"✅ Dashboard at http://localhost:8084")
-        print(f"   Channels: http://localhost:8084/api/channels")
-        print(f"   Platforms: http://localhost:8084/api/channels/platforms")
+        safe_print(f"{icon('✅', '[OK]')} Dashboard at http://localhost:8084")
+        safe_print("   Channels: http://localhost:8084/api/channels")
+        safe_print("   Platforms: http://localhost:8084/api/channels/platforms")
         host = os.getenv("HOST", "127.0.0.1")
         uvicorn.run(app, host=host, port=8084, log_level="info")
diff --git a/examples/python/22-agentmail-bot/app.py b/examples/python/22-agentmail-bot/app.py
index e380b04..2e28787 100644
--- a/examples/python/22-agentmail-bot/app.py
+++ b/examples/python/22-agentmail-bot/app.py
@@ -33,6 +33,9 @@ import asyncio
 import os
 import sys
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check prerequisites ─────────────────────────────────────────────
 
 try:
@@ -41,8 +44,8 @@ try:
     SDK_OK = True
 except ImportError:
     SDK_OK = False
-    print("⚠️  Missing dependencies. Install with:")
-    print("   pip install praisonai agentmail praisonaiagents")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Missing dependencies. Install with:")
+    safe_print("   pip install praisonai agentmail praisonaiagents")
     sys.exit(1)
 
 
@@ -69,9 +72,9 @@ agent = Agent(
 async def run_simple():
     """Start agentmail bot with automatic env var resolution."""
     bot = Bot("agentmail", agent=agent)
-    print(f"📩 Starting AgentMail bot")
-    print(f"   API key: {'✅ set' if os.getenv('AGENTMAIL_API_KEY') else '❌ not set'}")
-    print("   Listening for new emails... (Ctrl+C to stop)")
+    safe_print(f"{icon('📩', '[MAIL]')} Starting AgentMail bot")
+    safe_print(f"   API key: {icon('✅', '[OK]') + ' set' if os.getenv('AGENTMAIL_API_KEY') else icon('❌', '[X]') + ' not set'}")
+    safe_print("   Listening for new emails... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -99,9 +102,9 @@ async def run_with_handlers():
     @bot.on_message
     async def log_email(message):
         subject = message.metadata.get("subject", "(no subject)")
-        print(f"📨 From: {message.sender.username}")
-        print(f"   Subject: {subject}")
-        print(f"   Preview: {message.text[:100]}...")
+        safe_print(f"{icon('📨', '[MSG]')} From: {message.sender.username}")
+        safe_print(f"   Subject: {subject}")
+        safe_print(f"   Preview: {message.text[:100]}...")
 
     # Custom command: emails with subject starting with "/status"
     @bot.on_command("status")
@@ -121,10 +124,10 @@ async def run_with_handlers():
             reply_to=message.message_id,
         )
 
-    print(f"📩 Starting AgentMail bot with custom handlers")
-    print(f"   Email: {bot.email_address or '(will be assigned on start)'}")
-    print(f"   Poll interval: {config.polling_interval}s")
-    print("   Listening... (Ctrl+C to stop)")
+    safe_print(f"{icon('📩', '[MAIL]')} Starting AgentMail bot with custom handlers")
+    safe_print(f"   Email: {bot.email_address or '(will be assigned on start)'}")
+    safe_print(f"   Poll interval: {config.polling_interval}s")
+    safe_print("   Listening... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -136,30 +139,30 @@ async def run_inbox_demo():
         token=os.environ.get("AGENTMAIL_API_KEY", ""),
     )
 
-    print("📩 AgentMail Inbox Lifecycle Demo")
-    print("=" * 50)
+    safe_print(f"{icon('📩', '[MAIL]')} AgentMail Inbox Lifecycle Demo")
+    safe_print("=" * 50)
 
     # Create inbox
-    print("\n1. Creating inbox...")
+    safe_print("\n1. Creating inbox...")
     inbox = await bot.create_inbox(domain=os.getenv("AGENTMAIL_DOMAIN"))
-    print(f"   ✅ Created: {inbox.get('email_address', 'unknown')}")
-    print(f"   ID: {inbox.get('id', 'unknown')}")
+    safe_print(f"   {icon('✅', '[OK]')} Created: {inbox.get('email_address', 'unknown')}")
+    safe_print(f"   ID: {inbox.get('id', 'unknown')}")
 
     # List inboxes
-    print("\n2. Listing inboxes...")
+    safe_print("\n2. Listing inboxes...")
     inboxes = await bot.list_inboxes()
-    print(f"   Found: {len(inboxes)} inbox(es)")
+    safe_print(f"   Found: {len(inboxes)} inbox(es)")
     for ib in inboxes:
-        print(f"      📧 {ib.get('email_address', '?')} (id: {ib.get('id', '?')})")
+        safe_print(f"      {icon('📧', '[EMAIL]')} {ib.get('email_address', '?')} (id: {ib.get('id', '?')})")
 
     # Delete inbox
     inbox_id = inbox.get("id", "")
     if inbox_id:
-        print(f"\n3. Deleting inbox {inbox_id}...")
+        safe_print(f"\n3. Deleting inbox {inbox_id}...")
         deleted = await bot.delete_inbox(inbox_id)
-        print(f"   {'✅ Deleted' if deleted else '❌ Failed to delete'}")
+        safe_print(f"   {icon('✅', '[OK]') + ' Deleted' if deleted else icon('❌', '[X]') + ' Failed to delete'}")
 
-    print("\n✅ Inbox lifecycle complete")
+    safe_print(f"\n{icon('✅', '[OK]')} Inbox lifecycle complete")
 
 
 # ── Probe: Test connectivity ────────────────────────────────────────
@@ -167,17 +170,17 @@ async def run_inbox_demo():
 async def run_probe():
     """Test AgentMail API connectivity."""
     bot = Bot("agentmail", agent=agent)
-    print("🔍 Testing AgentMail API connectivity...")
+    safe_print(f"{icon('🔍', '[PROBE]')} Testing AgentMail API connectivity...")
     result = await bot.probe()
 
-    print(f"\n   Connection: {'✅ OK' if result.ok else '❌ FAILED'}")
-    print(f"   Platform:   {result.platform}")
+    safe_print(f"\n   Connection: {icon('✅', '[OK]') + ' OK' if result.ok else icon('❌', '[FAIL]') + ' FAILED'}")
+    safe_print(f"   Platform:   {result.platform}")
     if result.bot_username:
-        print(f"   Email:      {result.bot_username}")
+        safe_print(f"   Email:      {result.bot_username}")
     if result.elapsed_ms:
-        print(f"   Latency:    {result.elapsed_ms:.0f}ms")
+        safe_print(f"   Latency:    {result.elapsed_ms:.0f}ms")
     if result.error:
-        print(f"   Error:      {result.error}")
+        safe_print(f"   Error:      {result.error}")
 
 
 # ── Main ────────────────────────────────────────────────────────────
diff --git a/examples/python/23-agentmail-channel/app.py b/examples/python/23-agentmail-channel/app.py
index 4a527aa..1fe0136 100644
--- a/examples/python/23-agentmail-channel/app.py
+++ b/examples/python/23-agentmail-channel/app.py
@@ -40,7 +40,9 @@ import sys
 # ── Imports ─────────────────────────────────────────────────────────
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "src"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
+from _shared.console import icon, safe_print
 from praisonaiui.server import create_app
 import uvicorn
 
@@ -88,11 +90,11 @@ def seed_agentmail_channel():
         })
         if r.status_code in (200, 201):
             channel_id = r.json().get("id", "unknown")
-            print(f"   ✅ AgentMail channel registered: {channel_id}")
+            safe_print(f"   {icon('✅', '[OK]')} AgentMail channel registered: {channel_id}")
         else:
-            print(f"   ⚠️  Channel registration: {r.status_code} — {r.json()}")
+            safe_print(f"   {icon('⚠️', '[WARN]')}  Channel registration: {r.status_code} - {r.json()}")
     else:
-        print("   ✓ AgentMail channel already exists")
+        safe_print(f"   {icon('✓', '[OK]')} AgentMail channel already exists")
 
     # Also seed email + telegram for comparison in dashboard
     for name, platform in [("Support Email", "email"), ("Telegram Support", "telegram")]:
@@ -103,7 +105,7 @@ def seed_agentmail_channel():
                 "platform": platform,
                 "config": {},
             })
-            print(f"   ✓ Seeded {platform} channel")
+            safe_print(f"   {icon('✓', '[OK]')} Seeded {platform} channel")
 
 
 # ── Verify agentmail platform support ──────────────────────────────
@@ -120,9 +122,9 @@ def verify_agentmail_support():
         platforms = r.json().get("platforms", [])
         am_supported = "agentmail" in platforms
         email_supported = "email" in platforms
-        print(f"   Supported platforms: {platforms}")
-        print(f"   email supported: {'✅ Yes' if email_supported else '❌ No'}")
-        print(f"   agentmail supported: {'✅ Yes' if am_supported else '❌ No'}")
+        safe_print(f"   Supported platforms: {platforms}")
+        safe_print(f"   email supported: {icon('✅', '[OK]') + ' Yes' if email_supported else icon('❌', '[X]') + ' No'}")
+        safe_print(f"   agentmail supported: {icon('✅', '[OK]') + ' Yes' if am_supported else icon('❌', '[X]') + ' No'}")
         return am_supported
     return False
 
@@ -136,51 +138,51 @@ def test_channel_api():
     app = create_app()
     client = TestClient(app)
 
-    print("\n📋 Channel API Test")
-    print("=" * 40)
+    safe_print(f"\n{icon('📋', '[TEST]')} Channel API Test")
+    safe_print("=" * 40)
 
     # List channels
     r = client.get("/api/channels")
     channels = r.json().get("channels", [])
-    print(f"   GET /api/channels: {len(channels)} channels")
+    safe_print(f"   GET /api/channels: {len(channels)} channels")
     for ch in channels:
-        icon = {"email": "📧", "agentmail": "📩"}.get(ch.get("platform", ""), "💬")
-        print(f"      {icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
+        ch_icon = {"email": icon("📧", "[email]"), "agentmail": icon("📩", "[mail]")}.get(ch.get("platform", ""), icon("💬", "[chat]"))
+        safe_print(f"      {ch_icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
 
     # Get agentmail channel status
     am_ch = next((ch for ch in channels if ch.get("platform") == "agentmail"), None)
     if am_ch:
         ch_id = am_ch.get("id")
         r = client.get(f"/api/channels/{ch_id}/status")
-        print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
+        safe_print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
 
-    print()
+    safe_print()
 
 
 # ── Main ────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    print("📩 PraisonAIUI — AgentMail Channel Integration")
-    print("=" * 50)
-    print()
+    safe_print(f"{icon('📩', '[MAIL]')} PraisonAIUI - AgentMail Channel Integration")
+    safe_print("=" * 50)
+    safe_print()
 
     if "--test" in sys.argv:
         # Test mode: verify APIs + seed data
-        print("🔍 Verifying agentmail platform support...")
+        safe_print(f"{icon('🔍', '[VERIFY]')} Verifying agentmail platform support...")
         verify_agentmail_support()
-        print()
-        print("🌱 Seeding agentmail channel...")
+        safe_print()
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding agentmail channel...")
         seed_agentmail_channel()
         test_channel_api()
     else:
         # Run the dashboard
-        print("🌱 Seeding channels...")
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding channels...")
         seed_agentmail_channel()
-        print()
+        safe_print()
 
         app = create_app()
-        print(f"✅ Dashboard at http://localhost:8084")
-        print(f"   Channels: http://localhost:8084/api/channels")
-        print(f"   Platforms: http://localhost:8084/api/channels/platforms")
+        safe_print(f"{icon('✅', '[OK]')} Dashboard at http://localhost:8084")
+        safe_print("   Channels: http://localhost:8084/api/channels")
+        safe_print("   Platforms: http://localhost:8084/api/channels/platforms")
         host = os.getenv("HOST", "127.0.0.1")
         uvicorn.run(app, host=host, port=8084, log_level="info")
diff --git a/examples/python/_shared/console.py b/examples/python/_shared/console.py
new file mode 100644
index 0000000..a588e60
--- /dev/null
+++ b/examples/python/_shared/console.py
@@ -0,0 +1,40 @@
+"""Cross-platform console output for examples.
+
+Legacy Windows consoles default stdout to a non-UTF code page (cp1252/cp437)
+that cannot encode emoji such as U+1F680. Calling ``print`` with those
+characters raises ``UnicodeEncodeError`` and crashes the example before the
+server starts. These helpers mirror the ``_supports_unicode``/``_icon`` pattern
+in ``praisonaiui.cli`` so examples degrade to ASCII instead of crashing while
+preserving emoji on UTF-aware terminals.
+
+Usage:
+    from _shared.console import safe_print, icon
+
+    safe_print(f"{icon('🚀', '[START]')} Starting gateway...")
+"""
+
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+
+def supports_unicode() -> bool:
+    try:
+        encoding = sys.stdout.encoding or ""
+    except Exception:
+        return False
+    return "utf" in encoding.lower()
+
+
+def icon(symbol: str, fallback: str) -> str:
+    return symbol if supports_unicode() else fallback
+
+
+def safe_print(*args: Any, **kwargs: Any) -> None:
+    sep = kwargs.pop("sep", " ")
+    text = sep.join(str(a) for a in args)
+    try:
+        print(text, **kwargs)
+    except UnicodeEncodeError:
+        print(text.encode("ascii", errors="replace").decode("ascii"), **kwargs)
diff --git a/tests/unit/test_examples_console.py b/tests/unit/test_examples_console.py
new file mode 100644
index 0000000..153be9b
--- /dev/null
+++ b/tests/unit/test_examples_console.py
@@ -0,0 +1,70 @@
+"""Tests for the shared examples console helper.
+
+These verify that examples degrade emoji to ASCII on legacy Windows code pages
+(cp1252) instead of raising ``UnicodeEncodeError`` while preserving emoji on
+UTF-aware terminals, mirroring the CLI ``_icon``/``_supports_unicode`` pattern.
+"""
+
+import io
+import os
+import sys
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "examples", "python", "_shared"))
+from console import icon, safe_print, supports_unicode
+
+
+class TestSupportsUnicode:
+    def test_utf8_encoding_is_supported(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "utf-8"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is True
+
+    def test_cp1252_encoding_is_not_supported(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "cp1252"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is False
+
+    def test_missing_encoding_falls_back_to_ascii(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = None
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is False
+
+
+class TestIcon:
+    def test_returns_symbol_on_utf8(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "utf-8"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert icon("🚀", "[START]") == "🚀"
+
+    def test_returns_fallback_on_cp1252(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "cp1252"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert icon("🚀", "[START]") == "[START]"
+
+
+class TestSafePrint:
+    def test_never_raises_on_cp1252(self):
+        buffer = io.BytesIO()
+        stream = io.TextIOWrapper(buffer, encoding="cp1252", errors="strict")
+        with mock.patch("sys.stdout", stream):
+            safe_print("🚀 Starting")
+            stream.flush()
+        assert buffer.getvalue()
+
+    def test_preserves_unicode_on_utf8(self):
+        buffer = io.StringIO()
+        with mock.patch("sys.stdout", buffer):
+            safe_print("🚀 Starting")
+        assert "🚀" in buffer.getvalue()
+
+    def test_respects_sep(self):
+        buffer = io.StringIO()
+        with mock.patch("sys.stdout", buffer):
+            safe_print("a", "b", sep="-")
+        assert buffer.getvalue().strip() == "a-b"