Spaces:
Sleeping
Sleeping
File size: 1,480 Bytes
fc6208e |
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 |
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict
app = FastAPI()
class MCPMessage(BaseModel):
session: str
id: str
command: str
args: Dict[str, str]
@app.post("/mcp")
async def mcp_endpoint(message: MCPMessage, request: Request):
print(f"[MCP Server] Received: {message.command} | Args: {message.args}")
if message.command == "mcp-negotiate":
return {
"id": message.id,
"status": "ok",
"message": "Negotiation successful",
"session": message.session,
"version": message.args.get("version", "1.0"),
}
elif message.command == "mcp-message":
return {
"id": message.id,
"status": "ok",
"message": f"Delivered to {message.args.get('to')}",
"echo": message.args.get("body", ""),
}
elif message.command == "mcp-echo-upper":
original = message.args.get("body", "")
return {
"id": message.id,
"status": "ok",
"message": "Uppercase echo",
"original": original,
"upper": original.upper()
}
return JSONResponse(status_code=400, content={
"id": message.id,
"status": "error",
"message": f"Unknown command: {message.command}"
})
@app.get("/")
def root():
return {"message": "MCP Test Server is running"}
|