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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, transport: MCPTransport) -> None:
    self._transport = transport
    self._initialized = False
    self._capabilities: Dict[str, Any] = {}
    self._id_counter = itertools.count(1)
    # A client may be shared by server, scheduled, and channel agents.
    # Keep each transport request/response exchange atomic so stdio
    # readers cannot consume another thread's JSON-RPC response.
    self._request_lock = threading.RLock()
    # Closing must not wait for ``_request_lock``: transport.close() is
    # what interrupts a request that is blocked in a transport read.
    # An event lets queued requests fail before touching that transport,
    # while this separate lock keeps close itself idempotent.
    self._closed = threading.Event()
    self._transport_closed = threading.Event()
    self._close_lock = threading.Lock()
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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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 
86
87
88
89
90
91
92
93
94
95
96
97
98
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)
    )
    with self._request_lock:
        self._raise_if_closed()
        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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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", [])
    # MCP tools often wrap long-running pentest/scan commands. The default
    # ToolSpec timeout (30s) kills them mid-scan. Bump to 600s — individual
    # MCP servers can shorten via their own protocol if needed.
    return [
        ToolSpec(
            name=t["name"],
            description=t.get("description", ""),
            parameters=t.get("inputSchema", {}),
            timeout_seconds=600.0,
        )
        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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 
135
136
137
138
139
140
141
142
143
144
145
def close(self) -> None:
    """Close the transport connection."""
    # Do not acquire _request_lock here. A transport request can be stuck
    # waiting for a server response, and closing the underlying transport
    # is the mechanism that unblocks it.
    with self._close_lock:
        if self._transport_closed.is_set():
            return
        self._closed.set()
        self._transport.close()
        self._transport_closed.set()