Web Sockets, MCP, or Directly? who can give the best solution? Fsat API EX -
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, List, Optional
import json
import uuid
import uvicorn
app = FastAPI()
In-memory database for this example
items = {}
class ConnectionManager:
def init(self):
self.active_connections: List[WebSocket] =
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.get(“/”)
async def get():
with open(“index.html”, “r”) as f:
return f.read()
@app.websocket(“/ws”)
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
# Send all items when connection is established
items_html = generate_items_html()
await manager.send_personal_message(
f'<div id="items-list">{items_html}</div>',
websocket
)
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
action = message.get("action", "")
if action == "create":
# Create new item
item_id = str(uuid.uuid4())
items[item_id] = {
"id": item_id,
"title": message.get("title", ""),
"description": message.get("description", "")
}
# Broadcast updated items list to all clients
await manager.broadcast(generate_items_html())
elif action == "edit-form":
# Send edit form for specific item
item_id = message.get("id")
if item_id in items:
item = items[item_id]
edit_form = generate_edit_form(item)
await manager.send_personal_message(
f'<div id="edit-form-{item_id}" class="edit-form">{edit_form}</div>',
websocket
)
elif action == "update":
# Update existing item
item_id = message.get("id")
if item_id in items:
items[item_id]["title"] = message.get("title", items[item_id]["title"])
items[item_id]["description"] = message.get("description", items[item_id]["description"])
# Broadcast updated items list to all clients
await manager.broadcast(generate_items_html())
elif action == "delete":
# Delete an item
item_id = message.get("id")
if item_id in items:
del items[item_id]
# Broadcast updated items list to all clients
await manager.broadcast(generate_items_html())
elif action == "cancel-edit":
# Handle cancel edit - simply hide the form
item_id = message.get("id")
await manager.send_personal_message(
f'<div id="edit-form-{item_id}" class="edit-form hidden"></div>',
websocket
)
except WebSocketDisconnect:
manager.disconnect(websocket)
def generate_items_html() → str:
“”“Generate HTML for all items”“”
if not items:
return “
No items yet. Add one using the form above.
”html = ""
for item_id, item in items.items():
html += f"""
<div class="item" id="item-{item['id']}">
<h3>{item['title']}</h3>
<p>{item['description']}</p>
<div class="item-actions">
<button ws-send
hx-vals='{{"action": "edit-form", "id": "{item['id']}"}}''>
Edit
</button>
<button ws-send
hx-confirm="Are you sure you want to delete this item?"
hx-vals='{{"action": "delete", "id": "{item['id']}"}}''>
Delete
</button>
</div>
<div id="edit-form-{item['id']}" class="edit-form hidden"></div>
</div>
"""
return html
def generate_edit_form(item) → str:
“”“Generate HTML for an edit form”“”
return f"“”
Title:
Description:
{item[‘description’]}
Save Changes
<button type=“button” ws-send hx-vals=‘{{“action”: “cancel-edit”, “id”: “{item[‘id’]}”}}’'>Cancel
“”"
if name == “main”:
uvicorn.run(“app:app”, host=“0.0.0.0”, port=8000, reload=True)
HTMX WebSocket CRUD Example body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } .item { border: 1px solid #ddd; padding: 10px; margin-bottom: 10px; border-radius: 4px; } .item-actions { margin-top: 10px; } button { margin-right: 5px; cursor: pointer; } form { margin-bottom: 20px; } input, textarea { width: 100%; padding: 8px; margin-bottom: 10px; } .edit-form { background-color: #f7f7f7; padding: 10px; margin-top: 10px; } .hidden { display: none; }
WebSocket CRUD with HTMX
<!-- WebSocket connection container -->
<div hx-ext="ws" ws-connect="/ws">
<!-- Create form -->
<div>
<h2>Add New Item</h2>
<form ws-send id="createForm">
<input type="hidden" name="action" value="create">
<div>
<label for="title">Title:</label>
<input type="text" id="title" name="title" required>
</div>
<div>
<label for="description">Description:</label>
<textarea id="description" name="description" rows="3"></textarea>
</div>
<button type="submit">Add Item</button>
</form>
</div>
<!-- Items container - will be updated by the server -->
<div id="items-container">
<h2>Items</h2>
<div id="items-list">
<!-- Items will be populated here -->
<div class="loading">Loading items...</div>
</div>
</div>
<!-- Template for individual item -->
<template id="item-template">
<div class="item" id="item-{id}">
<h3>{title}</h3>
<p>{description}</p>
<div class="item-actions">
<button ws-send
hx-vals='{"action": "edit-form", "id": "{id}"}'>
Edit
</button>
<button ws-send
hx-confirm="Are you sure you want to delete this item?"
hx-vals='{"action": "delete", "id": "{id}"}'>
Delete
</button>
</div>
<div id="edit-form-{id}" class="edit-form hidden"></div>
</div>
</template>
<!-- Template for edit form -->
<template id="edit-form-template">
<form ws-send>
<input type="hidden" name="action" value="update">
<input type="hidden" name="id" value="{id}">
<div>
<label for="edit-title-{id}">Title:</label>
<input type="text" id="edit-title-{id}" name="title" value="{title}" required>
</div>
<div>
<label for="edit-description-{id}">Description:</label>
<textarea id="edit-description-{id}" name="description" rows="3">{description}</textarea>
</div>
<button type="submit">Save Changes</button>
<button type="button" ws-send hx-vals='{"action": "cancel-edit", "id": "{id}"}'>Cancel</button>
</form>
</template>
</div>
<script>
// Initialize WebSocket connection
document.addEventListener('DOMContentLoaded', function() {
// Reset create form after submission
document.getElementById('createForm').addEventListener('htmx:afterRequest', function(event) {
if (event.detail.successful) {
this.reset();
}
});
});
</script>