Pin text I/O to UTF-8 and fail CI on locale-dependent reads/writes (#3296)

Co-authored-by: ShuQingDollarVoyager <57471784+ShuQingDollarVoyager@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: 3 people

.github/workflows/shared.yml

@@ -99,6 +99,9 @@ jobs:
           # tests/examples/test_stories_smoke.py is gated on this var; it spawns real
           # stdio + uvicorn subprocesses, so run it on exactly one matrix cell.
           MCP_EXAMPLES_SMOKE: ${{ matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12' && matrix.dep-resolution.name == 'locked' && '1' || '' }}
+          # PEP 597: text I/O without encoding= raises EncodingWarning, which pytest's "error"
+          # filter makes fatal on every cell. An env var (not -X) so pytest-xdist workers inherit it.
+          PYTHONWARNDEFAULTENCODING: "1"
         run: |
           uv run --frozen --no-sync coverage erase
           uv run --frozen --no-sync coverage run -m pytest -n auto

AGENTS.md

@@ -46,6 +46,11 @@
   dependencies and obscure circular-import bugs. Only exception: when a
   top-level import genuinely can't work (lazy-loading optional deps, or
   tests that re-import a module).
+- Always pass `encoding=` to text-mode `open()`, `Path.read_text()`/`write_text()`,
+  `tempfile` and `subprocess` text pipes — normally `"utf-8"`, or `"locale"` when the
+  platform encoding is genuinely intended; the default is the process locale, not UTF-8.
+  CI and `scripts/test` run pytest with `PYTHONWARNDEFAULTENCODING=1` (PEP 597), which
+  makes any omission an error under the `error` filter.
 
 ## Testing
 

docs_src/uri_templates/tutorial002.py

@@ -11,4 +11,4 @@
 @mcp.resource("manuals://{+path}")
 def read_manual(path: str) -> str:
     """A staff manual page, served from a directory on disk."""
-    return safe_join(DOCS_ROOT, path).read_text()
+    return safe_join(DOCS_ROOT, path).read_text(encoding="utf-8")

examples/clients/simple-chatbot/mcp_simple_chatbot/main.py

@@ -44,7 +44,7 @@ def load_config(file_path: str) -> dict[str, Any]:
             FileNotFoundError: If configuration file doesn't exist.
             JSONDecodeError: If configuration file is invalid JSON.
         """
-        with open(file_path, "r") as f:
+        with open(file_path, encoding="utf-8") as f:
             return json.load(f)
 
     @property

examples/stories/_harness.py

@@ -123,7 +123,7 @@ async def _self_hosted(name: str, cfg: dict[str, Any]) -> AsyncIterator[str]:
 
 def _story_cfg(name: str) -> dict[str, Any]:
     """The manifest entry for the story ``name`` with ``[defaults]`` applied."""
-    manifest: dict[str, Any] = tomllib.loads((Path(__file__).parent / "manifest.toml").read_text())
+    manifest: dict[str, Any] = tomllib.loads((Path(__file__).parent / "manifest.toml").read_text(encoding="utf-8"))
     return manifest["defaults"] | manifest["story"].get(name, {})
 
 

pyproject.toml

@@ -279,6 +279,9 @@ filterwarnings = [
     # 2026-07-28 drops ping; Client.send_ping() is advisory-deprecated and the
     # legacy interaction/transport tests still drive it.
     "ignore:ping is removed as of 2026-07-28.*:mcp.MCPDeprecationWarning",
+    # CI and scripts/test set PYTHONWARNDEFAULTENCODING=1, so "error" rejects any text I/O
+    # of ours that omits encoding=; pytest-examples' own unguarded text I/O isn't ours.
+    "ignore:'encoding' argument not specified:EncodingWarning:pytest_examples",
 ]
 
 [tool.markdown.lint]

scripts/gen_surface_types.py

@@ -124,7 +124,7 @@
 
 def load_pinned() -> list[dict[str, str]]:
     """Read `schema/PINNED.json` and verify each vendored file's sha256."""
-    entries: list[dict[str, str]] = json.loads((SCHEMA_DIR / "PINNED.json").read_text())
+    entries: list[dict[str, str]] = json.loads((SCHEMA_DIR / "PINNED.json").read_text(encoding="utf-8"))
     for entry in entries:
         path = SCHEMA_DIR / f"{entry['protocol_version']}.json"
         actual = hashlib.sha256(path.read_bytes()).hexdigest()
@@ -189,7 +189,7 @@ def run_codegen(schema_path: Path, output_path: Path) -> None:
             "--type-mappings", "byte=string", "uri=string", "uri-template=string",
             "--disable-timestamp",
         ],
-        capture_output=True, text=True,
+        capture_output=True, encoding="utf-8", errors="replace",
     )
     # fmt: on
     if result.returncode != 0:
@@ -222,16 +222,16 @@ def patch(match: re.Match[str]) -> str:
 def build(entry: dict[str, str]) -> str:
     """Generate, post-process, and format one version's surface module text."""
     version = entry["protocol_version"]
-    schema = json.loads((SCHEMA_DIR / f"{version}.json").read_text())
+    schema = json.loads((SCHEMA_DIR / f"{version}.json").read_text(encoding="utf-8"))
     patch_schema(schema, SCHEMA_PATCHES.get(version, []))
     make_server_info_opaque(schema)
 
     with tempfile.TemporaryDirectory() as tmp:
         patched = Path(tmp) / "schema.json"
