Update app.py

app.py

@@ -1,16 +1,42 @@
-from fastapi import FastAPI, HTTPException, Header
-from pydantic import BaseModel, Field
-from typing import Optional, Dict, Any, List
+import os
+import asyncio
 from uuid import uuid4
 from datetime import datetime, timezone
-import asyncio
+from typing import Optional, Dict, Any, List
+
+from fastapi import FastAPI, HTTPException, Header, Request
+from fastapi.responses import JSONResponse
+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.)
+# ============================================================
+commands: Dict[str, Dict[str, Any]] = {}
+queue: List[str] = []
+queue_lock = asyncio.Lock()
 
-app = FastAPI(title="Create3 Robot Command Bridge", version="1.0.0")
 
-# ====== Simple shared secret auth (optional but recommended) ======
-API_TOKEN = "change-me-please"  # set this to your own token
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
 
-def check_token(auth_header: Optional[str]):
+
+def check_bearer(auth_header: Optional[str]):
     if not API_TOKEN:
         return
     if not auth_header or not auth_header.startswith("Bearer "):
@@ -19,30 +45,6 @@ def check_token(auth_header: Optional[str]):
     if token != API_TOKEN:
         raise HTTPException(status_code=403, detail="Invalid token")
 
-# ====== In-memory command store ======
-# For HF Space demo/testing this is OK.
-# If you restart the app, queue clears.
-commands: Dict[str, Dict[str, Any]] = {}
-queue: List[str] = []
-queue_lock = asyncio.Lock()
-
-# ====== Models ======
-class CommandCreate(BaseModel):
-    command: str = Field(..., description="Robot command name")
-    args: Dict[str, Any] = Field(default_factory=dict, description="Command arguments")
-    source: Optional[str] = Field(default="ai", description="Who sent the command")
-    priority: int = Field(default=0, description="Higher = earlier in queue")
-    ttl_seconds: Optional[int] = Field(default=120, description="Optional expiration")
-
-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
-
-# ====== Helpers ======
-def now_iso() -> str:
-    return datetime.now(timezone.utc).isoformat()
 
 def command_expired(cmd: Dict[str, Any]) -> bool:
     ttl = cmd.get("ttl_seconds")
@@ -52,74 +54,270 @@ def command_expired(cmd: Dict[str, Any]) -> bool:
     age = (datetime.now(timezone.utc) - created).total_seconds()
     return age > ttl
 
-# ====== API ======
-
-@app.get("/")
-async def root():
-    return {"ok": True, "service": "create3-command-bridge"}
-
-@app.post("/commands")
-async def create_command(cmd: CommandCreate, authorization: Optional[str] = Header(default=None)):
-    check_token(authorization)
 
+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]:
     cmd_id = str(uuid4())
     record = {
         "id": cmd_id,
-        "command": cmd.command,
-        "args": cmd.args,
-        "source": cmd.source,
-        "priority": cmd.priority,
-        "ttl_seconds": cmd.ttl_seconds,
+        "command": command,
+        "args": args or {},
+        "source": source,
+        "priority": priority,
+        "ttl_seconds": ttl_seconds,
         "status": "queued",
         "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
 
-        # Insert by priority (higher first)
+        # priority insert (higher first)
         inserted = False
         for i, queued_id in enumerate(queue):
-            if record["priority"] > commands[queued_id]["priority"]:
+            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)
 
-    return {"ok": True, "command_id": cmd_id, "status": "queued"}
+    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 = FastMCP(
+    "Create3 Robot Bridge",
+    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."
+    ),
+    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}
+
+
+@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)
+    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,
+    )
+    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)
+    return {"ok": True, "queued": rec}
+
+
+# ============================================================
+# FASTAPI APP (REST endpoints for robot client + MCP mount)
+# ============================================================
+app = FastAPI(title="Create3 MCP + Queue Bridge", version="2.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)
+
+
+@app.get("/")
+async def root():
+    return {
+        "ok": True,
+        "service": "Create3 MCP + Queue Bridge",
+        "mcp_endpoint": "/mcp",
+        "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}
+
 
 @app.get("/commands")
 async def list_commands(limit: int = 20, authorization: Optional[str] = Header(default=None)):
-    check_token(authorization)
-    # latest updated first
-    items = sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)[:limit]
+    check_bearer(authorization)
+    items = sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)[: max(1, min(limit, 100))]
     return {"ok": True, "items": items}
 
+
 @app.get("/commands/{command_id}")
-async def get_command(command_id: str, authorization: Optional[str] = Header(default=None)):
-    check_token(authorization)
+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),
-):
+async def claim_next_command(robot_id: str, authorization: Optional[str] = Header(default=None)):
     """
-    Client polls this endpoint to claim the next command.
-    Returns one command and marks it running.
+    Robot client polls this endpoint to claim one command.
     """
-    check_token(authorization)
+    check_bearer(authorization)
 
     async with queue_lock:
-        # Clean expired queued commands while scanning
         while queue:
             cmd_id = queue.pop(0)
             cmd = commands.get(cmd_id)
@@ -141,58 +339,53 @@ async def claim_next_command(
 
     return {"ok": True, "item": None}
 
+
 @app.post("/commands/{command_id}/status")
 async def update_command_status(
     command_id: str,
     update: CommandStatusUpdate,
     authorization: Optional[str] = Header(default=None),
 ):
-    check_token(authorization)
+    check_bearer(authorization)
 
     cmd = commands.get(command_id)
     if not cmd:
         raise HTTPException(status_code=404, detail="Command not found")
 
-    # Basic state update
     cmd["status"] = update.status
     cmd["result"] = update.result
     cmd["error"] = update.error
     if update.robot_id:
         cmd["claimed_by"] = update.robot_id
     cmd["updated_at"] = now_iso()
-
     return {"ok": True}
 
+
 @app.post("/commands/{command_id}/cancel")
 async def cancel_command(command_id: str, authorization: Optional[str] = Header(default=None)):
-    check_token(authorization)
+    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"], "message": "Already finished"}
+        return {"ok": True, "status": cmd["status"]}
 
     cmd["status"] = "failed"
     cmd["error"] = "Cancelled by operator"
     cmd["updated_at"] = now_iso()
 
-    # Remove from queue if still queued
     async with queue_lock:
         if command_id in queue:
             queue.remove(command_id)
 
     return {"ok": True, "status": "failed"}
 
-@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(),
-    }
+
+# ============================================================
+# MOUNT MCP ASGI APP AT /mcp (streamable HTTP transport)
+# ============================================================
+# 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()))