diff --git a/runners/kafka-streams/job-server/build.gradle b/runners/kafka-streams/job-server/build.gradle
new file mode 100644
index 000000000000..45d50cd1a788
--- /dev/null
+++ b/runners/kafka-streams/job-server/build.gradle
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Kafka Streams Runner JobServer build file.
+ *
+ * Packages the runner and everything it needs into one jar, so a pipeline from an SDK other than
+ * Java can start a job server without a Beam source tree. The Python KafkaStreamsRunner builds and
+ * launches this jar for the user.
+ */
+
+apply plugin: 'org.apache.beam.module'
+apply plugin: 'application'
+// Must be set before the shadow plugin is applied.
+mainClassName = "org.apache.beam.runners.kafka.streams.KafkaStreamsJobServerDriver"
+
+applyJavaNature(
+  automaticModuleName: 'org.apache.beam.runners.kafka.streams.jobserver',
+  validateShadowJar: false,
+  exportJavadoc: false,
+  shadowClosure: {
+    // Kafka's clients and Streams libraries ship reference.conf-style resources that have to be
+    // concatenated rather than overwritten when everything lands in one jar.
+    append "reference.conf"
+  },
+)
+
+def kafkaStreamsRunnerProject = ":runners:kafka-streams"
+
+description = "Apache Beam :: Runners :: Kafka Streams :: Job Server"
+
+dependencies {
+  implementation project(kafkaStreamsRunnerProject)
+  permitUnusedDeclared project(kafkaStreamsRunnerProject)
+  // A binding, or the job server starts but logs nothing at all, which is unhelpful for something
+  // a user runs in the foreground and reads to see what their pipeline is doing.
+  runtimeOnly library.java.slf4j_simple
+  runtimeOnly project(":sdks:java:extensions:google-cloud-platform-core")
+}
+
+// The runner's classes only exist in the shadow jar, so the job server has to be started through
+// runShadow rather than the plain run task.
+runShadow {
+  args = []
+  if (project.hasProperty('jobHost'))
+    args += ["--job-host=${project.property('jobHost')}"]
+  if (project.hasProperty('jobPort'))
+    args += ["--job-port=${project.property('jobPort')}"]
+  if (project.hasProperty('artifactPort'))
+    args += ["--artifact-port=${project.property('artifactPort')}"]
+  if (project.hasProperty('expansionPort'))
+    args += ["--expansion-port=${project.property('expansionPort')}"]
+  if (project.hasProperty('artifactsDir'))
+    args += ["--artifacts-dir=${project.property('artifactsDir')}"]
+  if (project.hasProperty('cleanArtifactsPerJob'))
+    args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
+}
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 2533083f7e7e..b0d0bdfbaa64 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -732,6 +732,7 @@ class StandardOptions(PipelineOptions):
       'apache_beam.runners.interactive.interactive_runner.InteractiveRunner',
       'apache_beam.runners.portability.flink_runner.FlinkRunner',
       'apache_beam.runners.portability.fn_api_runner.FnApiRunner',
+      'apache_beam.runners.portability.kafka_streams_runner.KafkaStreamsRunner',
       'apache_beam.runners.portability.portable_runner.PortableRunner',
       'apache_beam.runners.portability.prism_runner.PrismRunner',
       'apache_beam.runners.portability.spark_runner.SparkRunner',
@@ -2064,6 +2065,25 @@ def _add_argparse_args(cls, parser):
         ' and the number of key groups used for partitioned state.')
 
 
+class KafkaStreamsRunnerOptions(PipelineOptions):
+  @classmethod
+  def _add_argparse_args(cls, parser):
+    parser.add_argument(
+        '--bootstrap_servers',
+        default='localhost:9092',
+        help='Comma-separated list of host:port Kafka brokers the pipeline '
+        'connects to.')
+    parser.add_argument(
+        '--application_id',
+        help='Kafka Streams application.id for the pipeline. Must be unique '
+        'per pipeline, since it identifies the consumer group and the '
+        'runner\'s internal topics.')
+    parser.add_argument(
+        '--kafka_streams_job_server_jar',
+        help='Path or URL to a Beam Kafka Streams job server jar. If unset, '
+        'the jar is built from the Beam source tree.')
+
+
 class SparkRunnerOptions(PipelineOptions):
   @classmethod
   def _add_argparse_args(cls, parser):
