How to work with lifespan?

[MicaelJarniac]

The FastMCP 1.0 implementation in modelcontextprotocol/python-sdk has the following:

"""Example showing lifespan support for startup/shutdown with strong typing."""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession


# Mock database class for example
class Database:
    """Mock database class for example."""

    @classmethod
    async def connect(cls) -> "Database":
        """Connect to database."""
        return cls()

    async def disconnect(self) -> None:
        """Disconnect from database."""
        pass

    def query(self) -> str:
        """Execute a query."""
        return "Query result"


@dataclass
class AppContext:
    """Application context with typed dependencies."""

    db: Database


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context."""
    # Initialize on startup
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        # Cleanup on shutdown
        await db.disconnect()


# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)


# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context[ServerSession, AppContext]) -> str:
    """Tool that uses initialized resources."""
    db = ctx.request_context.lifespan_context.db
    return db.query()

https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#server

I wanted to know how to do the same on this FastMCP 2.0.

Especially, how to get it correctly type-hinted and everything like that.

I couldn't find it in the docs, and from testing locally, the type-hints aren't working.

Is there a better approach for initializing a resource, like in the example above?

[mnbf9rca]

mcp.http_app() returns a StarletteWithLifespan that already has a lifespan managing its StreamableHTTPSessionManager. so we append to it:

  from contextlib import asynccontextmanager
  import asyncio

  def wrap_lifespan_with_background_task(original_lifespan):
      """
      Wraps FastMCP's existing lifespan to add custom background tasks.
      Preserves the session manager lifespan while adding custom logic.
      """
      @asynccontextmanager
      async def combined_lifespan(app):
          # Run FastMCP's original lifespan (manages session manager)
          async with original_lifespan(app):
              # Startup: Start your background task
              app.state.my_background_task = asyncio.create_task(
                  my_background_function()
              )
              logger.info("Background task started")

              yield  # App runs here

              # Shutdown: Stop background task gracefully
              if bg_task := getattr(app.state, "my_background_task", None):
                  bg_task.cancel()
                  try:
                      await asyncio.wait_for(bg_task, timeout=5.0)
                  except asyncio.CancelledError:
                      pass  # Expected
                  except asyncio.TimeoutError:
                      logger.warning("Task cancellation timed out")
                  except Exception as exc:
                      logger.exception("Error cancelling task: %s", exc)
                  logger.info("Background task stopped")

      return combined_lifespan

  def main():
      # Create FastMCP app
      app = mcp.http_app(path="/mcp")

      # Wrap the existing lifespan
      original_lifespan = app.router.lifespan_context
      if original_lifespan:
          app.router.lifespan_context = wrap_lifespan_with_background_task(
              original_lifespan
          )
      else:
          logger.warning("No FastMCP lifespan found!")

      uvicorn.run(app, host="0.0.0.0", port=8000)