Update app.py

app.py

@@ -1,31 +1,16 @@
-import os
 import asyncio
 from uuid import uuid4
 from datetime import datetime, timezone
-from typing import Optional, Dict, Any, List
+from typing import Dict, Any, Optional, List
 
-from fastapi import FastAPI, HTTPException, Header, Request
-from fastapi.responses import JSONResponse
+from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel, Field
-from starlette.routing import Mount
 
-# MCP (official Python SDK)
 from mcp.server.fastmcp import FastMCP, Context
 
-# ============================================================
-# CONFIG
-# ============================================================
-API_TOKEN = os.getenv("ROBOT_API_TOKEN", "change-me-please")
-ALLOWED_ORIGINS = {
-    # Add your domains here (HF space URL, localhost during testing, etc.)
-    "https://your-space-name.hf.space",
-    "http://localhost:7860",
-    "http://127.0.0.1:7860",
-}
 
 # ============================================================
-# IN-MEMORY COMMAND STORE
-# (Good for MVP / HF Space demos. Use Redis later if you want persistence.)
+# In-memory command queue (local dev)
 # ============================================================
 commands: Dict[str, Dict[str, Any]] = {}
 queue: List[str] = []
@@ -36,287 +21,107 @@ def now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
-def check_bearer(auth_header: Optional[str]):
-    if not API_TOKEN:
-        return
-    if not auth_header or not auth_header.startswith("Bearer "):
-        raise HTTPException(status_code=401, detail="Missing bearer token")
-    token = auth_header.split(" ", 1)[1].strip()
-    if token != API_TOKEN:
-        raise HTTPException(status_code=403, detail="Invalid token")
-
-
-def command_expired(cmd: Dict[str, Any]) -> bool:
-    ttl = cmd.get("ttl_seconds")
-    if not ttl:
-        return False
-    created = datetime.fromisoformat(cmd["created_at"])
-    age = (datetime.now(timezone.utc) - created).total_seconds()
-    return age > ttl
-
-
-async def enqueue_command(
-    command: str,
-    args: Optional[Dict[str, Any]] = None,
-    source: str = "mcp",
-    priority: int = 0,
-    ttl_seconds: int = 120,
-    meta: Optional[Dict[str, Any]] = None,
-) -> Dict[str, Any]:
+async def enqueue(command: str, args: Optional[Dict[str, Any]] = None, source: str = "mcp") -> Dict[str, Any]:
     cmd_id = str(uuid4())
     record = {
         "id": cmd_id,
         "command": command,
         "args": args or {},
         "source": source,
-        "priority": priority,
-        "ttl_seconds": ttl_seconds,
-        "status": "queued",
+        "status": "queued",      # queued | running | done | failed
         "result": None,
         "error": None,
         "created_at": now_iso(),
         "updated_at": now_iso(),
         "claimed_by": None,
-        "meta": meta or {},
     }
-
     async with queue_lock:
         commands[cmd_id] = record
-
-        # priority insert (higher first)
-        inserted = False
-        for i, queued_id in enumerate(queue):
-            q = commands.get(queued_id)
-            if not q:
-                continue
-            if record["priority"] > q.get("priority", 0):
-                queue.insert(i, cmd_id)
-                inserted = True
-                break
-        if not inserted:
-            queue.append(cmd_id)
-
+        queue.append(cmd_id)
     return record
 
 
 # ============================================================
-# FASTAPI MODELS (robot runner REST side)
-# ============================================================
-class CommandStatusUpdate(BaseModel):
-    status: str = Field(..., description="queued | running | done | failed | expired")
-    result: Optional[Dict[str, Any]] = None
-    error: Optional[str] = None
-    robot_id: Optional[str] = None
-
-
-class CommandCreateREST(BaseModel):
-    command: str
-    args: Dict[str, Any] = Field(default_factory=dict)
-    source: str = "api"
-    priority: int = 0
-    ttl_seconds: int = 120
-
-
-# ============================================================
-# MCP SERVER (tools for Claude/ChatGPT/etc)
+# MCP server (SSE transport)
 # ============================================================
 mcp = FastMCP(
-    "Create3 Robot Bridge",
+    "Create3 Local Bridge (SSE)",
     instructions=(
-        "Send safe, short Create 3 robot commands by enqueueing them for a local robot runner. "
-        "This server does not execute commands directly; it queues them for a trusted client."
+        "Tools queue robot commands for a local Create 3 runner. "
+        "This MCP server does not control the robot directly."
     ),
-    stateless_http=True,
-    json_response=True,
 )
 