diff --git a/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py b/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py
new file mode 100644
index 000000000000..fa785781ad49
--- /dev/null
+++ b/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py
@@ -0,0 +1,123 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+import logging
+import tempfile
+import unittest
+
+import mock
+
+from apache_beam.options import pipeline_options
+from apache_beam.runners.portability.kafka_streams_runner import KafkaStreamsJarJobServer
+from apache_beam.runners.portability.kafka_streams_runner import KafkaStreamsRunner
+
+
+class KafkaStreamsTestPipelineOptions(pipeline_options.PipelineOptions):
+  def view_as(self, cls):
+    # Ensure only KafkaStreamsRunnerOptions and JobServerOptions are used when
+    # calling default_job_server. If other options classes are needed, the
+    # cache key must include them to prevent incorrect hits.
+    assert (
+        cls is pipeline_options.KafkaStreamsRunnerOptions or
+        cls is pipeline_options.JobServerOptions)
+    return super().view_as(cls)
+
+
+class KafkaStreamsJavaJobServerTest(unittest.TestCase):
+  def test_job_server_cache(self):
+    # Multiple KafkaStreamsRunner instances may be created, so job servers have
+    # to be cached across runner instances: each one is an external Java
+    # process, and starting a second for the same configuration would fail to
+    # bind the same ports.
+
+    # Options that do not affect job server configuration, such as
+    # sdk_worker_parallelism, should still hit the same cache entry.
+    job_server1 = KafkaStreamsRunner().default_job_server(
+        KafkaStreamsTestPipelineOptions(['--sdk_worker_parallelism=1']))
+    job_server2 = KafkaStreamsRunner().default_job_server(
+        KafkaStreamsTestPipelineOptions(['--sdk_worker_parallelism=2']))
+    self.assertIs(job_server2, job_server1)
+
+    # JobServerOptions do affect it, so a different port is a different server.
+    job_server3 = KafkaStreamsRunner().default_job_server(
+        KafkaStreamsTestPipelineOptions(['--job_port=1234']))
+    self.assertIsNot(job_server3, job_server1)
+
+    # So do the runner's own options.
+    job_server4 = KafkaStreamsRunner().default_job_server(
+        KafkaStreamsTestPipelineOptions(['--bootstrap_servers=other:9092']))
+    self.assertIsNot(job_server4, job_server1)
+    self.assertIsNot(job_server4, job_server3)
+
+    job_server5 = KafkaStreamsRunner().default_job_server(
+        KafkaStreamsTestPipelineOptions(['--application_id=other-pipeline']))
+    self.assertIsNot(job_server5, job_server1)
+    self.assertIsNot(job_server5, job_server4)
+
+  def test_java_arguments(self):
+    # These are what the job server driver is launched with, so they have to be
+    # options it accepts.
+    job_server = KafkaStreamsJarJobServer(
+        pipeline_options.PipelineOptions(['--application_id=test-pipeline']))
+    self.assertEqual([
+        '--artifacts-dir',
+        '/tmp/artifacts',
+        '--job-port',
+        8099,
+        '--artifact-port',
+        8098,
+        '--expansion-port',
+        8097
+    ],
+                     job_server.java_arguments(
+                         8099, 8098, 8097, '/tmp/artifacts'))
+
+  def test_path_to_jar_defaults_to_the_job_server_module(self):
+    job_server = KafkaStreamsJarJobServer(pipeline_options.PipelineOptions([]))
+    # Without an explicit jar the runner resolves the one built by the job
+    # server module, which is what lets a user run a pipeline without having
+    # built or started anything first. Resolving it for real would either
+    # download or demand a built jar, so only the target is checked here.
+    with mock.patch.object(job_server, 'path_to_beam_jar') as path_to_beam_jar:
+      job_server.path_to_jar()
+    path_to_beam_jar.assert_called_once_with(
+        ':runners:kafka-streams:job-server:shadowJar')
+
+  def test_path_to_jar_uses_an_explicit_jar(self):
+    with tempfile.NamedTemporaryFile(suffix='.jar') as jar:
+      job_server = KafkaStreamsJarJobServer(
+          pipeline_options.PipelineOptions(
+              ['--kafka_streams_job_server_jar=%s' % jar.name]))
+      self.assertEqual(jar.name, job_server.path_to_jar())
+
+  def test_path_to_jar_rejects_an_unusable_path(self):
+    job_server = KafkaStreamsJarJobServer(
+        pipeline_options.PipelineOptions(
+            ['--kafka_streams_job_server_jar=/no/such/jar.jar']))
+    # A path that is neither an existing file nor a URL cannot be recovered
+    # from, so it fails with the command that would produce a jar rather than
+    # letting the job server fail to start later.
+    with self.assertRaises(ValueError) as context:
+      job_server.path_to_jar()
+    self.assertIn('job-server:shadowJar', str(context.exception))
+
+
+if __name__ == '__main__':
+  logging.getLogger().setLevel(logging.INFO)
+  unittest.main()
diff --git a/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
new file mode 100644
index 000000000000..a2d043ca174e
--- /dev/null
+++ b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
@@ -0,0 +1,106 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""A runner for executing portable pipelines on Kafka Streams."""
+
+# pytype: skip-file
+
+import os
+import urllib
+
+from apache_beam.options import pipeline_options
+from apache_beam.runners.portability import job_server
+from apache_beam.runners.portability import portable_runner
+
+# A Java job server is a heavyweight external process, so reuse one across
+# pipelines configured the same way.
+JOB_SERVER_CACHE = {}
+
+
+class KafkaStreamsRunner(portable_runner.PortableRunner):
+  """A runner for executing pipelines on Kafka Streams.
+
+  Starts a job server automatically, so a pipeline can be submitted without
+  running one by hand:
+
+      python my_pipeline.py \\
+          --runner=KafkaStreamsRunner \\
+          --bootstrap_servers=localhost:9092 \\
+          --application_id=my-pipeline
+
+  Pass --job_endpoint instead to submit to a job server that is already
+  running.
+  """
+
+  # Inherits run_portable_pipeline from PortableRunner.
+
+  def default_environment(self, options):
+    portable_options = options.view_as(pipeline_options.PortableOptions)
+    if (not portable_options.environment_type and
+        not portable_options.output_executable_path):
+      # The job server runs on this machine, so the SDK harness can too, which
+      # saves the user from needing Docker for a local run.
+      portable_options.environment_type = 'LOOPBACK'
+    return super().default_environment(options)
+
+  def default_job_server(self, options):
+    # Only these two option groups affect how the job server is configured, so
+    # they are what the cache is keyed on.
+    kafka_streams_options = options.view_as(
+        pipeline_options.KafkaStreamsRunnerOptions)
+    job_server_options = options.view_as(pipeline_options.JobServerOptions)
+    options_str = str(kafka_streams_options) + str(job_server_options)
+    if options_str not in JOB_SERVER_CACHE:
+      JOB_SERVER_CACHE[options_str] = job_server.StopOnExitJobServer(
+          KafkaStreamsJarJobServer(options))
+    return JOB_SERVER_CACHE[options_str]
+
+
+class KafkaStreamsJarJobServer(job_server.JavaJarJobServer):
+  def __init__(self, options):
+    super().__init__(options)
+    kafka_streams_options = options.view_as(
+        pipeline_options.KafkaStreamsRunnerOptions)
+    self._jar = kafka_streams_options.kafka_streams_job_server_jar
+
+  def path_to_jar(self):
+    if self._jar:
+      if not os.path.exists(self._jar):
+        url = urllib.parse.urlparse(self._jar)
+        if not url.scheme:
+          raise ValueError(
+              'Unable to parse jar URL "%s". If using a full URL, make sure '
+              'the scheme is specified. If using a local file path, make sure '
+              'the file exists; you may have to first build the job server '
+              'using `./gradlew runners:kafka-streams:job-server:shadowJar`.' %
+              self._jar)
+      return self._jar
+    return self.path_to_beam_jar(
+        ':runners:kafka-streams:job-server:shadowJar')
+
+  def java_arguments(
+      self, job_port, artifact_port, expansion_port, artifacts_dir):
+    return [
+        '--artifacts-dir',
+        artifacts_dir,
+        '--job-port',
+        job_port,
+        '--artifact-port',
+        artifact_port,
+        '--expansion-port',
+        expansion_port
+    ]
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 9a9cb1dd1acb..cd1134d685c9 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -145,6 +145,7 @@ include(":runners:java-job-service")
 include(":runners:jet")
 include(":runners:kafka-streams")
 include(":runners:kafka-streams:proto")
+include(":runners:kafka-streams:job-server")
 include(":runners:local-java")
 include(":runners:portability:java")
 include(":runners:prism")