File size: 11,944 Bytes
0d33002
 
08794a2
4c1f058
0d33002
b6176a5
0d33002
154abca
 
 
 
 
b6176a5
d09d7f6
025e46b
70e3e8c
69d9f94
154abca
 
 
0d33002
154abca
987be7e
59f2792
0d33002
341332a
0d33002
 
154abca
4c1f058
154abca
0d33002
 
d09d7f6
c219134
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
670f900
1534a61
0d33002
 
 
 
59f2792
 
e4bf853
59f2792
 
 
1534a61
 
 
0d33002
1534a61
0d33002
1534a61
0d33002
 
1534a61
0d33002
 
 
 
 
 
 
 
4c1f058
 
 
 
0d33002
154abca
 
08794a2
987be7e
08794a2
58bd353
08794a2
59f2792
08794a2
 
 
 
 
 
 
 
 
 
 
 
 
 
59f2792
58bd353
 
 
 
 
08794a2
 
154abca
d09d7f6
154abca
 
3959209
154abca
987be7e
 
59f2792
154abca
 
 
 
 
 
3959209
154abca
987be7e
 
154abca
e033dfc
154abca
6ad1252
987be7e
 
 
154abca
 
 
e033dfc
 
 
154abca
 
 
 
 
 
 
8c0df0f
7befcc3
8c0df0f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7befcc3
 
 
b6176a5
08794a2
987be7e
154abca
 
b6176a5
 
 
154abca
 
 
987be7e
154abca
 
b6176a5
 
 
154abca
 
 
08794a2
9077360
 
154abca
987be7e
154abca
 
 
 
 
 
987be7e
154abca
 
 
 
 
 
e033dfc
987be7e
 
125f50b
987be7e
 
 
 
154abca
08794a2
86a15b9
987be7e
 
58bd353
 
86a15b9
 
a506ee1
86a15b9
 
4c1f058
 
 
d09d7f6
4c1f058
70e3e8c
3959209
4c1f058
 
 
987be7e
 
59f2792
4c1f058
 
 
 
 
 
3959209
4c1f058
 
 
987be7e
 
4c1f058
 
 
 
987be7e
 
 
4c1f058
 
 
 
 
 
 
 
c219134
 
4c1f058
6ad1252
7befcc3
8c0df0f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4c1f058
7befcc3
 
4c1f058
86eb39a
bf9967e
c219134
86eb39a
c219134
 
9a6f944
4c1f058
 
 
86eb39a
bf9967e
c219134
86eb39a
c219134
4c1f058
 
 
 
987be7e
 
125f50b
987be7e
 
 
 
4c1f058
 
 
 
 
 
 
08794a2
86a15b9
987be7e
 
58bd353
 
 
86a15b9
 
 
86eb39a
a506ee1
86a15b9
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
from __future__ import annotations

from collections.abc import AsyncGenerator, Callable, Generator
from contextlib import asynccontextmanager, contextmanager
from contextvars import ContextVar
from typing import TYPE_CHECKING, cast

from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
from mcp.server.auth.middleware.bearer_auth import (
    BearerAuthBackend,
    RequireAuthMiddleware,
)
from mcp.server.auth.provider import TokenVerifier as TokenVerifierProtocol
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.sse import SseServerTransport
from mcp.server.streamable_http import EventStore
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import BaseRoute, Mount, Route
from starlette.types import Lifespan, Receive, Scope, Send

from fastmcp.server.auth import AuthProvider
from fastmcp.utilities.logging import get_logger

if TYPE_CHECKING:
    from fastmcp.server.server import FastMCP

logger = get_logger(__name__)


