diff --git a/test-server/.gitignore b/test-server/.gitignore
new file mode 100644
index 000000000..7cd1e642f
--- /dev/null
+++ b/test-server/.gitignore
@@ -0,0 +1,9 @@
+# Cloned Python ESDK source (fetched by the orchestrator / Makefile at run time)
+.deps/
+# Python virtual environment
+.venv/
+# Python caches / build artifacts
+__pycache__/
+*.pyc
+.pytest_cache/
+*.egg-info/
diff --git a/test-server/README.md b/test-server/README.md
new file mode 100644
index 000000000..167206778
--- /dev/null
+++ b/test-server/README.md
@@ -0,0 +1,46 @@
+# ESDK TestServer — Python Language_Server (hand-implemented)
+
+A hand-implemented `Language_Server` for the ESDK TestServer. Smithy server
+codegen is not used for Python, so this server hand-writes the marshalling layer
+that conforms to the single Smithy model's **rpcv2Cbor** wire contract
+(Requirement 1.8) and delegates to the real **AWS Encryption SDK for Python**.
+
+## What it implements
+
+- The five operations `CreateClient`, `Encrypt`, `Decrypt`, `EncryptStream`,
+  `DecryptStream`, routed at `POST /service/ESDKTestServer/operation/<Op>`.
+- A thread-safe `ClientId -> client` registry (`CreateClient` returns a UUID).
+- The tagged-union-via-optional-members config (keyrings, CMMs, recursive
+  multi-keyring / required-EC CMM) → real Material Providers Library keyrings and
+  cryptographic materials managers, mirroring the Java server's `EsdkClientFactory`.
+- The two modeled errors serialized as a CBOR map carrying the `__type`
+  discriminator (the error's absolute shape id), so the stock generated Java
+  `Test_Client` maps them back to `GenericServerError` / `ESDKClientError`
+  (matching the Java server's `DiscriminatingCbor`).
+- Streaming: the `*Stream` operations carry a plain blob on the wire and drive the
+  ESDK streaming API server-side (Streaming_Capable).
+
+## Dependencies
+
+- `cbor2`, `boto3` (declared in `pyproject.toml`).
+- The **AWS Encryption SDK for Python** + **Material Providers Library**, installed
+  from a clone of https://github.com/aws/aws-encryption-sdk-python (its repo URL is
+  part of the Configuration). The orchestrator/Makefile clones it and installs it
+  editable from the relative path, so the server runs against live source with no
+  compile step.
+
+## Run locally
+
+```bash
+python3 -m venv .venv
+.venv/bin/pip install -e .deps/aws-encryption-sdk-python \
+    'aws-cryptographic-material-providers>=1.7.4,<=1.11.2' cbor2 boto3
+.venv/bin/pip install -e .
+.venv/bin/python -m esdk_test_server 8091   # listens on http://127.0.0.1:8091
+```
+
+Point the Tests at it (alongside the Java server) via runtime configuration:
+
+```
+-Desdk.testserver.targets=java:3=http://127.0.0.1:8080,python:4=http://127.0.0.1:8091
+```
diff --git a/test-server/bug-config.json b/test-server/bug-config.json
new file mode 100644
index 000000000..0db2acd90
--- /dev/null
+++ b/test-server/bug-config.json
@@ -0,0 +1,5 @@
+[
+  "decrypt-accepts-trailing-bytes-commit-key",
+  "decrypt-accepts-trailing-bytes-commit-key-ecdsa",
+  "encrypt-accepts-reserved-prefix-encryption-context-key"
+]
diff --git a/test-server/esdk_test_server/__init__.py b/test-server/esdk_test_server/__init__.py
new file mode 100644
index 000000000..086ae56af
--- /dev/null
+++ b/test-server/esdk_test_server/__init__.py
@@ -0,0 +1,10 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Hand-implemented Python Language_Server for the ESDK TestServer.
+
+Speaks the same rpcv2Cbor wire contract as the single Smithy model (the source of
+truth) and delegates to the real AWS Encryption SDK for Python. There is no
+generated code here: the marshalling layer is hand-written to conform to the
+model's request/response shapes, the five operations, and the two modeled errors
+(Requirement 1.8).
+"""
diff --git a/test-server/esdk_test_server/__main__.py b/test-server/esdk_test_server/__main__.py
new file mode 100644
index 000000000..34698f6ea
--- /dev/null
+++ b/test-server/esdk_test_server/__main__.py
@@ -0,0 +1,25 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Entry point: ``python -m esdk_test_server <port>`` starts the Python server.
+
+The port is the first positional argument (mirroring the Java server's
+``runServer <port>``), defaulting to the ESDK_TESTSERVER_PORT env var, then 8081.
+"""
+
+import os
+import sys
+
+from .app import serve
+
+
+def main(argv=None):
+    argv = list(sys.argv[1:] if argv is None else argv)
+    if argv:
+        port = int(argv[0])
+    else:
+        port = int(os.environ.get("ESDK_TESTSERVER_PORT", "8081"))
+    serve(port)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/test-server/esdk_test_server/app.py b/test-server/esdk_test_server/app.py
new file mode 100644
index 000000000..d19cc8987
--- /dev/null
+++ b/test-server/esdk_test_server/app.py
@@ -0,0 +1,228 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""The hand-implemented rpcv2Cbor HTTP server for the Python Language_Server.
+
+Implements the wire contract of the single Smithy model by hand (no generated
+code): routes ``POST /service/ESDKTestServer/operation/<Op>`` requests, decodes
+the CBOR request body, dispatches to the operation handler, and encodes the CBOR
+response. Every operation's outcome is exactly one of a modeled success response,
+a ``GenericServerError``, or an ``ESDKClientError`` — the latter two serialized as
+a CBOR map carrying the ``__type`` discriminator (the error's absolute shape id)
+so the stock generated Java Test_Client maps them back to the correct modeled
+type (Requirements 5.1-5.6, 6.1-6.4), matching the Java server's DiscriminatingCbor.
+"""
+
+import threading
+import uuid
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import cbor2
+
+from .esdk_bridge import ClientError, ServerError, build_client
+
+_NAMESPACE = "aws.cryptography.esdk.testserver"
+_GENERIC_SERVER_ERROR = _NAMESPACE + "#GenericServerError"
+_ESDK_CLIENT_ERROR = _NAMESPACE + "#ESDKClientError"
+_SMITHY_PROTOCOL = "rpc-v2-cbor"
+_CBOR_CONTENT_TYPE = "application/cbor"
+
+
+class ClientRegistry:
+    """Thread-safe in-memory ClientId -> EsdkClientBundle registry (Requirement 3)."""
+
+    def __init__(self):
+        self._clients = {}
+        self._lock = threading.Lock()
+
+    def register(self, bundle):
+        client_id = str(uuid.uuid4())
+        with self._lock:
+            self._clients[client_id] = bundle
+        return client_id
+
+    def resolve(self, client_id):
+        with self._lock:
+            return self._clients.get(client_id)
+
+
+def _require_client(registry, request):
+    """Resolve the referenced client or reject with a GenericServerError (Req 3.9)."""
+    client_id = request.get("clientId")
+    if not client_id:
+        raise ServerError("clientId is required and must be non-empty")
+    bundle = registry.resolve(client_id)
+    if bundle is None:
+        raise ServerError(f"no client registered for clientId: {client_id}")
+    return bundle
+
+
+# ---------------------------------------------------------------------------
+# Operation handlers. Each returns the response member map or raises
+# ClientError (-> ESDKClientError) / ServerError (-> GenericServerError).
+# ---------------------------------------------------------------------------
+def _describe_exception(exc):
+    """Flatten an exception to a message, appending any nested ``list`` of
+    encountered exceptions (the MPL ``CollectionOfErrors`` raised when e.g. no
+    configured key could decrypt) so the underlying causes are visible rather
+    than only the top-level "the list ... is available via `list`"."""
+    message = str(exc)
+    nested = getattr(exc, "list", None)
+    if isinstance(nested, (list, tuple)) and nested:
+        causes = "; ".join(_describe_exception(cause) for cause in nested)
+        message = f"{message} [encountered: {causes}]"
+    return message
+
+
+def _create_client(registry, request):
+    config = request.get("config")
+    if config is None:
+        raise ServerError("config is required")
+    try:
+        bundle = build_client(config)
+    except (ClientError, ServerError):
+        raise
+    except Exception as exc:  # noqa: BLE001 - construction failure -> GenericServerError (Req 3.6)
+        raise ServerError(f"CreateClient failed to construct the ESDK client: {_describe_exception(exc)}") from exc
+    return {"clientId": registry.register(bundle)}
+
+
+def _encrypt(registry, request):
+    bundle = _require_client(registry, request)
+    try:
+        ciphertext = bundle.encrypt(
+            request["plaintext"],
+            request.get("encryptionContext"),
+            request.get("algorithmSuiteId"),
+            request.get("frameLength"),
+        )
+    except Exception as exc:  # noqa: BLE001 - ESDK-thrown -> ESDKClientError (Req 4.10, 5.6)
+        raise ClientError(_describe_exception(exc)) from exc
+    return {"ciphertext": ciphertext}
+
+
+def _decrypt(registry, request):
+    bundle = _require_client(registry, request)
+    try:
+        plaintext, encryption_context, algorithm_suite_id = bundle.decrypt(
+            request["ciphertext"], request.get("encryptionContext"))
+    except Exception as exc:  # noqa: BLE001
+        raise ClientError(_describe_exception(exc)) from exc
+    response = {"plaintext": plaintext}
+    if encryption_context:
+        response["encryptionContext"] = encryption_context
+    if algorithm_suite_id:
+        response["algorithmSuiteId"] = algorithm_suite_id
+    return response
+
+
+def _encrypt_stream(registry, request):
+    bundle = _require_client(registry, request)
+    try:
+        ciphertext = bundle.encrypt_stream(
+            request["plaintext"],
+            request.get("encryptionContext"),
+            request.get("algorithmSuiteId"),
+            request.get("frameLength"),
+            request.get("plaintextLengthBound"),
+        )
+    except Exception as exc:  # noqa: BLE001
+        raise ClientError(_describe_exception(exc)) from exc
+    return {"ciphertext": ciphertext}
+
+
+def _decrypt_stream(registry, request):
+    bundle = _require_client(registry, request)
+    try:
+        plaintext, encryption_context, algorithm_suite_id = bundle.decrypt_stream(
+            request["ciphertext"], request.get("encryptionContext"))
+    except Exception as exc:  # noqa: BLE001
+        raise ClientError(_describe_exception(exc)) from exc
+    response = {"plaintext": plaintext}
+    if encryption_context:
+        response["encryptionContext"] = encryption_context
+    if algorithm_suite_id:
+        response["algorithmSuiteId"] = algorithm_suite_id
+    return response
+
+
+_OPERATIONS = {
+    "CreateClient": _create_client,
+    "Encrypt": _encrypt,
+    "Decrypt": _decrypt,
+    "EncryptStream": _encrypt_stream,
+    "DecryptStream": _decrypt_stream,
+}
+
+
+def _make_handler(registry):
+    class RpcV2CborHandler(BaseHTTPRequestHandler):
+        # Speak HTTP/1.1 with keep-alive so the smithy-java client's pooled
+        # connections stay valid across the many requests the Tests make. Every
+        # response carries a Content-Length (see _send_cbor), which is what lets
+        # the base handler keep the connection open rather than closing it after
+        # each response (the HTTP/1.0 default, which caused intermittent
+        # "received no bytes" transport errors on reused connections).
+        protocol_version = "HTTP/1.1"
+
+        # Quiet the default per-request stderr logging.
+        def log_message(self, *args):  # noqa: D401
+            pass
+
+        def _send_cbor(self, status, payload):
+            body = cbor2.dumps(payload)
+            self.send_response(status)
+            self.send_header("smithy-protocol", _SMITHY_PROTOCOL)
+            self.send_header("Content-Type", _CBOR_CONTENT_TYPE)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def _send_error(self, type_id, message):
+            # Modeled @error("client") shapes: HTTP 400 with a CBOR body carrying
+            # the __type discriminator so the client picks the right modeled error.
+            self._send_cbor(400, {"__type": type_id, "message": message})
+
+        def do_POST(self):  # noqa: N802 - required name
+            operation = self.path.rsplit("/", 1)[-1] if "/operation/" in self.path else None
+            handler = _OPERATIONS.get(operation)
+            if handler is None:
+                self._send_error(_GENERIC_SERVER_ERROR, f"unknown operation: {self.path}")
+                return
+            try:
+                length = int(self.headers.get("Content-Length") or 0)
+                raw = self.rfile.read(length) if length else b""
+                request = cbor2.loads(raw) if raw else {}
+                response = handler(registry, request)
+                self._send_cbor(200, response)
+            except ClientError as exc:
+                self._send_error(_ESDK_CLIENT_ERROR, str(exc))
+            except ServerError as exc:
+                self._send_error(_GENERIC_SERVER_ERROR, str(exc))
+            except Exception as exc:  # noqa: BLE001 - never leak a bare HTTP error (Req 6.1-6.4)
+                self._send_error(_GENERIC_SERVER_ERROR, f"unexpected server error: {exc}")
+
+    return RpcV2CborHandler
+
+
+class _EsdkTestThreadingHTTPServer(ThreadingHTTPServer):
+    # A generous listen backlog so bursts of new connections from the
+    # heavily-parameterized Tests are not reset (the socketserver default of 5 is
+    # far too small and manifested as intermittent "received no bytes" errors).
+    request_queue_size = 128
+    # Do not let a lingering worker thread block process shutdown.
+    daemon_threads = True
+    # Free the port immediately on restart.
+    allow_reuse_address = True
+
+
+def serve(port, host="127.0.0.1"):
+    """Start the Python Language_Server on ``host:port`` and serve until stopped."""
+    registry = ClientRegistry()
+    server = _EsdkTestThreadingHTTPServer((host, port), _make_handler(registry))
+    print(f"esdk-test-server (python) listening at http://{host}:{port}", flush=True)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
diff --git a/test-server/esdk_test_server/esdk_bridge.py b/test-server/esdk_test_server/esdk_bridge.py
new file mode 100644
index 000000000..41a015a02
--- /dev/null
+++ b/test-server/esdk_test_server/esdk_bridge.py
@@ -0,0 +1,362 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Delegation to the real AWS Encryption SDK for Python.
+
+Translates the modeled ``ESDKClientConfig`` (a tagged-union-via-optional-members
+CBOR map, mirroring the single Smithy model at the lowest faithful API layer) into
+a real ESDK client backed by a Material Providers Library keyring / cryptographic
+materials manager, and drives the one-shot and streaming encrypt/decrypt APIs.
+
+The MPL model shapes and enums mirror the Java server's ``EsdkClientFactory``
+one-for-one, so the two languages build equivalent clients from the same config.
+"""
+
+import os
+
+import boto3
+import aws_encryption_sdk
+from aws_encryption_sdk import CommitmentPolicy
+from aws_encryption_sdk.identifiers import Algorithm
+
+from aws_cryptographic_material_providers.mpl import AwsCryptographicMaterialProviders
+from aws_cryptographic_material_providers.mpl.config import MaterialProvidersConfig
+from aws_cryptographic_material_providers.mpl import models as mpl
+from aws_cryptographic_material_providers.keystore import KeyStore
+from aws_cryptographic_material_providers.keystore.config import KeyStoreConfig
+from aws_cryptographic_material_providers.keystore.models import KMSConfigurationKmsKeyArn
+
+
+class ClientError(Exception):
+    """A failure originating in the ESDK itself: forwarded as ESDKClientError."""
+
+
+class ServerError(Exception):
+    """A framework/config failure: forwarded as GenericServerError."""
+
+
+_MATERIAL_PROVIDERS = AwsCryptographicMaterialProviders(config=MaterialProvidersConfig())
+
+
+def _region():
+    """Resolve the AWS region for KMS clients (mirrors the Java server default)."""
+    return (
+        os.environ.get("AWS_REGION")
+        or os.environ.get("AWS_DEFAULT_REGION")
+        or os.environ.get("ESDK_TESTSERVER_KMS_REGION")
+        or "us-west-2"
+    )
+
+
+def _kms_client():
+    return boto3.client("kms", region_name=_region())
+
+
+def _kms_client_for_key(kms_key_id):
+    """A KMS client in the key's own region.
+
+    KMS rejects an ARN whose region differs from the client's region ("Invalid
+    arn <region>"), so a us-east-1 key needs a us-east-1 client even when the
+    default region is us-west-2. Falls back to the default region for a bare key
+    id or alias that carries no region.
+    """
+    region = _region()
+    if isinstance(kms_key_id, str) and kms_key_id.startswith("arn:"):
+        parts = kms_key_id.split(":")
+        if len(parts) > 3 and parts[3]:
+            region = parts[3]
+    return boto3.client("kms", region_name=region)
+
+
+def to_algorithm(model_suite_id):
+    """Map a modeled ``ESDKAlgorithmSuiteId`` to a Python ``Algorithm`` member.
+
+    The model ids are the Python names prefixed with ``ALG_``; the non-KDF suites
+    additionally carry a ``_NO_KDF`` suffix the Python enum omits.
+    """
+    name = model_suite_id
+    if name.startswith("ALG_"):
+        name = name[len("ALG_"):]
+    if name.endswith("_NO_KDF"):
+        name = name[: -len("_NO_KDF")]
+    try:
+        return getattr(Algorithm, name)
+    except AttributeError as exc:
+        raise ServerError(f"unknown algorithm suite id: {model_suite_id}") from exc
+
+
+# Modeled ESDKAlgorithmSuiteId names, used to invert to_algorithm() so decrypt can
+# report the suite it determined from the message header.
+_MODEL_SUITE_IDS = (
+    "ALG_AES_128_GCM_IV12_TAG16_NO_KDF",
+    "ALG_AES_192_GCM_IV12_TAG16_NO_KDF",
+    "ALG_AES_256_GCM_IV12_TAG16_NO_KDF",
+    "ALG_AES_128_GCM_IV12_TAG16_HKDF_SHA256",
+    "ALG_AES_192_GCM_IV12_TAG16_HKDF_SHA256",
+    "ALG_AES_256_GCM_IV12_TAG16_HKDF_SHA256",
+    "ALG_AES_128_GCM_IV12_TAG16_HKDF_SHA256_ECDSA_P256",
+    "ALG_AES_192_GCM_IV12_TAG16_HKDF_SHA384_ECDSA_P384",
+    "ALG_AES_256_GCM_IV12_TAG16_HKDF_SHA384_ECDSA_P384",
+    "ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY",
+    "ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY_ECDSA_P384",
+)
+
+
+def from_algorithm(algorithm):
+    """Inverse of to_algorithm: a Python ``Algorithm`` -> modeled ESDKAlgorithmSuiteId name."""
+    for model_suite_id in _MODEL_SUITE_IDS:
+        if to_algorithm(model_suite_id) == algorithm:
+            return model_suite_id
+    return None
+
+
+def _commitment_policy(value):
+    try:
+        return getattr(CommitmentPolicy, value)
+    except AttributeError as exc:
+        raise ServerError(f"unknown commitment policy: {value}") from exc
+
+
+def _one_variant(tagged, what):
+    """Return the single (name, value) set on a tagged-union map, else reject.
+
+    Enforces the "exactly one optional variant member set" invariant at runtime
+    (Requirements 2.3, 2.4).
+    """
+    present = [(k, v) for k, v in tagged.items() if v is not None]
+    if len(present) != 1:
+        raise ClientError(
+            f"exactly one {what} variant must be set, found {len(present)}: "
+            f"{sorted(k for k, _ in present)}"
+        )
+    return present[0]
+
+
+# ---------------------------------------------------------------------------
+# Keyring construction (tagged union via optional members; recursive for Multi).
+# ---------------------------------------------------------------------------
+def _build_keyring(keyring):
+    name, cfg = _one_variant(keyring, "keyring")
+    if name == "RawAes":
+        return _MATERIAL_PROVIDERS.create_raw_aes_keyring(
+            input=mpl.CreateRawAesKeyringInput(
+                key_namespace=cfg["keyNamespace"],
+                key_name=cfg["keyName"],
+                wrapping_key=cfg["wrappingKey"],
+                wrapping_alg=getattr(mpl.AesWrappingAlg, cfg["wrappingAlg"]),
+            )
+        )
+    if name == "RawRsa":
+        return _MATERIAL_PROVIDERS.create_raw_rsa_keyring(
+            input=mpl.CreateRawRsaKeyringInput(
+                key_namespace=cfg["keyNamespace"],
+                key_name=cfg["keyName"],
+                padding_scheme=getattr(mpl.PaddingScheme, cfg["paddingScheme"]),
+                public_key=cfg.get("publicKey"),
+                private_key=cfg.get("privateKey"),
+            )
+        )
+    if name == "AwsKms":
+        return _MATERIAL_PROVIDERS.create_aws_kms_keyring(
+            input=mpl.CreateAwsKmsKeyringInput(
+                kms_key_id=cfg["kmsKeyId"],
+                kms_client=_kms_client_for_key(cfg["kmsKeyId"]),
+                grant_tokens=cfg.get("grantTokens"),
+            )
+        )
+    if name == "AwsKmsMrk":
+        return _MATERIAL_PROVIDERS.create_aws_kms_mrk_keyring(
+            input=mpl.CreateAwsKmsMrkKeyringInput(
+                kms_key_id=cfg["kmsKeyId"],
+                kms_client=_kms_client_for_key(cfg["kmsKeyId"]),
+                grant_tokens=cfg.get("grantTokens"),
+            )
+        )
+    if name == "AwsKmsMultiKeyring":
+        return _MATERIAL_PROVIDERS.create_aws_kms_multi_keyring(
+            input=mpl.CreateAwsKmsMultiKeyringInput(
+                generator=cfg.get("generator"),
+                kms_key_ids=cfg.get("kmsKeyIds"),
+            )
+        )
+    if name == "AwsKmsMrkMultiKeyring":
+        # MRK-aware multi-keyring: an optional MRK generator + child MRK key ids.
+        # Mirrors AwsKmsMultiKeyring (the MPL builds its own default KMS client
+        # supplier); the MRK-aware form matches multi-region keys on decrypt.
+        return _MATERIAL_PROVIDERS.create_aws_kms_mrk_multi_keyring(
+            input=mpl.CreateAwsKmsMrkMultiKeyringInput(
+                generator=cfg.get("generator"),
+                kms_key_ids=cfg.get("kmsKeyIds"),
+            )
+        )
+    if name == "AwsKmsDiscovery":
+        return _MATERIAL_PROVIDERS.create_aws_kms_discovery_keyring(
+            input=mpl.CreateAwsKmsDiscoveryKeyringInput(
+                kms_client=_kms_client(),
+                discovery_filter=_discovery_filter(cfg.get("discoveryFilter")),
+                grant_tokens=cfg.get("grantTokens"),
+            )
+        )
+    if name == "AwsKmsMrkDiscovery":
+        return _MATERIAL_PROVIDERS.create_aws_kms_mrk_discovery_keyring(
+            input=mpl.CreateAwsKmsMrkDiscoveryKeyringInput(
+                kms_client=boto3.client("kms", region_name=cfg["region"]),
+                region=cfg["region"],
+                discovery_filter=_discovery_filter(cfg.get("discoveryFilter")),
+                grant_tokens=cfg.get("grantTokens"),
+            )
+        )
+    if name == "AwsKmsRsa":
+        kms_client = _kms_client_for_key(cfg["kmsKeyId"])
+        public_key = cfg.get("publicKey")
+        if public_key is None:
+            # Fetch the RSA public key from KMS (as the Java server does) so the
+            # keyring can OnEncrypt. KMS GetPublicKey returns DER (X.509
+            # SubjectPublicKeyInfo); CreateAwsKmsRsaKeyring expects PEM.
+            der = kms_client.get_public_key(KeyId=cfg["kmsKeyId"])["PublicKey"]
+            public_key = _der_to_public_key_pem(der)
+        return _MATERIAL_PROVIDERS.create_aws_kms_rsa_keyring(
+            input=mpl.CreateAwsKmsRsaKeyringInput(
+                kms_key_id=cfg["kmsKeyId"],
+                encryption_algorithm=cfg.get("encryptionAlgorithm"),
+                public_key=public_key,
+                kms_client=kms_client,
+                grant_tokens=cfg.get("grantTokens"),
+            )
+        )
+    if name == "Multi":
+        children = [_build_keyring(k) for k in cfg["childKeyrings"]]
+        generator = _build_keyring(cfg["generator"]) if cfg.get("generator") else None
+        return _MATERIAL_PROVIDERS.create_multi_keyring(
+            input=mpl.CreateMultiKeyringInput(child_keyrings=children, generator=generator)
+        )
+    if name == "AwsKmsHierarchical":
+        key_store = KeyStore(
+            config=KeyStoreConfig(
+                ddb_client=boto3.client("dynamodb", region_name=_region()),
+                ddb_table_name=cfg["keyStoreTableName"],
+                logical_key_store_name=cfg["logicalKeyStoreName"],
+                kms_client=_kms_client(),
+                kms_configuration=KMSConfigurationKmsKeyArn(value=cfg["kmsKeyArn"]),
+            )
+        )
+        return _MATERIAL_PROVIDERS.create_aws_kms_hierarchical_keyring(
+            input=mpl.CreateAwsKmsHierarchicalKeyringInput(
+                key_store=key_store,
+                branch_key_id=cfg["branchKeyId"],
+                ttl_seconds=cfg["ttlSeconds"],
+                cache=mpl.CacheTypeDefault(value=mpl.DefaultCache(entry_capacity=100)),
+            )
+        )
+    raise ClientError(f"unsupported keyring variant: {name}")
+
+
+def _discovery_filter(filt):
+    if not filt:
+        return None
+    return mpl.DiscoveryFilter(partition=filt["partition"], account_ids=filt["accountIds"])
+
+
+def _der_to_public_key_pem(der_bytes):
+    """Wrap DER (X.509 SubjectPublicKeyInfo) bytes as a PEM ``PUBLIC KEY`` block.
+
+    Pass through unchanged if the bytes already look like PEM.
+    """
+    if der_bytes[:11] == b"-----BEGIN ":
+        return der_bytes
+    import base64
+
+    b64 = base64.b64encode(der_bytes).decode("ascii")
+    lines = [b64[i:i + 64] for i in range(0, len(b64), 64)]
+    pem = "-----BEGIN PUBLIC KEY-----\n" + "\n".join(lines) + "\n-----END PUBLIC KEY-----\n"
+    return pem.encode("ascii")
+
+
+# ---------------------------------------------------------------------------
+# CMM construction (tagged union; recursive for RequiredEncryptionContext).
+# ---------------------------------------------------------------------------
+def _build_cmm(cmm):
+    name, cfg = _one_variant(cmm, "cmm")
+    if name == "Default":
+        return _MATERIAL_PROVIDERS.create_default_cryptographic_materials_manager(
+            input=mpl.CreateDefaultCryptographicMaterialsManagerInput(
+                keyring=_build_keyring(cfg["keyring"])
+            )
+        )
+    if name == "RequiredEncryptionContext":
+        return _MATERIAL_PROVIDERS.create_required_encryption_context_cmm(
+            input=mpl.CreateRequiredEncryptionContextCMMInput(
+                underlying_cmm=_build_cmm(cfg["underlyingCMM"]),
+                required_encryption_context_keys=cfg["requiredEncryptionContextKeys"],
+            )
+        )
+    if name == "Caching":
+        raise ClientError(
+            "Caching CMM is not exercised by the round-trip Tests and is not supported here"
+        )
+    raise ClientError(f"unsupported cmm variant: {name}")
+
+
+class EsdkClientBundle:
+    """A configured ESDK client plus its materials manager and commitment policy."""
+
+    def __init__(self, client, materials_manager):
+        self._client = client
+        self._cmm = materials_manager
+
+    def _common_kwargs(self, encryption_context, algorithm_suite_id, frame_length):
+        kwargs = {"materials_manager": self._cmm}
+        if encryption_context:
+            kwargs["encryption_context"] = encryption_context
+        if algorithm_suite_id:
+            kwargs["algorithm"] = to_algorithm(algorithm_suite_id)
+        if frame_length is not None:
+            kwargs["frame_length"] = frame_length
+        return kwargs
+
+    def encrypt(self, plaintext, encryption_context, algorithm_suite_id, frame_length):
+        kwargs = self._common_kwargs(encryption_context, algorithm_suite_id, frame_length)
+        ciphertext, _ = self._client.encrypt(source=plaintext, **kwargs)
+        return ciphertext
+
+    def decrypt(self, ciphertext, encryption_context):
+        kwargs = {"materials_manager": self._cmm}
+        if encryption_context:
+            kwargs["encryption_context"] = encryption_context
+        plaintext, header = self._client.decrypt(source=ciphertext, **kwargs)
+        return plaintext, dict(header.encryption_context or {}), from_algorithm(header.algorithm)
+
+    def encrypt_stream(self, plaintext, encryption_context, algorithm_suite_id, frame_length,
+                       plaintext_length_bound=None):
+        kwargs = self._common_kwargs(encryption_context, algorithm_suite_id, frame_length)
+        if plaintext_length_bound is not None:
+            # The Python ESDK enforces the plaintext length bound via source_length:
+            # the total plaintext encrypted is not allowed to exceed it.
+            kwargs["source_length"] = plaintext_length_bound
+        with self._client.stream(mode="e", source=plaintext, **kwargs) as encryptor:
+            return encryptor.read()
+
+    def decrypt_stream(self, ciphertext, encryption_context):
+        kwargs = {"materials_manager": self._cmm}
+        if encryption_context:
+            kwargs["encryption_context"] = encryption_context
+        with self._client.stream(mode="d", source=ciphertext, **kwargs) as decryptor:
+            plaintext = decryptor.read()
+            header = decryptor.header
+        return plaintext, dict(header.encryption_context or {}), from_algorithm(header.algorithm)
+
+
+def build_client(config):
+    """Build an :class:`EsdkClientBundle` from a modeled ``ESDKClientConfig`` map.
+
+    A construction failure surfaces as a ServerError (GenericServerError,
+    Requirement 3.6), except an exactly-one-variant violation which is a
+    ClientError (ESDKClientError, Requirements 2.3, 2.4).
+    """
+    commitment = _commitment_policy(config["commitmentPolicy"])
+    cmm = _build_cmm(config["cmm"])
+    client_kwargs = {"commitment_policy": commitment}
+    max_edks = config.get("maxEncryptedDataKeys")
+    if max_edks is not None:
+        client_kwargs["max_encrypted_data_keys"] = max_edks
+    client = aws_encryption_sdk.EncryptionSDKClient(**client_kwargs)
+    return EsdkClientBundle(client, cmm)
diff --git a/test-server/feature-config.json b/test-server/feature-config.json
new file mode 100644
index 000000000..1020db2b7
--- /dev/null
+++ b/test-server/feature-config.json
@@ -0,0 +1,23 @@
+{
+  "supportedFeatures": [
+    "streaming",
+    "MPL",
+    "hierarchical",
+    "raw-aes",
+    "raw-rsa",
+    "multi",
+    "aws-kms",
+    "aws-kms-multi",
+    "aws-kms-discovery",
+    "aws-kms-mrk",
+    "aws-kms-mrk-multi",
+    "aws-kms-mrk-discovery",
+    "aws-kms-rsa",
+    "required-encryption-context"
+  ],
+  "unsupportedFeatures": [
+    "raw-ecdh",
+    "aws-kms-ecdh",
+    "caching"
+  ]
+}
diff --git a/test-server/pyproject.toml b/test-server/pyproject.toml
new file mode 100644
index 000000000..5a1ac1010
--- /dev/null
+++ b/test-server/pyproject.toml
@@ -0,0 +1,26 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "esdk-test-server-python"
+version = "0.0.0"
+description = "Hand-implemented Python Language_Server for the ESDK TestServer (rpcv2Cbor)"
+requires-python = ">=3.11"
+# The AWS Encryption SDK for Python and the Material Providers Library are NOT
+# pinned here: the orchestrator/Makefile installs the ESDK from the cloned
+# aws-encryption-sdk-python source (its repo URL is part of the Configuration),
+# via an editable relative-path install, so the server exercises live source.
+# Only the wire-protocol codec and AWS SDK client are declared as ordinary deps.
+dependencies = [
+    "cbor2>=5.6",
+    "boto3>=1.34",
+]
+
+[project.scripts]
+esdk-test-server-python = "esdk_test_server.__main__:main"
+
+[tool.setuptools]
+packages = ["esdk_test_server"]
diff --git a/test-server/server-config.json b/test-server/server-config.json
new file mode 100644
index 000000000..180128721
--- /dev/null
+++ b/test-server/server-config.json
@@ -0,0 +1,8 @@
+{
+  "commonsRepository": {
+    "name": "aws-crypto-tools-commons",
+    "url": "git@github.com:aws/aws-crypto-tools-commons.git",
+    "branch": "lucmcdon/esdk-test-server-all-languages"
+  },
+  "product": "esdk"
+}