Skip to content

Transports API

The Mojentic MCP library provides transport classes for communicating with MCP servers using different protocols. These transport classes implement the McpTransport interface.

mojentic_mcp.transports.McpTransport

Bases: ABC

Abstract base class for MCP transports.

Source code in src/mojentic_mcp/transports.py 
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
class McpTransport(abc.ABC):
    """Abstract base class for MCP transports."""

    @abc.abstractmethod
    def send_request(self, rpc_request: JsonRpcRequest) -> Dict[str, Any]:
        """Sends a JSON-RPC request and returns the response.

        Args:
            rpc_request: The JSON-RPC request object.

        Returns:
            The JSON-RPC response as a dictionary.

        Raises:
            McpTransportError: If a transport-level error occurs.
            JsonRpcError: If the server returns a JSON-RPC error.
        """
        pass

    def initialize(self) -> None:
        """Initializes the transport (e.g., connect, start subprocess)."""
        pass

    def shutdown(self) -> None:
        """Shuts down the transport (e.g., disconnect, stop subprocess)."""
        pass

    def __enter__(self):
        self.initialize()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown()

initialize()

Initializes the transport (e.g., connect, start subprocess).

Source code in src/mojentic_mcp/transports.py 
33
34
35
def initialize(self) -> None:
    """Initializes the transport (e.g., connect, start subprocess)."""
    pass

send_request(rpc_request) abstractmethod

Sends a JSON-RPC request and returns the response.

Args: rpc_request: The JSON-RPC request object.

Returns: The JSON-RPC response as a dictionary.

Raises: McpTransportError: If a transport-level error occurs. JsonRpcError: If the server returns a JSON-RPC error.

Source code in src/mojentic_mcp/transports.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@abc.abstractmethod
def send_request(self, rpc_request: JsonRpcRequest) -> Dict[str, Any]:
    """Sends a JSON-RPC request and returns the response.

    Args:
        rpc_request: The JSON-RPC request object.

    Returns:
        The JSON-RPC response as a dictionary.

    Raises:
        McpTransportError: If a transport-level error occurs.
        JsonRpcError: If the server returns a JSON-RPC error.
    """
    pass

shutdown()

Shuts down the transport (e.g., disconnect, stop subprocess).

Source code in src/mojentic_mcp/transports.py 
37
38
39
def shutdown(self) -> None:
    """Shuts down the transport (e.g., disconnect, stop subprocess)."""
    pass

mojentic_mcp.transports.HttpTransport

Bases: McpTransport

MCP Transport using HTTP.

Source code in src/mojentic_mcp/transports.py 
 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
class HttpTransport(McpTransport):
    """MCP Transport using HTTP."""

    def __init__(self, url: str = None, host: str = None, port: int = None, path: str = "/jsonrpc", timeout: float = 30.0, http_gateway: Optional[HttpClientGateway] = None):
        """Initialize the HTTP transport.

        Args:
            url (str, optional): The full URL to the JSON-RPC endpoint. If provided, this takes precedence over host, port, and path.
            host (str, optional): The host to connect to. Required if url is not provided.
            port (int, optional): The port to connect to. Required if url is not provided.
            path (str, optional): The path to the JSON-RPC endpoint. Defaults to "/jsonrpc".
            timeout (float, optional): The timeout for HTTP requests in seconds. Defaults to 30.0.
            http_gateway (HttpGateway, optional): The HTTP gateway to use. If not provided, a new one will be created.

        Raises:
            ValueError: If neither url nor both host and port are provided.
        """
        if url:
            self._url = url
        elif host and port:
            self._url = f"http://{host}:{port}{path}"
        else:
            raise ValueError("Either url or both host and port must be provided")

        self._timeout = timeout
        self._http_gateway = http_gateway or HttpClientGateway(timeout=timeout)

    def initialize(self) -> None:
        self._http_gateway.initialize()
        logger.info("HttpTransport initialized", url=self._url)

    def shutdown(self) -> None:
        self._http_gateway.shutdown()
        logger.info("HttpTransport shutdown", url=self._url)

    def send_request(self, rpc_request: JsonRpcRequest) -> Dict[str, Any]:
        request_payload = rpc_request.model_dump(exclude_none=True)
        logger.debug("Sending HTTP request", url=self._url, payload=request_payload)
        try:
            response_json = self._http_gateway.post(self._url, request_payload)
            logger.debug("Received HTTP response", response=response_json)

            if "error" in response_json:
                err = response_json["error"]
                raise JsonRpcError(code=err.get("code"), message=err.get("message"), data=err.get("data"))
            return response_json
        except RuntimeError as e:
            logger.error("HTTP gateway error", exc_info=True)
            raise McpTransportError(f"HTTP client not initialized: {e}") from e
        except Exception as e:
            logger.error("HTTP request error", exc_info=True)
            raise McpTransportError(f"HTTP request failed: {e}") from e

