Skip to content

woodglue.client

Async JSON-RPC 2.0 client with typed responses.

Async JSON-RPC 2.0 client for woodglue servers.

WoodglueClient makes typed RPC calls, optionally resolving return types from x-global-ref in the OpenAPI spec. Uses Tornado's AsyncHTTPClient.

WoodglueRpcError

Bases: Exception

Raised when the server returns a JSON-RPC error response.

Source code in src/woodglue/client.py 
19
20
21
22
23
24
25
class WoodglueRpcError(Exception):
    """Raised when the server returns a JSON-RPC error response."""

    def __init__(self, code: int, message: str):
        self.code: int = code
        self.message: str = message
        super().__init__(f"JSON-RPC error {code}: {message}")

WoodglueClient

Async client for woodglue JSON-RPC servers.

Optionally loads the OpenAPI spec to auto-resolve return types from x-global-ref vendor extensions.

Resolution priority on call(): 1. Explicit return_type parameter 2. resolver callable (receives the x-global-ref string) 3. Type from load_spec() if loaded 4. Return raw dict/primitive

Source code in src/woodglue/client.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class WoodglueClient:
    """
    Async client for woodglue JSON-RPC servers.

    Optionally loads the OpenAPI spec to auto-resolve return types from
    `x-global-ref` vendor extensions.

    Resolution priority on `call()`:
    1. Explicit `return_type` parameter
    2. `resolver` callable (receives the `x-global-ref` string)
    3. Type from `load_spec()` if loaded
    4. Return raw dict/primitive
    """

    def __init__(self, base_url: str, *, token: str | None = None, data_dir: Path | None = None):
        self._base_url: str = base_url.rstrip("/")
        self._http: AsyncHTTPClient = AsyncHTTPClient()
        self._request_id: int = 0
        self._return_types: dict[str, type[BaseModel]] = {}
        self._return_grefs: dict[str, str] = {}
        if token is not None:
            self._token: str | None = token
        elif data_dir is not None:
            from woodglue.token_store import get_single_token

            auth_db = data_dir / "auth.db"
            self._token = get_single_token(auth_db) if auth_db.exists() else None
        else:
            self._token = None

    async def load_spec(self, strict: bool = False) -> None:
        """
        Fetch `/docs/openapi.json` and resolve `x-global-ref` types.

        With `strict=True`, raises `ImportError` if any gref cannot be
        resolved. With `strict=False`, skips unresolvable grefs.
        """
        from lythonic import GlobalRef

        headers: dict[str, str] = {}
        if self._token:
            headers["Authorization"] = f"Bearer {self._token}"
        resp = await self._http.fetch(f"{self._base_url}/docs/openapi.json", headers=headers)
        spec = json.loads(resp.body)

        for _path, path_item in spec.get("paths", {}).items():
            for _http_method, operation in path_item.items():
                op_id = operation.get("operationId")
                if not op_id:
                    continue

                resp_content = (
                    operation.get("responses", {})
                    .get("200", {})
                    .get("content", {})
                    .get("application/json", {})
                )
                schema = resp_content.get("schema", {})
                gref_str = schema.get("x-global-ref")
                if not gref_str:
                    continue

                self._return_grefs[op_id] = gref_str

                try:
                    gref = GlobalRef(gref_str)
                    cls = gref.get_instance()
                    if isinstance(cls, type) and issubclass(cls, BaseModel):
                        self._return_types[op_id] = cls
                except Exception as exc:
                    if strict:
                        raise ImportError(
                            f"Cannot resolve x-global-ref '{gref_str}' for method '{op_id}'"
                        ) from exc

    async def call(
        self,
        method: str,
        *,
        return_type: type[BaseModel] | None = None,
        resolver: Callable[[str], type[BaseModel] | None] | None = None,
        **kwargs: Any,
    ) -> Any:
        """
        Call a JSON-RPC method and return the deserialized result.

        `kwargs` are sent as the JSON-RPC `params` object. BaseModel
        values in kwargs are serialized via `model_dump(mode="json")`.
        """
        self._request_id += 1

        params: dict[str, Any] = {}
        for key, value in kwargs.items():
            if isinstance(value, BaseModel):
                params[key] = value.model_dump(mode="json")
            else:
                params[key] = value

        body = json.dumps(
            {
                "jsonrpc": "2.0",
                "method": method,
                "params": params,
                "id": self._request_id,
            }
        )

        headers: dict[str, str] = {"Content-Type": "application/json"}
        if self._token:
            headers["Authorization"] = f"Bearer {self._token}"

        req = HTTPRequest(
            f"{self._base_url}/rpc",
            method="POST",
            body=body,
            headers=headers,
        )
        resp = await self._http.fetch(req)
        data = json.loads(resp.body)

        if "error" in data:
            err = data["error"]
            raise WoodglueRpcError(err["code"], err["message"])

        result = data.get("result")

        resolved_type = return_type
        if resolved_type is None and resolver is not None:
            gref_str = self._return_grefs.get(method)
            if gref_str:
                resolved_type = resolver(gref_str)
        if resolved_type is None:
            resolved_type = self._return_types.get(method)

        if resolved_type is not None and isinstance(result, dict):
            return resolved_type.model_validate(result)
        return result

