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

.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

@@ -266,6 +266,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)},

src/mcp/shared/path_security.py

@@ -12,7 +12,7 @@
 
     @mcp.resource("file://docs/{+path}")
     def read_doc(path: str) -> str:
-        return safe_join("/data/docs", path).read_text()
+        return safe_join("/data/docs", path).read_text(encoding="utf-8")
 """
 
 import string

tests/cli/test_claude.py

@@ -64,7 +64,7 @@ def raise_not_found(distribution_name: str) -> str:
 
 
 def _read_server(config_dir: Path, name: str) -> dict[str, Any]:
-    config = json.loads((config_dir / "claude_desktop_config.json").read_text())
+    config = json.loads((config_dir / "claude_desktop_config.json").read_text(encoding="utf-8"))
     return config["mcpServers"][name]
 
 
@@ -121,7 +121,7 @@ def test_env_vars_written(config_dir: Path):
 def test_existing_env_vars_merged_new_wins(config_dir: Path):
     """Re-installing should merge env vars, with new values overriding existing ones."""
     (config_dir / "claude_desktop_config.json").write_text(
-        json.dumps({"mcpServers": {"s": {"env": {"OLD": "keep", "KEY": "old"}}}})
+        json.dumps({"mcpServers": {"s": {"env": {"OLD": "keep", "KEY": "old"}}}}), encoding="utf-8"
     )
 
     assert update_claude_config(file_spec="s.py:app", server_name="s", env_vars={"KEY": "new"})
@@ -131,22 +131,37 @@ def test_existing_env_vars_merged_new_wins(config_dir: Path):
 
 def test_existing_env_vars_preserved_without_new(config_dir: Path):
     """Re-installing without env_vars should keep the existing env block intact."""
-    (config_dir / "claude_desktop_config.json").write_text(json.dumps({"mcpServers": {"s": {"env": {"KEEP": "me"}}}}))
+    (config_dir / "claude_desktop_config.json").write_text(
+        json.dumps({"mcpServers": {"s": {"env": {"KEEP": "me"}}}}), encoding="utf-8"
+    )
 
     assert update_claude_config(file_spec="s.py:app", server_name="s")
 
     assert _read_server(config_dir, "s")["env"] == {"KEEP": "me"}
 
 
 def test_other_servers_preserved(config_dir: Path):
-    """Installing a new server should not clobber existing mcpServers entries."""
-    (config_dir / "claude_desktop_config.json").write_text(json.dumps({"mcpServers": {"other": {"command": "x"}}}))
+    """Installing a new server must not clobber existing entries, non-ASCII text included (the file is UTF-8)."""
+    other = {"command": "C:\\Users\\张伟\\uv.exe", "env": {"CITY": "Zürich"}}
+    config_file = config_dir / "claude_desktop_config.json"
+    config_file.write_text(json.dumps({"mcpServers": {"文件": other}}, ensure_ascii=False), encoding="utf-8")
+
+    assert update_claude_config(file_spec="s.py:app", server_name="s")
+
+    config = json.loads(config_file.read_text(encoding="utf-8"))
+    assert set(config["mcpServers"]) == {"文件", "s"}
+    assert config["mcpServers"]["文件"] == other
+
+
+@pytest.mark.parametrize("codec", ["utf-8-sig", "utf-16"])
+def test_existing_config_with_a_bom_is_accepted(config_dir: Path, codec: str):
+    """A config saved by Windows tooling (UTF-8 with BOM, or PowerShell 5's UTF-16 `>`) can still be installed into."""
+    config_file = config_dir / "claude_desktop_config.json"
+    config_file.write_bytes(json.dumps({"mcpServers": {"other": {"command": "x"}}}).encode(codec))
 
     assert update_claude_config(file_spec="s.py:app", server_name="s")
 
-    config = json.loads((config_dir / "claude_desktop_config.json").read_text())
-    assert set(config["mcpServers"]) == {"other", "s"}
-    assert config["mcpServers"]["other"] == {"command": "x"}
+    assert set(json.loads(config_file.read_bytes())["mcpServers"]) == {"other", "s"}
 
 
 def test_raises_when_config_dir_missing(monkeypatch: pytest.MonkeyPatch):

tests/cli/test_utils.py

@@ -28,7 +28,7 @@ def fake_version(distribution_name: str) -> str:
 def test_parse_file_path_accepts_valid_specs(tmp_path: Path, spec: str, expected_obj: str | None):
     """Should accept valid file specs."""
     file = tmp_path / spec.split(":")[0]
-    file.write_text("x = 1")
+    file.write_text("x = 1", encoding="utf-8")
     path, obj = _parse_file_path(f"{file}:{expected_obj}" if ":" in spec else str(file))
     assert path == file.resolve()
     assert obj == expected_obj

tests/docs_src/test_apps.py

@@ -100,4 +100,4 @@ async def test_a_file_resource_is_served_with_the_app_mime_type_filled_in() -> N
     contents = result.contents[0]
     assert isinstance(contents, TextResourceContents)
     assert contents.mime_type == APP_MIME_TYPE
-    assert contents.text == tutorial003.REPORT_HTML.read_text()
+    assert contents.text == tutorial003.REPORT_HTML.read_text(encoding="utf-8")

tests/docs_src/test_uri_templates.py

@@ -129,9 +129,9 @@ def test_dotdot_is_a_component_check_not_a_substring_scan() -> None:
 async def test_safe_join_serves_a_file_inside_the_base_directory(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """tutorial002: `safe_join(DOCS_ROOT, path).read_text()` returns the file under the base."""
+    """tutorial002: `safe_join(DOCS_ROOT, path).read_text(encoding="utf-8")` returns the file under the base."""
     (tmp_path / "printing").mkdir()
-    (tmp_path / "printing" / "setup.md").write_text("# Printer setup")
+    (tmp_path / "printing" / "setup.md").write_text("# Printer setup", encoding="utf-8")
     monkeypatch.setattr(tutorial002, "DOCS_ROOT", tmp_path)
     async with Client(tutorial002.mcp) as client:
         (content,) = (await client.read_resource("manuals://printing/setup.md")).contents

tests/interaction/transports/test_stdio.py

@@ -71,7 +71,7 @@ async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
     async def collect(params: LoggingMessageNotificationParams) -> None:
         received.append(params)
 
-    with tempfile.TemporaryFile(mode="w+") as errlog:
+    with tempfile.TemporaryFile(mode="w+", encoding="utf-8", errors="replace") as errlog:
         transport = stdio_client(
             StdioServerParameters(
                 command=sys.executable,

tests/server/mcpserver/resources/test_file_resources.py

@@ -16,7 +16,7 @@ def temp_file():
     File is automatically cleaned up after the test if it still exists.
     """
     content = "test content"
-    with NamedTemporaryFile(mode="w", delete=False) as f:
+    with NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as f:
         f.write(content)
         path = Path(f.name).resolve()
     yield path

tests/server/mcpserver/servers/test_file_server.py

@@ -13,9 +13,9 @@ def test_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
     tmp = tmp_path_factory.mktemp("test_files")
 
     # Create test files
-    (tmp / "example.py").write_text("print('hello world')")
-    (tmp / "readme.md").write_text("# Test Directory\nThis is a test.")
-    (tmp / "config.json").write_text('{"test": true}')
+    (tmp / "example.py").write_text("print('hello world')", encoding="utf-8")
+    (tmp / "readme.md").write_text("# Test Directory\nThis is a test.", encoding="utf-8")
+    (tmp / "config.json").write_text('{"test": true}', encoding="utf-8")
 
     return tmp
 
@@ -38,23 +38,23 @@ def list_test_dir() -> list[str]:
     def read_example_py() -> str:
         """Read the example.py file"""
         try:
-            return (test_dir / "example.py").read_text()
+            return (test_dir / "example.py").read_text(encoding="utf-8")
         except FileNotFoundError:
             return "File not found"
 
     @mcp.resource("file://test_dir/readme.md")
     def read_readme_md() -> str:
         """Read the readme.md file"""
         try:  # pragma: no cover
-            return (test_dir / "readme.md").read_text()
+            return (test_dir / "readme.md").read_text(encoding="utf-8")
         except FileNotFoundError:  # pragma: no cover
             return "File not found"
 
     @mcp.resource("file://test_dir/config.json")
     def read_config_json() -> str:
         """Read the config.json file"""
         try:  # pragma: no cover
-            return (test_dir / "config.json").read_text()
+            return (test_dir / "config.json").read_text(encoding="utf-8")
         except FileNotFoundError:  # pragma: no cover
             return "File not found"
 

tests/server/mcpserver/test_server.py

@@ -819,7 +819,7 @@ async def test_file_resource_text(self, tmp_path: Path):
 
         # Create a text file
         text_file = tmp_path / "test.txt"
-        text_file.write_text("Hello from file!")
+        text_file.write_text("Hello from file!", encoding="utf-8")
 
         resource = FileResource(uri="file://test.txt", name="test.txt", path=text_file)
         mcp.add_resource(resource)

tests/test_types.py

@@ -495,9 +495,9 @@ def test_bare_import_mcp_binds_the_types_submodule():
     # A regression hangs forever, so the bound only has to beat never (matches the suite's
     # other subprocess.run calls).
     result = subprocess.run(
-        [sys.executable, "-c", "import mcp; print(mcp.types.Tool.__name__)"],
+        [sys.executable, "-X", "utf8", "-c", "import mcp; print(mcp.types.Tool.__name__)"],
         capture_output=True,
-        text=True,
+        encoding="utf-8",
         check=False,
         timeout=20,
     )