Option to mount SSE server to existing ASGI server (#312)

* Option to mount SSE server to existing ASGI server

Fixes #311

Add option to mount SSE server to an existing ASGI server.

* Add a new method `sse_app` in `src/mcp/server/fastmcp/server.py` to return an instance of the SSE server app.
* Update the `run_sse_async` method in `src/mcp/server/fastmcp/server.py` to use the new `sse_app` method.
* Update the documentation in `README.md` to include instructions on how to mount the SSE server to an existing ASGI server.
* Fix the example in `README.md` to use `app.mount('/', mcp.sse_app())` instead.

---

For more details, open the [Copilot Workspace session](https://copilot-workspace.githubnext.com/modelcontextprotocol/python-sdk/issues/311?shareId=XXXX-XXXX-XXXX-XXXX).

* Add `sse_app` method and update `run_sse_async` method in `server.py`

* Add `sse_app` method to return an instance of the SSE server app
* Update `run_sse_async` method to use the new `sse_app` method

Update `README.md` to include instructions for mounting SSE server

* Add section on mounting the SSE server to an existing ASGI server

* fix: Move import statements to the top of the file/

* docs: Update README to reflect changes in mounting SSE server with Starlette

* docs: Formatting of SSE server mounting example in README
This commit is contained in:
comfuture
2025-03-19 13:26:49 +09:00
committed by GitHub
parent 6b6f34eaa6
commit a9aca20205
2 changed files with 41 additions and 12 deletions

View File

@@ -31,6 +31,7 @@
- [Development Mode](#development-mode) - [Development Mode](#development-mode)
- [Claude Desktop Integration](#claude-desktop-integration) - [Claude Desktop Integration](#claude-desktop-integration)
- [Direct Execution](#direct-execution) - [Direct Execution](#direct-execution)
- [Mounting to an Existing ASGI Server](#mounting-to-an-existing-asgi-server)
- [Examples](#examples) - [Examples](#examples)
- [Echo Server](#echo-server) - [Echo Server](#echo-server)
- [SQLite Explorer](#sqlite-explorer) - [SQLite Explorer](#sqlite-explorer)
@@ -346,6 +347,31 @@ python server.py
mcp run server.py mcp run server.py
``` ```
### Mounting to an Existing ASGI Server
You can mount the SSE server to an existing ASGI server using the `sse_app` method. This allows you to integrate the SSE server with other ASGI applications.
```python
from starlette.applications import Starlette
from starlette.routes import Mount, Host
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
# Mount the SSE server to the existing ASGI server
app = Starlette(
routes=[
Mount('/', app=mcp.sse_app()),
]
)
# or dynamically mount as host
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
```
For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).
## Examples ## Examples
### Echo Server ### Echo Server

View File

@@ -19,6 +19,8 @@ import uvicorn
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from pydantic.networks import AnyUrl from pydantic.networks import AnyUrl
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from mcp.server.fastmcp.exceptions import ResourceError from mcp.server.fastmcp.exceptions import ResourceError
from mcp.server.fastmcp.prompts import Prompt, PromptManager from mcp.server.fastmcp.prompts import Prompt, PromptManager
@@ -461,9 +463,19 @@ class FastMCP:
async def run_sse_async(self) -> None: async def run_sse_async(self) -> None:
"""Run the server using SSE transport.""" """Run the server using SSE transport."""
from starlette.applications import Starlette starlette_app = self.sse_app()
from starlette.routing import Mount, Route
config = uvicorn.Config(
starlette_app,
host=self.settings.host,
port=self.settings.port,
log_level=self.settings.log_level.lower(),
)
server = uvicorn.Server(config)
await server.serve()
def sse_app(self) -> Starlette:
"""Return an instance of the SSE server app."""
sse = SseServerTransport("/messages/") sse = SseServerTransport("/messages/")
async def handle_sse(request): async def handle_sse(request):
@@ -476,7 +488,7 @@ class FastMCP:
self._mcp_server.create_initialization_options(), self._mcp_server.create_initialization_options(),
) )
starlette_app = Starlette( return Starlette(
debug=self.settings.debug, debug=self.settings.debug,
routes=[ routes=[
Route("/sse", endpoint=handle_sse), Route("/sse", endpoint=handle_sse),
@@ -484,15 +496,6 @@ class FastMCP:
], ],
) )
config = uvicorn.Config(
starlette_app,
host=self.settings.host,
port=self.settings.port,
log_level=self.settings.log_level.lower(),
)
server = uvicorn.Server(config)
await server.serve()
async def list_prompts(self) -> list[MCPPrompt]: async def list_prompts(self) -> list[MCPPrompt]:
"""List all available prompts.""" """List all available prompts."""
prompts = self._prompt_manager.list_prompts() prompts = self._prompt_manager.list_prompts()