ZeroToken/mcp_server_http.py at main · AMOS144/ZeroToken · GitHub

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
"""
ZeroToken MCP Server - Streamable HTTP transport.
Runs the same MCP server as stdio mode but over HTTP for OpenClaw/MCPorter.
Service stays resident so browser state is preserved across tool calls.
"""

import asyncio
import contextlib
import os

from starlette.applications import Starlette
from starlette.routing import Route

from mcp.server.streamable_http_manager import StreamableHTTPSessionManager

from mcp_server import server


class _StreamableHTTPASGIApp:
    """ASGI app that delegates to StreamableHTTPSessionManager."""

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

    async def __call__(self, scope, receive, send):
        await self.session_manager.handle_request(scope, receive, send)


def _create_app() -> Starlette:
    """Create Starlette app with Streamable HTTP MCP endpoint."""
    session_manager = StreamableHTTPSessionManager(
        app=server,
        event_store=None,
        json_response=True,
        stateless=False,
        security_settings=None,
        retry_interval=None,
    )
    mcp_app = _StreamableHTTPASGIApp(session_manager)

    @contextlib.asynccontextmanager
    async def lifespan(app: Starlette):
        async with session_manager.run():
            yield

    return Starlette(
        routes=[
            Route("/mcp", endpoint=mcp_app, methods=["GET", "POST", "DELETE"]),
        ],
        lifespan=lifespan,
    )


def run(host: str | None = None, port: int | None = None):
    """Entry point for zerotoken-mcp-http console script."""
    import argparse
    import sys
    parser = argparse.ArgumentParser(description="ZeroToken MCP over Streamable HTTP")
    parser.add_argument("--host", default=None, help="Host to bind (default: 0.0.0.0)")
    parser.add_argument("--port", type=int, default=None, help="Port (default: 8000)")
    args, _ = parser.parse_known_args(sys.argv[1:])
    host = host or args.host or os.environ.get("ZEROTOKEN_HTTP_HOST", "0.0.0.0")
    port = port or args.port or int(os.environ.get("ZEROTOKEN_HTTP_PORT", "8000"))

    app = _create_app()

    async def serve():
        import uvicorn
        config = uvicorn.Config(
            app,
            host=host,
            port=port,
            log_level="info",
        )
        server_instance = uvicorn.Server(config)
        await server_instance.serve()

    asyncio.run(serve())


if __name__ == "__main__":
    run()