Skip to content

HTTP Server API

The Mojentic MCP library provides an HTTP server implementation for exposing MCP functionality over HTTP using FastAPI.

mojentic_mcp.mcp_http.HttpMcpServer

An MCP server that communicates over HTTP using JSON-RPC.

Source code in src/mojentic_mcp/mcp_http.py 
 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
class HttpMcpServer:
    """An MCP server that communicates over HTTP using JSON-RPC."""

    def __init__(self, rpc_handler: JsonRpcHandler, path: str = "/jsonrpc"):
        """Initialize the HTTP MCP server.

        Args:
            rpc_handler (JsonRpcHandler): The JSON-RPC handler to use.
            path (str, optional): The path to serve the JSON-RPC endpoint on. Defaults to "/jsonrpc".
        """
        self.rpc_handler = rpc_handler
        self.path = path
        self.app = FastAPI(title="Codebase Examiner MCP", description="MCP server for examining Python codebases")

        # Register the JSON-RPC endpoint
        self.app.post(self.path)(self.handle_jsonrpc)

    async def handle_jsonrpc(self, request: Request) -> Response:
        """Handle JSON-RPC 2.0 requests.

        Args:
            request (Request): The FastAPI request object

        Returns:
            Response: The JSON-RPC response
        """
        try:
            # Parse the request body
            body = await request.json()

            # Convert to JsonRpcRequest model
            rpc_request = JsonRpcRequest(**body)

            # Handle the JSON-RPC request
            response = self.rpc_handler.handle_request(rpc_request)

            # Return the response
            return Response(
                content=json.dumps(response),
                media_type="application/json"
            )
        except json.JSONDecodeError:
            # Invalid JSON
            error_response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {
                    "code": -32700,
                    "message": "Parse error"
                }
            }
            return Response(
                content=json.dumps(error_response),
                media_type="application/json",
                status_code=400
            )
        except ValidationError as e:
            # Invalid Request (validation error)
            error_response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {
                    "code": -32600,
                    "message": "Invalid Request",
                    "data": str(e)
                }
            }
            return Response(
                content=json.dumps(error_response),
                media_type="application/json",
                status_code=400
            )
        except Exception as e:
            # Server error
            error_response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {
                    "code": -32603,
                    "message": f"Internal error: {str(e)}"
                }
            }
            return Response(
                content=json.dumps(error_response),
                media_type="application/json",
                status_code=500
            )

    def run(self, host: str = "0.0.0.0", port: int = 8080):
        """Run the HTTP MCP server.

        Args:
            host (str): The host to bind to
            port (int): The port to listen on
        """
        uvicorn.run(self.app, host=host, port=port)

__init__(rpc_handler, path='/jsonrpc')

Initialize the HTTP MCP server.

Args: rpc_handler (JsonRpcHandler): The JSON-RPC handler to use. path (str, optional): The path to serve the JSON-RPC endpoint on. Defaults to "/jsonrpc".

Source code in src/mojentic_mcp/mcp_http.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
def __init__(self, rpc_handler: JsonRpcHandler, path: str = "/jsonrpc"):
    """Initialize the HTTP MCP server.

    Args:
        rpc_handler (JsonRpcHandler): The JSON-RPC handler to use.
        path (str, optional): The path to serve the JSON-RPC endpoint on. Defaults to "/jsonrpc".
    """
    self.rpc_handler = rpc_handler
    self.path = path
    self.app = FastAPI(title="Codebase Examiner MCP", description="MCP server for examining Python codebases")

    # Register the JSON-RPC endpoint
    self.app.post(self.path)(self.handle_jsonrpc)

handle_jsonrpc(request) async

Handle JSON-RPC 2.0 requests.

Args: request (Request): The FastAPI request object

Returns: Response: The JSON-RPC response

