(
process: subprocess.Popen[bytes],
responses: "queue.Queue[dict[str, Any]]",
request_id: int,
method: str,
params: dict[str, Any],
timeout: float,
accept_tool_error: bool = False,
)
| 138 | |
| 139 | |
| 140 | def request( |
| 141 | process: subprocess.Popen[bytes], |
| 142 | responses: "queue.Queue[dict[str, Any]]", |
| 143 | request_id: int, |
| 144 | method: str, |
| 145 | params: dict[str, Any], |
| 146 | timeout: float, |
| 147 | accept_tool_error: bool = False, |
| 148 | ) -> dict[str, Any]: |
| 149 | send( |
| 150 | process, |
| 151 | { |
| 152 | "jsonrpc": "2.0", |
| 153 | "id": request_id, |
| 154 | "method": method, |
| 155 | "params": params, |
| 156 | }, |
| 157 | ) |
| 158 | return wait_response(process, responses, request_id, timeout, accept_tool_error) |
| 159 | |
| 160 | |
| 161 | def grouped_search_qualified_name( |
no test coverage detected