Skip to content

transport

transport

MCP transport implementations.

Classes

MCPTransport

Bases: ABC

Abstract transport layer for MCP communication.

Functions
send abstractmethod 
send(request: MCPRequest) -> MCPResponse

Send a request and return the response.

Source code in src/openjarvis/mcp/transport.py 
19
20
21
@abstractmethod
def send(self, request: MCPRequest) -> MCPResponse:
    """Send a request and return the response."""
close abstractmethod 
close() -> None

Release transport resources.

Source code in src/openjarvis/mcp/transport.py 
23
24
25
@abstractmethod
def close(self) -> None:
    """Release transport resources."""

InProcessTransport 

InProcessTransport(server: MCPServer)

Bases: MCPTransport

Direct in-process transport for testing.

Routes requests directly to an MCPServer instance without serialization overhead.

Source code in src/openjarvis/mcp/transport.py 
35
36
def __init__(self, server: MCPServer) -> None:
    self._server = server
Functions
send 
send(request: MCPRequest) -> MCPResponse

Dispatch request directly to the server.

Source code in src/openjarvis/mcp/transport.py 
38
39
40
def send(self, request: MCPRequest) -> MCPResponse:
    """Dispatch request directly to the server."""
    return self._server.handle(request)
close 
close() -> None

No resources to release.

Source code in src/openjarvis/mcp/transport.py 
42
43
def close(self) -> None:
    """No resources to release."""

StdioTransport 

StdioTransport(command: List[str])

Bases: MCPTransport

JSON-RPC over stdin/stdout subprocess transport.

Launches a subprocess and communicates via JSON lines on stdin/stdout.

Source code in src/openjarvis/mcp/transport.py 
53
54
55
56
def __init__(self, command: List[str]) -> None:
    self._command = command
    self._process: Optional[subprocess.Popen[str]] = None
    self._start()
Functions
send 
send(request: MCPRequest) -> MCPResponse

Write request as JSON line, read response line.

Source code in src/openjarvis/mcp/transport.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def send(self, request: MCPRequest) -> MCPResponse:
    """Write request as JSON line, read response line."""
    proc = self._process
    if proc is None or proc.stdin is None or proc.stdout is None:
        raise RuntimeError("Transport process is not running")

    line = request.to_json() + "\n"
    proc.stdin.write(line)
    proc.stdin.flush()

    response_line = proc.stdout.readline()
    if not response_line:
        raise RuntimeError("No response from subprocess")
    return MCPResponse.from_json(response_line.strip())
close 
close() -> None

Terminate the subprocess.

Source code in src/openjarvis/mcp/transport.py 
83
84
85
86
87
88
def close(self) -> None:
    """Terminate the subprocess."""
    if self._process is not None:
        self._process.terminate()
        self._process.wait(timeout=5)
        self._process = None

SSETransport 

SSETransport(url: str)

Bases: MCPTransport

JSON-RPC over HTTP with Server-Sent Events.

Sends requests via HTTP POST and reads SSE responses.

Source code in src/openjarvis/mcp/transport.py 
97
98
def __init__(self, url: str) -> None:
    self._url = url
Functions
send 
send(request: MCPRequest) -> MCPResponse

Send request via HTTP POST.

Source code in src/openjarvis/mcp/transport.py 
100
101
102
103
104
105
106
107
108
109
110
111
def send(self, request: MCPRequest) -> MCPResponse:
    """Send request via HTTP POST."""
    import httpx

    response = httpx.post(
        self._url,
        json=json.loads(request.to_json()),
        headers={"Content-Type": "application/json"},
        timeout=30.0,
    )
    response.raise_for_status()
    return MCPResponse.from_json(response.text)
close 
close() -> None

No persistent connection to close.

Source code in src/openjarvis/mcp/transport.py 
113
114
def close(self) -> None:
    """No persistent connection to close."""