-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Open
Labels
duplicateDuplicates an existing open issue. Reference the original issue when applying.Duplicates an existing open issue. Reference the original issue when applying.enhancementImprovement to existing functionality. For issues and smaller PR improvements.Improvement to existing functionality. For issues and smaller PR improvements.serverRelated to FastMCP server implementation or server-side functionality.Related to FastMCP server implementation or server-side functionality.
Description
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: shutdownReactions are currently unavailable
Metadata
Metadata
Assignees
Labels
duplicateDuplicates an existing open issue. Reference the original issue when applying.Duplicates an existing open issue. Reference the original issue when applying.enhancementImprovement to existing functionality. For issues and smaller PR improvements.Improvement to existing functionality. For issues and smaller PR improvements.serverRelated to FastMCP server implementation or server-side functionality.Related to FastMCP server implementation or server-side functionality.