diff --git a/pyflowlauncher/__init__.py b/pyflowlauncher/__init__.py
index bff0204..cea85fa 100644
--- a/pyflowlauncher/__init__.py
+++ b/pyflowlauncher/__init__.py
@@ -4,6 +4,7 @@
 from .result import Result, send_results
 from .method import Method
 from .response import handle_response
+from .launcher import FlowLauncherV1, FlowLauncherV2
 
 
 logger = logging.getLogger(__name__)
@@ -15,4 +16,6 @@
     "Result",
     "Method",
     "handle_response",
+    "FlowLauncherV1",
+    "FlowLauncherV2",
 ]
diff --git a/pyflowlauncher/api.py b/pyflowlauncher/api.py
index 37359e3..e1d5d30 100644
--- a/pyflowlauncher/api.py
+++ b/pyflowlauncher/api.py
@@ -77,3 +77,21 @@ def open_url(url: str, in_private: bool = False) -> JsonRPCRequest:
 def open_uri(uri: str) -> JsonRPCRequest:
     """Open a URI."""
     return _send_action("OpenAppUri", uri)
+
+
+class Api:
+    """Flow Launcher API calls, accessible via plugin.launcher.api."""
+    change_query = staticmethod(change_query)
+    shell_run = staticmethod(shell_run)
+    close_app = staticmethod(close_app)
+    hide_app = staticmethod(hide_app)
+    show_app = staticmethod(show_app)
+    show_msg = staticmethod(show_msg)
+    open_setting_dialog = staticmethod(open_setting_dialog)
+    start_loading_bar = staticmethod(start_loading_bar)
+    stop_loading_bar = staticmethod(stop_loading_bar)
+    reload_plugins = staticmethod(reload_plugins)
+    copy_to_clipboard = staticmethod(copy_to_clipboard)
+    open_directory = staticmethod(open_directory)
+    open_url = staticmethod(open_url)
+    open_uri = staticmethod(open_uri)
diff --git a/pyflowlauncher/icons.py b/pyflowlauncher/icons.py
index 73edd83..8915a63 100644
--- a/pyflowlauncher/icons.py
+++ b/pyflowlauncher/icons.py
@@ -21,6 +21,24 @@ def _get_icon(icon_name: str, file_ext: str = "png") -> Optional[str]:
     return None
 
 
+class Icons:
+    """Flow Launcher icon paths, accessible via launcher.icons.<name>."""
+    _IMAGE_DIR = "Images"
+
+    def __init__(self, program_dir: Optional[Path]) -> None:
+        self._dir = Path(program_dir) / self._IMAGE_DIR if program_dir else None
+
+    def _get(self, name: str, ext: str = "png") -> Optional[str]:
+        if self._dir is None:
+            return None
+        return str(self._dir / f"{name}.{ext}")
+
+    def __getattr__(self, name: str) -> Optional[str]:
+        if name.startswith('__') and name.endswith('__'):
+            raise AttributeError(name)
+        return self._get(name)
+
+
 ADMIN = _get_icon("admin")
 APP = _get_icon("app")
 APP_ERROR = _get_icon("app_error")
