main.py•5.34 kB
import argparse
import os
import warnings
from typing import Any
import uvicorn
from mcp.server import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.server.fastmcp.server import StreamableHTTPASGIApp
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse
from starlette.middleware.cors import CORSMiddleware
from starlette.staticfiles import StaticFiles
from starlette.routing import Route, Mount
# Import tools from separate module
from src.tools.handler import handle_tool_call
from src.tools import TOOLS
from src.middlewares.auth import AuthenticationMiddleware
# Suppress websockets deprecation warnings
warnings.filterwarnings("ignore", category=DeprecationWarning, module="websockets")
warnings.filterwarnings("ignore", category=DeprecationWarning, module="uvicorn.protocols.websockets")
# Create low-level MCP server
server = Server("MCP Server Template")
# Register tool handlers
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""List available tools."""
return TOOLS
@server.call_tool()
async def handle_call_tool_wrapper(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls by delegating to tools module."""
return await handle_tool_call(name, arguments)
# Custom route handlers
async def root(request):
return RedirectResponse(url="/docs")
async def health_check(request):
return JSONResponse({"status": "ok"})
async def mcp_info(request):
tools_list = await handle_list_tools()
return JSONResponse(
{
"status": "running",
"protocol": "mcp",
"server_name": "MCP Server Template",
"description": "Template for building custom Model Context Protocol servers",
"mcp_endpoint": "/mcp",
"tools_available": len(tools_list),
"note": "This is an MCP server template. Customize the tools in src/tools.py",
}
)
async def list_tools_endpoint(request):
tools_list = await handle_list_tools()
tools = []
for tool in tools_list:
tools.append(
{
"name": tool.name,
"description": tool.description or "No description available",
"parameters": tool.inputSchema or {},
}
)
return JSONResponse({"tools": tools})
async def docs(request):
from pathlib import Path
tools_list = await handle_list_tools()
# Generate tools HTML
tools_html = ""
for tool in tools_list:
description = tool.description or "No description available"
tools_html += f"""
<div class="bg-slate-800/50 rounded-lg p-5 border border-slate-700 hover:border-indigo-500 transition-colors">
<h3 class="font-semibold text-base mb-2 text-indigo-300">{tool.name}</h3>
<p class="text-gray-400 text-sm leading-relaxed">{description}</p>
</div>
"""
# Read template and replace placeholders
template_path = Path(__file__).parent / "src" / "templates" / "docs.html"
with open(template_path, "r") as f:
template = f.read()
html_content = template.replace("{{ tools_count }}", str(len(tools_list)))
html_content = html_content.replace("{{ tools_html }}", tools_html)
return HTMLResponse(content=html_content)
if __name__ == "__main__":
from contextlib import asynccontextmanager
parser = argparse.ArgumentParser(description="MCP Server Template")
args = parser.parse_args()
port = int(os.environ.get("PORT", 8000))
from pathlib import Path
static_dir = Path(__file__).parent / "src" / "public"
# Create session manager for streamable HTTP transport
session_manager = StreamableHTTPSessionManager(app=server)
# Create the ASGI handler
streamable_http_app = StreamableHTTPASGIApp(session_manager)
# Create lifespan context manager to run session manager
@asynccontextmanager
async def lifespan(app):
async with session_manager.run():
yield
# Create Starlette app with custom routes
app = Starlette(
routes=[
Route("/", root),
Route("/health", health_check),
Route("/info", mcp_info),
Route("/tools", list_tools_endpoint),
Route("/docs", docs),
Mount("/static", StaticFiles(directory=str(static_dir)), name="static"),
Route("/mcp", streamable_http_app), # MCP endpoint at /mcp
],
lifespan=lifespan
)
# Add middleware in the correct order (they execute in reverse order)
# First add authentication middleware
app.add_middleware(
AuthenticationMiddleware,
protected_routes=["/mcp", "/tools"]
)
# Then add CORS middleware (will execute first, before auth)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["GET", "POST", "OPTIONS", "DELETE"],
allow_headers=["*"],
expose_headers=["Mcp-Session-Id"],
max_age=86400,
)
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")