content: """
  Run from the repository root:
      uv run examples/snippets/servers/lowlevel/lifespan.py
  """
  
  from collections.abc import AsyncIterator
  from contextlib import asynccontextmanager
  from typing import Any
  
  import mcp.server.stdio
  import mcp.types as types
  from mcp.server.lowlevel import NotificationOptions, Server
  from mcp.server.models import InitializationOptions
  
  
  # Mock database class for example
  class Database:
      """Mock database class for example."""
  
      @classmethod
      async def connect(cls) -> "Database":
          """Connect to database."""
          print("Database connected")
          return cls()
  
      async def disconnect(self) -> None:
          """Disconnect from database."""
          print("Database disconnected")
  
      async def query(self, query_str: str) -> list[dict[str, str]]:
          """Execute a query."""
          # Simulate database query
          return [{"id": "1", "name": "Example", "query": query_str}]
  
  
  @asynccontextmanager
  async def server_lifespan(_server: Server) -> AsyncIterator[dict[str, Any]]:
      """Manage server startup and shutdown lifecycle."""
      # Initialize resources on startup
      db = await Database.connect()
      try:
          yield {"db": db}
      finally:
          # Clean up on shutdown
          await db.disconnect()
  
  
  # Pass lifespan to server
  server = Server("example-server", lifespan=server_lifespan)
  
  
  @server.list_tools()
  async def handle_list_tools() -> list[types.Tool]:
      """List available tools."""
      return [
          types.Tool(
              name="query_db",
              description="Query the database",
              inputSchema={
                  "type": "object",
                  "properties": {"query": {"type": "string", "description": "SQL query to execute"}},
                  "required": ["query"],
              },
          )
      ]
  
  
  @server.call_tool()
  async def query_db(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
      """Handle database query tool call."""
      if name != "query_db":
          raise ValueError(f"Unknown tool: {name}")
  
      # Access lifespan context
      ctx = server.request_context
      db = ctx.lifespan_context["db"]
  
      # Execute query
      results = await db.query(arguments["query"])
  
      return [types.TextContent(type="text", text=f"Query results: {results}")]
  
  
  async def run():
      """Run the server with lifespan management."""
      async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
          await server.run(
              read_stream,
              write_stream,
              InitializationOptions(
                  server_name="example-server",
                  server_version="0.1.0",
                  capabilities=server.get_capabilities(
                      notification_options=NotificationOptions(),
                      experimental_capabilities={},
                  ),
              ),
          )
  
  
  if __name__ == "__main__":
      import asyncio
  
      asyncio.run(run())
  