"""MCP Server implementation."""
import logging
from functools import cached_property
from typing import Callable, Optional
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from .base import BaseServer, default_health_check
from ..settings import ServerSettings
class MCPServer(BaseServer):
"""MCP Server implementation."""
def __init__(
self,
settings: ServerSettings,
readiness_check: Optional[Callable[[], bool]] = None,
liveness_check: Optional[Callable[[], bool]] = None,
):
"""Initialize MCP server."""
super().__init__(settings, readiness_check, liveness_check)
self.logger = logging.getLogger(__name__)
@cached_property
def mcp(self) -> FastMCP:
"""FastMCP instance."""
return FastMCP(base_url=self.settings.MCP_BASE, stateless_http=True, json_response=True)
@cached_property
def rest(self) -> Starlette:
"""Creates rest server and adds all the routes to the rest server."""
app = self.mcp.streamable_http_app()
self.add_health_check(app)
self.add_middleware(app)
return app
