solidworks_mcp.setup.handshake

solidworks_mcp.setup.handshake

Start the server the way a client will and confirm it lists its tools.

This is the MCP stdio handshake a client performs: initialize, the initialized notification, then tools/list. Setup runs it before writing any config, so a broken launcher is caught here rather than as a silent failure inside the client.

Attributes

_INITIALIZE module-attribute

_INITIALIZE = {'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2025-06-18', 'capabilities': {}, 'clientInfo': {'name': 'solidworks-mcp-setup', 'version': '1'}}}

_INITIALIZED module-attribute

_INITIALIZED = {'jsonrpc': '2.0', 'method': 'notifications/initialized'}

_LIST_TOOLS module-attribute

_LIST_TOOLS = {'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list'}

_Lines module-attribute

_Lines = queue.Queue[bytes | None]

Classes

HandshakeResult dataclass

HandshakeResult(ok: bool, tools: int, message: str)

The outcome of the start-up check.

Attributes:

Name	Type	Description
ok	bool	Whether the server answered and listed at least one tool.
tools	int	How many tools it listed.
message	str	One line for the user.

Functions:

_converse

_converse(process: Popen[bytes], deadline: float) -> HandshakeResult

Send the handshake and read the two replies.

Parameters:

Name	Type	Description	Default
process	Popen[bytes]	The started server, with piped stdin and stdout.	required
deadline	float	time.monotonic() value to give up at.	required

Returns:

Type	Description
HandshakeResult	The check's outcome.

Source code in src/solidworks_mcp/setup/handshake.py

def _converse(process: subprocess.Popen[bytes], deadline: float) -> HandshakeResult:
    """Send the handshake and read the two replies.

    Args:
        process: The started server, with piped stdin and stdout.
        deadline: ``time.monotonic()`` value to give up at.

    Returns:
        The check's outcome.
    """
    lines = _line_reader(process.stdout)
    _send(process, _INITIALIZE)
    if _response(lines, 1, deadline) is None:
        return _failed(process, "did not answer the MCP initialize request")
    _send(process, _INITIALIZED)
    _send(process, _LIST_TOOLS)
    reply = _response(lines, 2, deadline)
    if reply is None:
        return _failed(process, "did not answer the tools/list request")
    tools = reply.get("result", {}).get("tools")
    if not isinstance(tools, list) or not tools:
        return HandshakeResult(False, 0, "The server answered but listed no tools.")
    return HandshakeResult(
        True, len(tools), f"The server started and listed {len(tools)} tools."
    )

_failed

_failed(process: Popen[bytes], what: str) -> HandshakeResult

Describe a missing reply, with the exit code if the server has exited.

Parameters:

Name	Type	Description	Default
process	Popen[bytes]	The server process.	required
what	str	What it failed to do.	required

Returns:

Type	Description
HandshakeResult	A failed result.

Source code in src/solidworks_mcp/setup/handshake.py

def _failed(process: subprocess.Popen[bytes], what: str) -> HandshakeResult:
    """Describe a missing reply, with the exit code if the server has exited.

    Args:
        process: The server process.
        what: What it failed to do.

    Returns:
        A failed result.
    """
    # stdout can close a moment before the process has finished exiting, so
    # give it a second to report its exit code.
    try:
        code: int | None = process.wait(timeout=1)
    except subprocess.TimeoutExpired:
        code = None
    exited = f" (it exited with code {code})" if code is not None else ""
    return HandshakeResult(False, 0, f"The server {what}{exited}.")

_line_reader

_line_reader(stream: IO[bytes] | None) -> _Lines

Read stream line by line on a daemon thread.

A thread lets the reader wait with a timeout: readline on a pipe blocks with no timeout of its own on Windows.

Parameters:

Name	Type	Description	Default
stream	IO[bytes] | None	The server's stdout.	required

Returns:

Type	Description
_Lines	A queue of lines, ending with None when the stream closes.

Source code in src/solidworks_mcp/setup/handshake.py

