Skip to content

client

client

MCP Client — connects to MCP servers and discovers/calls tools.

Classes

MCPClient

MCPClient(transport: MCPTransport)

Client that communicates with an MCP server via a transport.

PARAMETER DESCRIPTION
transport

The transport layer to use for communication.

TYPE: MCPTransport

Source code in src/openjarvis/mcp/client.py
def __init__(self, transport: MCPTransport) -> None:
    self._transport = transport
    self._initialized = False
    self._capabilities: Dict[str, Any] = {}
    self._id_counter = itertools.count(1)
Functions
initialize
initialize() -> Dict[str, Any]

Perform the MCP initialize handshake.

Sends the required client info and protocol version, then confirms with a notifications/initialized notification as required by the MCP specification.

Returns the server capabilities.

Source code in src/openjarvis/mcp/client.py
def initialize(self) -> Dict[str, Any]:
    """Perform the MCP initialize handshake.

    Sends the required client info and protocol version, then
    confirms with a ``notifications/initialized`` notification
    as required by the MCP specification.

    Returns the server capabilities.
    """
    params = {
        "protocolVersion": "2025-03-26",
        "capabilities": {},
        "clientInfo": {"name": "openjarvis", "version": "0.1.0"},
    }
    response = self._send("initialize", params)
    self._initialized = True
    self._capabilities = response.result.get("capabilities", {})
    # Send the required initialized notification per MCP spec
    self.notify("notifications/initialized")
    return response.result
notify
notify(method: str, params: Dict[str, Any] | None = None) -> None

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

Per JSON-RPC 2.0 spec, notifications omit the id field entirely.

Source code in src/openjarvis/mcp/client.py
def notify(self, method: str, params: Dict[str, Any] | None = None) -> None:
    """Send a JSON-RPC notification (no response expected).

    Per JSON-RPC 2.0 spec, notifications omit the ``id`` field entirely.
    """
    request = MCPRequest(
        method=method,
        params=params or {},
        id=None,  # None → no id field in JSON (notification)
    )
    self._transport.send_notification(request)
list_tools
list_tools() -> List[ToolSpec]

Discover available tools from the server.

Returns a list of ToolSpec objects.

Source code in src/openjarvis/mcp/client.py
def list_tools(self) -> List[ToolSpec]:
    """Discover available tools from the server.

    Returns a list of ``ToolSpec`` objects.
    """
    response = self._send("tools/list")
    tools = response.result.get("tools", [])
    return [
        ToolSpec(
            name=t["name"],
            description=t.get("description", ""),
            parameters=t.get("inputSchema", {}),
        )
        for t in tools
    ]
call_tool
call_tool(name: str, arguments: Dict[str, Any] | None = None) -> Dict[str, Any]

Call a tool on the server.

Returns the result dictionary with content and isError fields.

Source code in src/openjarvis/mcp/client.py
def call_tool(
    self,
    name: str,
    arguments: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    """Call a tool on the server.

    Returns the result dictionary with ``content`` and ``isError`` fields.
    """
    response = self._send(
        "tools/call",
        {"name": name, "arguments": arguments or {}},
    )
    return response.result
close
close() -> None

Close the transport connection.

Source code in src/openjarvis/mcp/client.py
def close(self) -> None:
    """Close the transport connection."""
    self._transport.close()