Add a Python Unity HTTP bridge

.llm/mcp/README.md

@@ -183,3 +183,6 @@ directly to the host bridge. Match health's `bridgePid` to the listener owner; h
 
 The universal configurator runs in typed Python through `uv run scripts/mcp_configure.py`. The legacy Node entry remains a launcher with Python 3.9+ fallback. The sync wrapper no
 longer requires Node.
+
+Run the typed Python bridge directly with `uv run scripts/unity_mcp_host.py` on the host. It uses the same settings, health identity, route checks, and dynamic port publication as
+the legacy Node command.

scripts/tests/test_unity_mcp_host.py

@@ -15,14 +15,15 @@
 import time
 import unittest
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 from unittest import mock
 from urllib.error import HTTPError
 from urllib.request import Request, urlopen
 from unittest.mock import Mock, patch
 
 sys.dont_write_bytecode = True
 REPO = Path(__file__).resolve().parents[2]
+PYTHON = str(getattr(sys, "_base_executable", sys.executable))
 
 
 def cleanup_fixture_directory(path: Path) -> None:
@@ -126,21 +127,31 @@ def release_after_locked_cleanup(path: Path) -> None:
             self.assertFalse(root.exists())
 
 FAKE_CLI = r"""
-const readline = require('node:readline');
-readline.createInterface({ input: process.stdin }).on('line', line => {
-    const message = JSON.parse(line);
-    if (message.method === 'crash') {
-        process.stdout.write('partial output');
-        process.exit(1);
-    }
-    if (message.id === undefined) return;
-    const result = message.method === 'initialize'
-        ? { serverInfo: { name: 'fixture' }, capabilities: {}, protocolVersion: '2025-03-26' }
-        : { method: message.method, params: message.params };
-    setTimeout(() => process.stdout.write(JSON.stringify({ jsonrpc: '2.0', id: message.id, result }) + '\n'), 5);
-});
+import json, os, sys, time
+for line in sys.stdin:
+    message = json.loads(line)
+    if message.get('method') == 'crash':
+        sys.stdout.write('partial output')
+        sys.stdout.flush()
+        os._exit(1)
+    if message.get('method') == 'never-respond':
+        continue
+    if 'id' not in message:
+        continue
+    result = dict(serverInfo=dict(name='fixture'), capabilities={}, protocolVersion='2025-03-26') if message.get('method') == 'initialize' else dict(method=message.get('method'), params=message.get('params'))
+    time.sleep(0.005)
+    print(json.dumps(dict(jsonrpc='2.0', id=message['id'], result=result)), flush=True)
+"""
+
+PYTHON_CONFIG_SYNC_FIXTURE = """
+import os, time
+from pathlib import Path
+Path('sync-port.pending').write_text(os.environ['UNITY_MCP_HTTP_PORT'])
+Path('sync-port.pending').replace('sync-port')
+time.sleep(1.8)
 """
 
