diff --git a/README.md b/README.md
index 2a583c4..5840906 100644
--- a/README.md
+++ b/README.md
@@ -33,6 +33,7 @@ Different versions of the API client support different features depending on you
 | **midclt CLI** |
 | stdin payload (`-` argument) | ❌ | ❌ | ❌ | ✅ |
 | `--insecure` CLI flag | ❌ | ❌ | ❌ | ✅ |
+| Persistent daemon mode (experimental) | ❌ | ❌ | ❌ | ✅ |
 | **Python API** |
 | New-style jobs support | ❌ | ❌ | ✅ | ✅ |
 
@@ -61,6 +62,7 @@ Different versions of the API client support different features depending on you
 - **midclt:** API key file path support - `-K` can now accept file paths directly
 - **midclt:** `--insecure` flag for development/testing with self-signed certificates
 - **midclt:** stdin processing for sensitive payloads (`-` argument)
+- **midclt:** Persistent daemon mode (experimental) for reduced authentication overhead
 
 ## Getting Started
 
@@ -80,6 +82,34 @@ The client's default behavior is to connect to the localhost's middlewared socke
 
 **Note on Performance:** Each `midclt` command invocation incurs significant authentication and auditing overhead. Workloads that poll API endpoints or call endpoints frequently should use the Python API client directly with a persistent authenticated websocket connection (see [Scripting](#scripting) section below) rather than repeatedly invoking `midclt`.
 
+#### Persistent Daemon Mode (Experimental)
+
+**EXPERIMENTAL: This feature is under active development and may change.**
+
+To reduce authentication overhead for frequent `midclt` invocations, you can run a persistent daemon that maintains an authenticated connection. Subsequent `midclt` calls use the `-d` flag to communicate with the daemon via Unix domain socket.
+
+```bash
+# Start daemon in background (authenticates once, logs to ~/.midclt/logs/)
+midclt -u wss://192.168.1.108/api/current -U admin -K /path/to/key.json daemon --lifetime 600 &
+
+# Use daemon for calls (no re-authentication needed, but URI/username required to find socket)
+midclt -u wss://192.168.1.108/api/current -U admin -d call system.info
+midclt -u wss://192.168.1.108/api/current -U admin -d -j call pool.dataset.lock mypool/mydataset
+
+# Stop daemon
+midclt -u wss://192.168.1.108/api/current -U admin daemon-stop
+```
+
+**Daemon Options:**
+- `--lifetime SECONDS`: Idle timeout before daemon exits (default: 600, 0 = no timeout)
+- `--log-file PATH`: Log to file instead of default (`~/.midclt/logs/daemon-{id}.log`)
+
+**Notes:**
+- Daemon uses Unix domain sockets (Linux/macOS/BSD only)
+- Each URI/username combination gets a unique daemon instance
+- Socket and logs stored in `~/.midclt/`
+- Subscription commands not supported via daemon
+
 #### Disable SSL certificate verification
 
 The `--insecure` option disables SSL certificate verification when connecting to a remote TrueNAS instance. This is useful for development or testing environments with self-signed certificates.
diff --git a/tests/daemon/__init__.py b/tests/daemon/__init__.py
new file mode 100644
index 0000000..374730a
--- /dev/null
+++ b/tests/daemon/__init__.py
@@ -0,0 +1 @@
+"""Tests for daemon functionality."""
diff --git a/tests/daemon/test_client.py b/tests/daemon/test_client.py
new file mode 100644
index 0000000..3484741
--- /dev/null
+++ b/tests/daemon/test_client.py
@@ -0,0 +1,239 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Unit tests for daemon client message parsing."""
+
+import json
+import unittest
+from unittest.mock import patch
+
+from truenas_api_client.daemon.client import send_to_daemon
+from truenas_api_client.daemon.constants import MessageType, Command
+
+
+class MockSocket:
+    """Mock socket that returns predefined responses."""
+
+    def __init__(self, responses):
+        """Initialize with list of response strings to return."""
+        self.responses = responses
+        self.response_index = 0
+        self.sent_data = []
+        self.closed = False
+
+    def connect(self, path):
+        """Mock connect."""
+        pass
+
+    def settimeout(self, timeout):
+        """Mock settimeout."""
+        pass
+
+    def sendall(self, data):
+        """Mock sendall - record what was sent."""
+        self.sent_data.append(data)
+
+    def recv(self, bufsize):
+        """Mock recv - return next chunk of response."""
+        if self.response_index < len(self.responses):
+            data = self.responses[self.response_index]
+            self.response_index += 1
+            return data
+        return b''  # EOF
+
+    def close(self):
+        """Mock close."""
+        self.closed = True
+
+
+class TestMessageParsing(unittest.TestCase):
+    """Test message parsing logic."""
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_simple_result_message(self, mock_path, mock_socket_class):
+        """Test parsing a simple result message."""
+        response = {'type': MessageType.RESULT, 'result': 'pong'}
+        response_bytes = (json.dumps(response) + '\n').encode('utf-8')
+
+        mock_sock = MockSocket([response_bytes])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        result = send_to_daemon({'command': Command.PING})
+
+        self.assertEqual(result['type'], MessageType.RESULT)
+        self.assertEqual(result['result'], 'pong')
+        self.assertTrue(mock_sock.closed)
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_error_message(self, mock_path, mock_socket_class):
+        """Test parsing an error message."""
+        response = {
+            'type': MessageType.ERROR,
+            'error': 'Something went wrong',
+            'error_type': 'ValueError'
+        }
+        response_bytes = (json.dumps(response) + '\n').encode('utf-8')
+
+        mock_sock = MockSocket([response_bytes])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        result = send_to_daemon({'command': Command.CALL})
+
+        self.assertEqual(result['type'], MessageType.ERROR)
+        self.assertEqual(result['error'], 'Something went wrong')
+        self.assertTrue(mock_sock.closed)
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_progress_then_result(self, mock_path, mock_socket_class):
+        """Test parsing progress messages followed by result."""
+        progress1 = {
+            'type': MessageType.PROGRESS,
+            'percent': 50,
+            'description': 'Halfway there'
+        }
+        progress2 = {
+            'type': MessageType.PROGRESS,
+            'percent': 100,
+            'description': 'Complete'
+        }
+        result = {'type': MessageType.RESULT, 'result': True}
+
+        # Send as separate chunks
+        responses = [
+            (json.dumps(progress1) + '\n').encode('utf-8'),
+            (json.dumps(progress2) + '\n').encode('utf-8'),
+            (json.dumps(result) + '\n').encode('utf-8'),
+        ]
+
+        mock_sock = MockSocket(responses)
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        progress_updates = []
+
+        def progress_callback(msg):
+            progress_updates.append(msg)
+
+        final_result = send_to_daemon(
+            {'command': Command.CALL},
+            progress_callback=progress_callback
+        )
+
+        # Should have received 2 progress updates
+        self.assertEqual(len(progress_updates), 2)
+        self.assertEqual(progress_updates[0]['percent'], 50)
+        self.assertEqual(progress_updates[1]['percent'], 100)
+
+        # Final result should be the result message
+        self.assertEqual(final_result['type'], MessageType.RESULT)
+        self.assertEqual(final_result['result'], True)
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_progress_without_callback(self, mock_path, mock_socket_class):
+        """Test that progress messages are skipped when no callback provided."""
+        progress = {
+            'type': MessageType.PROGRESS,
+            'percent': 50,
+            'description': 'Halfway'
+        }
+        result = {'type': MessageType.RESULT, 'result': 'done'}
+
+        responses = [
+            (json.dumps(progress) + '\n').encode('utf-8'),
+            (json.dumps(result) + '\n').encode('utf-8'),
+        ]
+
+        mock_sock = MockSocket(responses)
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        # No callback provided - progress should be skipped
+        final_result = send_to_daemon({'command': Command.CALL})
+
+        self.assertEqual(final_result['type'], MessageType.RESULT)
+        self.assertEqual(final_result['result'], 'done')
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_chunked_message(self, mock_path, mock_socket_class):
+        """Test parsing a message received in multiple chunks."""
+        result = {'type': MessageType.RESULT, 'result': 'test'}
+        result_str = json.dumps(result) + '\n'
+        result_bytes = result_str.encode('utf-8')
+
+        # Split into multiple chunks
+        chunk1 = result_bytes[:10]
+        chunk2 = result_bytes[10:20]
+        chunk3 = result_bytes[20:]
+
+        mock_sock = MockSocket([chunk1, chunk2, chunk3])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        final_result = send_to_daemon({'command': Command.PING})
+
+        self.assertEqual(final_result['type'], MessageType.RESULT)
+        self.assertEqual(final_result['result'], 'test')
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_multiple_messages_in_one_chunk(self, mock_path, mock_socket_class):
+        """Test parsing when multiple messages arrive in one chunk."""
+        progress = {'type': MessageType.PROGRESS, 'percent': 50}
+        result = {'type': MessageType.RESULT, 'result': 'done'}
+
+        # Both messages in one chunk
+        combined = (json.dumps(progress) + '\n' + json.dumps(result) + '\n').encode('utf-8')
+
+        mock_sock = MockSocket([combined])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        progress_updates = []
+
+        final_result = send_to_daemon(
+            {'command': Command.CALL},
+            progress_callback=lambda m: progress_updates.append(m)
+        )
+
+        # Should have parsed both messages correctly
+        self.assertEqual(len(progress_updates), 1)
+        self.assertEqual(final_result['type'], MessageType.RESULT)
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_unknown_message_type(self, mock_path, mock_socket_class):
+        """Test handling of unknown message type."""
+        response = {'type': 'unknown_type', 'data': 'something'}
+        response_bytes = (json.dumps(response) + '\n').encode('utf-8')
+
+        mock_sock = MockSocket([response_bytes])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        with self.assertRaises(Exception) as ctx:
+            send_to_daemon({'command': Command.PING})
+
+        self.assertIn('Unknown message type', str(ctx.exception))
+
+    @patch('truenas_api_client.daemon.client.socket.socket')
+    @patch('truenas_api_client.daemon.client.get_daemon_socket_path')
+    def test_connection_closed_prematurely(self, mock_path, mock_socket_class):
+        """Test handling when daemon closes connection without sending response."""
+        # Return empty bytes immediately (connection closed)
+        mock_sock = MockSocket([b''])
+        mock_socket_class.return_value = mock_sock
+        mock_path.return_value = '/tmp/test.sock'
+
+        with self.assertRaises(Exception) as ctx:
+            send_to_daemon({'command': Command.PING})
+
+        self.assertIn('closed connection without response', str(ctx.exception))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/daemon/test_constants.py b/tests/daemon/test_constants.py
new file mode 100644
index 0000000..c83ebbe
--- /dev/null
+++ b/tests/daemon/test_constants.py
@@ -0,0 +1,75 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Unit tests for daemon constants."""
+
+import unittest
+
+from truenas_api_client.daemon.constants import (
+    Command,
+    MessageType,
+    DAEMON_ACCEPT_TIMEOUT,
+    DAEMON_CONNECT_TIMEOUT,
+    DAEMON_REQUEST_TIMEOUT,
+    SOCKET_PERMISSIONS,
+    SOCKET_RECV_BUFFER_SIZE,
+)
+
+
+class TestMessageTypeEnum(unittest.TestCase):
+    """Test MessageType enum."""
+
+    def test_enum_values(self):
+        """MessageType enum should have expected values."""
+        self.assertEqual(MessageType.PROGRESS, 'progress')
+        self.assertEqual(MessageType.RESULT, 'result')
+        self.assertEqual(MessageType.ERROR, 'error')
+
+
+class TestCommandEnum(unittest.TestCase):
+    """Test Command enum."""
+
+    def test_enum_values(self):
+        """Command enum should have expected values."""
+        self.assertEqual(Command.PING, 'ping')
+        self.assertEqual(Command.CALL, 'call')
+        self.assertEqual(Command.STOP, 'stop')
+
+
+class TestTimeoutConstants(unittest.TestCase):
+    """Test timeout constants."""
+
+    def test_request_timeout(self):
+        """Request timeout should be reasonable."""
+        self.assertGreater(DAEMON_REQUEST_TIMEOUT, 0)
+        self.assertGreaterEqual(DAEMON_REQUEST_TIMEOUT, 60)  # At least 1 minute
+
+    def test_accept_timeout(self):
+        """Accept timeout should be short for quick checks."""
+        self.assertGreater(DAEMON_ACCEPT_TIMEOUT, 0)
+        self.assertLessEqual(DAEMON_ACCEPT_TIMEOUT, 10)
+
+    def test_connect_timeout(self):
+        """Connect timeout should be short for quick checks."""
+        self.assertGreater(DAEMON_CONNECT_TIMEOUT, 0)
+        self.assertLessEqual(DAEMON_CONNECT_TIMEOUT, 5)
+
+
+class TestBufferSize(unittest.TestCase):
+    """Test buffer size constant."""
+
+    def test_power_of_two(self):
+        """Buffer size should be power of 2 (common for network buffers)."""
+        # Check if power of 2: n & (n-1) == 0 for powers of 2
+        # This also ensures it's positive
+        self.assertEqual(SOCKET_RECV_BUFFER_SIZE & (SOCKET_RECV_BUFFER_SIZE - 1), 0)
+
+
+class TestSocketPermissions(unittest.TestCase):
+    """Test socket permissions constant."""
+
+    def test_owner_only(self):
+        """Socket permissions should be owner-only (0o600)."""
+        self.assertEqual(SOCKET_PERMISSIONS, 0o600)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/daemon/test_utils.py b/tests/daemon/test_utils.py
new file mode 100644
index 0000000..b33e513
--- /dev/null
+++ b/tests/daemon/test_utils.py
@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Unit tests for daemon utility functions."""
+
+import unittest
+from unittest.mock import patch
+
+from truenas_api_client.daemon.utils import (
+    get_daemon_identifier,
+    get_daemon_socket_path,
+    get_daemon_pid_path,
+)
+
+
+class TestDaemonIdentifier(unittest.TestCase):
+    """Test daemon identifier generation."""
+
+    def test_local_identifier(self):
+        """Local connections should use 'local' identifier."""
+        self.assertEqual(get_daemon_identifier(uri=None, username=None), "local")
+        self.assertEqual(get_daemon_identifier(uri=None, username="admin"), "local")
+
+    def test_deterministic(self):
+        """Same URI/username should produce same identifier."""
+        uri = "wss://192.168.1.100/api/current"
+        username = "admin"
+
+        id1 = get_daemon_identifier(uri, username)
+        id2 = get_daemon_identifier(uri, username)
+
+        self.assertEqual(id1, id2)
+        self.assertEqual(len(id1), 12)  # First 12 chars of SHA256
+
+    def test_unique_combinations(self):
+        """Different URI/username combinations should produce different identifiers."""
+        uri1 = "wss://192.168.1.100/api/current"
+        uri2 = "wss://192.168.1.101/api/current"
+        username1 = "admin"
+        username2 = "user"
+
+        id_uri1_user1 = get_daemon_identifier(uri1, username1)
+        id_uri2_user1 = get_daemon_identifier(uri2, username1)
+        id_uri1_user2 = get_daemon_identifier(uri1, username2)
+
+        # All should be different
+        self.assertNotEqual(id_uri1_user1, id_uri2_user1)
+        self.assertNotEqual(id_uri1_user1, id_uri1_user2)
+        self.assertNotEqual(id_uri2_user1, id_uri1_user2)
+
+
+class TestDaemonPaths(unittest.TestCase):
+    """Test daemon path generation."""
+
+    def test_socket_path_format_local(self):
+        """Socket path for local connection should use 'local' identifier."""
+        with patch('truenas_api_client.daemon.utils.get_daemon_dir', return_value='/tmp/test'):
+            path = get_daemon_socket_path(uri=None, username=None)
+            self.assertEqual(path, '/tmp/test/daemon-local.sock')
+
+    def test_socket_path_format_remote(self):
+        """Socket path for remote connection should have hash identifier."""
+        with patch('truenas_api_client.daemon.utils.get_daemon_dir', return_value='/tmp/test'):
+            path = get_daemon_socket_path(uri="ws://test/api/current", username="admin")
+            self.assertTrue(path.startswith('/tmp/test/daemon-'))
+            self.assertTrue(path.endswith('.sock'))
+            # Should not be 'local'
+            self.assertNotIn('daemon-local', path)
+
+    def test_pid_path_format_local(self):
+        """PID path for local connection should use 'local' identifier."""
+        with patch('truenas_api_client.daemon.utils.get_daemon_dir', return_value='/tmp/test'):
+            path = get_daemon_pid_path(uri=None, username=None)
+            self.assertEqual(path, '/tmp/test/daemon-local.pid')
+
+    def test_pid_path_format_remote(self):
+        """PID path for remote connection should have hash identifier."""
+        with patch('truenas_api_client.daemon.utils.get_daemon_dir', return_value='/tmp/test'):
+            path = get_daemon_pid_path(uri="ws://test/api/current", username="admin")
+            self.assertTrue(path.startswith('/tmp/test/daemon-'))
+            self.assertTrue(path.endswith('.pid'))
+
+    def test_socket_and_pid_paths_match(self):
+        """Socket and PID paths should have matching identifiers."""
+        uri = "wss://192.168.1.100/api/current"
+        username = "admin"
+
+        socket_path = get_daemon_socket_path(uri, username)
+        pid_path = get_daemon_pid_path(uri, username)
+
+        # Extract identifiers from paths
+        socket_id = socket_path.split('daemon-')[1].split('.sock')[0]
+        pid_id = pid_path.split('daemon-')[1].split('.pid')[0]
+
+        self.assertEqual(socket_id, pid_id)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/truenas_api_client/__init__.py b/truenas_api_client/__init__.py
index a860b4c..b588ffb 100644
--- a/truenas_api_client/__init__.py
+++ b/truenas_api_client/__init__.py
@@ -57,6 +57,13 @@
 from . import ejson as json
 from .auth_api_key import api_key_authenticate, APIKeyAuthMech
 from .config import CALL_TIMEOUT
+from .daemon import (
+    call_via_daemon,
+    ping_via_daemon,
+    run_daemon,
+    setup_daemon_logging,
+    stop_daemon,
+)
 from .exc import ReserveFDException, ClientException, ErrnoMixin, ValidationErrors, CallTimeout  # noqa
 from .legacy import LegacyClient
 from .jsonrpc import CollectionUpdateParams, ErrorObj, JobFields, JSONRPCError, JSONRPCMessage, TruenasError
@@ -1007,6 +1014,9 @@ def get_parser():
     parser.add_argument('-P', '--password')
     parser.add_argument('-K', '--api-key')
     parser.add_argument('-t', '--timeout', type=int)
+    parser.add_argument('-d', '--use-daemon',
+                        help='Use persistent daemon connection instead of creating a new connection',
+                        action='store_true')
     parser.add_argument('--insecure',
                         help='Disable SSL verification (WARNING: not for production)',
                         action='store_true')
@@ -1036,6 +1046,23 @@ def get_parser():
     iparser.add_argument('-n', '--number', type=int, help='Number of events to wait before exit')
     iparser.add_argument('-t', '--timeout', type=int)
 
+    # daemon options
+    iparser = subparsers.add_parser('daemon', help='Start a persistent authenticated connection daemon')
+    iparser.add_argument(
+        '--lifetime',
+        type=int,
+        default=600,
+        help='Idle timeout in seconds before daemon exits (default: 600, 0 = no timeout)'
+    )
+    iparser.add_argument(
+        '--log-file',
+        type=str,
+        help='Log to file instead of syslog'
+    )
+
+    # daemon stop
+    subparsers.add_parser('daemon-stop', help='Stop a running daemon')
+
     return parser
 
 
@@ -1080,6 +1107,10 @@ def from_json(args):
                     yield i
 
     if args.name == 'call':
+        # Use daemon if requested
+        if args.use_daemon:
+            sys.exit(call_via_daemon(args, from_json))
+
         try:
             with Client(uri=args.uri, verify_ssl=not args.insecure) as c:
                 try:
@@ -1157,6 +1188,9 @@ def callback(job: _JobDict):
             print('Failed to run middleware call. Daemon not running?', file=sys.stderr)
             sys.exit(1)
     elif args.name == 'ping':
+        if args.use_daemon:
+            sys.exit(ping_via_daemon(args))
+
         with Client(uri=args.uri, verify_ssl=not args.insecure) as c:
             if not (result := c.ping()):
                 sys.exit(1)
@@ -1184,6 +1218,39 @@ def cb(mtype: str, **message):
             if 'error' in subscribe_payload and subscribe_payload['error']:
                 raise ValueError(subscribe_payload['error'])
             sys.exit(0)
+    elif args.name == 'daemon':
+        # Start a persistent daemon
+        # Set up logging for daemon
+        log_file = getattr(args, 'log_file', None)
+        log_file_path = setup_daemon_logging(log_file=log_file, uri=args.uri, username=args.username)
+
+        try:
+            with Client(uri=args.uri, verify_ssl=not args.insecure) as c:
+                # Authenticate
+                try:
+                    if args.username and args.password:
+                        if not c.call('auth.login', args.username, args.password):
+                            raise ValueError('Invalid username or password')
+                    elif args.api_key:
+                        c.login_with_api_key(args.username, args.api_key)
+                except Exception as e:
+                    print(f"Failed to login: {e}", file=sys.stderr)
+                    sys.exit(1)
+
+                # Run daemon with authenticated client
+                run_daemon(c, args.lifetime, uri=args.uri, username=args.username, log_file=log_file_path)
+        except OSError as e:
+            print(f"Error starting daemon: {e}", file=sys.stderr)
+            sys.exit(1)
+        except (FileNotFoundError, ConnectionRefusedError):
+            print('Failed to connect to middleware. Is it running?', file=sys.stderr)
+            sys.exit(1)
+    elif args.name == 'daemon-stop':
+        # Stop a running daemon
+        if stop_daemon(uri=args.uri, username=args.username):
+            sys.exit(0)
+        else:
+            sys.exit(1)
     else:
         parser.print_help()
 
diff --git a/truenas_api_client/daemon/README.md b/truenas_api_client/daemon/README.md
new file mode 100644
index 0000000..00576f6
--- /dev/null
+++ b/truenas_api_client/daemon/README.md
@@ -0,0 +1,100 @@
+# Daemon Module
+
+Persistent connection daemon for `midclt` to reduce authentication overhead.
+
+## Purpose
+
+Each `midclt` invocation creates a new WebSocket connection and authenticates, which has significant overhead. The daemon maintains a persistent authenticated connection that multiple `midclt` calls can reuse via Unix domain sockets.
+
+## Architecture
+
+```
+midclt -d call  -->  Daemon Server  -->  Middleware Server
+   (IPC via Unix socket)  (WebSocket JSON-RPC)
+```
+
+The daemon server runs in a separate process, maintains one authenticated WebSocket connection to middleware, and accepts multiple concurrent client connections via Unix domain socket.
+
+## Modules
+
+### `server.py`
+- `DaemonServer`: ThreadingUnixStreamServer with client lock for thread safety
+- `DaemonRequestHandler`: Handles client requests, forwards to middleware
+- `run_daemon()`: Main server loop with idle timeout
+
+### `client.py`
+- `send_to_daemon()`: Send request to daemon, handle streaming responses
+- `is_daemon_running()`: Check if daemon is alive
+- `stop_daemon()`: Send stop command to daemon
+
+### `cli.py`
+- `call_via_daemon()`: Execute midclt call command via daemon
+- `ping_via_daemon()`: Execute midclt ping command via daemon
+- Progress bar handling for job callbacks
+
+### `utils.py`
+- Path generation: `get_daemon_socket_path()`, `get_daemon_pid_path()`
+- Identifier generation: SHA256 hash of URI/username
+
+### `constants.py`
+- `MessageType`: PROGRESS, RESULT, ERROR
+- `Command`: PING, CALL, STOP
+- Timeout and buffer size constants
+
+### `log_config.py`
+- `setup_daemon_logging()`: Configure logging to file or stderr
+
+## IPC Protocol
+
+Communication uses JSON over Unix domain socket, one message per line.
+
+### Request Format
+```json
+{
+  "command": "call|ping|stop",
+  "method": "api.method.name",
+  "params": [...],
+  "kwargs": {"timeout": 60, "job": true}
+}
+```
+
+### Response Format
+
+**Simple response:**
+```json
+{"type": "result", "result": <value>}
+{"type": "error", "error": "message", "error_type": "ExceptionClass"}
+```
+
+**Job with progress streaming:**
+```json
+{"type": "progress", "percent": 25, "description": "Starting..."}
+{"type": "progress", "percent": 50, "description": "Processing..."}
+{"type": "result", "result": <value>}
+```
+
+## Socket Paths
+
+- **Local**: `~/.midclt/daemon-local.sock`
+- **Remote**: `~/.midclt/daemon-{hash}.sock` (hash = first 12 chars of SHA256 of `uri:username`)
+- **PID**: Same pattern with `.pid` extension
+- **Logs**: `~/.midclt/logs/daemon-{identifier}.log`
+
+## Thread Safety
+
+The daemon uses a single lock (`client_lock`) to serialize all middleware API calls. This ensures the WebSocket connection is not accessed concurrently, as the underlying `JSONRPCClient` is not thread-safe.
+
+## Lifecycle
+
+1. Daemon starts, authenticates to middleware
+2. Binds Unix socket with 0o600 permissions (owner-only)
+3. Writes PID file with fsync
+4. Accepts connections in threaded mode
+5. Each request holds `client_lock` for entire request/response cycle
+6. Exits after idle timeout or on SIGTERM/SIGINT
+
+## Limitations
+
+- Unix domain sockets only (Linux/macOS/BSD)
+- One daemon per URI/username combination
+- Subscription commands not supported via daemon
diff --git a/truenas_api_client/daemon/__init__.py b/truenas_api_client/daemon/__init__.py
new file mode 100644
index 0000000..0e798a1
--- /dev/null
+++ b/truenas_api_client/daemon/__init__.py
@@ -0,0 +1,20 @@
+"""Daemon functionality for persistent midclt connections."""
+
+from .cli import call_via_daemon, ping_via_daemon
+from .client import is_daemon_running, send_to_daemon, stop_daemon
+from .log_config import setup_daemon_logging
+from .server import run_daemon
+from .utils import get_daemon_identifier, get_daemon_pid_path, get_daemon_socket_path
+
+__all__ = [
+    'call_via_daemon',
+    'ping_via_daemon',
+    'is_daemon_running',
+    'send_to_daemon',
+    'stop_daemon',
+    'run_daemon',
+    'setup_daemon_logging',
+    'get_daemon_identifier',
+    'get_daemon_socket_path',
+    'get_daemon_pid_path',
+]
diff --git a/truenas_api_client/daemon/cli.py b/truenas_api_client/daemon/cli.py
new file mode 100644
index 0000000..a34cc5d
--- /dev/null
+++ b/truenas_api_client/daemon/cli.py
@@ -0,0 +1,134 @@
+"""CLI helper functions for daemon integration."""
+import pprint
+import sys
+
+from .. import ejson as json
+from ..utils import ProgressBar
+from .client import is_daemon_running, send_to_daemon
+from .constants import Command, MessageType
+
+
+def call_via_daemon(args, from_json_func):
+    """Execute a midclt call command via the daemon.
+
+    Args:
+        args: Parsed command-line arguments.
+        from_json_func: Function to parse JSON arguments.
+
+    Returns:
+        Exit code (0 for success, 1 for error).
+    """
+    # Check if daemon is running
+    if not is_daemon_running(uri=args.uri, username=args.username):
+        print("Error: No daemon is running. Start one with 'midclt daemon'", file=sys.stderr)
+        return 1
+
+    try:
+        # Prepare call parameters
+        params = list(from_json_func(args.method[1:]))
+        kwargs = {}
+        if args.timeout:
+            kwargs['timeout'] = args.timeout
+        if args.job:
+            kwargs['job'] = True
+
+        # Set up progress callback if this is a job
+        progress_bar = None
+
+        if args.job:
+            if args.job_print == 'progressbar':
+                progress_bar = ProgressBar()
+                progress_bar.__enter__()
+
+                def progress_callback(progress_msg):
+                    """Update progress bar with info from daemon."""
+                    try:
+                        progress_bar.update(
+                            progress_msg.get('percent', 0),
+                            progress_msg.get('description', '')
+                        )
+                    except Exception as e:
+                        print(f'Failed to update progress bar: {e!s}', file=sys.stderr)
+            else:
+                lastdesc = ['']  # Use list to avoid nonlocal in nested function
+
+                def progress_callback(progress_msg):
+                    """Print job description to stderr if it has changed."""
+                    desc = progress_msg.get('description', '')
+                    if desc and desc != lastdesc[0]:
+                        print(desc, file=sys.stderr)
+                        lastdesc[0] = desc
+        else:
+            progress_callback = None
+
+        # Send to daemon
+        response = send_to_daemon({
+            'command': Command.CALL,
+            'method': args.method[0],
+            'params': params,
+            'kwargs': kwargs
+        }, uri=args.uri, username=args.username, progress_callback=progress_callback)
+
+        # Finish progress bar if we used one
+        if progress_bar:
+            progress_bar.finish()
+            progress_bar.__exit__(None, None, None)
+
+        # Handle response
+        if response.get('type') == MessageType.ERROR:
+            if not args.quiet:
+                print(response['error'], file=sys.stderr)
+                if response.get('trace'):
+                    print(response['trace'].get('formatted', ''), file=sys.stderr)
+                if response.get('extra'):
+                    pprint.pprint(response['extra'], stream=sys.stderr)
+            return 1
+
+        # Print result
+        rv = response['result']
+        if isinstance(rv, (int, str)):
+            print(rv)
+        else:
+            print(json.dumps(rv))
+        return 0
+
+    except Exception as e:
+        if progress_bar:
+            progress_bar.__exit__(None, None, None)
+        print(f"Error communicating with daemon: {e}", file=sys.stderr)
+        return 1
+
+
+def ping_via_daemon(args):
+    """Execute a midclt ping command via the daemon.
+
+    Args:
+        args: Parsed command-line arguments.
+
+    Returns:
+        Exit code (0 for success, 1 for error).
+    """
+    # Check if daemon is running
+    if not is_daemon_running(uri=args.uri, username=args.username):
+        print("Error: No daemon is running. Start one with 'midclt daemon'", file=sys.stderr)
+        return 1
+
+    try:
+        # Send ping to daemon
+        response = send_to_daemon({
+            'command': 'ping',
+            'timeout': getattr(args, 'timeout', 10)
+        }, uri=args.uri, username=args.username)
+
+        # Handle response
+        if response.get('type') == MessageType.ERROR:
+            print(f"Ping failed: {response['error']}", file=sys.stderr)
+            return 1
+
+        # Print result
+        print(response['result'])
+        return 0
+
+    except Exception as e:
+        print(f"Error communicating with daemon: {e}", file=sys.stderr)
+        return 1
diff --git a/truenas_api_client/daemon/client.py b/truenas_api_client/daemon/client.py
new file mode 100644
index 0000000..d88b859
--- /dev/null
+++ b/truenas_api_client/daemon/client.py
@@ -0,0 +1,136 @@
+"""Client-side functions for communicating with daemon."""
+import os
+import socket
+import sys
+
+from .. import ejson as json
+from .constants import (
+    DAEMON_CONNECT_TIMEOUT,
+    DAEMON_REQUEST_TIMEOUT,
+    Command,
+    MessageType,
+    SOCKET_RECV_BUFFER_SIZE,
+)
+from .utils import cleanup_stale_files, get_daemon_pid_path, get_daemon_socket_path
+
+
+def is_daemon_running(uri=None, username=None):
+    """Check if a daemon is currently running and responsive.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+    """
+    socket_path = get_daemon_socket_path(uri, username)
+    pid_path = get_daemon_pid_path(uri, username)
+
+    if not os.path.exists(socket_path):
+        return False
+
+    # Check if PID file exists and process is alive
+    if os.path.exists(pid_path):
+        try:
+            with open(pid_path, 'r') as f:
+                pid = int(f.read().strip())
+            # Check if process exists (signal 0 doesn't kill, just checks)
+            os.kill(pid, 0)
+        except (OSError, ValueError):
+            # Process doesn't exist, clean up stale files
+            cleanup_stale_files(uri, username)
+            return False
+
+    # Try to connect to verify it's responsive
+    try:
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.settimeout(DAEMON_CONNECT_TIMEOUT)
+        sock.connect(socket_path)
+        sock.close()
+        return True
+    except (socket.error, socket.timeout, AttributeError):
+        return False
+
+
+def send_to_daemon(request_data, uri=None, username=None, progress_callback=None):
+    """Send a request to the daemon and receive the response.
+
+    Args:
+        request_data: Dictionary containing the request to forward to middleware.
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+        progress_callback: Optional callback function for job progress updates.
+            Called with dict containing 'percent', 'description', 'state'.
+
+    Returns:
+        The response from middleware (dict with 'type' and either 'result' or 'error').
+
+    Raises:
+        Exception: If communication with daemon fails.
+    """
+    socket_path = get_daemon_socket_path(uri, username)
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+
+    try:
+        sock.settimeout(DAEMON_REQUEST_TIMEOUT)
+        sock.connect(socket_path)
+
+        # Send request as JSON
+        request_json = json.dumps(request_data) + '\n'
+        sock.sendall(request_json.encode('utf-8'))
+
+        # Receive response(s) - may be multiple messages for jobs with progress
+        buffer = b''
+        while True:
+            # Read until we have a complete message (ends with newline)
+            while b'\n' not in buffer:
+                chunk = sock.recv(SOCKET_RECV_BUFFER_SIZE)
+                if not chunk:
+                    raise Exception("Daemon closed connection without response")
+                buffer += chunk
+
+            # Extract one message
+            line, buffer = buffer.split(b'\n', 1)
+            if not line:
+                continue
+
+            response = json.loads(line.decode('utf-8'))
+            msg_type = response.get('type')
+
+            # Handle progress updates
+            if msg_type == MessageType.PROGRESS:
+                if progress_callback:
+                    progress_callback(response)
+                continue  # Keep reading for final result
+
+            # Handle result or error (final message)
+            elif msg_type in (MessageType.RESULT, MessageType.ERROR):
+                return response
+
+            else:
+                raise Exception(f"Unknown message type from daemon: {msg_type}")
+
+    finally:
+        sock.close()
+
+
+def stop_daemon(uri=None, username=None):
+    """Stop a running daemon.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+    """
+    if not is_daemon_running(uri, username):
+        print("No daemon is currently running", file=sys.stderr)
+        return False
+
+    try:
+        response = send_to_daemon({'command': Command.STOP}, uri, username)
+        if MessageType.RESULT in response:
+            print(response['result'], file=sys.stderr)
+            return True
+        else:
+            print(f"Error stopping daemon: {response.get('error', 'Unknown error')}", file=sys.stderr)
+            return False
+    except Exception as e:
+        print(f"Error communicating with daemon: {e}", file=sys.stderr)
+        return False
diff --git a/truenas_api_client/daemon/constants.py b/truenas_api_client/daemon/constants.py
new file mode 100644
index 0000000..b3debbd
--- /dev/null
+++ b/truenas_api_client/daemon/constants.py
@@ -0,0 +1,28 @@
+"""Constants and enums for daemon functionality."""
+from enum import StrEnum
+
+
+class MessageType(StrEnum):
+    """Types of messages in daemon IPC protocol."""
+    PROGRESS = 'progress'
+    RESULT = 'result'
+    ERROR = 'error'
+
+
+class Command(StrEnum):
+    """Commands that can be sent to the daemon."""
+    PING = 'ping'
+    CALL = 'call'
+    STOP = 'stop'
+
+
+# Timeout constants (in seconds)
+DAEMON_REQUEST_TIMEOUT = 300  # 5 minutes for long operations
+DAEMON_ACCEPT_TIMEOUT = 1.0  # Check for shutdown/timeout every second
+DAEMON_CONNECT_TIMEOUT = 1  # Quick check if daemon is running
+
+# Buffer sizes
+SOCKET_RECV_BUFFER_SIZE = 4096
+
+# File permissions
+SOCKET_PERMISSIONS = 0o600  # Only user can access
diff --git a/truenas_api_client/daemon/log_config.py b/truenas_api_client/daemon/log_config.py
new file mode 100644
index 0000000..ae0fef1
--- /dev/null
+++ b/truenas_api_client/daemon/log_config.py
@@ -0,0 +1,49 @@
+"""Logging configuration for daemon."""
+import logging
+import os
+import sys
+
+
+def setup_daemon_logging(log_file=None, uri=None, username=None):
+    """Set up logging for daemon process.
+
+    Args:
+        log_file: Optional path to log file. If None, defaults to daemon log directory.
+        uri: The middleware URI (for default log file naming).
+        username: Optional username (for default log file naming).
+
+    Returns:
+        str: The log file path being used.
+    """
+    from .utils import get_daemon_dir, get_daemon_identifier
+
+    logger = logging.getLogger()
+    logger.setLevel(logging.INFO)
+
+    formatter = logging.Formatter(
+        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S'
+    )
+
+    if log_file is None:
+        # Default to log file in daemon directory
+        daemon_dir = get_daemon_dir()
+        log_dir = os.path.join(daemon_dir, 'logs')
+        os.makedirs(log_dir, exist_ok=True)
+
+        identifier = get_daemon_identifier(uri, username)
+        log_file = os.path.join(log_dir, f'daemon-{identifier}.log')
+
+    try:
+        handler = logging.FileHandler(log_file)
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+        logger.info("Logging to %s", log_file)
+        return log_file
+    except (OSError, PermissionError) as e:
+        # Fall back to stderr if we can't write to file
+        handler = logging.StreamHandler(sys.stderr)
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+        logger.error("Failed to open log file %s: %s. Logging to stderr.", log_file, e)
+        return None
diff --git a/truenas_api_client/daemon/server.py b/truenas_api_client/daemon/server.py
new file mode 100644
index 0000000..5ebc1e7
--- /dev/null
+++ b/truenas_api_client/daemon/server.py
@@ -0,0 +1,246 @@
+"""Daemon server implementation."""
+import logging
+import os
+import signal
+import socket
+import socketserver
+import sys
+import time
+from threading import Event, Lock
+from typing import cast
+
+from .. import ejson as json
+from ..exc import ClientException
+from .constants import (
+    DAEMON_ACCEPT_TIMEOUT,
+    Command,
+    MessageType,
+    SOCKET_PERMISSIONS,
+)
+from .utils import cleanup_stale_files, get_daemon_pid_path, get_daemon_socket_path
+
+logger = logging.getLogger(__name__)
+
+
+class DaemonRequestHandler(socketserver.StreamRequestHandler):
+    """Handler for daemon client requests."""
+
+    def handle(self):
+        """Process a single request from a client."""
+        server = cast(DaemonServer, self.server)
+        try:
+            # Update last activity time (using monotonic for timeout tracking)
+            server.last_activity = time.monotonic()
+
+            # Read request (one line of JSON)
+            request_line = self.rfile.readline().decode('utf-8').strip()
+            if not request_line:
+                return
+
+            request = json.loads(request_line)
+            command = request.get('command')
+
+            logger.debug("Received command: %s", command)
+
+            # Get the authenticated client from the server
+            client = server.client
+
+            # Serialize all client operations to ensure thread safety
+            # Hold lock for entire request/response cycle on the websocket
+            with server.client_lock:
+                # Handle different commands
+                if command == Command.PING:
+                    try:
+                        result = client.ping(timeout=request.get('timeout', 10))
+                        response = {'type': MessageType.RESULT, 'result': result}
+                    except Exception as e:
+                        logger.error("Ping failed: %s", e)
+                        response = {'type': MessageType.ERROR, 'error': str(e), 'error_type': type(e).__name__}
+
+                elif command == Command.CALL:
+                    method = request['method']
+                    params = request.get('params', [])
+                    kwargs = request.get('kwargs', {})
+                    is_job = kwargs.get('job', False)
+
+                    logger.debug("Calling method: %s (job=%s)", method, is_job)
+
+                    try:
+                        # For jobs, set up progress callback to stream updates
+                        if is_job:
+                            def job_callback(job_dict):
+                                """Send job progress updates to client."""
+                                try:
+                                    progress_msg = json.dumps({
+                                        'type': MessageType.PROGRESS,
+                                        'percent': job_dict.get('progress', {}).get('percent', 0),
+                                        'description': job_dict.get('progress', {}).get('description', ''),
+                                        'state': job_dict.get('state', ''),
+                                    }) + '\n'
+                                    self.wfile.write(progress_msg.encode('utf-8'))
+                                    self.wfile.flush()
+                                except Exception as e:
+                                    logger.error("Error sending progress update: %s", e)
+
+                            kwargs['callback'] = job_callback
+
+                        result = client.call(method, *params, **kwargs)
+                        response = {'type': MessageType.RESULT, 'result': result}
+                    except ClientException as e:
+                        logger.error("Call to %s failed: %s", method, e)
+                        response = {
+                            'type': MessageType.ERROR,
+                            'error': e.error or str(e),
+                            'error_type': 'ClientException',
+                            'trace': e.trace,
+                            'extra': e.extra,
+                        }
+                    except Exception as e:
+                        logger.error("Call to %s failed: %s", method, e, exc_info=True)
+                        response = {
+                            'type': MessageType.ERROR,
+                            'error': str(e),
+                            'error_type': type(e).__name__,
+                        }
+
+                elif command == Command.STOP:
+                    logger.info("Received stop command")
+                    response = {'type': MessageType.RESULT, 'result': 'Daemon shutting down'}
+                    # Send response before shutting down
+                    response_json = json.dumps(response) + '\n'
+                    self.wfile.write(response_json.encode('utf-8'))
+                    self.wfile.flush()
+                    # Trigger shutdown
+                    server.shutdown_event.set()
+                    return
+
+                else:
+                    logger.warning("Unknown command: %s", command)
+                    response = {
+                        'type': MessageType.ERROR,
+                        'error': f'Unknown command: {command}',
+                        'error_type': 'ValueError'
+                    }
+
+            # Send response (outside lock, but inside try block)
+            response_json = json.dumps(response) + '\n'
+            self.wfile.write(response_json.encode('utf-8'))
+            self.wfile.flush()
+
+        except Exception as e:
+            logger.error("Error handling request: %s", e, exc_info=True)
+            # Send error response
+            try:
+                error_response = json.dumps({
+                    'type': MessageType.ERROR,
+                    'error': str(e),
+                    'error_type': type(e).__name__,
+                }) + '\n'
+                self.wfile.write(error_response.encode('utf-8'))
+                self.wfile.flush()
+            except Exception:
+                pass
+
+
+class DaemonServer(socketserver.ThreadingUnixStreamServer):
+    """Unix socket server for the midclt daemon."""
+
+    def __init__(self, socket_path, client, lifetime):
+        """Initialize the daemon server.
+
+        Args:
+            socket_path: Path to Unix socket.
+            client: Authenticated Client instance.
+            lifetime: Idle timeout in seconds (0 = no timeout).
+        """
+        self.client = client
+        self.client_lock = Lock()  # Serialize all client operations
+        self.lifetime = lifetime
+        self.last_activity = time.monotonic()
+        self.shutdown_event = Event()
+
+        super().__init__(socket_path, DaemonRequestHandler)
+
+
+def run_daemon(client, lifetime, uri=None, username=None, log_file=None):
+    """Run the daemon server that forwards requests to middleware.
+
+    Args:
+        client: Authenticated Client instance with persistent connection.
+        lifetime: Maximum idle time in seconds before daemon exits (0 = no timeout).
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+        log_file: Path to log file (for stdout message).
+
+    Raises:
+        OSError: Platform does not support Unix domain sockets.
+    """
+    # Check if platform supports Unix domain sockets
+    if not hasattr(socket, 'AF_UNIX'):
+        raise OSError("Unix domain sockets not supported on this platform")
+
+    # Additional platform check for better error messages
+    if sys.platform not in ('linux', 'darwin', 'freebsd', 'openbsd', 'netbsd'):
+        # Still allow it if AF_UNIX exists (e.g., newer Windows), but warn about potential issues
+        logger.warning("Daemon mode on %s is experimental", sys.platform)
+
+    socket_path = get_daemon_socket_path(uri, username)
+    pid_path = get_daemon_pid_path(uri, username)
+
+    # Clean up any stale socket
+    cleanup_stale_files(uri, username)
+
+    # Write PID file with proper flushing and syncing
+    with open(pid_path, 'w') as f:
+        f.write(str(os.getpid()))
+        f.flush()
+        os.fsync(f.fileno())
+
+    try:
+        # Create server
+        server = DaemonServer(socket_path, client, lifetime)
+
+        # Set permissions so only user can connect (Unix only)
+        if hasattr(os, 'chmod'):
+            os.chmod(socket_path, SOCKET_PERMISSIONS)
+
+        logger.info("Daemon started with PID %d", os.getpid())
+        logger.info("Socket: %s", socket_path)
+        if lifetime > 0:
+            logger.info("Idle timeout: %d seconds", lifetime)
+        else:
+            logger.info("Idle timeout: disabled")
+
+        # Print to stdout for user visibility
+        if log_file:
+            print(f"Accepting connections. Log file: {log_file}", flush=True)
+        else:
+            print("Accepting connections. Logging to stderr.", flush=True)
+
+        # Signal handler for graceful shutdown
+        def signal_handler(signum, frame):
+            logger.info("Received signal %d, shutting down...", signum)
+            server.shutdown_event.set()
+
+        signal.signal(signal.SIGTERM, signal_handler)
+        signal.signal(signal.SIGINT, signal_handler)
+
+        # Main loop: serve requests and check for timeout
+        server.timeout = DAEMON_ACCEPT_TIMEOUT
+
+        while not server.shutdown_event.is_set():
+            server.handle_request()
+
+            # Check idle timeout (using monotonic time)
+            if lifetime > 0 and (time.monotonic() - server.last_activity) > lifetime:
+                logger.info("Idle timeout of %d seconds exceeded, shutting down", lifetime)
+                break
+
+    finally:
+        # Cleanup
+        try:
+            server.server_close()
+        except Exception as e:
+            logger.error("Error closing server: %s", e)
+        cleanup_stale_files(uri, username)
+        logger.info("Daemon stopped")
diff --git a/truenas_api_client/daemon/utils.py b/truenas_api_client/daemon/utils.py
new file mode 100644
index 0000000..44865fb
--- /dev/null
+++ b/truenas_api_client/daemon/utils.py
@@ -0,0 +1,68 @@
+"""Utility functions for daemon paths and identifiers."""
+import hashlib
+import os
+
+
+def get_daemon_dir():
+    """Get the directory for daemon socket and PID file."""
+    daemon_dir = os.path.expanduser('~/.midclt')
+    os.makedirs(daemon_dir, exist_ok=True)
+    return daemon_dir
+
+
+def get_daemon_identifier(uri=None, username=None):
+    """Get unique identifier for daemon based on connection parameters.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+
+    Returns:
+        A unique identifier string for this daemon configuration.
+    """
+    if uri is None:
+        return "local"
+
+    # Create unique identifier from URI and username
+    key = f"{uri}:{username}" if username else uri
+    hash_digest = hashlib.sha256(key.encode()).hexdigest()
+    return hash_digest[:12]  # Use first 12 chars for reasonable uniqueness
+
+
+def get_daemon_socket_path(uri=None, username=None):
+    """Get the path to the daemon Unix socket.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+    """
+    identifier = get_daemon_identifier(uri, username)
+    return os.path.join(get_daemon_dir(), f'daemon-{identifier}.sock')
+
+
+def get_daemon_pid_path(uri=None, username=None):
+    """Get the path to the daemon PID file.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+    """
+    identifier = get_daemon_identifier(uri, username)
+    return os.path.join(get_daemon_dir(), f'daemon-{identifier}.pid')
+
+
+def cleanup_stale_files(uri=None, username=None):
+    """Remove stale socket and PID files.
+
+    Args:
+        uri: The middleware URI (None for local).
+        username: Optional username for the connection.
+    """
+    try:
+        os.unlink(get_daemon_socket_path(uri, username))
+    except OSError:
+        pass
+    try:
+        os.unlink(get_daemon_pid_path(uri, username))
+    except OSError:
+        pass