-        patched.write_text(json.dumps(schema))
+        patched.write_text(json.dumps(schema), encoding="utf-8")
         raw = Path(tmp) / "raw.py"
         run_codegen(patched, raw)
-        source = raw.read_text()
+        source = raw.read_text(encoding="utf-8")
 
     source = re.sub(r"\A# generated by datamodel-codegen:\n#[^\n]*\n", "", source)
     source = re.sub(r"^class Model\(RootModel\[Any\]\):\n {4}root: Any\n+", "", source, count=1, flags=re.MULTILINE)
@@ -253,12 +253,12 @@ def build(entry: dict[str, str]) -> str:
 
     staging = TYPES_DIR / f"_staging_{version}.py"
     try:
-        staging.write_text(source)
+        staging.write_text(source, encoding="utf-8")
         subprocess.run(
             ["uv", "run", "--frozen", "ruff", "format", "--no-cache", str(staging)],
             cwd=REPO_ROOT, capture_output=True, check=True,
         )  # fmt: skip
-        return staging.read_text()
+        return staging.read_text(encoding="utf-8")
     finally:
         staging.unlink(missing_ok=True)
 
@@ -275,10 +275,10 @@ def main(argv: list[str] | None = None) -> int:
         candidate = build(entry)
         if not args.check:
             target.parent.mkdir(parents=True, exist_ok=True)
-            target.write_text(candidate)
+            target.write_text(candidate, encoding="utf-8")
             print(f"{entry['protocol_version']}: wrote {target.relative_to(REPO_ROOT)} ({len(candidate)} bytes)")
             continue
-        committed = target.read_text() if target.is_file() else ""
+        committed = target.read_text(encoding="utf-8") if target.is_file() else ""
         if committed != candidate:
             drift = True
             sys.stderr.writelines(

scripts/test

@@ -3,7 +3,8 @@
 set -ex
 
 uv run --frozen coverage erase
-uv run --frozen coverage run -m pytest -n auto $@
+# PYTHONWARNDEFAULTENCODING=1 mirrors CI: text I/O without encoding= fails under pytest's "error" filter.
+PYTHONWARNDEFAULTENCODING=1 uv run --frozen coverage run -m pytest -n auto "$@"
 uv run --frozen coverage combine
 uv run --frozen coverage report
 # strict-no-cover spawns `uv run coverage json` internally without --frozen;

scripts/update_readme_snippets.py

@@ -43,15 +43,7 @@ def process_snippet_block(match: re.Match[str], check_mode: bool = False) -> str
     file_path = match.group(2)
 
     try:
-        # Read the entire file. A missing source file must be fatal: a "Warning"
-        # that returns the stale block lets --check pass with exit 0, so a
-        # renamed or deleted snippet is invisible to CI. SystemExit deliberately
-        # escapes the `except Exception` below.
-        file = Path(file_path)
-        if not file.exists():
-            sys.exit(f"Error: snippet-source file not found: {file_path}")
-
-        code = file.read_text().rstrip()
+        code = Path(file_path).read_text(encoding="utf-8").rstrip()
         github_url = get_github_url(file_path)
 
         # Build the replacement block
@@ -88,9 +80,9 @@ def process_snippet_block(match: re.Match[str], check_mode: bool = False) -> str
 
         return replacement
 
-    except Exception as e:
-        print(f"Error processing {file_path}: {e}")
-        return full_match
+    except (OSError, UnicodeDecodeError) as e:
+        # Fatal, not "warn and keep the stale block": that would let --check pass with exit 0.
+        sys.exit(f"Error processing {file_path}: {e}")
 
 
 def update_readme_snippets(check_mode: bool = False) -> bool:
@@ -107,7 +99,7 @@ def update_readme_snippets(check_mode: bool = False) -> bool:
         print(f"Error: README file not found: {readme_path}")
         return False
 
-    content = readme_path.read_text()
+    content = readme_path.read_text(encoding="utf-8")
     original_content = content
 
     # Pattern to match snippet-source blocks
@@ -129,14 +121,14 @@ def update_readme_snippets(check_mode: bool = False) -> bool:
             )
             return False
         else:
-            print(f"✓ {readme_path} code snippets are up to date")
+            print(f"{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}")
+            readme_path.write_text(updated_content, encoding="utf-8")
+            print(f"Updated {readme_path}")
         else:
-            print(f"✓ {readme_path} already up to date")
+            print(f"{readme_path} already up to date")
         return True
 
 

src/mcp/cli/claude.py

@@ -92,7 +92,7 @@ def update_claude_config(
     config_file = config_dir / "claude_desktop_config.json"
     if not config_file.exists():  # pragma: lax no cover
         try:
-            config_file.write_text("{}")
+            config_file.write_text("{}", encoding="utf-8")
         except Exception:
             logger.exception(
                 "Failed to create Claude config file",
@@ -103,7 +103,7 @@ def update_claude_config(
             return False
 
     try:
-        config = json.loads(config_file.read_text())
+        config = json.loads(config_file.read_bytes())
         if "mcpServers" not in config:
             config["mcpServers"] = {}
 
@@ -154,7 +154,7 @@ def update_claude_config(
 
         config["mcpServers"][server_name] = server_config
 
-        config_file.write_text(json.dumps(config, indent=2))
+        config_file.write_text(json.dumps(config, indent=2), encoding="utf-8")
         logger.info(
             f"Added server '{server_name}' to Claude config",
             extra={"config_file": str(config_file)},