class StreamableHTTPASGIApp:
    """ASGI application wrapper for Streamable HTTP server transport."""

    def __init__(self, session_manager):
        self.session_manager = session_manager

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        try:
            await self.session_manager.handle_request(scope, receive, send)
        except RuntimeError as e:
            if str(e) == "Task group is not initialized. Make sure to use run().":
                logger.error(
                    f"Original RuntimeError from mcp library: {e}", exc_info=True
                )
                new_error_message = (
                    "FastMCP's StreamableHTTPSessionManager task group was not initialized. "
                    "This commonly occurs when the FastMCP application's lifespan is not "
                    "passed to the parent ASGI application (e.g., FastAPI or Starlette). "
                    "Please ensure you are setting `lifespan=mcp_app.lifespan` in your "
                    "parent app's constructor, where `mcp_app` is the application instance "
                    "returned by `fastmcp_instance.http_app()`. \\n"
                    "For more details, see the FastMCP ASGI integration documentation: "
                    "https://gofastmcp.com/deployment/asgi"
                )
                # Raise a new RuntimeError that includes the original error's message
                # for full context, but leads with the more helpful guidance.
                raise RuntimeError(f"{new_error_message}\\nOriginal error: {e}") from e
            else:
                # Re-raise other RuntimeErrors if they don't match the specific message
                raise


_current_http_request: ContextVar[Request | None] = ContextVar(  # type: ignore[assignment]
    "http_request",
    default=None,
)


class StarletteWithLifespan(Starlette):
    @property
    def lifespan(self) -> Lifespan[Starlette]:
        return self.router.lifespan_context


@contextmanager
def set_http_request(request: Request) -> Generator[Request, None, None]:
    token = _current_http_request.set(request)
    try:
        yield request
    finally:
        _current_http_request.reset(token)


class RequestContextMiddleware:
    """
    Middleware that stores each request in a ContextVar
    """

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            with set_http_request(Request(scope)):
                await self.app(scope, receive, send)
        else:
            await self.app(scope, receive, send)


def create_base_app(
    routes: list[BaseRoute],
    middleware: list[Middleware],
    debug: bool = False,
    lifespan: Callable | None = None,
) -> StarletteWithLifespan:
    """Create a base Starlette app with common middleware and routes.

    Args:
        routes: List of routes to include in the app
        middleware: List of middleware to include in the app
        debug: Whether to enable debug mode
        lifespan: Optional lifespan manager for the app

    Returns:
        A Starlette application
    """
    # Always add RequestContextMiddleware as the outermost middleware
    middleware.append(Middleware(RequestContextMiddleware))

    return StarletteWithLifespan(
        routes=routes,
        middleware=middleware,
        debug=debug,
        lifespan=lifespan,
    )


def create_sse_app(
    server: FastMCP[LifespanResultT],
    message_path: str,
    sse_path: str,
    auth: AuthProvider | None = None,
    debug: bool = False,
    routes: list[BaseRoute] | None = None,
    middleware: list[Middleware] | None = None,
) -> StarletteWithLifespan:
    """Return an instance of the SSE server app.

    Args:
        server: The FastMCP server instance
        message_path: Path for SSE messages
        sse_path: Path for SSE connections
        auth: Optional authentication provider (AuthProvider)
        debug: Whether to enable debug mode
        routes: Optional list of custom routes
        middleware: Optional list of middleware
    Returns:
        A Starlette application with RequestContextMiddleware
    """

    server_routes: list[BaseRoute] = []
    server_middleware: list[Middleware] = []

    # Set up SSE transport
    sse = SseServerTransport(message_path)

    # Create handler for SSE connections
    async def handle_sse(scope: Scope, receive: Receive, send: Send) -> Response:
        async with sse.connect_sse(scope, receive, send) as streams:
            await server._mcp_server.run(
                streams[0],
                streams[1],
                server._mcp_server.create_initialization_options(),
            )
        return Response()

    # Set up auth if enabled
    if auth:
        # Create auth middleware
        auth_middleware = [
            Middleware(
                AuthenticationMiddleware,
                backend=BearerAuthBackend(auth),
            ),
            Middleware(AuthContextMiddleware),
        ]

        # Get auth routes and scopes
        auth_routes = auth.get_routes()
        required_scopes = getattr(auth, "required_scopes", None) or []

        # Get resource metadata URL for WWW-Authenticate header
        resource_metadata_url = auth.get_resource_metadata_url()

        server_routes.extend(auth_routes)
        server_middleware.extend(auth_middleware)

        # Auth is enabled, wrap endpoints with RequireAuthMiddleware
        server_routes.append(
            Route(
                sse_path,
                endpoint=RequireAuthMiddleware(
                    handle_sse, required_scopes, resource_metadata_url
                ),
                methods=["GET"],
            )
        )
        server_routes.append(
            Mount(
                message_path,
                app=RequireAuthMiddleware(
                    sse.handle_post_message, required_scopes, resource_metadata_url
                ),
            )
        )
    else:
        # No auth required
        async def sse_endpoint(request: Request) -> Response:
            return await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]

        server_routes.append(
            Route(
                sse_path,
                endpoint=sse_endpoint,
                methods=["GET"],
            )
        )
        server_routes.append(
            Mount(
                message_path,
                app=sse.handle_post_message,
            )
        )

    # Add custom routes with lowest precedence
    if routes:
        server_routes.extend(routes)
    server_routes.extend(server._get_additional_http_routes())

    # Add middleware
    if middleware:
        server_middleware.extend(middleware)

    # Create and return the app
    app = create_base_app(
        routes=server_routes,
        middleware=server_middleware,
        debug=debug,
    )
    # Store the FastMCP server instance on the Starlette app state
    app.state.fastmcp_server = server
    app.state.path = sse_path

    return app


