(self, method, params=None, timeout=60)
| 85 | return json.loads(line.decode("utf-8", "strict")) |
| 86 | |
| 87 | def request(self, method, params=None, timeout=60): |
| 88 | self._id += 1 |
| 89 | rid = self._id |
| 90 | self._send({"jsonrpc": "2.0", "id": rid, "method": method, |
| 91 | "params": params or {}}) |
| 92 | deadline = time.time() + timeout |
| 93 | while True: |
| 94 | msg = self._read_message(timeout=max(1, deadline - time.time())) |
| 95 | if msg.get("id") == rid: |
| 96 | return msg |
| 97 | if time.time() > deadline: |
| 98 | raise McpError("timeout waiting for id=%d" % rid) |
| 99 | |
| 100 | def notify(self, method, params=None): |
| 101 | self._send({"jsonrpc": "2.0", "method": method, "params": params or {}}) |
no test coverage detected