Update app.py

app.py

@@ -4,32 +4,27 @@ from datetime import datetime, timezone
 from typing import Dict, Any, Optional, List
 
 from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel, Field
-from starlette.routing import Mount
-
+from pydantic import BaseModel
 from mcp.server.fastmcp import FastMCP, Context
 
-
-# ============================================================
+# ---------------------------
 # In-memory queue
-# ============================================================
+# ---------------------------
 commands: Dict[str, Dict[str, Any]] = {}
 queue: List[str] = []
 queue_lock = asyncio.Lock()
 
-
 def now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
-
 async def enqueue(command: str, args: Optional[Dict[str, Any]] = None, source: str = "mcp") -> Dict[str, Any]:
-    cmd_id = str(uuid4())
+    cid = str(uuid4())
     rec = {
-        "id": cmd_id,
+        "id": cid,
         "command": command,
         "args": args or {},
         "source": source,
-        "status": "queued",  # queued | running | done | failed
+        "status": "queued",
         "result": None,
         "error": None,
         "created_at": now_iso(),
@@ -37,64 +32,48 @@ async def enqueue(command: str, args: Optional[Dict[str, Any]] = None, source: s
         "claimed_by": None,
     }
     async with queue_lock:
-        commands[cmd_id] = rec
-        queue.append(cmd_id)
+        commands[cid] = rec
+        queue.append(cid)
     return rec
 
-
-# ============================================================
-# MCP server (Streamable HTTP) — better for ChatGPT connector
-# ============================================================
+# ---------------------------
+# MCP server (3 tools only)
+# ---------------------------
 mcp = FastMCP(
     "Create3 Robot Bridge",
-    instructions="Queues robot commands for a local Create 3 runner.",
-    stateless_http=True,   # good for hosted environments
-    json_response=False,   # keep MCP transport behavior, not plain JSON API
+    instructions="Queues commands for a local Create 3 runner.",
+    stateless_http=True,
 )
 
-# Exactly 3 tools
 @mcp.tool()
 async def dock(ctx: Context) -> dict:
-    """Queue a dock command for Create 3."""
-    rec = await enqueue("dock", {}, source="mcp")
+    rec = await enqueue("dock", {}, "mcp")
     return {"ok": True, "queued": rec}
 
 @mcp.tool()
 async def undock(ctx: Context) -> dict:
-    """Queue an undock command for Create 3."""
-    rec = await enqueue("undock", {}, source="mcp")
+    rec = await enqueue("undock", {}, "mcp")
     return {"ok": True, "queued": rec}
 
 @mcp.tool()
 async def move_cm(cm: float, ctx: Context) -> dict:
-    """Queue a move command in centimeters (negative moves backward)."""
-    rec = await enqueue("move_cm", {"cm": cm}, source="mcp")
+    rec = await enqueue("move_cm", {"cm": cm}, "mcp")
     return {"ok": True, "queued": rec}
 
-
-# ============================================================
-# FastAPI app + queue endpoints for local Create 3 client
-# ============================================================
-app = FastAPI(title="Create3 MCP + Queue", version="1.0.0")
-
+# ---------------------------
+# FastAPI app
+# ---------------------------
+app = FastAPI(title="Create3 MCP Bridge")
 
 class StatusUpdate(BaseModel):
-    status: str = Field(..., description="running | done | failed")
+    status: str
     result: Optional[Dict[str, Any]] = None
     error: Optional[str] = None
     robot_id: Optional[str] = None
 
-
 @app.get("/")
 async def root():
-    return {
-        "ok": True,
-        "mcp_endpoint": "/mcp",
-        "health": "/health",
-        "note": "Use the ChatGPT MCP connector URL ending in /mcp",
-        "time": now_iso(),
-    }
-
+    return {"ok": True, "mcp": "/mcp", "health": "/health", "time": now_iso()}
 
 @app.get("/health")
 async def health():
@@ -106,19 +85,16 @@ async def health():
         "time": now_iso(),
     }
 
-
 @app.get("/commands")
-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}
-
+async def list_commands():
+    return {"ok": True, "items": sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)}
 
 @app.post("/commands/next")
 async def claim_next_command(robot_id: str):
     async with queue_lock:
         while queue:
-            cmd_id = queue.pop(0)
-            cmd = commands.get(cmd_id)
+            cid = queue.pop(0)
+            cmd = commands.get(cid)
             if not cmd or cmd["status"] != "queued":
                 continue
             cmd["status"] = "running"
@@ -127,7 +103,6 @@ async def claim_next_command(robot_id: str):
             return {"ok": True, "item": cmd}
     return {"ok": True, "item": None}
 
-
 @app.post("/commands/{command_id}/status")
 async def update_status(command_id: str, update: StatusUpdate):
     cmd = commands.get(command_id)
@@ -141,6 +116,5 @@ async def update_status(command_id: str, update: StatusUpdate):
     cmd["updated_at"] = now_iso()
     return {"ok": True}
 
-
-# Mount MCP at /mcp (Streamable HTTP transport)
-app.router.routes.append(Mount("/mcp", app=mcp.streamable_http_app()))
+# Mount MCP at /mcp
+app.mount("/mcp", mcp.streamable_http_app())