Spaces:
Sleeping
Sleeping
File size: 3,051 Bytes
cf66711 ea832c3 | 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 | import asyncio
from uuid import uuid4
from datetime import datetime, timezone
from typing import Dict, Any, Optional, List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP, Context
commands: Dict[str, Dict[str, Any]] = {}
queue: List[str] = []
queue_lock = asyncio.Lock()
def now_iso():
return datetime.now(timezone.utc).isoformat()
async def enqueue(command: str, args: Optional[Dict[str, Any]] = None):
cid = str(uuid4())
rec = {
"id": cid,
"command": command,
"args": args or {},
"status": "queued",
"result": None,
"error": None,
"created_at": now_iso(),
"updated_at": now_iso(),
"claimed_by": None,
}
async with queue_lock:
commands[cid] = rec
queue.append(cid)
return rec
mcp = FastMCP(
"Create3 Robot Bridge",
instructions="Queues robot commands for a local Create 3 runner.",
stateless_http=True,
)
@mcp.tool()
async def dock(ctx: Context) -> dict:
return {"ok": True, "queued": await enqueue("dock", {})}
@mcp.tool()
async def undock(ctx: Context) -> dict:
return {"ok": True, "queued": await enqueue("undock", {})}
@mcp.tool()
async def move_cm(cm: float, ctx: Context) -> dict:
return {"ok": True, "queued": await enqueue("move_cm", {"cm": cm})}
app = FastAPI()
class StatusUpdate(BaseModel):
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": "/mcp", "health": "/health"}
@app.get("/health")
async def health():
return {"ok": True, "time": now_iso()}
@app.get("/commands")
async def list_commands():
return {"ok": True, "items": list(commands.values())}
@app.post("/commands/next")
async def claim_next_command(robot_id: str):
async with queue_lock:
while queue:
cid = queue.pop(0)
cmd = commands.get(cid)
if not cmd or cmd["status"] != "queued":
continue
cmd["status"] = "running"
cmd["claimed_by"] = robot_id
cmd["updated_at"] = now_iso()
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)
if not cmd:
raise HTTPException(status_code=404, detail="Command not found")
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}
# Disable slash redirects to avoid bad proxy redirects (http vs https)
app.router.redirect_slashes = False
# Mount both MCP transports
app.mount("/mcp", mcp.streamable_http_app()) # for streamable HTTP clients
app.mount("/sse", mcp.sse_app()) # for ChatGPT connector / SSE clients |