diff --git a/pyflowlauncher/launcher.py b/pyflowlauncher/launcher.py
new file mode 100644
index 0000000..f700512
--- /dev/null
+++ b/pyflowlauncher/launcher.py
@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Optional
+
+from .api import Api
+from .base import pyFlowLauncherObject
+from .icons import Icons
+from .jsonrpc import JsonRPCClient
+
+
+class Launcher(pyFlowLauncherObject, ABC):
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._settings: dict = {}
+        self.api = Api()
+        self._program_dir: Optional[Path] = self._find_program_dir()
+        self.icons = Icons(self._program_dir)
+
+    @property
+    def settings(self) -> dict:
+        return self._settings
+
+    @property
+    def program_dir(self) -> Optional[Path]:
+        return self._program_dir
+
+    def _find_program_dir(self) -> Optional[Path]:
+        env = os.getenv("FLOW_PROGRAM_DIRECTORY")
+        if env:
+            return Path(env)
+        appdata = os.getenv("APPDATA")
+        if appdata:
+            candidate = Path(appdata) / "FlowLauncher"
+            if candidate.exists():
+                return candidate
+        local = os.getenv("LOCALAPPDATA")
+        if local:
+            candidate = Path(local) / "FlowLauncher"
+            if candidate.exists():
+                return candidate
+        current = Path.cwd()
+        for _ in range(5):
+            if (current / "Flow.Launcher.exe").exists():
+                return current
+            current = current.parent
+        self.logger.warning("Could not locate Flow Launcher program directory.")
+        return None
+
+    @abstractmethod
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None: ...
+
+
+class FlowLauncherV1(Launcher):
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._client = JsonRPCClient()
+
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None:
+        request = self._client.recieve()
+        self._settings = request.get('settings', {})
+        result = await dispatch(request['method'], request.get('parameters', []))
+        if result:
+            self._client.send(result)
+
+
+class FlowLauncherV2(Launcher):
+
+    _LIFECYCLE_METHODS = {'initialize', 'reload_data'}
+
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None:
+        loop = asyncio.get_event_loop()
+        while True:
+            line = await loop.run_in_executor(None, sys.stdin.readline)
+            if not line:
+                break
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                request = json.loads(line)
+            except json.JSONDecodeError:
+                self.logger.warning("Bad JSON-RPC line: %r", line)
+                continue
+
+            request_id = request.get('id')
+            method = request.get('method', '')
+            incoming_settings = request.get('settings')
+            if incoming_settings is not None:
+                self._settings = incoming_settings
+
+            if method == 'close':
+                self._send({'id': request_id, 'result': {}, 'error': None})
+                break
+
+            if method in self._LIFECYCLE_METHODS:
+                self._send({'id': request_id, 'result': {}, 'error': None})
+                continue
+
+            params = request.get('parameters', [])
+            if method == 'query' and params and isinstance(params[0], dict):
+                params = [params[0].get('search') or params[0].get('trimmedQuery', '')]
+
+            try:
+                result = await dispatch(method, params)
+            except Exception:
+                self.logger.exception("Unhandled error dispatching %r", method)
+                self._send({'id': request_id, 'result': [], 'error': None, 'debugMessage': ''})
+                continue
+
+            self._send_response(request_id, method, result)
+
+    def _send(self, data: dict) -> None:
+        sys.stdout.write(json.dumps(data) + '\n')
+        sys.stdout.flush()
+
+    def _send_response(self, request_id: Any, method: str, result: Any) -> None:
+        if result is None:
+            self._send({'id': request_id, 'result': {}, 'error': None})
+            return
+        if isinstance(result, dict) and 'Result' in result:
+            payload = {
+                'id': request_id,
+                'result': result['Result'],
+                'settingsChange': result.get('SettingsChange'),
+                'debugMessage': '',
+            }
+        else:
+            payload = {'id': request_id, 'hide': True}
+        self._send(payload)
diff --git a/pyflowlauncher/plugin.py b/pyflowlauncher/plugin.py
index 0e2b16e..ffe66a4 100644
--- a/pyflowlauncher/plugin.py
+++ b/pyflowlauncher/plugin.py
@@ -9,8 +9,9 @@
 from .base import pyFlowLauncherObject
 
 from .event import EventHandler
+from .launcher import Launcher, FlowLauncherV1, FlowLauncherV2
+from .jsonrpc import JsonRPCRequest
 from .response import handle_response
-from .jsonrpc import JsonRPCClient, JsonRPCRequest
 from .models.plugin_manifest import FILE_NAME
 from .manifest import Manifest
 
@@ -20,14 +21,21 @@
 
 class Plugin(pyFlowLauncherObject):
 
-    def __init__(self, methods: list[Method] | None = None) -> None:
+    def __init__(self, methods: list[Method] | None = None, launcher: Optional[Launcher] = None) -> None:
         super().__init__()
-        self._client = JsonRPCClient()
+        self._launcher: Launcher = launcher if launcher is not None else self._detect_launcher()
         self._event_handler = EventHandler()
-        self._settings: dict[str, Any] = {}
         if methods:
             self.add_methods(methods)
 
+    def _detect_launcher(self) -> Launcher:
+        try:
+            if (self.manifest.language or '').lower() == 'python_v2':
+                return FlowLauncherV2()
+        except FileNotFoundError:
+            pass
+        return FlowLauncherV1()
+
     def add_method(self, method: Method) -> str:
         """Add a method to the event handler."""
         @wraps(method)
@@ -65,25 +73,23 @@ def action(self, method: Method, parameters: Optional[List] = None) -> JsonRPCRe
         method_name = self.add_method(method)
         return {"method": method_name, "parameters": parameters or []}
 
