-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
63 lines (50 loc) · 1.72 KB
/
Copy pathmain.py
File metadata and controls
63 lines (50 loc) · 1.72 KB
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
import contextlib
import os
import uvicorn
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route
from src.tools import register_tools
def create_mcp_app():
"""Create and configure the MCP server"""
mcp = FastMCP("GitHubManager", stateless_http=True, json_response=True)
register_tools(mcp)
return mcp
app = create_mcp_app()
async def health_check(request):
"""Render health probe endpoint."""
return JSONResponse({"status": "ok"})
class McpEndpoint:
"""Handle the bare /mcp path without a mount redirect."""
async def __call__(self, scope, receive, send):
await app.session_manager.handle_request(scope, receive, send)
@contextlib.asynccontextmanager
async def lifespan(starlette_app: Starlette):
async with app.session_manager.run():
yield
def get_cors_origins() -> list[str]:
origins = os.environ.get("CORS_ALLOWED_ORIGINS", "*")
return [origin.strip() for origin in origins.split(",") if origin.strip()]
starlette_app = Starlette(
routes=[
Route("/health", health_check),
Route("/mcp", McpEndpoint(), methods=["GET", "POST", "DELETE"]),
Mount("/", app=app.streamable_http_app()),
],
lifespan=lifespan,
)
http_app = CORSMiddleware(
starlette_app,
allow_origins=get_cors_origins(),
allow_methods=["GET", "POST", "DELETE"],
allow_headers=["*"],
expose_headers=["Mcp-Session-Id"],
)
def main():
"""Main entry point"""
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(http_app, host="0.0.0.0", port=port)
if __name__ == "__main__":
main()