+
 CONFIG_SYNC_FIXTURE = """
 import fs from 'node:fs';
 fs.writeFileSync('sync-port.pending', process.env.UNITY_MCP_HTTP_PORT);
@@ -193,14 +204,17 @@ def test_port_is_published_only_after_the_write_completes(self) -> None:
 
 @unittest.skipUnless(shutil.which('node'), 'node is required')
 class HostTests(unittest.TestCase):
+    python_bridge = False
+
     def setUp(self) -> None:
         self.root = Path(tempfile.mkdtemp(prefix='unity-mcp-host-'))
         self.addCleanup(cleanup_fixture_directory, self.root)
         script_dir = self.root / '.llm' / 'mcp'
         script_dir.mkdir(parents=True)
-        self.script = script_dir / 'unity-mcp-host.mjs'
-        shutil.copyfile(REPO / '.llm' / 'mcp' / self.script.name, self.script)
         (self.root / 'scripts').mkdir()
+        source = REPO / ('scripts/unity_mcp_host.py' if self.python_bridge else '.llm/mcp/unity-mcp-host.mjs')
+        self.script = self.root / source.relative_to(REPO)
+        shutil.copyfile(source, self.script)
         shutil.copyfile(REPO / 'scripts' / 'unity_mcp_settings.py', self.root / 'scripts' / 'unity_mcp_settings.py')
         (self.root / 'mcp').write_text(FAKE_CLI, encoding='utf-8')
         self.home = self.root / 'home'
@@ -214,11 +228,11 @@ def setUp(self) -> None:
         self.env = {key: value for key, value in os.environ.items() if not key.startswith('UNITY_')}
         self.env.update(HOME=str(self.home), USERPROFILE=str(self.home), UNITY_MCP_AUTO_SYNC='0')
         self.env_file = self.root / '.env.local'
-        self.env_file.write_text(f'\ufeffexport UNITY_CLI="{shutil.which("node")}"\nUNITY_MCP_PROJECT="{self.project}"\nUNITY_MCP_TOKEN=fixture-secret\nUNITY_MCP_TIMEOUT_MS=1500\n', encoding='utf-8')
+        self.env_file.write_text(f'\ufeffexport UNITY_CLI="{PYTHON}"\nUNITY_MCP_PROJECT="{self.project}"\nUNITY_MCP_TOKEN=fixture-secret\nUNITY_MCP_TIMEOUT_MS=1500\n', encoding='utf-8')
         self.log = (self.root / 'bridge.log').open('w+', encoding='utf-8')
         self.addCleanup(self.log.close)
         self.node_options: list[str] = []
-        if self._testMethodName == 'test_auth_notifications_and_parallel_request_ids':
+        if not self.python_bridge and self._testMethodName == 'test_auth_notifications_and_parallel_request_ids':
             observer = self.root / 'observe-response-end.mjs'
             observer.write_text("""
 import http from 'node:http';
@@ -252,7 +266,8 @@ def launch(self, command: Optional[list[str]] = None) -> None:
             time.sleep(0.02)
 
     def command(self) -> list[str]:
-        return [str(shutil.which('node')), *self.node_options, str(self.script), '--host', '127.0.0.1', '--port', str(self.port)]
+        launcher = [PYTHON] if self.python_bridge else [str(shutil.which('node')), *self.node_options]
+        return [*launcher, str(self.script), '--host', '127.0.0.1', '--port', str(self.port)]
 
     def logs(self) -> str:
         self.log.flush()
@@ -283,7 +298,7 @@ def stop(self) -> None:
             if taskkill_failure is not None:
                 self.fail(taskkill_failure)
 
-    def http(self, route: str, payload: Optional[Dict[str, Any]] = None, token: str = 'fixture-secret', method: Optional[str] = None) -> tuple[int, Any]:
+    def http(self, route: str, payload: Optional[object] = None, token: str = 'fixture-secret', method: Optional[str] = None) -> tuple[int, Any]:
         headers = {'content-type': 'application/json', 'authorization': f'Bearer {token}'}
         data = None if payload is None else json.dumps(payload).encode()
         request = Request(f'http://127.0.0.1:{self.port}{route}', data=data, headers=headers, method=method)
@@ -353,7 +368,8 @@ def test_auth_notifications_and_parallel_request_ids(self) -> None:
                     self.assertEqual(client.recv(1), b'', 'Close only after consuming the complete request.')
                 observations = [json.loads(line.split('[drain-probe]', 1)[1])
                                 for line in self.logs().splitlines() if '[drain-probe]' in line]
-                self.assertIn({'probe': probe, 'drained': True}, observations)
+                if not self.python_bridge:
+                    self.assertIn({'probe': probe, 'drained': True}, observations)
 
         # An abandoning client must leave the bridge responsive, without a retry.
         with socket.create_connection(('127.0.0.1', self.port), timeout=5) as client:
@@ -364,6 +380,25 @@ def test_auth_notifications_and_parallel_request_ids(self) -> None:
                 self.assertEqual(json.loads(response.read()), {'error': 'unauthorized'})
         self.assertEqual(self.http('/healthz')[0], 200)
 
