diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8407e4a1b..939dd68a5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
  * Added `==` as an alias to `=` (#859)
  * Added custom exception formatting for `basilisp.lang.compiler.exception.CompilerException` and `basilisp.lang.reader.SyntaxError` to show more useful details to users on errors (#870)
  * Added `merge-with` core function (#860)
+ * Added proxy hooks to Python's `sys.stdin`, `sys.stdout`, and `sys.stderr` so they track the values of the corresponding `basilisp.core` stream Var values (#839) 
 
 ### Changed
  * Cause exceptions arising from compilation issues during macroexpansion will no longer be nested for each level of macroexpansion (#852)
diff --git a/src/basilisp/core.lpy b/src/basilisp/core.lpy
index 721692ae2..5ad2ecebf 100644
--- a/src/basilisp/core.lpy
+++ b/src/basilisp/core.lpy
@@ -4120,6 +4120,23 @@
   value."
   sys/stderr)
 
+;; Proxy objects for the Python standard streams.
+;;
+;; Without holding a reference to these objects as a Var, they are garbage collected
+;; any time their reference count otherwise reaches 0 (as can be the case while
+;; running PyTest, which commonly reassigns the standard streams to capture output).
+;; The IOProxy class will attempt to close the underlying stream when it is garbage
+;; collected, which is problematic for most use cases.
+
+(def ^:private py-stdin-proxy
+  (basilisp.lang.runtime.IOProxy/for-standard-stream #'*in* "stdin"))
+
+(def ^:private py-stdout-proxy
+  (basilisp.lang.runtime.IOProxy/for-standard-stream #'*out* "stdout"))
+
+(def ^:private py-stderr-proxy
+  (basilisp.lang.runtime.IOProxy/for-standard-stream #'*err* "stderr"))
+
 (def ^:dynamic *print-sep* " ")
 
 (def ^:dynamic *flush-on-newline*
diff --git a/src/basilisp/lang/runtime.py b/src/basilisp/lang/runtime.py
index 75b12b35d..6a9aae9f1 100644
--- a/src/basilisp/lang/runtime.py
+++ b/src/basilisp/lang/runtime.py
@@ -5,6 +5,7 @@
 import functools
 import importlib.metadata
 import inspect
+import io
 import itertools
 import logging
 import math
@@ -2030,6 +2031,171 @@ def print_generated_python() -> bool:
     )
 
 
+############
+# IO Proxy #
+############
+
+
+class IOProxy(io.TextIOWrapper):  # pragma: no cover
+    """
+    Proxy class which proxies all `io.TextIOWrapper` methods down to the current
+    (potentially thread-local) value of the wrapped Var (which should itself be either
+    an `io.TextIOWrapper` or some other file-like object).
+
+    This is intended to be installed in place of the Python standard streams (such
+    as `sys.stdout`) to proxy calls back to the buffer stored in the corresponding
+    `basilisp.core` dynamic Var (such as `*out*`). Because Python functions still
+    rely on the `sys` references (and are obviously unaware of Basilisp), setting
+    those references to a proxy class which uses the buffer underlying Basilisp's
+    own stream Vars enables Basilisp code to capture inputs and outputs of both Python
+    and Basilisp code.
+
+    This also reduces the risk of drifting between `sys` and `basilisp.core` streams,
+    since the Python variant should always track Basilisp unless some other code
+    hijacks the `sys` streams.
+    """
+
+    def __init__(self, var: Var):
+        self._var = var
+
+    @classmethod
+    def for_standard_stream(cls, var: Var, member: str) -> "IOProxy":
+        """Create an IOProxy object for one of the 3 Python standard streams:
+        `sys.stdin`, `sys.stdout`, or `sys.stderr`."""
+        assert member in {"stdin", "stdout", "stderr"}
+        proxy = cls(var)
+        setattr(sys, member, proxy)
+        return proxy
+
+    @property
+    def name(self):
+        return self._var.value.name
+
+    @property
+    def buffer(self):
+        return self._var.value.buffer
+
+    @property
+    def encoding(self):
+        return self._var.value.encoding
+
+    @property
+    def errors(self):
+        return self._var.value.errors
+
+    @property
+    def newlines(self):
+        return self._var.value.newlines
+
+    @property
+    def mode(self):
+        return self._var.value.mode
+
+    @property
+    def closed(self):
+        return self._var.value.closed
+
+    @property
+    def line_buffering(self):
+        return self._var.value.line_buffering
+
+    @property
+    def write_through(self):
+        return self._var.value.write_through
+
+    def reconfigure(  # pylint: disable=too-many-arguments
+        self,
+        *,
+        encoding=None,
+        errors=None,
+        newline=None,
+        line_buffering=None,
+        write_through=None,
+    ):
+        self._var.value.reconfigure(
+            encoding=encoding,
+            errors=errors,
+            newline=newline,
+            line_buffering=line_buffering,
+            write_through=write_through,
+        )
+
+    def __enter__(self):
+        return self._var.value.__enter__()
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._var.value.__exit__(exc_type, exc_val, exc_tb)
+
+    def __iter__(self):
+        return self._var.value.__iter__()
+
+    def __next__(self):
+        return self._var.value.__next__()
+
+    def __del__(self):
+        self._var.value.__del__()
+
+    def __str__(self):
+        return repr(self)
+
+    def __repr__(self):
+        return (
+            f"<{self.__class__.__module__}.{self.__class__.__qualname__} "
+            f"{self._var} buffer={self._var.value.__repr__()}>"
+        )
+
+    def __hash__(self):
+        return self._var.value.__hash__()
+
+    def writelines(self, __lines):
+        self._var.value.writelines(__lines)
+
+    def readline(self, __size=-1):
+        return self._var.value.readline(__size)
+
+    def readlines(self, __hint=-1):
+        return self._var.value.readlines(__hint)
+
+    def seek(self, __cookie, __whence=0):
+        return self._var.value.seek(__cookie, __whence)
+
+    def detach(self):
+        return self._var.value.detach()
+
+    def write(self, __s):
+        return self._var.value.write(__s)
+
+    def read(self, __size=...):
+        return self._var.value.read(__size)
+
+    def close(self):
+        self._var.value.close()
+
+    def fileno(self):
+        return self._var.value.fileno()
+
+    def flush(self):
+        self._var.value.flush()
+
+    def isatty(self):
+        return self._var.value.isatty()
+
+    def readable(self):
+        return self._var.value.readable()
+
+    def seekable(self):
+        return self._var.value.seekable()
+
+    def tell(self):
+        return self._var.value.tell()
+
+    def truncate(self, __size=...):
+        return self._var.value.truncate(__size)
+
+    def writable(self):
+        return self._var.value.writable()
+
+
 #########################
 # Bootstrap the Runtime #
 #########################
diff --git a/tests/basilisp/runtime_test.py b/tests/basilisp/runtime_test.py
index 680e8dc28..61708ffd5 100644
--- a/tests/basilisp/runtime_test.py
+++ b/tests/basilisp/runtime_test.py
@@ -1,4 +1,7 @@
+import io
+import os
 import platform
+import subprocess
 import sys
 from decimal import Decimal
 from fractions import Fraction
@@ -17,7 +20,7 @@
 from basilisp.lang import vector as vec
 from basilisp.lang.compiler.constants import SpecialForm
 from basilisp.lang.interfaces import ISeq
-from tests.basilisp.helpers import get_or_create_ns
+from tests.basilisp.helpers import CompileFn, get_or_create_ns
 
 
 def test_is_supported_python_version():
@@ -747,3 +750,44 @@ def test_resolve_alias(self, core_ns):
             ) == runtime.resolve_alias(
                 sym.symbol("non-existent-alias-var", ns="wee.woo"), ns=ns
             )
+
+
+class TestIOProxy:
+    """PyTest captures the standard streams in a way that makes testing Basilisp's
+    own stream hooks nearly impossible. Testing with a subprocess is easier."""
+
+    def test_hooked_stdout(self):
+        res = subprocess.run(
+            [
+                "basilisp",
+                "run",
+                "-c",
+                '(print (with-out-str (python/print "Hello from Python stdout!")))',
+            ],
+            check=True,
+            capture_output=True,
+        )
+        assert res.stdout.decode("utf-8") == f"Hello from Python stdout!{os.linesep}"
+        assert res.stderr == b""
+
+    def test_hooked_stderr(self):
+        res = subprocess.run(
+            [
+                "basilisp",
+                "run",
+                "-c",
+                """
+                (import io
+                        sys)
+
+                (print
+                  (binding [*err* (io/StringIO)]
+                    (python/print "Hello from Python stderr!" ** :file sys/stderr)
+                    (.getvalue *err*)))
+                """,
+            ],
+            check=True,
+            capture_output=True,
+        )
+        assert res.stdout.decode("utf-8") == f"Hello from Python stderr!{os.linesep}"
+        assert res.stderr == b""
diff --git a/tests/basilisp/testrunner_test.py b/tests/basilisp/testrunner_test.py
index 39f1f134f..1962681c1 100644
--- a/tests/basilisp/testrunner_test.py
+++ b/tests/basilisp/testrunner_test.py
@@ -126,27 +126,15 @@ def test_failure_repr(self, run_result: pytest.RunResult):
         ),
     )
     def test_error_repr(self, run_result: pytest.RunResult):
-        if sys.version_info < (3, 11):
-            expected = [
-                "ERROR in (assertion-test) (test_testrunner.lpy:12)",
-                "",
-                "Traceback (most recent call last):",
-                '  File "*test_testrunner.lpy", line 12, in assertion_test',
-                '    (is (throw (ex-info "Uncaught exception" {})))',
-                "basilisp.lang.exception.ExceptionInfo: Uncaught exception {}",
-            ]
-        else:
-            expected = [
+        run_result.stdout.fnmatch_lines(
+            [
                 "ERROR in (assertion-test) (test_testrunner.lpy:12)",
                 "",
                 "Traceback (most recent call last):",
                 '  File "*test_testrunner.lpy", line 12, in assertion_test',
                 '    (is (throw (ex-info "Uncaught exception" {})))',
                 "basilisp.lang.exception.ExceptionInfo: Uncaught exception {}",
-            ]
-
-        run_result.stdout.fnmatch_lines(
-            expected,
+            ],
             consecutive=True,
         )