__init__(url=None, host=None, port=None, path='/jsonrpc', timeout=30.0, http_gateway=None)

Initialize the HTTP transport.

Args: url (str, optional): The full URL to the JSON-RPC endpoint. If provided, this takes precedence over host, port, and path. host (str, optional): The host to connect to. Required if url is not provided. port (int, optional): The port to connect to. Required if url is not provided. path (str, optional): The path to the JSON-RPC endpoint. Defaults to "/jsonrpc". timeout (float, optional): The timeout for HTTP requests in seconds. Defaults to 30.0. http_gateway (HttpGateway, optional): The HTTP gateway to use. If not provided, a new one will be created.

Raises: ValueError: If neither url nor both host and port are provided.

Source code in src/mojentic_mcp/transports.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(self, url: str = None, host: str = None, port: int = None, path: str = "/jsonrpc", timeout: float = 30.0, http_gateway: Optional[HttpClientGateway] = None):
    """Initialize the HTTP transport.

    Args:
        url (str, optional): The full URL to the JSON-RPC endpoint. If provided, this takes precedence over host, port, and path.
        host (str, optional): The host to connect to. Required if url is not provided.
        port (int, optional): The port to connect to. Required if url is not provided.
        path (str, optional): The path to the JSON-RPC endpoint. Defaults to "/jsonrpc".
        timeout (float, optional): The timeout for HTTP requests in seconds. Defaults to 30.0.
        http_gateway (HttpGateway, optional): The HTTP gateway to use. If not provided, a new one will be created.

    Raises:
        ValueError: If neither url nor both host and port are provided.
    """
    if url:
        self._url = url
    elif host and port:
        self._url = f"http://{host}:{port}{path}"
    else:
        raise ValueError("Either url or both host and port must be provided")

    self._timeout = timeout
    self._http_gateway = http_gateway or HttpClientGateway(timeout=timeout)

mojentic_mcp.transports.StdioTransport

Bases: McpTransport

MCP Transport using STDIO with a subprocess.

Source code in src/mojentic_mcp/transports.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
class StdioTransport(McpTransport):
    """MCP Transport using STDIO with a subprocess."""

    def __init__(self, command: List[str], stdio_gateway: Optional[StdioGateway] = None):
        """Initialize the STDIO transport.

        Args:
            command (List[str]): The command to run.
            stdio_gateway (StdioGateway, optional): The STDIO gateway to use. If not provided, a new one will be created.
        """
        self._command = command
        self._stdio_gateway = stdio_gateway or StdioGateway()
        self._request_id_counter = 1  # Simple counter for stdio requests if not provided
        self._lock = threading.Lock()  # To ensure one request/response cycle at a time
        self._pid = None

    def initialize(self) -> None:
        try:
            self._pid = self._stdio_gateway.start_process(self._command)
            logger.info("StdioTransport initialized", command=self._command, pid=self._pid)
        except FileNotFoundError:
            logger.error("Stdio command not found", command=self._command)
            raise McpTransportError(f"Command not found: {self._command[0]}")
        except Exception as e:
            logger.error("Failed to start Stdio process", command=self._command, exc_info=True)
            raise McpTransportError(f"Failed to start subprocess for command '{self._command[0]}': {e}") from e

    def shutdown(self) -> None:
        if self._pid:
            logger.info("Shutting down StdioTransport", pid=self._pid)
            try:
                # Send exit request before terminating
                if self._stdio_gateway.is_process_running():
                    try:
                        exit_request = JsonRpcRequest(jsonrpc="2.0", id=self._request_id_counter, method="exit", params={})
                        self._request_id_counter += 1
                        self._stdio_gateway.write_line(json.dumps(exit_request.model_dump(exclude_none=True)))
                    except Exception:
                        logger.warn("Failed to send exit command for stdio process", pid=self._pid, exc_info=True)

                # Terminate the process
                self._stdio_gateway.terminate_process()
            except Exception as e:
                logger.warn("Error during StdioTransport shutdown", pid=self._pid, exc_info=True)

            self._pid = None
        logger.info("StdioTransport shutdown complete")

    def send_request(self, rpc_request: JsonRpcRequest) -> Dict[str, Any]:
        if not self._pid or not self._stdio_gateway.is_process_running():
            raise McpTransportError("STDIO process not running or process terminated.")

        with self._lock:
            if rpc_request.id is None:
                rpc_request.id = self._request_id_counter
                self._request_id_counter += 1

            request_payload_str = json.dumps(rpc_request.model_dump(exclude_none=True))
            logger.debug("Sending STDIO request", payload=request_payload_str, pid=self._pid)

            try:
                self._stdio_gateway.write_line(request_payload_str)

                try:
                    response_line = self._stdio_gateway.read_line()
                except EOFError:
                    stderr_output = self._stdio_gateway.get_stderr_output()
                    logger.error("No response from STDIO process", pid=self._pid, stderr=stderr_output)
                    raise McpTransportError(f"No response from STDIO process (PID: {self._pid}) or process terminated.")

                logger.debug("Received STDIO response line", line=response_line, pid=self._pid)
                response_json = json.loads(response_line)

                if "error" in response_json:
                    err = response_json["error"]
                    raise JsonRpcError(code=err.get("code"), message=err.get("message"), data=err.get("data"))

                if rpc_request.id is not None and response_json.get("id") != rpc_request.id:
                    logger.warn("Received STDIO response with mismatched ID",
                                expected_id=rpc_request.id, actual_id=response_json.get("id"), pid=self._pid)

                return response_json

            except ValueError as e:
                logger.error("STDIO process error", pid=self._pid, exc_info=True)
                raise McpTransportError(f"STDIO process error: {e}") from e
            except BrokenPipeError:
                logger.error("Broken pipe with STDIO process. Process likely terminated.", pid=self._pid, exc_info=True)
                stderr_content = self._stdio_gateway.get_stderr_output()
                raise McpTransportError(f"Broken pipe with STDIO process (PID: {self._pid}). Stderr: {stderr_content[:500]}") # Limit stderr length
            except json.JSONDecodeError as e:
                logger.error("Failed to decode JSON response from STDIO", pid=self._pid, exc_info=True, response_line=response_line)
                raise McpTransportError(f"Invalid JSON response from STDIO server: {e}") from e
            except Exception as e:
                logger.error("Error during STDIO communication", pid=self._pid, exc_info=True)
                raise McpTransportError(f"STDIO communication error: {e}") from e

