import argparse
import uvicorn
from mcp.server import InitializationOptions, NotificationOptions
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from server.mcp_app import server, NAME, VERSION
SSE_PREFIX = '/ragnar'
sse = SseServerTransport(f"{SSE_PREFIX}/messages/")
# Define handler functions
async def handle_sse(request):
# Create an SSE transport at an endpoint
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await server.run(
streams[0],
streams[1],
InitializationOptions(
server_name=NAME,
server_version=VERSION,
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
def main(host, port):
# Create Starlette routes for SSE and message handling
routes = [
Route(f"{SSE_PREFIX}/sse", endpoint=handle_sse),
Mount(f"{SSE_PREFIX}/messages/", app=sse.handle_post_message),
]
# Create and run Starlette app
starlette_app = Starlette(routes=routes)
uvicorn.run(starlette_app, host=host, port=port)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run the Starlette SSE server.")
parser.add_argument(
"--host",
type=str,
default="localhost",
help="Host address to bind the server to (default: localhost)"
)
parser.add_argument(
"--port",
type=int,
default=8001,
help="Port number to bind the server to (default: 8001)"
)
# Parse arguments from the command line
args = parser.parse_args()
main(host=args.host, port=args.port)