-# Optional: simple helper tool
-@mcp.tool()
-async def robot_status_summary(ctx: Context) -> dict:
-    """Get queue and command status summary."""
-    queued = sum(1 for c in commands.values() if c["status"] == "queued")
-    running = sum(1 for c in commands.values() if c["status"] == "running")
-    done = sum(1 for c in commands.values() if c["status"] == "done")
-    failed = sum(1 for c in commands.values() if c["status"] == "failed")
-    expired = sum(1 for c in commands.values() if c["status"] == "expired")
-    return {
-        "ok": True,
-        "server": "Create3 Robot Bridge",
-        "queued": queued,
-        "running": running,
-        "done": done,
-        "failed": failed,
-        "expired": expired,
-        "total": len(commands),
-        "time": now_iso(),
-    }
-
-
-@mcp.tool()
-async def list_recent_commands(limit: int = 20, ctx: Optional[Context] = None) -> dict:
-    """List recent robot commands and their status."""
-    items = sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)[: max(1, min(limit, 100))]
-    return {"ok": True, "items": items}
-
-
-@mcp.tool()
-async def get_command(command_id: str, ctx: Optional[Context] = None) -> dict:
-    """Get the status of one command by ID."""
-    cmd = commands.get(command_id)
-    if not cmd:
-        return {"ok": False, "error": "Command not found", "command_id": command_id}
-    return {"ok": True, "item": cmd}
-
-
-# ---- Robot command tools (enqueue only) ----
-@mcp.tool()
-async def undock(priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Undock the Create 3 from its dock."""
-    rec = await enqueue_command("undock", {}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
-    return {"ok": True, "queued": rec}
-
-
-@mcp.tool()
-async def dock(priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Dock the Create 3 to its dock."""
-    rec = await enqueue_command("dock", {}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
-    return {"ok": True, "queued": rec}
-
-
-@mcp.tool()
-async def move_cm(cm: float, priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Move forward (or backward if negative) in centimeters."""
-    rec = await enqueue_command("move_cm", {"cm": cm}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
-    return {"ok": True, "queued": rec}
-
-
-@mcp.tool()
-async def turn_left(deg: float = 90, priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Turn the robot left by degrees."""
-    rec = await enqueue_command("turn_left", {"deg": deg}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
-    return {"ok": True, "queued": rec}
-
+# --- Exactly 3 tools ---
 
 @mcp.tool()
-async def turn_right(deg: float = 90, priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Turn the robot right by degrees."""
-    rec = await enqueue_command("turn_right", {"deg": deg}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
+async def dock(ctx: Context) -> dict:
+    """Dock the Create 3."""
+    rec = await enqueue("dock", {}, source="mcp")
     return {"ok": True, "queued": rec}
 
 
 @mcp.tool()
-async def drive(left: float, right: float, seconds: float = 1.0, priority: int = 0, ttl_seconds: int = 120, ctx: Optional[Context] = None) -> dict:
-    """Direct wheel drive command. Use small values for safety."""
-    rec = await enqueue_command(
-        "drive",
-        {"left": left, "right": right, "seconds": seconds},
-        source="mcp",
-        priority=priority,
-        ttl_seconds=ttl_seconds,
-    )
+async def undock(ctx: Context) -> dict:
+    """Undock the Create 3."""
+    rec = await enqueue("undock", {}, source="mcp")
     return {"ok": True, "queued": rec}
 
 
 @mcp.tool()
-async def stop(priority: int = 100, ttl_seconds: int = 30, ctx: Optional[Context] = None) -> dict:
-    """Emergency stop. High priority."""
-    rec = await enqueue_command("stop", {}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
-    return {"ok": True, "queued": rec}
-
-
-@mcp.tool()
-async def say(text: str, priority: int = 0, ttl_seconds: int = 60, ctx: Optional[Context] = None) -> dict:
-    """Queue a local console 'say' message (placeholder for TTS if you add it client-side)."""
-    rec = await enqueue_command("say", {"text": text}, source="mcp", priority=priority, ttl_seconds=ttl_seconds)
+async def move_cm(cm: float, ctx: Context) -> dict:
+    """Move forward (negative cm moves backward)."""
+    rec = await enqueue("move_cm", {"cm": cm}, source="mcp")
     return {"ok": True, "queued": rec}
 
 
 # ============================================================
-# FASTAPI APP (REST endpoints for robot client + MCP mount)
+# FastAPI app + queue endpoints for local robot runner
 # ============================================================
-app = FastAPI(title="Create3 MCP + Queue Bridge", version="2.0.0")
-
+app = FastAPI(title="Create3 MCP SSE + Queue (Local)", version="1.0.0")
 
-# --- Security middleware for MCP HTTP endpoint (Origin + Bearer) ---
-@app.middleware("http")
-async def mcp_security_middleware(request: Request, call_next):
-    # MCP is mounted under /mcp (via Starlette mount below)
-    if request.url.path.startswith("/mcp"):
-        # Origin validation for browser-based calls (recommended by MCP transport spec)
-        origin = request.headers.get("origin")
-        if origin:
-            if origin not in ALLOWED_ORIGINS:
-                return JSONResponse(status_code=403, content={"detail": "Origin not allowed"})
 
-        # Bearer token auth
-        try:
-            check_bearer(request.headers.get("authorization"))
-        except HTTPException as e:
-            return JSONResponse(status_code=e.status_code, content={"detail": e.detail})
-
-    # Also protect REST command endpoints
-    if request.url.path.startswith("/commands"):
-        try:
-            check_bearer(request.headers.get("authorization"))
-        except HTTPException as e:
-            return JSONResponse(status_code=e.status_code, content={"detail": e.detail})
-
-    return await call_next(request)
+class StatusUpdate(BaseModel):
+    status: str = Field(..., description="running | done | failed")
+    result: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None
+    robot_id: Optional[str] = None
 
 
 @app.get("/")
 async def root():
     return {
         "ok": True,
-        "service": "Create3 MCP + Queue Bridge",
-        "mcp_endpoint": "/mcp",
+        "mcp_sse_endpoint": "/sse",
+        "note": "Point your MCP connector to http://localhost:8000/sse",
         "time": now_iso(),
     }
 
 
 @app.get("/health")
 async def health():
-    queued = sum(1 for c in commands.values() if c["status"] == "queued")
-    running = sum(1 for c in commands.values() if c["status"] == "running")
-    return {"ok": True, "queued": queued, "running": running, "total": len(commands), "time": now_iso()}
-
-
-# ---- REST enqueue endpoint (optional, for testing / non-MCP senders) ----
-@app.post("/commands")
-async def create_command(cmd: CommandCreateREST, authorization: Optional[str] = Header(default=None)):
-    check_bearer(authorization)
-    rec = await enqueue_command(
-        cmd.command,
-        args=cmd.args,
-        source=cmd.source,
-        priority=cmd.priority,
-        ttl_seconds=cmd.ttl_seconds,
-    )
-    return {"ok": True, "queued": rec}
+    return {
+        "ok": True,
+        "queued": sum(1 for c in commands.values() if c["status"] == "queued"),
+        "running": sum(1 for c in commands.values() if c["status"] == "running"),
+        "total": len(commands),
+        "time": now_iso(),
+    }
 
 
 @app.get("/commands")
-async def list_commands(limit: int = 20, authorization: Optional[str] = Header(default=None)):
-    check_bearer(authorization)
-    items = sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)[: max(1, min(limit, 100))]
+async def list_commands(limit: int = 50):
+    items = sorted(
+        commands.values(),
+        key=lambda x: x["updated_at"],
+        reverse=True
+    )[: max(1, min(limit, 200))]
     return {"ok": True, "items": items}
 
 
-@app.get("/commands/{command_id}")
-async def get_command_rest(command_id: str, authorization: Optional[str] = Header(default=None)):
-    check_bearer(authorization)
-    if command_id not in commands:
-        raise HTTPException(status_code=404, detail="Command not found")
-    return {"ok": True, "item": commands[command_id]}
-
-
 @app.post("/commands/next")
-async def claim_next_command(robot_id: str, authorization: Optional[str] = Header(default=None)):
-    """
-    Robot client polls this endpoint to claim one command.
-    """
-    check_bearer(authorization)
-
+async def claim_next_command(robot_id: str):
+    """Robot runner polls this to claim the next queued command."""
     async with queue_lock:
         while queue:
             cmd_id = queue.pop(0)
@@ -326,12 +131,6 @@ async def claim_next_command(robot_id: str, authorization: Optional[str] = Heade
             if cmd["status"] != "queued":
                 continue
 
-            if command_expired(cmd):
-                cmd["status"] = "expired"
-                cmd["error"] = "TTL expired before execution"
-                cmd["updated_at"] = now_iso()
-                continue
-
             cmd["status"] = "running"
             cmd["claimed_by"] = robot_id
             cmd["updated_at"] = now_iso()
@@ -341,13 +140,7 @@ async def claim_next_command(robot_id: str, authorization: Optional[str] = Heade
 
 
 @app.post("/commands/{command_id}/status")
-async def update_command_status(
-    command_id: str,
-    update: CommandStatusUpdate,
-    authorization: Optional[str] = Header(default=None),
-):
-    check_bearer(authorization)
-
+async def update_status(command_id: str, update: StatusUpdate):
     cmd = commands.get(command_id)
     if not cmd:
         raise HTTPException(status_code=404, detail="Command not found")
@@ -361,31 +154,7 @@ async def update_command_status(
     return {"ok": True}
 
 
-@app.post("/commands/{command_id}/cancel")
-async def cancel_command(command_id: str, authorization: Optional[str] = Header(default=None)):
-    check_bearer(authorization)
-
-    cmd = commands.get(command_id)
-    if not cmd:
-        raise HTTPException(status_code=404, detail="Command not found")
-
-    if cmd["status"] in ("done", "failed", "expired"):
-        return {"ok": True, "status": cmd["status"]}
-
-    cmd["status"] = "failed"
-    cmd["error"] = "Cancelled by operator"
-    cmd["updated_at"] = now_iso()
-
-    async with queue_lock:
-        if command_id in queue:
-            queue.remove(command_id)
-
-    return {"ok": True, "status": "failed"}
-
-
 # ============================================================
-# MOUNT MCP ASGI APP AT /mcp (streamable HTTP transport)
+# Mount MCP SSE app at /sse
 # ============================================================
-# FastMCP's streamable_http_app() returns an ASGI app implementing MCP over HTTP.
-# This is what Claude/ChatGPT-style MCP clients connect to.
-app.router.routes.append(Mount("/mcp", app=mcp.streamable_http_app()))
+app.mount("/sse", mcp.sse_app())