Merge pull request #302 from jlowin/context

Add method for retrieving current starlette request to FastMCP context

docs/servers/context.mdx

@@ -269,11 +269,31 @@ async def advanced_tool(ctx: Context) -> str:
     return f"Server: {server_name}"
 ```
 
+For web applications, you can access the underlying HTTP request:
+
+```python
+@mcp.tool()
+async def handle_web_request(ctx: Context) -> dict:
+    """Access HTTP request information from the Starlette request."""
+    request = ctx.get_starlette_request()
+    
+    # Access HTTP headers, query parameters, etc.
+    user_agent = request.headers.get("user-agent", "Unknown")
+    client_ip = request.client.host if request.client else "Unknown"
+    
+    return {
+        "user_agent": user_agent,
+        "client_ip": client_ip,
+        "path": request.url.path,
+    }
+```
+
 **Advanced Properties:**
 
 - **`ctx.fastmcp -> FastMCP`**: Access the server instance the context belongs to
 - **`ctx.session`**: Access the raw `mcp.server.session.ServerSession` object
 - **`ctx.request_context`**: Access the raw `mcp.shared.context.RequestContext` object
+- **`ctx.get_starlette_request() -> Request`**: Access the active Starlette request object (when running with a web server)
 
 <Warning>
 Direct use of `session` or `request_context` requires understanding the low-level MCP Python SDK and may be less stable than using the methods provided directly on the `Context` object.

src/fastmcp/server/context.py

@@ -13,10 +13,12 @@ from mcp.types import (
     SamplingMessage,
     TextContent,
 )
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 from pydantic.networks import AnyUrl
+from starlette.requests import Request
 
 from fastmcp.server.server import FastMCP
+from fastmcp.utilities.http import get_current_starlette_request
 from fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -59,6 +61,8 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
     _request_context: RequestContext[ServerSessionT, LifespanContextT] | None
     _fastmcp: FastMCP | None
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     def __init__(
         self,
         *,
@@ -222,3 +226,10 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
         )
 
         return result.content
+
+    def get_starlette_request(self) -> Request:
+        """Get the active starlette request."""
+        request = get_current_starlette_request()
+        if request is None:
+            raise ValueError("Request is not available outside a Starlette request")
+        return request

src/fastmcp/server/server.py

@@ -59,6 +59,7 @@ from fastmcp.resources.template import ResourceTemplate
 from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import Tool
 from fastmcp.utilities.decorators import DecoratedFunction
+from fastmcp.utilities.http import RequestMiddleware
 from fastmcp.utilities.logging import configure_logging, get_logger
 
 if TYPE_CHECKING:
@@ -822,10 +823,11 @@ class FastMCP(Generic[LifespanResultT]):
         log_level: str | None = None,
     ) -> None:
         """Run the server using SSE transport."""
-        starlette_app = self.sse_app()
+        app = self.sse_app()
+        app = RequestMiddleware(app)
 
         config = uvicorn.Config(
-            starlette_app,
+            app,
             host=host or self.settings.host,
             port=port or self.settings.port,
             log_level=log_level or self.settings.log_level.lower(),

src/fastmcp/utilities/http.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from contextlib import (
+    asynccontextmanager,
+)
+from contextvars import ContextVar
+
+from starlette.requests import Request
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+_current_starlette_request: ContextVar[Request | None] = ContextVar(
+    "starlette_request",
+    default=None,
+)
+
+
+@asynccontextmanager
+async def starlette_request_context(request: Request):
+    token = _current_starlette_request.set(request)
+    try:
+        yield
+    finally:
+        _current_starlette_request.reset(token)
+
+
+def get_current_starlette_request() -> Request | None:
+    return _current_starlette_request.get()
+
+
+class RequestMiddleware:
+    """
+    Middleware that stores each request in a ContextVar
+    """
+
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        async with starlette_request_context(Request(scope)):
+            await self.app(scope, receive, send)