MervinPraison/PraisonAIUI
1
0 Fork 0
mirror of https://github.com/MervinPraison/PraisonAIUI.git synced 2026-11-07 05:56:17 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

feat: safe_print for examples to avoid cp1252 UnicodeEncodeError (fixes #231)

Browse Source 
Add examples/python/_shared/console.py with safe_print/icon helpers that
degrade emoji to ASCII on legacy Windows code pages (cp1252) instead of
crashing, mirroring the CLI _supports_unicode/_icon pattern. Wire the helper
into the high-severity gateway (16, 17, 18) and channel (20-23) examples so
they start on default Windows PowerShell without PYTHONUTF8.

Co-authored-by: Mervin Praison <[email protected]>
This commit is contained in:
praisonai-triage-agent[bot]andMervin Praison committed 2026-07-27 09:04:48 +00:00
1 parent 01a31f7c8b
commit 6fd27d0ed7
9 files changed
+275 -145
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

examples/python/16-gateway-integration/app.py
+16 -13
View File
@@ -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
examples/python/17-three-column-demo/app.py
+21 -18
View File
@@ -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)
examples/python/18-full-chat/app.py
+22 -18
View File
@@ -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)
examples/python/20-email-bot/app.py
+21 -18
View File
@@ -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 ────────────────────────────────────────────────────────────
examples/python/21-email-channel/app.py
+26 -24
View File
@@ -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")
examples/python/22-agentmail-bot/app.py
+32 -29
View File
@@ -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 ────────────────────────────────────────────────────────────
examples/python/23-agentmail-channel/app.py
+27 -25
View File
@@ -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")
examples/python/_shared/console.py
+40
View File
@@ -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)
tests/unit/test_examples_console.py
+70
View File
@@ -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"