+    def test_timeout_parse_errors_and_batch_requests_keep_the_bridge_usable(self) -> None:
+        response = self.rpc('never-respond')
+        self.assertEqual(response['error']['code'], -32000)
+        self.assertIn('Timed out after 1500ms', response['error']['message'])
+        batch = [dict(jsonrpc='2.0', id='first', method='first'),
+                 dict(jsonrpc='2.0', method='notifications/initialized'),
+                 dict(jsonrpc='2.0', id='second', method='second')]
+        status, result = self.http('/mcp', batch)
+        self.assertEqual(status, 200)
+        self.assertEqual([item['id'] for item in result], ['first', 'second'])
+        request = Request(f'http://127.0.0.1:{self.port}/mcp', data=b'{',
+                          headers={'Authorization': 'Bearer fixture-secret'})
+        with self.assertRaises(HTTPError) as raised:
+            urlopen(request, timeout=5)
+        with raised.exception as error:
+            self.assertEqual(error.code, 400)
+            self.assertEqual(json.load(error)['error']['code'], -32700)
+        self.assertEqual(self.rpc('initialize')['result']['serverInfo']['name'], 'fixture')
+
     def test_cli_crash_does_not_poison_restarted_stdout(self) -> None:
         self.rpc('initialize')
         result = self.rpc('crash')
@@ -461,6 +496,8 @@ def test_loopback_occupant_forces_wildcard_bridge_to_another_port(self) -> None:
             self.assertEqual(self.rpc('initialize')['result']['serverInfo']['name'], 'fixture')
 
     def test_failed_self_check_exits_before_publishing_port(self) -> None:
+        if self.python_bridge:
+            self.skipTest("Python startup failures are covered by route-injection unit tests.")
         self.stop()
         preload = self.root / 'redirect-health.mjs'
         preload.write_text("import http from 'node:http'; const get = http.get; http.get = (options, callback) => get({ ...options, path: '/wrong-endpoint' }, callback);", encoding='utf-8')
@@ -473,6 +510,8 @@ def test_failed_self_check_exits_before_publishing_port(self) -> None:
         self.assertEqual(before, self.env_file.read_text(encoding='utf-8'))
 
     def test_intercepted_docker_route_selects_and_publishes_a_verified_port(self) -> None:
+        if self.python_bridge:
+            self.skipTest("Python startup failures are covered by route-injection unit tests.")
         self.stop()
         for mode in ('wrong-instance', 'silent', 'missing-dns'):
             with self.subTest(mode=mode), contextlib.ExitStack() as cleanup:
@@ -527,6 +566,8 @@ def test_intercepted_docker_route_selects_and_publishes_a_verified_port(self) ->
                 self.assertEqual(self.rpc('initialize')['result']['serverInfo']['name'], 'fixture')
 
     def test_existing_bridge_with_intercepted_docker_route_is_not_reused(self) -> None:
+        if self.python_bridge:
+            self.skipTest("Python startup failures are covered by route-injection unit tests.")
         old = self.process
         self.addCleanup(old.wait, 5)
         self.addCleanup(old.terminate)
@@ -563,8 +604,8 @@ def test_repeated_start_recognizes_absolute_cli_path(self) -> None:
 
     def test_slow_config_sync_keeps_http_responsive_and_uses_selected_port(self) -> None:
         self.stop()
-        configure = self.script.parent / 'configure.mjs'
-        configure.write_text(CONFIG_SYNC_FIXTURE, encoding='utf-8')
+        configure = self.root / 'scripts/mcp_configure.py' if self.python_bridge else self.script.parent / 'configure.mjs'
+        configure.write_text(PYTHON_CONFIG_SYNC_FIXTURE if self.python_bridge else CONFIG_SYNC_FIXTURE, encoding='utf-8')
         self.env['UNITY_MCP_AUTO_SYNC'] = '1'
         self.env['UNITY_MCP_HTTP_PORT'] = '9020'
         self.env_file.write_text(self.env_file.read_text(encoding='utf-8').replace(f'UNITY_MCP_HTTP_PORT={self.port}', 'UNITY_MCP_HTTP_PORT=9020'), encoding='utf-8')
