main.py•2.7 kB
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import APIKeyHeader
from fastapi.staticfiles import StaticFiles
from mcp_sandbox.core.mcp_tools import SandboxToolsPlugin
from mcp_sandbox.api.routes import configure_app
from mcp_sandbox.api.auth_routes import router as auth_router
from mcp_sandbox.db.database import db
from mcp_sandbox.middleware.auth_middleware import AuthMiddleware
from mcp_sandbox.utils.config import logger, HOST, PORT, REQUIRE_AUTH
# API Key validation for SSE connections
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
async def get_api_key(request: Request = None, api_key_header: str = Depends(API_KEY_HEADER)):
"""Get API key from header or query parameters"""
# First try to get from header
if api_key_header:
return api_key_header
# Then try to get from query params if request is provided
if request and "api_key" in request.query_params:
return request.query_params.get("api_key")
# No API key found
return None
async def get_api_key_user(request: Request = None, api_key: str = Depends(get_api_key)):
"""Validate API key and get associated user"""
if api_key is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API Key is required",
)
# Find user with provided API key
for user in db.get_all_users():
if user.get("api_key") == api_key:
return user
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key",
)
def main():
"""Main entry point for the application"""
# Create FastAPI app
app = FastAPI(title="MCP Sandbox")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust in production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Add authentication middleware
app.add_middleware(AuthMiddleware)
# Include authentication routes
app.include_router(auth_router)
# Initialize sandbox tools
sandbox_plugin = SandboxToolsPlugin()
# Access the MCP server directly for configure_app
# We pass the plugin itself so we can access its user context methods
configure_app(app, sandbox_plugin)
# Start FastAPI server
auth_status = "enabled" if REQUIRE_AUTH else "disabled"
logger.info(f"Starting MCP Sandbox with authentication {auth_status}")
uvicorn.run(app, host=HOST, port=PORT)
if __name__ == "__main__":
main()