Source code in src/mojentic_mcp/mcp_http.py 
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
async def handle_jsonrpc(self, request: Request) -> Response:
    """Handle JSON-RPC 2.0 requests.

    Args:
        request (Request): The FastAPI request object

    Returns:
        Response: The JSON-RPC response
    """
    try:
        # Parse the request body
        body = await request.json()

        # Convert to JsonRpcRequest model
        rpc_request = JsonRpcRequest(**body)

        # Handle the JSON-RPC request
        response = self.rpc_handler.handle_request(rpc_request)

        # Return the response
        return Response(
            content=json.dumps(response),
            media_type="application/json"
        )
    except json.JSONDecodeError:
        # Invalid JSON
        error_response = {
            "jsonrpc": "2.0",
            "id": None,
            "error": {
                "code": -32700,
                "message": "Parse error"
            }
        }
        return Response(
            content=json.dumps(error_response),
            media_type="application/json",
            status_code=400
        )
    except ValidationError as e:
        # Invalid Request (validation error)
        error_response = {
            "jsonrpc": "2.0",
            "id": None,
            "error": {
                "code": -32600,
                "message": "Invalid Request",
                "data": str(e)
            }
        }
        return Response(
            content=json.dumps(error_response),
            media_type="application/json",
            status_code=400
        )
    except Exception as e:
        # Server error
        error_response = {
            "jsonrpc": "2.0",
            "id": None,
            "error": {
                "code": -32603,
                "message": f"Internal error: {str(e)}"
            }
        }
        return Response(
            content=json.dumps(error_response),
            media_type="application/json",
            status_code=500
        )

run(host='0.0.0.0', port=8080)

Run the HTTP MCP server.

Args: host (str): The host to bind to port (int): The port to listen on

Source code in src/mojentic_mcp/mcp_http.py 
100
101
102
103
104
105
106
107
def run(self, host: str = "0.0.0.0", port: int = 8080):
    """Run the HTTP MCP server.

    Args:
        host (str): The host to bind to
        port (int): The port to listen on
    """
    uvicorn.run(self.app, host=host, port=port)

mojentic_mcp.mcp_http.start_server(port, rpc_handler, path='/jsonrpc')

Start the MCP server.

Args: port (int): The port to run the server on rpc_handler (JsonRpcHandler): The JSON-RPC handler to use. path (str, optional): The path to serve the JSON-RPC endpoint on. Defaults to "/jsonrpc".

Source code in src/mojentic_mcp/mcp_http.py 
110
111
112
113
114
115
116
117
118
119
def start_server(port: int, rpc_handler: JsonRpcHandler, path: str = "/jsonrpc"):
    """Start the MCP server.

    Args:
        port (int): The port to run the server on
        rpc_handler (JsonRpcHandler): The JSON-RPC handler to use.
        path (str, optional): The path to serve the JSON-RPC endpoint on. Defaults to "/jsonrpc".
    """
    server = HttpMcpServer(rpc_handler, path=path)
    server.run(port=port)

JSON-RPC Endpoint

The HTTP server exposes a single JSON-RPC endpoint. By default, this endpoint is at /jsonrpc, but it can be customized using the path parameter when creating the server. This endpoint accepts POST requests with JSON-RPC 2.0 payloads.

Request Format

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "method-name",
  "params": {}
}

Response Format

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "result": {}
}

Error Response Format

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "error": {
    "code": -32000,
    "message": "Error message"
  }
}

Error Handling

The HTTP server handles various error conditions:

  1. Invalid JSON: If the request body is not valid JSON, a parse error is returned (code -32700).

  2. Invalid Request: If the request does not conform to the JSON-RPC 2.0 specification, an invalid request error is returned (code -32600).

  3. Internal Error: If an unexpected error occurs while processing the request, an internal error is returned (code -32603).

Environment Variables

The HTTP server can be configured using environment variables:

  • PORT: The port to listen on. This overrides the port parameter passed to the run method.

Example: 

PORT=8001 python -m mojentic_mcp.mcp_http

Command Line Usage

The HTTP server can be run directly from the command line:

python -m mojentic_mcp.mcp_http <port>

Example: 

python -m mojentic_mcp.mcp_http 8000

Best Practices

  1. Use a Dedicated Port: Choose a port that is not used by other services.

  2. Configure Logging: Set up appropriate logging to monitor server activity.

  3. Secure the Server: If exposing the server to the internet, consider using HTTPS and authentication.

  4. Handle Errors Gracefully: Ensure that your tools handle errors gracefully to avoid server crashes.

See Also