feat: add lifespan support to FastMCP server

Adds support for the lifespan API to FastMCP server, enabling:
- Simple setup with FastMCP constructor
- Type-safe context passing to tools and handlers
- Configuration via Settings class
This commit is contained in:
David Soria Parra
2025-02-11 12:15:08 +00:00
parent 2c7bd8343e
commit d3ea9009b0

View File

@@ -3,8 +3,13 @@
import inspect import inspect
import json import json
import re import re
from collections.abc import AsyncIterator
from contextlib import (
AbstractAsyncContextManager,
asynccontextmanager,
)
from itertools import chain from itertools import chain
from typing import Any, Callable, Literal, Sequence from typing import Any, Callable, Generic, Literal, Sequence
import anyio import anyio
import pydantic_core import pydantic_core
@@ -19,8 +24,16 @@ from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceMan
from mcp.server.fastmcp.tools import ToolManager from mcp.server.fastmcp.tools import ToolManager
from mcp.server.fastmcp.utilities.logging import configure_logging, get_logger from mcp.server.fastmcp.utilities.logging import configure_logging, get_logger
from mcp.server.fastmcp.utilities.types import Image from mcp.server.fastmcp.utilities.types import Image
from mcp.server.lowlevel import Server as MCPServer
from mcp.server.lowlevel.helper_types import ReadResourceContents from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import (
LifespanResultT,
)
from mcp.server.lowlevel.server import (
Server as MCPServer,
)
from mcp.server.lowlevel.server import (
lifespan as default_lifespan,
)
from mcp.server.sse import SseServerTransport from mcp.server.sse import SseServerTransport
from mcp.server.stdio import stdio_server from mcp.server.stdio import stdio_server
from mcp.shared.context import RequestContext from mcp.shared.context import RequestContext
@@ -50,7 +63,7 @@ from mcp.types import (
logger = get_logger(__name__) logger = get_logger(__name__)
class Settings(BaseSettings): class Settings(BaseSettings, Generic[LifespanResultT]):
"""FastMCP server settings. """FastMCP server settings.
All settings can be configured via environment variables with the prefix FASTMCP_. All settings can be configured via environment variables with the prefix FASTMCP_.
@@ -85,13 +98,36 @@ class Settings(BaseSettings):
description="List of dependencies to install in the server environment", description="List of dependencies to install in the server environment",
) )
lifespan: (
Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None
) = Field(None, description="Lifespan contexte manager")
def lifespan_wrapper(
app: "FastMCP",
lifespan: Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]],
) -> Callable[[MCPServer], AbstractAsyncContextManager[object]]:
@asynccontextmanager
async def wrap(s: MCPServer) -> AsyncIterator[object]:
async with lifespan(app) as context:
yield context
return wrap
class FastMCP: class FastMCP:
def __init__( def __init__(
self, name: str | None = None, instructions: str | None = None, **settings: Any self, name: str | None = None, instructions: str | None = None, **settings: Any
): ):
self.settings = Settings(**settings) self.settings = Settings(**settings)
self._mcp_server = MCPServer(name=name or "FastMCP", instructions=instructions)
self._mcp_server = MCPServer(
name=name or "FastMCP",
instructions=instructions,
lifespan=lifespan_wrapper(self, self.settings.lifespan)
if self.settings.lifespan
else default_lifespan,
)
self._tool_manager = ToolManager( self._tool_manager = ToolManager(
warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
) )