@@ -581,6 +622,11 @@ def test_slow_config_sync_keeps_http_responsive_and_uses_selected_port(self) ->
         time.sleep(1.8)
 
 
+class PythonHostTests(HostTests):
+    __unittest_skip__ = False
+    python_bridge = True
+
+
 class CleanupTests(unittest.TestCase):
     def test_taskkill_failure_requires_a_live_process(self) -> None:
         for state in ('already-exited', 'exits-while-waiting', 'remains-live'):

scripts/tests/test_unity_mcp_host_python.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Pin Python bridge startup publication and Docker-route failure handling."""
+from __future__ import annotations
+
+from pathlib import Path
+import socket
+import os
+import sys
+import threading
+import unittest
+from unittest.mock import Mock, patch
+from urllib.error import URLError
+
+sys.dont_write_bytecode = True
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import unity_mcp_host as host
+from unity_mcp_settings import BridgeSettings
+
+
+def settings() -> BridgeSettings:
+    return BridgeSettings(cliCommand='unused', cliArgs=['mcp'], projectPath='', token='', port='9020',
+                timeout='100', autoSync='0', warnings=[])
+
+
+class StartupTests(unittest.TestCase):
+    def test_docker_route_requires_the_same_instance_before_publication(self) -> None:
+        with patch.object(host.sys, 'platform', 'win32'), patch.object(host, 'log'):
+            for value in ('different-listener', None, 'expected'):
+                with self.subTest(value=value), patch.object(host, 'read_health', return_value={'instanceId': value}):
+                    if value == 'expected':
+                        host.verify_docker_host_route(9020, '0.0.0.0', 'expected')
+                    else:
+                        with self.assertRaises(host.DockerHostRouteError):
+                            host.verify_docker_host_route(9020, '0.0.0.0', 'expected')
+            for reason in (TimeoutError('silent listener'), socket.gaierror(socket.EAI_AGAIN, 'temporary DNS failure')):
+                with self.subTest(reason=reason), patch.object(host, 'read_health', side_effect=URLError(reason)):
+                    with self.assertRaises(host.DockerHostRouteError):
+                        host.verify_docker_host_route(9020, '::', 'expected')
+            with patch.object(host, 'read_health', side_effect=URLError(socket.gaierror(socket.EAI_NONAME, 'missing Docker DNS'))):
+                host.verify_docker_host_route(9020, '0.0.0.0', 'expected')
+            with patch.object(host, 'read_health') as read:
+                host.verify_docker_host_route(9020, '127.0.0.1', 'expected')
+                read.assert_not_called()
+
+    def test_intercepted_existing_bridge_is_not_reused(self) -> None:
+        with patch.object(host, 'read_health', return_value={'bridge': 'dataviz-unity-mcp', 'instanceId': 'existing'}), \
+                patch.object(host, 'verify_docker_host_route', side_effect=host.DockerHostRouteError('intercepted')):
+            self.assertFalse(host.existing_bridge(9020, '0.0.0.0'))
+
+    def test_failed_local_check_never_publishes_and_closes_the_candidate(self) -> None:
+        server = Mock()
+        with patch.object(host, 'existing_bridge', return_value=False), patch.object(host, 'port_free', return_value=True), \
+                patch.object(host, 'BridgeServer', return_value=server), patch.object(host.threading, 'Thread'), \
+                patch.object(host, 'read_health', side_effect=ValueError('wrong instance')), patch.object(host, 'persist_port') as persist:
+            with self.assertRaisesRegex(ValueError, 'Local HTTP self-check failed'):
+                host.serve(Path('.'), settings(), 9020, '127.0.0.1', threading.Event())
+            persist.assert_not_called()
+            server.shutdown.assert_called_once()
+            server.server_close.assert_called_once()
+
+    def test_intercepted_candidate_is_closed_and_only_verified_port_is_published(self) -> None:
+        servers: list[Mock] = []
+        bridges: list[host.Bridge] = []
+
+        def create(hostname: str, port: int, bridge: host.Bridge) -> Mock:
+            server = Mock()
+            servers.append(server)
+            bridges.append(bridge)
+            return server
+
+        def health(port: int, hostname: str) -> host.JsonMap:
+            return {'instanceId': bridges[-1].instance}
+
+        done = threading.Event()
+        done.set()
+        with patch.object(host, 'existing_bridge', return_value=False), patch.object(host, 'port_free', return_value=True), \
+                patch.object(host, 'BridgeServer', side_effect=create), patch.object(host.threading, 'Thread'), \
+                patch.object(host, 'read_health', side_effect=health), patch.object(host, 'log'), \
+                patch.object(host, 'verify_docker_host_route', side_effect=[host.DockerHostRouteError('silent'), None]), \
+                patch.object(host, 'persist_port', return_value=False) as persist:
+            self.assertEqual(host.serve(Path('.'), settings(), 9020, '0.0.0.0', done), 0)
+            persist.assert_called_once_with(Path('.'), 9021)
+        self.assertEqual(len(servers), 2)
+        for server in servers:
+            server.shutdown.assert_called_once()
+            server.server_close.assert_called_once()
+
+    def test_spawn_failure_preserves_an_earlier_inflight_request(self) -> None:
+        bridge = host.Bridge(settings())
+        self.addCleanup(bridge.close)
+        earlier = host.Pending('earlier')
+        bridge.next_id = 9
+        bridge.pending[9] = earlier
+        with patch.object(bridge, 'ensure_child', side_effect=ValueError('spawn cooldown')):
+            result = bridge.request({'id': 'later', 'method': 'test'})
+        self.assertEqual(result['id'], 'later')
+        self.assertIn('spawn cooldown', result['error']['message'])
+        self.assertIs(bridge.pending.get(9), earlier)
+        self.assertFalse(earlier.ready.is_set())
+        bridge.fail_pending('CLI exited')
+        self.assertTrue(earlier.ready.is_set())
+        self.assertEqual(earlier.response, host.rpc_error('earlier', -32000, 'CLI exited'))
+
+    @unittest.skipUnless(os.name == 'nt', 'Windows socket reuse semantics')
+    def test_reusable_windows_listener_is_not_available_to_probe_or_server(self) -> None:
+        with socket.socket() as occupant:
+            occupant.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            occupant.bind(('127.0.0.1', 0))
+            occupant.listen(8)
+            port = occupant.getsockname()[1]
+            self.assertFalse(host.port_free(port, '127.0.0.1'))
+            self.assertFalse(host.port_free(port, '0.0.0.0'))
+            bridge = host.Bridge(settings())
+            self.addCleanup(bridge.close)
+            with self.assertRaises(OSError):
+                with host.BridgeServer('127.0.0.1', port, bridge):
+                    pass
+
+    def test_invalid_json_and_batch_notifications_keep_protocol_responses(self) -> None:
+        bridge = host.Bridge(settings())
+        self.addCleanup(bridge.close)
+        self.assertEqual(bridge.handle(7), host.rpc_error(None, -32600, 'Invalid JSON-RPC message.'))
+        self.assertIsNone(bridge.handle([]))
+        response = {'jsonrpc': '2.0', 'id': 'original', 'result': {'value': '日本語'}}
+        with patch.object(bridge, 'request', return_value=response), patch.object(bridge, 'write') as write:
+            self.assertEqual(bridge.handle([{'id': 'original', 'method': 'test'}, {'method': 'initialized'}]), response)
+            write.assert_called_once_with({'method': 'initialized'})
+        with patch.object(bridge, 'write', side_effect=OSError('closed input')), patch.object(host, 'log'):
+            self.assertIsNone(bridge.handle({'method': 'initialized'}))
+
+
+if __name__ == '__main__':
+    unittest.main()