Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,6 @@ ignore = [

# Backwards-compatibility suppressions for existing code
"A001", # Variable shadows built-in
"A002", # Argument shadows built-in
"A004", # Import shadows built-in
"FBT001", # Boolean positional arg
"FBT002", # Boolean default value
Expand Down
60 changes: 30 additions & 30 deletions src/seclab_taskflow_agent/mcp_servers/codeql/jsonrpyc/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,18 +75,18 @@ class Spec:
"""

@classmethod
def check_id(cls, id: str | int | None, *, allow_empty: bool = False) -> None:
def check_id(cls, request_id: str | int | None, *, allow_empty: bool = False) -> None:
"""
Value check for *id* entries. When *allow_empty* is *True*, *id* is allowed to be *None*.
Raises a *TypeError* when *id* is neither an integer nor a string.
Value check for *request_id* entries. When *allow_empty* is *True*, *request_id* is allowed to be *None*.
Raises a *TypeError* when *request_id* is neither an integer nor a string.

:param id: The id to check.
:param allow_empty: Whether *id* is allowed to be *None*.
:param request_id: The id to check.
:param allow_empty: Whether *request_id* is allowed to be *None*.
:return: None.
:raises TypeError: When *id* is invalid.
:raises TypeError: When *request_id* is invalid.
"""
if (id is not None or not allow_empty) and not isinstance(id, (int, str)):
raise TypeError(f"id must be an integer or string, got {id} ({type(id)})")
if (request_id is not None or not allow_empty) and not isinstance(request_id, (int, str)):
raise TypeError(f"id must be an integer or string, got {request_id} ({type(request_id)})")

@classmethod
def check_method(cls, method: str, /) -> None:
Expand Down Expand Up @@ -120,37 +120,37 @@ def request(
cls,
method: str,
/,
id: str | int | None = None,
request_id: str | int | None = None,
*,
params: dict[str, Any] | None = None,
) -> str:
"""
Creates the string representation of a request that calls *method* with optional *params*
which are encoded by ``json.dumps``. When *id* is *None*, the request is considered a
which are encoded by ``json.dumps``. When *request_id* is *None*, the request is considered a
notification.

:param method: The method to call.
:param id: The id of the request.
:param request_id: The id of the request.
:param params: The parameters of the request.
:return: The request string.
:raises RPCInvalidRequest: When *method* or *id* are invalid.
:raises RPCInvalidRequest: When *method* or *request_id* are invalid.
:raises RPCParseError: When *params* could not be encoded.
"""
try:
cls.check_method(method)
cls.check_id(id, allow_empty=True)
cls.check_id(request_id, allow_empty=True)
except Exception as e:
raise RPCInvalidRequest(str(e))

# start building the request string
req = f'{{"jsonrpc":"2.0","method":"{method}"'

# add the id when given
if id is not None:
if request_id is not None:
# encode string ids
if isinstance(id, str):
id = json.dumps(id)
req += f',"id":{id}'
if isinstance(request_id, str):
request_id = json.dumps(request_id)
req += f',"id":{request_id}'

# add parameters when given
if params is not None:
Expand All @@ -165,29 +165,29 @@ def request(
return req

@classmethod
def response(cls, id: str | int | None, result: Any, /) -> str:
def response(cls, request_id: str | int | None, result: Any, /) -> str:
"""
Creates the string representation of a respone that was triggered by a request with *id*.
A *result* is required, even if it is *None*.

:param id: The id of the request that triggered this response.
:param request_id: The id of the request that triggered this response.
:param result: The result of the request.
:return: The response string.
:raises RPCInvalidRequest: When *id* is invalid.
:raises RPCParseError: When *result* could not be encoded.
"""
try:
cls.check_id(id)
cls.check_id(request_id)
except Exception as e:
raise RPCInvalidRequest(str(e))

# encode string ids
if isinstance(id, str):
id = json.dumps(id)
if isinstance(request_id, str):
request_id = json.dumps(request_id)

# build the response string
try:
res = f'{{"jsonrpc":"2.0","id":{id},"result":{json.dumps(result)}}}'
res = f'{{"jsonrpc":"2.0","id":{request_id},"result":{json.dumps(result)}}}'
except Exception as e:
raise RPCParseError(str(e))

Expand All @@ -196,7 +196,7 @@ def response(cls, id: str | int | None, result: Any, /) -> str:
@classmethod
def error(
cls,
id: str | int | None,
request_id: str | int | None,
code: int,
*,
data: Any | None = None,
Expand All @@ -206,15 +206,15 @@ def error(
*id*. *code* must lead to a registered :py:class:`RPCError`. *data* might contain
additional, detailed error information and is encoded by ``json.dumps`` when set.

:param id: The id of the request that triggered this error.
:param request_id: The id of the request that triggered this error.
:param code: The error code.
:param data: Additional error data.
:return: The error string.
:raises RPCInvalidRequest: When *id* or *code* are invalid.
:raises RPCParseError: When *data* could not be encoded.
"""
try:
cls.check_id(id)
cls.check_id(request_id)
cls.check_code(code)
except Exception as e:
raise RPCInvalidRequest(str(e))
Expand All @@ -233,11 +233,11 @@ def error(
err_data += "}"

# encode string ids
if isinstance(id, str):
id = json.dumps(id)
if isinstance(request_id, str):
request_id = json.dumps(request_id)

# start building the error string
err = f'{{"jsonrpc":"2.0","id":{id},"error":{err_data}}}'
err = f'{{"jsonrpc":"2.0","id":{request_id},"error":{err_data}}}'

return err

Expand Down Expand Up @@ -441,7 +441,7 @@ def call(

# create the request
params = params if params else {"args": args, "kwargs": kwargs}
req = Spec.request(method, id=id, params=params)
req = Spec.request(method, request_id=id, params=params)
print(f"-> {req}")
msg = f"Content-Length: {len(req)}\r\n\r\n{req}"
self._write(msg)
Expand Down
Loading