app.pyā¢2.25 kB
"""
Production entrypoint for Finizi B4B MCP Server on Cloud Run.
Exposes the MCP server via Streamable HTTP transport.
"""
import os
import sys
from contextlib import asynccontextmanager
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import JSONResponse
import structlog
# Configure structured logging
structlog.configure(
processors=[
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer()
],
logger_factory=structlog.PrintLoggerFactory(file=sys.stderr),
)
logger = structlog.get_logger(__name__)
# Import the MCP server instance
from src.finizi_b4b_mcp.server import mcp
# Health check endpoint for Cloud Run
async def health_check(request):
"""Health check endpoint for Cloud Run."""
return JSONResponse({
"status": "healthy",
"service": "finizi-b4b-mcp",
"version": "1.0.0"
})
# Root endpoint with service info
async def root(request):
"""Root endpoint with service information."""
return JSONResponse({
"name": "Finizi B4B MCP Server",
"version": "1.0.0",
"description": "MCP server for Finizi B4B API integration",
"endpoints": {
"mcp": "/mcp",
"health": "/health"
},
"transport": "streamable-http"
})
# Lifespan context manager to initialize FastMCP session manager
@asynccontextmanager
async def lifespan(app: Starlette):
"""Manage FastMCP session manager lifecycle."""
async with mcp.session_manager.run():
yield
# Create the ASGI application
# Configure for SSE transport
mcp.settings.streamable_http_path = "/"
app = Starlette(
routes=[
Route("/", root),
Route("/health", health_check),
Mount("/mcp", app=mcp.streamable_http_app()),
],
lifespan=lifespan,
)
# Log startup
logger.info(
"Finizi B4B MCP Server starting",
port=os.getenv("PORT", "8080"),
transport="streamable-http"
)
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", "8080"))
uvicorn.run(
"app:app",
host="0.0.0.0",
port=port,
log_level="info"
)