load_spec(strict=False) async

Fetch /docs/openapi.json and resolve x-global-ref types.

With strict=True, raises ImportError if any gref cannot be resolved. With strict=False, skips unresolvable grefs.

Source code in src/woodglue/client.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def load_spec(self, strict: bool = False) -> None:
    """
    Fetch `/docs/openapi.json` and resolve `x-global-ref` types.

    With `strict=True`, raises `ImportError` if any gref cannot be
    resolved. With `strict=False`, skips unresolvable grefs.
    """
    from lythonic import GlobalRef

    headers: dict[str, str] = {}
    if self._token:
        headers["Authorization"] = f"Bearer {self._token}"
    resp = await self._http.fetch(f"{self._base_url}/docs/openapi.json", headers=headers)
    spec = json.loads(resp.body)

    for _path, path_item in spec.get("paths", {}).items():
        for _http_method, operation in path_item.items():
            op_id = operation.get("operationId")
            if not op_id:
                continue

            resp_content = (
                operation.get("responses", {})
                .get("200", {})
                .get("content", {})
                .get("application/json", {})
            )
            schema = resp_content.get("schema", {})
            gref_str = schema.get("x-global-ref")
            if not gref_str:
                continue

            self._return_grefs[op_id] = gref_str

            try:
                gref = GlobalRef(gref_str)
                cls = gref.get_instance()
                if isinstance(cls, type) and issubclass(cls, BaseModel):
                    self._return_types[op_id] = cls
            except Exception as exc:
                if strict:
                    raise ImportError(
                        f"Cannot resolve x-global-ref '{gref_str}' for method '{op_id}'"
                    ) from exc

call(method, *, return_type=None, resolver=None, **kwargs) async

Call a JSON-RPC method and return the deserialized result.

kwargs are sent as the JSON-RPC params object. BaseModel values in kwargs are serialized via model_dump(mode="json").

Source code in src/woodglue/client.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
async def call(
    self,
    method: str,
    *,
    return_type: type[BaseModel] | None = None,
    resolver: Callable[[str], type[BaseModel] | None] | None = None,
    **kwargs: Any,
) -> Any:
    """
    Call a JSON-RPC method and return the deserialized result.

    `kwargs` are sent as the JSON-RPC `params` object. BaseModel
    values in kwargs are serialized via `model_dump(mode="json")`.
    """
    self._request_id += 1

    params: dict[str, Any] = {}
    for key, value in kwargs.items():
        if isinstance(value, BaseModel):
            params[key] = value.model_dump(mode="json")
        else:
            params[key] = value

    body = json.dumps(
        {
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
            "id": self._request_id,
        }
    )

    headers: dict[str, str] = {"Content-Type": "application/json"}
    if self._token:
        headers["Authorization"] = f"Bearer {self._token}"

    req = HTTPRequest(
        f"{self._base_url}/rpc",
        method="POST",
        body=body,
        headers=headers,
    )
    resp = await self._http.fetch(req)
    data = json.loads(resp.body)

    if "error" in data:
        err = data["error"]
        raise WoodglueRpcError(err["code"], err["message"])

    result = data.get("result")

    resolved_type = return_type
    if resolved_type is None and resolver is not None:
        gref_str = self._return_grefs.get(method)
        if gref_str:
            resolved_type = resolver(gref_str)
    if resolved_type is None:
        resolved_type = self._return_types.get(method)

    if resolved_type is not None and isinstance(result, dict):
        return resolved_type.model_validate(result)
    return result