def create_streamable_http_app(
    server: FastMCP[LifespanResultT],
    streamable_http_path: str,
    event_store: EventStore | None = None,
    auth: AuthProvider | None = None,
    json_response: bool = False,
    stateless_http: bool = False,
    debug: bool = False,
    routes: list[BaseRoute] | None = None,
    middleware: list[Middleware] | None = None,
) -> StarletteWithLifespan:
    """Return an instance of the StreamableHTTP server app.

    Args:
        server: The FastMCP server instance
        streamable_http_path: Path for StreamableHTTP connections
        event_store: Optional event store for session management
        auth: Optional authentication provider (AuthProvider)
        json_response: Whether to use JSON response format
        stateless_http: Whether to use stateless mode (new transport per request)
        debug: Whether to enable debug mode
        routes: Optional list of custom routes
        middleware: Optional list of middleware

    Returns:
        A Starlette application with StreamableHTTP support
    """
    server_routes: list[BaseRoute] = []
    server_middleware: list[Middleware] = []

    # Create session manager using the provided event store
    session_manager = StreamableHTTPSessionManager(
        app=server._mcp_server,
        event_store=event_store,
        json_response=json_response,
        stateless=stateless_http,
    )

    # Create the ASGI app wrapper
    streamable_http_app = StreamableHTTPASGIApp(session_manager)

    # Add StreamableHTTP routes with or without auth
    if auth:
        # Create auth middleware
        auth_middleware = [
            Middleware(
                AuthenticationMiddleware,
                backend=BearerAuthBackend(cast(TokenVerifierProtocol, auth)),
            ),
            Middleware(AuthContextMiddleware),
        ]

        # Get auth routes and scopes
        auth_routes = auth.get_routes()
        required_scopes = getattr(auth, "required_scopes", None) or []

        # Get resource metadata URL for WWW-Authenticate header
        resource_metadata_url = auth.get_resource_metadata_url()

        server_routes.extend(auth_routes)
        server_middleware.extend(auth_middleware)

        # Auth is enabled, wrap endpoint with RequireAuthMiddleware
        server_routes.append(
            Route(
                streamable_http_path,
                endpoint=RequireAuthMiddleware(
                    streamable_http_app, required_scopes, resource_metadata_url
                ),
            )
        )
    else:
        # No auth required
        server_routes.append(
            Route(
                streamable_http_path,
                endpoint=streamable_http_app,
            )
        )

    # Add custom routes with lowest precedence
    if routes:
        server_routes.extend(routes)
    server_routes.extend(server._get_additional_http_routes())

    # Add middleware
    if middleware:
        server_middleware.extend(middleware)

    # Create a lifespan manager to start and stop the session manager
    @asynccontextmanager
    async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
        async with session_manager.run():
            yield

    # Create and return the app with lifespan
    app = create_base_app(
        routes=server_routes,
        middleware=server_middleware,
        debug=debug,
        lifespan=lifespan,
    )
    # Store the FastMCP server instance on the Starlette app state
    app.state.fastmcp_server = server

    app.state.path = streamable_http_path

    return app