RDCIST-3853: Solve pyright issues

Author: OS-anasantos

scripts/update_readme_snippets.py

@@ -128,14 +128,14 @@ def update_readme_snippets(readme_path: Path = Path("README.md"), check_mode: bo
             )
             return False
         else:
-            print(f"✓ {readme_path} code snippets are up to date")
+            print(f"[OK] {readme_path} code snippets are up to date")
             return True
     else:
         if updated_content != original_content:
             readme_path.write_text(updated_content)
-            print(f"✓ Updated {readme_path}")
+            print(f"[OK] Updated {readme_path}")
         else:
-            print(f"✓ {readme_path} already up to date")
+            print(f"[OK] {readme_path} already up to date")
         return True
 
 

src/mcp/os/posix/utilities.py

@@ -29,20 +29,20 @@ async def terminate_posix_process_tree(process: Process, timeout_seconds: float
         return
 
     try:
-        pgid = os.getpgid(pid)
-        os.killpg(pgid, signal.SIGTERM)
+        pgid = os.getpgid(pid)  # type: ignore[attr-defined]
+        os.killpg(pgid, signal.SIGTERM)  # type: ignore[attr-defined]
 
         with anyio.move_on_after(timeout_seconds):
             while True:
                 try:
                     # Check if process group still exists (signal 0 = check only)
-                    os.killpg(pgid, 0)
+                    os.killpg(pgid, 0)  # type: ignore[attr-defined]
                     await anyio.sleep(0.1)
                 except ProcessLookupError:
                     return
 
         try:
-            os.killpg(pgid, signal.SIGKILL)
+            os.killpg(pgid, signal.SIGKILL)  # type: ignore[attr-defined]
         except ProcessLookupError:
             pass
 

src/mcp/os/win32/utilities.py

@@ -19,7 +19,6 @@
 
 # Windows-specific imports for Job Objects
 if sys.platform == "win32":
-    import pywintypes
     import win32api
     import win32con
     import win32job
@@ -28,7 +27,6 @@
     win32api = None
     win32con = None
     win32job = None
-    pywintypes = None
 
 JobHandle = int
 
@@ -127,6 +125,11 @@ def pid(self) -> int:
         """Return the process ID."""
         return self.popen.pid
 
+    @property
+    def returncode(self) -> int | None:
+        """Return the process exit code (None if still running)."""
+        return self.popen.returncode
+
 
 # ------------------------
 # Updated function
@@ -238,11 +241,17 @@ def _create_job_object() -> int | None:
         return None
 
     try:
-        job = win32job.CreateJobObject(None, "")
-        extended_info = win32job.QueryInformationJobObject(job, win32job.JobObjectExtendedLimitInformation)
+        job: JobHandle = win32job.CreateJobObject(None, "")  # type: ignore[arg-type]
+        extended_info = win32job.QueryInformationJobObject(  # type: ignore[misc]
+            job, win32job.JobObjectExtendedLimitInformation
+        )
 
         extended_info["BasicLimitInformation"]["LimitFlags"] |= win32job.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
-        win32job.SetInformationJobObject(job, win32job.JobObjectExtendedLimitInformation, extended_info)
+        win32job.SetInformationJobObject(  # pyright: ignore[reportUnknownMemberType]
+            job,
+            win32job.JobObjectExtendedLimitInformation,
+            extended_info,  # pyright: ignore[reportUnknownArgumentType]
+        )
         return job
     except Exception as e:
         logger.warning(f"Failed to create Job Object for process tree management: {e}")
@@ -269,7 +278,7 @@ def _maybe_assign_process_to_job(process: Process | FallbackProcess, job: JobHan
 
         try:
             win32job.AssignProcessToJobObject(job, process_handle)
-            process._job_object = job
+            process._job_object = job  # type: ignore[attr-defined]
         finally:
             win32api.CloseHandle(process_handle)
     except Exception as e:
@@ -295,7 +304,7 @@ async def terminate_windows_process_tree(process: Process | FallbackProcess, tim
     job = getattr(process, "_job_object", None)
     if job and win32job:
         try:
-            win32job.TerminateJobObject(job, 1)
+            win32job.TerminateJobObject(job, 1)  # type: ignore[misc]
         except Exception:
             # Job might already be terminated
             pass

tests/issues/test_1027_win_unreachable_cleanup.py

@@ -213,8 +213,10 @@ def echo(text: str) -> str:
                 await anyio.sleep(0.1)
 
         # Check if process is still running
-        if hasattr(process, "returncode") and process.returncode is not None:  # pragma: no cover
-            pytest.fail(f"Server process exited with code {process.returncode}")
+        # fmt: off
+        if hasattr(process, "returncode") and process.returncode is not None:  # pyright: ignore[reportUnknownMemberType,reportAttributeAccessIssue]  # pragma: no cover
+            pytest.fail(f"Server process exited with code {process.returncode}")  # pyright: ignore[reportUnknownMemberType,reportAttributeAccessIssue]
+        # fmt: on
 
         assert Path(startup_marker).exists(), "Server startup marker not created"