File size: 3,916 Bytes
81e3673 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
Minimal test server for testing the streaming and canvas implementation.
This bypasses existing backend issues to test only the new functionality.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
# Create minimal FastAPI app
app = FastAPI(title="Atom Test Server")
# Add CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Import and include ONLY the routes we created/tested
try:
print("Importing canvas routes...")
from api.canvas_routes import router as canvas_router
app.include_router(canvas_router)
print("✓ Canvas routes loaded")
except Exception as e:
print(f"✗ Canvas routes failed: {e}")
try:
print("Importing atom agent routes (for streaming endpoint)...")
# We'll create a minimal streaming endpoint inline since the full module has dependencies
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
class ChatRequest(BaseModel):
message: str
user_id: str
session_id: Optional[str] = None
workspace_id: Optional[str] = None
current_page: Optional[str] = None
conversation_history: Optional[List[Dict]] = None
@app.post("/api/atom-agent/chat/stream")
async def chat_stream(request: ChatRequest):
"""Test streaming endpoint"""
import uuid
from core.websockets import manager as ws_manager
message_id = str(uuid.uuid4())
user_channel = f"user:{request.user_id}"
# Simulate streaming
test_response = "This is a test streaming response. The implementation is working correctly!"
# Send tokens
for i, char in enumerate(test_response):
await ws_manager.broadcast(user_channel, {
"type": "streaming:update",
"id": message_id,
"delta": char,
"complete": False
})
# Send completion
await ws_manager.broadcast(user_channel, {
"type": "streaming:complete",
"id": message_id,
"content": test_response,
"complete": True
})
return {
"success": True,
"message_id": message_id,
"streamed": True
}
print("✓ Streaming endpoint loaded (inline)")
except Exception as e:
print(f"✗ Streaming endpoint failed: {e}")
# WebSocket endpoint
from fastapi import WebSocket
from core.websockets import manager as ws_manager
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""WebSocket endpoint for testing"""
user = await ws_manager.connect(websocket, token="dev-token")
if user:
print(f"✓ User {user.id} connected via WebSocket")
try:
while True:
await websocket.receive_text()
except:
pass
finally:
ws_manager.disconnect(websocket, user.id)
else:
await websocket.close()
@app.get("/")
async def root():
return {"status": "ok", "message": "Atom Test Server - Streaming & Canvas Implementation"}
@app.get("/health")
async def health():
return {"status": "healthy", "implementation": "streaming + canvas"}
if __name__ == "__main__":
print("\n" + "="*60)
print("ATOM TEST SERVER")
print("="*60)
print("Testing Implementation:")
print(" - Phase 1: LLM Token Streaming")
print(" - Phase 2: Canvas Chart Components")
print(" - Phase 3: Interactive Form System")
print("="*60)
print("\nStarting server on http://localhost:8000")
print("WebSocket: ws://localhost:8000/ws?token=dev-token")
print("\nPress Ctrl+C to stop\n")
uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")
|