+    @property
+    def launcher(self) -> Launcher:
+        return self._launcher
+
     @property
     def settings(self) -> dict:
-        if self._settings is None:
-            self._settings = {}
-        self._settings = self._client.recieve().get('settings', {})
-        return self._settings
+        return self._launcher.settings
 
     def run(self) -> None:
-        request = self._client.recieve()
-        method = request["method"]
-        parameters = request.get('parameters', [])
+        async def dispatch(method: str, parameters: list) -> Any:
+            return await self._event_handler.trigger_event(method, *parameters)
+
         if sys.version_info >= (3, 10, 0):
-            feedback = asyncio.run(self._event_handler.trigger_event(method, *parameters))
+            asyncio.run(self._launcher.run(dispatch))
         else:
             loop = asyncio.get_event_loop()
-            feedback = loop.run_until_complete(self._event_handler.trigger_event(method, *parameters))
-        if not feedback:
-            return
-        self._client.send(feedback)
+            loop.run_until_complete(self._launcher.run(dispatch))
 
     @property
     def run_dir(self) -> Path:
diff --git a/pyflowlauncher/response.py b/pyflowlauncher/response.py
index 8625037..560d0a1 100644
--- a/pyflowlauncher/response.py
+++ b/pyflowlauncher/response.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import inspect
 from typing import Any, Generator, Union
 
