sse explored

Author: Kamalabot

fastapi_sse/app/__init__.py

@@ -0,0 +1,35 @@
+from fastapi import FastAPI
+from app.sse import create_sse_server
+from mcp.server.fastmcp import FastMCP
+
+app = FastAPI()
+mcp = FastMCP("Echo")
+
+# Mount the Starlette SSE server onto the FastAPI app
+app.mount("/", create_sse_server(mcp))
+# this mounting allows the sse server to
+# handle the reqs to /sse and /message path
+
+
+@app.get("/")
+def read_root():
+    return {"Hello": "World"}
+
+
+# Add MCP functionality with decorators
+@mcp.resource("echo://{message}")
+def echo_resource(message: str) -> str:
+    """Echo a message as a resource"""
+    return f"Resource echo: {message}"
+
+
+@mcp.tool()
+def echo_tool(message: str) -> str:
+    """Echo a message as a tool"""
+    return f"Tool echo: {message}"
+
+
+@mcp.prompt()
+def echo_prompt(message: str) -> str:
+    """Create an echo prompt"""
+    return f"Please process this message: {message}"

fastapi_sse/app/main.py

@@ -0,0 +1,30 @@
+from mcp.server.fastmcp import FastMCP
+from mcp.server.sse import SseServerTransport
+from starlette.applications import Starlette
+from starlette.routing import Mount, Route
+
+
+def create_sse_server(mcp: FastMCP):
+    """Create a Starlette app that handles SSE connections and message handling"""
+    transport = SseServerTransport("/messages/")
+    # this takes care of the event streaming
+
+    # Define handler functions
+    async def handle_sse(request):
+        # estab connect
+        async with transport.connect_sse(
+            request.scope, request.receive, request._send
+        ) as streams:
+            # bi directional stream of conn
+            await mcp._mcp_server.run(
+                streams[0], streams[1], mcp._mcp_server.create_initialization_options()
+            )
+
+    # Create Starlette routes for SSE and message handling
+    routes = [
+        Route("/sse/", endpoint=handle_sse),
+        Mount("/messages/", app=transport.handle_post_message),
+    ]
+
+    # Create a Starlette app
+    return Starlette(routes=routes)

fastapi_sse/app/sse.py

@@ -0,0 +1,8 @@
+[project]
+name = "fastapi-sse"
+version = "0.1.0"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi[standard]>=0.115.12",
+    "mcp>=1.6.0",
+]