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

@abstractmethod
def send(self, request: MCPRequest) -> MCPResponse:
    """Send a request and return the response."""

send_notification

send_notification(request: MCPRequest) -> None

Send a JSON-RPC notification (no response expected).

The default implementation delegates to :meth:send and discards the response. Transports may override this when the server returns no body for notifications (e.g. HTTP 202 Accepted).

Source code in src/openjarvis/mcp/transport.py

def send_notification(self, request: MCPRequest) -> None:
    """Send a JSON-RPC notification (no response expected).

    The default implementation delegates to :meth:`send` and discards the
    response.  Transports may override this when the server returns no
    body for notifications (e.g. HTTP 202 Accepted).
    """
    self.send(request)

close abstractmethod

close() -> None

Release transport resources.

Source code in src/openjarvis/mcp/transport.py

@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

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

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

def close(self) -> None:
    """No resources to release."""

StdioTransport

StdioTransport(command: List[str], *, response_timeout: float = 600.0)

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

def __init__(
    self,
    command: List[str],
    *,
    response_timeout: float = 600.0,
) -> None:
    if response_timeout <= 0:
        raise ValueError("response_timeout must be positive")
    self._command = command
    self._response_timeout = response_timeout
    self._process: Optional[subprocess.Popen[str]] = None
    self._stdout_queue: queue.Queue[Any] = queue.Queue(
        maxsize=self._STDOUT_QUEUE_SIZE
    )
    self._reader_stop = threading.Event()
    self._reader_thread: Optional[threading.Thread] = None
    self._stderr_thread: Optional[threading.Thread] = None
    # A single stdout stream cannot safely serve multiple independent
    # readers: one request could consume and discard another request's
    # response. Serialize complete write/read exchanges (and notification
    # writes) so response correlation remains lossless.
    self._request_lock = threading.Lock()
    self._start()

Functions

send

send(request: MCPRequest) -> MCPResponse

Write request as JSON line, read lines until the matching response.

MCP servers may emit unsolicited notifications or stray/stale replies on stdout before the real response. Skip anything that isn't a well-formed JSON-RPC response carrying this request's id, rather than treating the first line as gospel (#751).

Source code in src/openjarvis/mcp/transport.py

def send(self, request: MCPRequest) -> MCPResponse:
    """Write request as JSON line, read lines until the matching response.

    MCP servers may emit unsolicited notifications or stray/stale
    replies on stdout before the real response. Skip anything that
    isn't a well-formed JSON-RPC response carrying this request's id,
    rather than treating the first line as gospel (#751).
    """
    with self._request_lock:
        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()

        deadline = time.monotonic() + self._response_timeout
        while True:
            # Check the wall-clock deadline even when a server continuously
            # floods stdout, so queued blank/noise/notification lines cannot
            # keep extending the request forever.
            response_line = self._next_stdout_line(deadline, request.id)
            response_line = response_line.strip()
            if not response_line:
                continue

            try:
                parsed = json.loads(response_line)
            except (json.JSONDecodeError, ValueError):
                parsed = None

            is_response = (
                isinstance(parsed, dict)
                and "id" in parsed
                and ("result" in parsed or "error" in parsed)
                and parsed["id"] == request.id
            )
            if is_response:
                return MCPResponse.from_json(response_line)

send_notification

send_notification(request: MCPRequest) -> None

Send a JSON-RPC notification — write only, never read.

Overrides the base implementation: stdio servers do not reply to notifications, so the default send() would block forever on proc.stdout.readline().

Source code in src/openjarvis/mcp/transport.py

def send_notification(self, request: MCPRequest) -> None:
    """Send a JSON-RPC notification — write only, never read.

    Overrides the base implementation: stdio servers do not reply
    to notifications, so the default ``send()`` would block forever
    on ``proc.stdout.readline()``.
    """
    with self._request_lock:
        proc = self._process
        if proc is None or proc.stdin is None:
            raise RuntimeError("Transport process is not running")
        line = request.to_json() + "\n"
        proc.stdin.write(line)
        proc.stdin.flush()

close

close() -> None

Terminate the subprocess.

Source code in src/openjarvis/mcp/transport.py

def close(self) -> None:
    """Terminate the subprocess."""
    if self._process is not None:
        self._reader_stop.set()
        # Wake a request waiting on the queue before stopping the process.
        try:
            self._stdout_queue.put_nowait(self._STDOUT_EOF)
        except queue.Full:
            try:
                self._stdout_queue.get_nowait()
                self._stdout_queue.put_nowait(self._STDOUT_EOF)
            except (queue.Empty, queue.Full):
                pass
        self._process.terminate()
        try:
            self._process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._process.kill()
            self._process.wait(timeout=5)
        if self._reader_thread is not None:
            self._reader_thread.join(timeout=1)
            self._reader_thread = None
        self._process = None
    if self._stderr_thread is not None:
        self._stderr_thread.join(timeout=5)
        self._stderr_thread = None

StreamableHTTPTransport

StreamableHTTPTransport(url: str, *, token: Optional[str] = None, connect_timeout: float = 10.0, request_timeout: float = 60.0)

Bases: MCPTransport

MCP Streamable HTTP transport (JSON-RPC over HTTP).

Uses a persistent httpx.Client session, tracks the Mcp-Session-Id header, and sends the Accept header required by the MCP Streamable HTTP specification.

Source code in src/openjarvis/mcp/transport.py

def __init__(
    self,
    url: str,
    *,
    token: Optional[str] = None,
    connect_timeout: float = 10.0,
    request_timeout: float = 60.0,
) -> None:
    import httpx

    self._url = url
    self._token = token
    self._session_id: Optional[str] = None
    self._client = httpx.Client(
        timeout=httpx.Timeout(
            connect=connect_timeout,
            read=request_timeout,
            write=request_timeout,
            pool=connect_timeout,
        ),
    )

Functions

send

send(request: MCPRequest) -> MCPResponse

Send request via HTTP POST following the MCP Streamable HTTP spec.

Handles both application/json and text/event-stream responses as allowed by the MCP Streamable HTTP specification.

Source code in src/openjarvis/mcp/transport.py

def send(self, request: MCPRequest) -> MCPResponse:
    """Send request via HTTP POST following the MCP Streamable HTTP spec.

    Handles both ``application/json`` and ``text/event-stream`` responses
    as allowed by the MCP Streamable HTTP specification.
    """
    response = self._post(request)
    content_type = response.headers.get("content-type", "")
    body = response.text
    if "text/event-stream" in content_type or body.lstrip().startswith("event:"):
        body = self._extract_json_from_sse(body)
    return MCPResponse.from_json(body)

send_notification

send_notification(request: MCPRequest) -> None

Send a notification — accept any 2xx, don't parse the body.

Source code in src/openjarvis/mcp/transport.py

def send_notification(self, request: MCPRequest) -> None:
    """Send a notification — accept any 2xx, don't parse the body."""
    # Track session id but don't try to parse a JSON-RPC response.
    # Servers may return 202 Accepted with an empty body.
    self._post(request)

close

close() -> None

Close the underlying httpx client.

Source code in src/openjarvis/mcp/transport.py

def close(self) -> None:
    """Close the underlying httpx client."""
    self._client.close()