diff --git a/tests/test_launcher.py b/tests/test_launcher.py
new file mode 100644
index 0000000..6d1c1f5
--- /dev/null
+++ b/tests/test_launcher.py
@@ -0,0 +1,301 @@
+import asyncio
+import json
+import sys
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from pyflowlauncher.launcher import FlowLauncherV1, FlowLauncherV2, Launcher
+from pyflowlauncher.plugin import Plugin
+from pyflowlauncher.result import Result, send_results
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_dispatch(result=None):
+    async def dispatch(method, params):
+        return result
+    return dispatch
+
+
+def _make_dispatch_raising(exc):
+    async def dispatch(method, params):
+        raise exc
+    return dispatch
+
+
+def _v2_run(launcher, lines, dispatch=None):
+    """Run FlowLauncherV2 with a fake stdin from a list of JSON-serialisable dicts."""
+    dispatch = dispatch or _make_dispatch(None)
+    stdin_text = '\n'.join(json.dumps(line) for line in lines) + '\n'
+
+    async def _run():
+        with patch('sys.stdin', StringIO(stdin_text)):
+            await launcher.run(dispatch)
+
+    asyncio.run(_run())
+
+
+# ---------------------------------------------------------------------------
+# FlowLauncherV1
+# ---------------------------------------------------------------------------
+
+class TestFlowLauncherV1:
+
+    def test_run_dispatches_and_sends(self, monkeypatch):
+        launcher = FlowLauncherV1()
+        request = {'method': 'query', 'parameters': ['hello'], 'settings': {}}
+        sent = []
+        monkeypatch.setattr(launcher._client, 'recieve', lambda: request)
+        monkeypatch.setattr(launcher._client, 'send', lambda data: sent.append(data))
+
+        result = send_results([Result(title='x')])
+        asyncio.run(launcher.run(_make_dispatch(result)))
+
+        assert sent == [result]
+
+    def test_run_no_result_skips_send(self, monkeypatch):
+        launcher = FlowLauncherV1()
+        request = {'method': 'query', 'parameters': ['hello'], 'settings': {}}
+        sent = []
+        monkeypatch.setattr(launcher._client, 'recieve', lambda: request)
+        monkeypatch.setattr(launcher._client, 'send', lambda data: sent.append(data))
+
+        asyncio.run(launcher.run(_make_dispatch(None)))
+
+        assert sent == []
+
+    def test_settings_cached_after_run(self, monkeypatch):
+        launcher = FlowLauncherV1()
+        request = {'method': 'query', 'parameters': [], 'settings': {'key': 'value'}}
+        monkeypatch.setattr(launcher._client, 'recieve', lambda: request)
+        monkeypatch.setattr(launcher._client, 'send', lambda data: None)
+
+        asyncio.run(launcher.run(_make_dispatch(None)))
+
+        assert launcher.settings == {'key': 'value'}
+
+    def test_is_default_for_unknown_manifest(self, monkeypatch):
+        monkeypatch.setattr('pyflowlauncher.plugin.Plugin.manifest',
+                            property(lambda self: (_ for _ in ()).throw(FileNotFoundError())),
+                            raising=False)
+        # patch cached_property by making root_dir raise
+        with patch.object(Plugin, 'root_dir', new_callable=lambda: property(
+            lambda self: (_ for _ in ()).throw(FileNotFoundError())
+        )):
+            plugin = Plugin()
+        assert isinstance(plugin._launcher, FlowLauncherV1)
+
+    def test_custom_launcher_injected(self):
+        v2 = FlowLauncherV2()
+        plugin = Plugin(launcher=v2)
+        assert plugin._launcher is v2
+
+
+# ---------------------------------------------------------------------------
+# Auto-detection
+# ---------------------------------------------------------------------------
+
+class TestAutoDetect:
+
+    def _plugin_with_language(self, language):
+        manifest = MagicMock()
+        manifest.language = language
+        with patch.object(Plugin, 'manifest', new_callable=lambda: property(lambda self: manifest)):
+            return Plugin()
+
+    def test_auto_detect_v1(self):
+        plugin = self._plugin_with_language('python')
+        assert isinstance(plugin._launcher, FlowLauncherV1)
+
+    def test_auto_detect_v2(self):
+        plugin = self._plugin_with_language('python_v2')
+        assert isinstance(plugin._launcher, FlowLauncherV2)
+
+    def test_auto_detect_case_insensitive(self):
+        plugin = self._plugin_with_language('Python_V2')
+        assert isinstance(plugin._launcher, FlowLauncherV2)
+
+    def test_explicit_overrides_auto_detect(self):
+        manifest = MagicMock()
+        manifest.language = 'python_v2'
+        v1 = FlowLauncherV1()
+        with patch.object(Plugin, 'manifest', new_callable=lambda: property(lambda self: manifest)):
+            plugin = Plugin(launcher=v1)
+        assert plugin._launcher is v1
+
+
+# ---------------------------------------------------------------------------
+# FlowLauncherV2
+# ---------------------------------------------------------------------------
+
+class TestFlowLauncherV2:
+
+    def test_query_request_dispatches_search_string(self):
+        launcher = FlowLauncherV2()
+        received = []
+
+        async def dispatch(method, params):
+            received.append((method, params))
+            return send_results([Result(title='x')])
+
+        _v2_run(launcher, [
+            {'id': 1, 'method': 'query', 'parameters': [{'search': 'hi', 'trimmedQuery': 'hi'}], 'settings': {}},
+            {'id': 2, 'method': 'close', 'parameters': []},
+        ], dispatch)
+
+        assert received == [('query', ['hi'])]
+
+    def test_response_echoes_id(self):
+        launcher = FlowLauncherV2()
+        output = []
+
+        async def dispatch(method, params):
+            return send_results([Result(title='r')])
+
+        stdin_text = json.dumps({'id': 42, 'method': 'query',
+                                 'parameters': [{'search': 'x'}], 'settings': {}}) + '\n'
+        stdin_text += json.dumps({'id': 99, 'method': 'close', 'parameters': []}) + '\n'
+
+        async def _run():
+            with patch('sys.stdin', StringIO(stdin_text)), \
+                 patch('sys.stdout', StringIO()) as mock_out:
+                await launcher.run(dispatch)
+                mock_out.seek(0)
+                for line in mock_out.read().splitlines():
+                    if line:
+                        output.append(json.loads(line))
+
+        asyncio.run(_run())
+
+        query_resp = next(r for r in output if r.get('result') is not None)
+        assert query_resp['id'] == 42
+
+    def test_eof_exits_cleanly(self):
+        launcher = FlowLauncherV2()
+        _v2_run(launcher, [])  # empty stdin → EOF immediately
+
+    def test_invalid_json_continues(self):
+        launcher = FlowLauncherV2()
+        received = []
+
+        async def dispatch(method, params):
+            received.append(method)
+            return None
+
+        stdin_io = StringIO('not-json\n' +
+                            json.dumps({'id': 1, 'method': 'query',
+                                        'parameters': [{'search': 'ok'}], 'settings': {}}) + '\n' +
+                            json.dumps({'id': 2, 'method': 'close', 'parameters': []}) + '\n')
+
+        async def _run():
+            with patch('sys.stdin', stdin_io):
+                await launcher.run(dispatch)
+
+        asyncio.run(_run())
+        assert received == ['query']
+
+    def test_settings_cached_per_request(self):
+        launcher = FlowLauncherV2()
+        _v2_run(launcher, [
+            {'id': 1, 'method': 'initialize', 'parameters': [], 'settings': {'k': 'v'}},
+            {'id': 2, 'method': 'close', 'parameters': []},
+        ])
+        assert launcher.settings == {'k': 'v'}
+
+    def test_lifecycle_initialize_not_dispatched(self):
+        launcher = FlowLauncherV2()
+        received = []
+
+        async def dispatch(method, params):
+            received.append(method)
+            return None
+
+        _v2_run(launcher, [
+            {'id': 1, 'method': 'initialize', 'parameters': []},
+            {'id': 2, 'method': 'close', 'parameters': []},
+        ], dispatch)
+
+        assert 'initialize' not in received
+
+    def test_lifecycle_close_exits_loop(self):
+        launcher = FlowLauncherV2()
+        received = []
+
+        async def dispatch(method, params):
+            received.append(method)
+            return None
+
+        _v2_run(launcher, [
+            {'id': 1, 'method': 'close', 'parameters': []},
+            {'id': 2, 'method': 'query', 'parameters': [{'search': 'after_close'}]},
+        ], dispatch)
+
+        assert received == []
+
+    def test_dispatch_exception_loop_continues(self):
+        launcher = FlowLauncherV2()
+        call_count = [0]
+
+        async def dispatch(method, params):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                raise RuntimeError("boom")
+            return None
+
+        _v2_run(launcher, [
+            {'id': 1, 'method': 'query', 'parameters': [{'search': 'a'}]},
+            {'id': 2, 'method': 'query', 'parameters': [{'search': 'b'}]},
+            {'id': 3, 'method': 'close', 'parameters': []},
+        ], dispatch)
+
+        assert call_count[0] == 2
+
+    def test_action_response_has_hide(self):
+        launcher = FlowLauncherV2()
+        output = []
+
+        async def dispatch(method, params):
+            return {'Method': 'some_action', 'Parameters': []}
+
+        stdin_text = json.dumps({'id': 5, 'method': 'on_click', 'parameters': [], 'settings': {}}) + '\n'
+        stdin_text += json.dumps({'id': 6, 'method': 'close', 'parameters': []}) + '\n'
+
+        async def _run():
+            with patch('sys.stdin', StringIO(stdin_text)), \
+                 patch('sys.stdout', StringIO()) as mock_out:
+                await launcher.run(dispatch)
+                mock_out.seek(0)
+                for line in mock_out.read().splitlines():
+                    if line:
+                        output.append(json.loads(line))
+
+        asyncio.run(_run())
+        action_resp = next(r for r in output if 'hide' in r)
+        assert action_resp == {'id': 5, 'hide': True}
+
+
+# ---------------------------------------------------------------------------
+# Launcher icons and program_dir
+# ---------------------------------------------------------------------------
+
+class TestLauncherIcons:
+
+    def test_icons_return_path_when_program_dir_set(self, tmp_path):
+        launcher = FlowLauncherV1.__new__(FlowLauncherV1)
+        from pyflowlauncher.icons import Icons
+        launcher.icons = Icons(tmp_path)
+        assert launcher.icons.admin == str(tmp_path / 'Images' / 'admin.png')
+
+    def test_icons_return_none_when_no_program_dir(self):
+        from pyflowlauncher.icons import Icons
+        icons = Icons(None)
+        assert icons.admin is None
+
+    def test_program_dir_from_env(self, tmp_path, monkeypatch):
+        monkeypatch.setenv('FLOW_PROGRAM_DIRECTORY', str(tmp_path))
+        launcher = FlowLauncherV1()
+        assert launcher.program_dir == tmp_path
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 4cc8d37..0807b8e 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -1,5 +1,6 @@
 import pytest
 from pyflowlauncher.plugin import Plugin
+from pyflowlauncher.launcher import Launcher
 
 
 def temp_method1():
@@ -28,8 +29,14 @@ def test_add_methods():
 
 
 def test_settings():
-    plugin = Plugin()
-    plugin._client.recieve = lambda: {'method': 'settings', 'parameters': [], 'settings': {'test': 'test'}}
+    class MockLauncher(Launcher):
+        @property
+        def settings(self):
+            return {'test': 'test'}
+        async def run(self, dispatch):
+            pass
+
+    plugin = Plugin(launcher=MockLauncher())
     assert plugin.settings == {'test': 'test'}