Skip to content

Make the starlet lifespan state publically accessible #3180

@jonha892

Description

@jonha892

Enhancement

Starlet has a concept of lifespan state where we can provide context values for the whole lifespan of a server, like singleton instances (e.g. a s3 client we don't want to create for each request)
https://starlette.dev/lifespan/#lifespan-state

Right now they are processed by the FastMCP server. However, they are only accessible via the _lifespan attribute.
If would be nice to have a public api in the fastmcp instance or in the context itself.

Would be even better if the lifespan access could somehow be typed, but afaik this is something the starlet devs also struggle with due to lacking support in the asynccontextmanager.

Example:

import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import TypedDict

from fastmcp import Client, FastMCP
from fastmcp.server.context import Context


class AppState(TypedDict):
    message: str
    counter: int


@asynccontextmanager
async def lifespan(app: FastMCP[AppState]) -> AsyncIterator[AppState]:
    print("Lifespan: starting", flush=True)
    try:
        yield {"message": "Hello from lifespan!", "counter": 42}
    finally:
        print("Lifespan: shutdown", flush=True)


mcp = FastMCP[AppState]("test-server", lifespan=lifespan)


@mcp.tool
async def get_message(ctx: Context) -> str:
    """Get the message from lifespan state."""
    # PROBLEM: _lifespan_result is private, no public API
    state = ctx.fastmcp._lifespan_result
    if state is None:
        raise RuntimeError("Lifespan state not available")
    return f"Message: {state['message']}, Counter: {state['counter']}"


async def main():
    async with Client(mcp) as client:
        result = await client.call_tool("get_message", {})
        print(f"Result: {result.content[0].text}")

        # Verify the state was accessible
        assert "Hello from lifespan!" in result.content[0].text
        assert "42" in result.content[0].text
        print("SUCCESS: Lifespan state is accessible via _lifespan_result")


if __name__ == "__main__":
    asyncio.run(main())

# Lifespan: starting
# Result: Message: Hello from lifespan!, Counter: 42
# SUCCESS: Lifespan state is accessible via _lifespan_result
# Lifespan: shutdown

Metadata

Metadata

Assignees

No one assigned

    Labels

    duplicateDuplicates an existing open issue. Reference the original issue when applying.enhancementImprovement to existing functionality. For issues and smaller PR improvements.serverRelated to FastMCP server implementation or server-side functionality.

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions