"""
Zintlr MCP Server - FastAPI Application
Remote MCP server for LLM integration with Zintlr sales intelligence platform.
Endpoints:
- POST / : MCP JSON-RPC endpoint
- GET /sse : Server-Sent Events for streaming (optional)
- OAuth endpoints for authentication passthrough
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import logging
from app.config import settings
from app.oauth import router as oauth_router
from app.mcp_handler import handle_mcp_request
from app.session import session_manager
# Configure logging
logging.basicConfig(
level=logging.DEBUG if settings.debug else logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# ============ Application Lifecycle ============
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application startup and shutdown events."""
# Startup
logger.info(f"Starting Zintlr MCP Server v{settings.mcp_server_version}")
logger.info(f"API Base URL: {settings.zintlr_api_base_url}")
logger.info(f"MCP Server URL: {settings.mcp_server_url}")
yield
# Shutdown
logger.info("Shutting down Zintlr MCP Server")
await session_manager.close()
# ============ FastAPI Application ============
app = FastAPI(
title="Zintlr MCP Server",
description="Remote MCP server for LLM integration with Zintlr",
version=settings.mcp_server_version,
lifespan=lifespan,
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include OAuth routes
app.include_router(oauth_router)
# ============ MCP Endpoints ============
@app.post("/")
async def mcp_endpoint(request: Request):
"""
Main MCP JSON-RPC endpoint.
LLM clients send JSON-RPC 2.0 messages here for:
- initialize: Protocol handshake
- tools/list: Get available tools
- tools/call: Execute a tool
- notifications/initialized: Client ready notification
"""
# Parse request body
try:
body = await request.json()
except Exception as e:
return JSONResponse(
status_code=400,
content={
"jsonrpc": "2.0",
"id": None,
"error": {
"code": -32700,
"message": f"Parse error: {str(e)}"
}
}
)
# Validate JSON-RPC structure
if not isinstance(body, dict) or body.get("jsonrpc") != "2.0":
return JSONResponse(
status_code=400,
content={
"jsonrpc": "2.0",
"id": body.get("id") if isinstance(body, dict) else None,
"error": {
"code": -32600,
"message": "Invalid Request: Not a valid JSON-RPC 2.0 request"
}
}
)
# Extract session ID from Authorization header
session_id = None
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
session_id = auth_header[7:]
# Handle the request
return await handle_mcp_request(request, body, session_id)
@app.get("/")
async def mcp_info():
"""
MCP Server information endpoint.
Returns basic info about the server for discovery.
"""
return {
"name": settings.mcp_server_name,
"version": settings.mcp_server_version,
"protocolVersion": settings.mcp_protocol_version,
"description": "Zintlr MCP Server - Sales intelligence for LLMs",
}
# ============ Health Check ============
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "version": settings.mcp_server_version}
# ============ Error Handlers ============
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""Handle HTTP exceptions."""
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Handle unexpected exceptions."""
logger.exception(f"Unexpected error: {exc}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error"}
)
# ============ Run Server ============
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host=settings.host,
port=settings.port,
reload=settings.debug,
)
