fastmcp icon indicating copy to clipboard operation
fastmcp copied to clipboard

Functions of Context class

Open s0214lx opened this issue 7 months ago • 8 comments

Enhancement Description

I want to share an object between different tools without affecting multiple users. Can I use the Context class to achieve this?

Use Case

No response

Proposed Implementation


s0214lx avatar Apr 29 '25 06:04 s0214lx

Can I use ctx.request_context.lifespan_context to store data that needs to be shared while the current user is using the tool?

s0214lx avatar Apr 29 '25 06:04 s0214lx

I think this is slightly related to #166 where the default MCP behavior is to create a new lifespan for every session and a session is often defined as "per API call".

One approach would be to use an external caching object to store objects and some sort of user ID to access it appropriately when tools are called. I'm not sure if the low-level SDK has fully exposed headers yet, but you could use a header to identify the user and hydrate their context potentially?

This could be an interesting thing to add to FastMCP...

from fastmcp import FastMCP, Context

mcp = FastMCP()

@mcp.tool()
def my_tool(name: str, context: Context) -> str:
    context.set(key=context.headers['user_id'], value=name)
    return f"Hello, {name}!"

@mcp.tool()
def greet(context: Context) -> str:
    return f"Hello, {context.get(key=context.headers['user_id'])}!"

Please note this is just a sketch, really need to think through the implications of storing unbounded state etc.

jlowin avatar Apr 30 '25 17:04 jlowin

I believe actually the session/lifespan persists for the length of the client's connection and thus the context object that gets passed around exists for the life of that client session.

from contextlib import asynccontextmanager

from fastmcp import Context, FastMCP
from pydantic import BaseModel


class SessionLifespan(BaseModel):
    """A simple model to represent session lifespan."""
    current_user: str | None = None

@asynccontextmanager
async def per_session_lifespan(app):
    """Lifespan context manager for the MCP server."""
    yield SessionLifespan()

mcp = FastMCP(
    name="Simple MCP Server",
    lifespan=per_session_lifespan
)

@mcp.tool()
def set_username(context: Context, username: str):
    """Set the current user in the session."""
    context.session.current_user = username

@mcp.tool()
def get_username(context: Context) -> str | None:
    """Get the current user from the session."""
    return context.session.current_user


if __name__ == "__main__":
    mcp.run(transport="sse", port=8001)

If you run two MCP clients you'll be able to run set username on both clients with different values and each client will get back its own username when calling get_username

strawgate avatar May 03 '25 00:05 strawgate

Good point @strawgate -- I've been trying to understand whether this is still the case with the new streamable http transport, which I think is designed to be stateless (which is why I think it's preferred over long-lived SSE)

jlowin avatar May 03 '25 14:05 jlowin

@jlowin @strawgate Thanks for your help, this is a very effective solution.

s0214lx avatar May 14 '25 05:05 s0214lx

I believe actually the session/lifespan persists for the length of the client's connection and thus the context object that gets passed around exists for the life of that client session.

from contextlib import asynccontextmanager

from fastmcp import Context, FastMCP
from pydantic import BaseModel


class SessionLifespan(BaseModel):
    """A simple model to represent session lifespan."""
    current_user: str | None = None

@asynccontextmanager
async def per_session_lifespan(app):
    """Lifespan context manager for the MCP server."""
    yield SessionLifespan()

mcp = FastMCP(
    name="Simple MCP Server",
    lifespan=per_session_lifespan
)

@mcp.tool()
def set_username(context: Context, username: str):
    """Set the current user in the session."""
    context.session.current_user = username

@mcp.tool()
def get_username(context: Context) -> str | None:
    """Get the current user from the session."""
    return context.session.current_user


if __name__ == "__main__":
    mcp.run(transport="sse", port=8001)

If you run two MCP clients you'll be able to run set username on both clients with different values and each client will get back its own username when calling get_username

How do I initialize the properties in the SessionLifespan class? I tried the following, but it doesn't seem to work:

class SessionLifespan(BaseModel):
    current_sa: object | None = None
    minio_client:object = None 

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.minio_client = Minio(
            os.getenv('minio_endpoint'),
            access_key=os.getenv('minio_access_key'),
            secret_key=os.getenv('minio_secret_key'),
            secure=False
        )

s0214lx avatar May 20 '25 02:05 s0214lx

At least for me It is working with ctx.request_context.lifespan_context instead of ctx.session as mentioned by @strawgate (fastmcp=2.3.4)

double-thinker avatar May 20 '25 07:05 double-thinker

At least for me It is working with ctx.request_context.lifespan_context instead of ctx.session as mentioned by @strawgate (fastmcp=2.3.4)

I think this is correct. I did a test, returning the attributes of the context.session object and the contents of context.request_context.lifespan_context , and I found that the attributes in the SessionLifespan class I defined were added to context.request_context.lifespan_context, and the initialization assignment I mentioned above was also successful. The context.session.current_user = username operation mentioned by @strawgate seems to just add an attribute to the context.session object and has nothing to do with the per_session_lifespan setting.

s0214lx avatar May 22 '25 04:05 s0214lx

Closing this as complete given updates to sessions, state, and storage

jlowin avatar Oct 15 '25 14:10 jlowin