Send a JSON-RPC request and return the result (raises MCPError on error).
(self, method: str, params: Optional[dict] = None)
| 119 | pass |
| 120 | |
| 121 | def request(self, method: str, params: Optional[dict] = None) -> Any: |
| 122 | """Send a JSON-RPC request and return the result (raises MCPError on error).""" |
| 123 | req_id = self._next_id() |
| 124 | self._send( |
| 125 | {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params or {}} |
| 126 | ) |
| 127 | deadline = time.time() + self.timeout |
| 128 | while True: |
| 129 | response = self._recv(deadline) |
| 130 | if response.get("id") == req_id: |
| 131 | if "error" in response: |
| 132 | err = response["error"] |
| 133 | raise MCPError( |
| 134 | err.get("code", -1), err.get("message", ""), err.get("data") |
| 135 | ) |
| 136 | return response.get("result") |
| 137 | |
| 138 | def notify(self, method: str, params: Optional[dict] = None) -> None: |
| 139 | """Send a JSON-RPC notification (no response expected).""" |
no test coverage detected