Fix EM102 linter errors: assign f-strings to variables before raising exceptions

Agent-Logs-Url: https://github.com/GitHubSecurityLab/seclab-taskflow-agent/sessions/ccb4b71f-fdce-4a18-bfc2-ef68f24b8187

Co-authored-by: kevinbackhouse <4358136+kevinbackhouse@users.noreply.github.com>

Author: Copilot

pyproject.toml

@@ -171,7 +171,6 @@ target-version = "py310"
 ignore = [
   # Style choices — deliberate project conventions
   "EM101",    # Exception string literals
-  "EM102",    # Exception f-strings
   "G004",     # Logging f-strings
   "T201",     # print() used for user output
   "TRY003",   # Raise with inline message strings

src/seclab_taskflow_agent/agent.py

@@ -172,7 +172,8 @@ def __init__(
         if token:
             resolved_token = os.getenv(token, "")
             if not resolved_token:
-                raise RuntimeError(f"Token env var {token!r} is not set")
+                msg = f"Token env var {token!r} is not set"
+                raise RuntimeError(msg)
         else:
             resolved_token = get_AI_token()
 

src/seclab_taskflow_agent/available_tools.py

@@ -108,18 +108,18 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
         # Resolve package and filename from dotted path
         components = toolname.rsplit(".", 1)
         if len(components) != 2:
-            raise BadToolNameError(
+            msg = (
                 f'Not a valid toolname: "{toolname}". '
                 f'Expected format: "packagename.filename"'
             )
+            raise BadToolNameError(msg)
         package, filename = components
 
         try:
             pkg_dir = importlib.resources.files(package)
             if not pkg_dir.is_dir():
-                raise BadToolNameError(
-                    f"Cannot load {toolname} because {pkg_dir} is not a valid directory."
-                )
+                msg = f"Cannot load {toolname} because {pkg_dir} is not a valid directory."
+                raise BadToolNameError(msg)
             filepath = pkg_dir.joinpath(filename + ".yaml")
             with filepath.open() as fh:
                 raw = yaml.safe_load(fh)
@@ -128,17 +128,17 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
             header = raw.get("seclab-taskflow-agent", {})
             filetype = header.get("filetype", "")
             if filetype != tooltype.value:
-                raise FileTypeException(
+                msg = (
                     f"Error in {filepath}: expected filetype {tooltype.value!r}, "
                     f"got {filetype!r}."
                 )
+                raise FileTypeException(msg)
 
             # Parse into the appropriate Pydantic model
             model_cls = DOCUMENT_MODELS.get(filetype)
             if model_cls is None:
-                raise BadToolNameError(
-                    f"Unknown filetype {filetype!r} in {toolname}"
-                )
+                msg = f"Unknown filetype {filetype!r} in {toolname}"
+                raise BadToolNameError(msg)
 
             try:
                 doc = model_cls(**raw)
@@ -147,9 +147,8 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
                 for err in exc.errors():
                     if "Unsupported version" in str(err.get("msg", "")):
                         raise VersionException(str(err["msg"])) from exc
-                raise BadToolNameError(
-                    f"Validation error loading {toolname}: {exc}"
-                ) from exc
+                msg = f"Validation error loading {toolname}: {exc}"
+                raise BadToolNameError(msg) from exc
 
             # Cache and return
             if tooltype not in self._cache:
@@ -158,10 +157,11 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
             return doc
 
         except ModuleNotFoundError as exc:
-            raise BadToolNameError(f"Cannot load {toolname}: {exc}") from exc
+            msg = f"Cannot load {toolname}: {exc}"
+            raise BadToolNameError(msg) from exc
         except FileNotFoundError:
-            raise BadToolNameError(
-                f"Cannot load {toolname} because {filepath} is not a valid file."
-            )
+            msg = f"Cannot load {toolname} because {filepath} is not a valid file."
+            raise BadToolNameError(msg)
         except ValueError as exc:
-            raise BadToolNameError(f"Cannot load {toolname}: {exc}") from exc
+            msg = f"Cannot load {toolname}: {exc}"
+            raise BadToolNameError(msg) from exc

src/seclab_taskflow_agent/cli.py

@@ -36,7 +36,8 @@
 def _parse_global(value: str) -> tuple[str, str]:
     """Parse a ``KEY=VALUE`` string into a (key, value) pair."""
     if "=" not in value:
-        raise typer.BadParameter(f"Invalid global variable format: {value!r}. Expected KEY=VALUE.")
+        msg = f"Invalid global variable format: {value!r}. Expected KEY=VALUE."
+        raise typer.BadParameter(msg)
     key, _, val = value.partition("=")
     return key.strip(), val.strip()
 

src/seclab_taskflow_agent/mcp_lifecycle.py

@@ -116,7 +116,8 @@ def _print_err(line: str) -> None:
                     client_session_timeout_seconds=client_session_timeout,
                 )
             case _:
-                raise ValueError(f"Unsupported MCP transport: {params['kind']}")
+                msg = f"Unsupported MCP transport: {params['kind']}"
+                raise ValueError(msg)
 
         entries.append(MCPServerEntry(MCPNamespaceWrap(confirms, mcp_server), server_proc, name=tb))
 

src/seclab_taskflow_agent/mcp_servers/codeql/client.py

@@ -467,7 +467,8 @@ def _file_uri_to_path(uri):
     # note: don't try to parse paths like "file://a/b" because that returns "/b", should be "file:///a/b"
     parsed = urlparse(uri)
     if parsed.scheme != "file":
-        raise ValueError(f"Not a file:// uri: {uri}")
+        msg = f"Not a file:// uri: {uri}"
+        raise ValueError(msg)
     path = unquote(parsed.path)
     region = None
     if ":" in path:
@@ -605,7 +606,8 @@ def run_query(
     if target:
         target_pos = get_query_position(query_path, target)
         if not target_pos:
-            raise ValueError(f"Could not resolve quick eval target for {target}")
+            msg = f"Could not resolve quick eval target for {target}"
+            raise ValueError(msg)
     try:
         with (
             QueryServer(database, keep_alive=keep_alive, log_stderr=log_stderr) as server,
@@ -635,5 +637,6 @@ def run_query(
                 case _:
                     raise ValueError("Unsupported output format {fmt}")
     except Exception as e:
-        raise RuntimeError(f"Error in run_query: {e}") from e
+        msg = f"Error in run_query: {e}"
+        raise RuntimeError(msg) from e
     return result

src/seclab_taskflow_agent/mcp_servers/codeql/jsonrpyc/__init__.py

@@ -86,7 +86,8 @@ def check_id(cls, id: str | int | None, *, allow_empty: bool = False) -> None:
         :raises TypeError: When *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)})")
+            msg = f"id must be an integer or string, got {id} ({type(id)})"
+            raise TypeError(msg)
 
     @classmethod
     def check_method(cls, method: str, /) -> None:
@@ -98,7 +99,8 @@ def check_method(cls, method: str, /) -> None:
         :raises TypeError: When *method* is invalid.
         """
         if not isinstance(method, str):
-            raise TypeError(f"method must be a string, got {method} ({type(method)})")
+            msg = f"method must be a string, got {method} ({type(method)})"
+            raise TypeError(msg)
 
     @classmethod
     def check_code(cls, code: int, /) -> None:
@@ -113,7 +115,8 @@ def check_code(cls, code: int, /) -> None:
         try:
             get_error(code)
         except Exception:
-            raise TypeError(f"invalid error code, got {code} ({type(code)})")
+            msg = f"invalid error code, got {code} ({type(code)})"
+            raise TypeError(msg)
 
     @classmethod
     def request(
@@ -502,7 +505,8 @@ def _handle_method(self, req: dict[str, Any]) -> None:
         try:
             method = req["method"]
             if method not in self.method_handlers:
-                raise ValueError(f"No handler defined for method: {method}")
+                msg = f"No handler defined for method: {method}"
+                raise ValueError(msg)
             result = self.method_handlers[method](req["params"])
             if "id" in req:
                 res = Spec.response(req["id"], result)
@@ -752,7 +756,8 @@ def run(self) -> None:
                         print(f"Incoming jsonrpc message: {decoded_msg}")
                     self.rpc._handle(decoded_msg)
                 else:
-                    raise ValueError(f"Don't know how to handle: {decoded_line}")
+                    msg = f"Don't know how to handle: {decoded_line}"
+                    raise ValueError(msg)
             else:
                 self._stopper.wait(self.interval)
 
@@ -836,13 +841,16 @@ class MyCustomRPCError(RPCError):
             title = "My custom error"
     """
     if not issubclass(cls, RPCError):
-        raise TypeError(f"'{cls}' is not a subclass of RPCError")
+        msg = f"'{cls}' is not a subclass of RPCError"
+        raise TypeError(msg)
 
     # check duplicates
     if cls.code in error_map_code:
-        raise AttributeError(f"duplicate RPC error code {cls.code}")
+        msg = f"duplicate RPC error code {cls.code}"
+        raise AttributeError(msg)
     if cls.code_range in error_map_code_range:
-        raise AttributeError(f"duplicate RPC error code range {cls.code_range}")
+        msg = f"duplicate RPC error code range {cls.code_range}"
+        raise AttributeError(msg)
 
     # register
     error_map_code[cls.code] = cls
@@ -867,7 +875,8 @@ def get_error(code: int) -> type[RPCError]:
         if lower <= code <= upper:
             return cls
 
-    raise ValueError(f"unknown error code '{code}' ({type(code)})")
+    msg = f"unknown error code '{code}' ({type(code)})"
+    raise ValueError(msg)
 
 
 @register_error

src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py

@@ -53,10 +53,12 @@
 def _resolve_query_path(language: str, query: str) -> Path:
     global TEMPLATED_QUERY_PATHS
     if language not in TEMPLATED_QUERY_PATHS:
-        raise RuntimeError(f"Error: Language `{language}` not supported!")
+        msg = f"Error: Language `{language}` not supported!"
+        raise RuntimeError(msg)
     query_path = TEMPLATED_QUERY_PATHS[language].get(query)
     if not query_path:
-        raise RuntimeError(f"Error: query `{query}` not supported for `{language}`!")
+        msg = f"Error: query `{query}` not supported for `{language}`!"
+        raise RuntimeError(msg)
     return Path(query_path)
 
 
@@ -69,7 +71,8 @@ def _resolve_db_path(relative_db_path: str | Path):
     absolute_path = CODEQL_DBS_BASE_PATH / relative_db_path
     if not absolute_path.is_dir():
         _debug_log(f"Database path not found: {absolute_path}")
-        raise RuntimeError(f"Error: Database not found at {absolute_path}!")
+        msg = f"Error: Database not found at {absolute_path}!"
+        raise RuntimeError(msg)
     return absolute_path
 
 

src/seclab_taskflow_agent/mcp_transport.py

@@ -109,7 +109,8 @@ async def async_wait_for_connection(
         host = parsed.hostname
         port = parsed.port
         if host is None or port is None:
-            raise ValueError(f"URL must include a host and port: {self.url}")
+            msg = f"URL must include a host and port: {self.url}"
+            raise ValueError(msg)
         deadline = asyncio.get_event_loop().time() + timeout
         while True:
             try:
@@ -119,7 +120,8 @@ async def async_wait_for_connection(
                 return
             except (OSError, ConnectionRefusedError):
                 if asyncio.get_event_loop().time() > deadline:
-                    raise TimeoutError(f"Could not connect to {host}:{port} after {timeout} seconds")
+                    msg = f"Could not connect to {host}:{port} after {timeout} seconds"
+                    raise TimeoutError(msg)
                 await asyncio.sleep(poll_interval)
 
     def wait_for_connection(
@@ -139,15 +141,17 @@ def wait_for_connection(
         host = parsed.hostname
         port = parsed.port
         if host is None or port is None:
-            raise ValueError(f"URL must include a host and port: {self.url}")
+            msg = f"URL must include a host and port: {self.url}"
+            raise ValueError(msg)
         deadline = time.time() + timeout
         while True:
             try:
                 with socket.create_connection((host, port), timeout=2):
                     return
             except OSError:
                 if time.time() > deadline:
-                    raise TimeoutError(f"Could not connect to {host}:{port} after {timeout} seconds")
+                    msg = f"Could not connect to {host}:{port} after {timeout} seconds"
+                    raise TimeoutError(msg)
                 time.sleep(poll_interval)
 
     def run(self) -> None:

src/seclab_taskflow_agent/mcp_utils.py

@@ -202,7 +202,8 @@ def mcp_client_params(
                     logging.debug(f"Initializing streamable toolbox: {tb}\nargs:\n{args}\nenv:\n{env}\n")
                     exe = shutil.which(sp.command)
                     if exe is None:
-                        raise FileNotFoundError(f"Could not resolve path to {sp.command}")
+                        msg = f"Could not resolve path to {sp.command}"
+                        raise FileNotFoundError(msg)
                     start_cmd = [exe]
                     if args:
                         for i, v in enumerate(args):
@@ -220,7 +221,8 @@ def mcp_client_params(
                     server_params["env"] = env
 
             case _:
-                raise ValueError(f"Unsupported MCP transport {kind}")
+                msg = f"Unsupported MCP transport {kind}"
+                raise ValueError(msg)
 
         client_params[tb] = (
             server_params,