def _line_reader(stream: IO[bytes] | None) -> _Lines:
    """Read ``stream`` line by line on a daemon thread.

    A thread lets the reader wait with a timeout: ``readline`` on a pipe
    blocks with no timeout of its own on Windows.

    Args:
        stream: The server's stdout.

    Returns:
        A queue of lines, ending with ``None`` when the stream closes.
    """
    lines: _Lines = queue.Queue()

    def pump() -> None:
        """Move each line into the queue, then the end marker."""
        assert stream is not None
        for line in iter(stream.readline, b""):
            lines.put(line)
        lines.put(None)

    threading.Thread(target=pump, daemon=True).start()
    return lines

_response

_response(lines: _Lines, request_id: int, deadline: float) -> dict[str, Any] | None

Wait for the JSON-RPC reply to request_id.

Lines that aren't JSON (a stray banner) and other messages (log notifications) are skipped.

Parameters:

Name	Type	Description	Default
lines	_Lines	The server's stdout lines.	required
request_id	int	The id of the request being answered.	required
deadline	float	time.monotonic() value to give up at.	required

Returns:

Type	Description
dict[str, Any] | None	The reply, or None on timeout or when the stream ends.

Source code in src/solidworks_mcp/setup/handshake.py

def _response(lines: _Lines, request_id: int, deadline: float) -> dict[str, Any] | None:
    """Wait for the JSON-RPC reply to ``request_id``.

    Lines that aren't JSON (a stray banner) and other messages (log
    notifications) are skipped.

    Args:
        lines: The server's stdout lines.
        request_id: The ``id`` of the request being answered.
        deadline: ``time.monotonic()`` value to give up at.

    Returns:
        The reply, or ``None`` on timeout or when the stream ends.
    """
    while (remaining := deadline - time.monotonic()) > 0:
        try:
            line = lines.get(timeout=remaining)
        except queue.Empty:
            return None
        if line is None:
            return None
        try:
            message = json.loads(line)
        except ValueError:
            continue
        if isinstance(message, dict) and message.get("id") == request_id:
            return message
    return None

_send

_send(process: Popen[bytes], message: dict[str, Any]) -> None

Write one JSON-RPC message as a line on the server's stdin.

Parameters:

Name	Type	Description	Default
process	Popen[bytes]	The server process.	required
message	dict[str, Any]	The JSON-RPC message.	required

Source code in src/solidworks_mcp/setup/handshake.py

def _send(process: subprocess.Popen[bytes], message: dict[str, Any]) -> None:
    """Write one JSON-RPC message as a line on the server's stdin.

    Args:
        process: The server process.
        message: The JSON-RPC message.
    """
    assert process.stdin is not None
    try:
        process.stdin.write((json.dumps(message) + "\n").encode())
        process.stdin.flush()
    except OSError:
        # The server has exited; the missing reply reports it.
        pass

check_server

check_server(command: str, args: list[str], timeout: float = 90.0) -> HandshakeResult

Start command args and run the MCP handshake over its stdio.

The process is always killed afterwards. Its stderr goes to a temporary file: an unread pipe can fill up and stall the server.

Parameters:

Name	Type	Description	Default
command	str	The launcher to start.	required
args	list[str]	Its arguments (setup passes --mock).	required
timeout	float	Seconds to wait for both replies, in total.	90.0

Returns:

Type	Description
HandshakeResult	Whether it worked, the tool count and a message.

Source code in src/solidworks_mcp/setup/handshake.py

def check_server(
    command: str, args: list[str], timeout: float = 90.0
) -> HandshakeResult:
    """Start ``command args`` and run the MCP handshake over its stdio.

    The process is always killed afterwards. Its stderr goes to a temporary
    file: an unread pipe can fill up and stall the server.

    Args:
        command: The launcher to start.
        args: Its arguments (setup passes ``--mock``).
        timeout: Seconds to wait for both replies, in total.

    Returns:
        Whether it worked, the tool count and a message.
    """
    with tempfile.TemporaryFile() as stderr:
        try:
            process = subprocess.Popen(
                [command, *args],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=stderr,
            )
        except OSError as exc:
            return HandshakeResult(
                False, 0, f"Could not start {command}: {exc.strerror or exc}."
            )
        try:
            return _converse(process, time.monotonic() + timeout)
        finally:
            process.kill()
            process.wait()