__init__(command, stdio_gateway=None)

Initialize the STDIO transport.

Args: command (List[str]): The command to run. stdio_gateway (StdioGateway, optional): The STDIO gateway to use. If not provided, a new one will be created.

Source code in src/mojentic_mcp/transports.py 
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(self, command: List[str], stdio_gateway: Optional[StdioGateway] = None):
    """Initialize the STDIO transport.

    Args:
        command (List[str]): The command to run.
        stdio_gateway (StdioGateway, optional): The STDIO gateway to use. If not provided, a new one will be created.
    """
    self._command = command
    self._stdio_gateway = stdio_gateway or StdioGateway()
    self._request_id_counter = 1  # Simple counter for stdio requests if not provided
    self._lock = threading.Lock()  # To ensure one request/response cycle at a time
    self._pid = None

mojentic_mcp.transports.McpTransportError

Bases: Exception

Base exception for transport-related errors.

Source code in src/mojentic_mcp/transports.py 
49
50
51
class McpTransportError(Exception):
    """Base exception for transport-related errors."""
    pass

Usage Examples

HTTP Transport

from mojentic_mcp.transports import HttpTransport
from mojentic_mcp.rpc import JsonRpcRequest

# Create an HTTP transport with a full URL
http_transport = HttpTransport(url="http://localhost:8000/jsonrpc")

# Create an HTTP transport with host and port (using default path "/jsonrpc")
http_transport_host_port = HttpTransport(host="localhost", port=8000)

# Create an HTTP transport with host, port, and custom path
http_transport_custom_path = HttpTransport(host="localhost", port=8000, path="/custom-path")

# Create an HTTP transport with a custom timeout
http_transport_with_timeout = HttpTransport(url="http://localhost:8000/jsonrpc", timeout=60.0)

# Use the transport as a context manager
with HttpTransport(url="http://localhost:8000/jsonrpc") as transport:
    # Create a JSON-RPC request
    rpc_request = JsonRpcRequest(method="tools/list", params={})

    # Send the request and get the response
    response = transport.send_request(rpc_request)

STDIO Transport

import sys
from mojentic_mcp.transports import StdioTransport
from mojentic_mcp.rpc import JsonRpcRequest

# Create a STDIO transport that runs a Python script
stdio_transport = StdioTransport(command=[sys.executable, "stdio_server.py"])

# Create a STDIO transport that runs a compiled executable
stdio_transport_executable = StdioTransport(command=["./mcp_server"])

# Use the transport as a context manager
with StdioTransport(command=[sys.executable, "stdio_server.py"]) as transport:
    # Create a JSON-RPC request
    rpc_request = JsonRpcRequest(method="tools/list", params={})

    # Send the request and get the response
    response = transport.send_request(rpc_request)

Best Practices

  1. Use Context Managers: Always use transports with a context manager (with statement) to ensure proper initialization and cleanup.

  2. Handle Errors: Always handle McpTransportError exceptions when sending requests.

  3. Choose the Right Transport: Use HttpTransport for communicating with HTTP servers and StdioTransport for communicating with STDIO servers.

  4. Set Appropriate Timeouts: For HttpTransport, set an appropriate timeout based on the expected response time of the server.

See Also