VSCodeClient

The main entry point for connecting to VS Code.

VSCodeClient

VSCodeClient(pipe_path: Optional[str] = None)

Main client for interacting with VS Code via Sockpuppet extension.

Initialize the VS Code client.

Parameters:

Name	Type	Description	Default
pipe_path	Optional[str]	Path to the named pipe/socket. If None, checks VSCODE_SOCKPUPPET_PIPE environment variable, then falls back to platform default.	None

Source code in vscode_sockpuppet/client.py

def __init__(self, pipe_path: Optional[str] = None):
    """
    Initialize the VS Code client.

    Args:
        pipe_path: Path to the named pipe/socket.
                  If None, checks VSCODE_SOCKPUPPET_PIPE environment
                  variable, then falls back to platform default.
    """
    if pipe_path is None:
        # First try environment variable (set by other extensions)
        pipe_path = os.environ.get("VSCODE_SOCKPUPPET_PIPE")

    if pipe_path is None:
        # Use platform-specific default paths
        if os.name == "nt":  # Windows
            pipe_path = r"\\.\pipe\vscode-sockpuppet"
        else:  # Unix/Linux/Mac
            pipe_path = os.path.join(tempfile.gettempdir(), "vscode-sockpuppet.sock")

    self.pipe_path = pipe_path
    self.sock: Optional[Union[socket.socket, Any]] = None
    self._request_id = 0
    self._lock = threading.Lock()
    self._buffer = ""
    # Emitters map: event name -> EventEmitter
    self._emitters: Dict[str, EventEmitter] = {}
    self._event_thread: Optional[threading.Thread] = None
    self._running = False
    self._session_listeners: list[Callable[[str, Dict[str, Any]], None]] = []
    self._session_listener_lock = threading.Lock()

    # API namespaces
    self.window = Window(self)
    self.workspace = Workspace(self)
    self.editor = Editor(self)
    self.fs = FileSystem(self)
    self.languages = Languages(self)
    self.lm = LanguageModel(self)

connect

connect() -> None

Connect to the VS Code extension via named pipe/socket.

Source code in vscode_sockpuppet/client.py

def connect(self) -> None:
    """Connect to the VS Code extension via named pipe/socket."""
    if os.name == "nt":  # Windows named pipe
        try:
            # On Windows, use standard file operations for named pipes
            # Python 3 can open named pipes like regular files
            import time

            # Wait for pipe to be available (retry for up to 5 seconds)
            max_retries = 50
            retry_delay = 0.1

            for attempt in range(max_retries):
                try:
                    # Open named pipe with read/write binary mode
                    self.sock = open(self.pipe_path, "r+b", buffering=0)
                    break
                except FileNotFoundError as e:
                    if attempt < max_retries - 1:
                        time.sleep(retry_delay)
                    else:
                        raise ConnectionError(
                            f"Named pipe not found: {self.pipe_path}. "
                            "Make sure VS Code extension is running."
                        ) from e
                except PermissionError as e:
                    raise ConnectionError(f"Permission denied accessing pipe: {e}") from e
        except Exception as e:
            if not isinstance(e, ConnectionError):
                raise ConnectionError(
                    f"Could not connect to VS Code. Make sure extension is running. Error: {e}"
                ) from e
            raise
    else:  # Unix domain socket
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            self.sock.connect(self.pipe_path)
        except Exception as e:
            raise ConnectionError(
                f"Could not connect to VS Code. Make sure extension is running. Error: {e}"
            ) from e

disconnect

disconnect() -> None

Disconnect from the VS Code extension.

Source code in vscode_sockpuppet/client.py

def disconnect(self) -> None:
    """Disconnect from the VS Code extension."""
    if self.sock:
        try:
            if hasattr(self.sock, "close"):
                self.sock.close()
        except Exception:
            pass
        self.sock = None
    self._notify_session_listeners("event-loop-stopped", {})

is_connected

is_connected() -> bool

Check if connected to VS Code.

Source code in vscode_sockpuppet/client.py

def is_connected(self) -> bool:
    """Check if connected to VS Code."""
    return self.sock is not None

execute_command

execute_command(command: str, *args) -> Any

Execute a VS Code command.

Parameters:

Name	Type	Description	Default
command	str	The command identifier	required
*args		Arguments to pass to the command	()

Returns:

Type	Description
Any	The result of the command execution

Source code in vscode_sockpuppet/client.py

def execute_command(self, command: str, *args) -> Any:
    """
    Execute a VS Code command.

    Args:
        command: The command identifier
        *args: Arguments to pass to the command

    Returns:
        The result of the command execution
    """
    return self._send_request(
        "commands.executeCommand", {"command": command, "args": list(args)}
    )

get_commands

get_commands(filter_internal: bool = False) -> list[str]

Get all available VS Code commands.

Parameters:

Name	Type	Description	Default
filter_internal	bool	Whether to filter internal commands	False

Returns:

Type	Description
list[str]	List of command identifiers

Source code in vscode_sockpuppet/client.py

def get_commands(self, filter_internal: bool = False) -> list[str]:
    """
    Get all available VS Code commands.

    Args:
        filter_internal: Whether to filter internal commands

    Returns:
        List of command identifiers
    """
    return self._send_request("commands.getCommands", {"filterInternal": filter_internal})

add_event_listener

add_event_listener(event: str, handler: Callable[[Any], None]) -> Callable[[], None]

Register a handler for a server-sent event and return an unsubscribe callable.

This replaces the old subscribe() API. The first listener for an event will trigger a server-side subscription (events.subscribe) and the last listener removed will trigger events.unsubscribe.

Source code in vscode_sockpuppet/client.py

def add_event_listener(self, event: str, handler: Callable[[Any], None]) -> Callable[[], None]:
    """Register a handler for a server-sent event and return an unsubscribe callable.

    This replaces the old `subscribe()` API. The first listener for an event
    will trigger a server-side subscription (`events.subscribe`) and the last
    listener removed will trigger `events.unsubscribe`.
    """
    # Ensure the background event loop is running
    if not self._running and self._event_thread is None:
        self._running = True
        self._event_thread = threading.Thread(target=self._event_loop, daemon=True)
        self._event_thread.start()

    emitter = self.get_emitter(event)
    return emitter.event(handler)

remove_event_listener

remove_event_listener(event: str, handler: Optional[Callable] = None) -> None

Remove a previously registered event handler or all handlers for an event.

If handler is None, explicitly unsubscribe from the server and remove the emitter.

Source code in vscode_sockpuppet/client.py

def remove_event_listener(self, event: str, handler: Optional[Callable] = None) -> None:
    """Remove a previously registered event handler or all handlers for an event.

    If handler is None, explicitly unsubscribe from the server and remove the emitter.
    """
    if event not in self._emitters:
        return

    emitter = self._emitters[event]

    if handler is None:
        # Remove all handlers and explicitly unsubscribe from server
        try:
            # avoid double-unsubscribe race by directly sending and dropping emitter
            self._send_request("events.unsubscribe", {"event": event})
        finally:
            del self._emitters[event]
    else:
        emitter.remove(handler)
        # If emitter has no listeners, remove it from map (on_no_listeners already sent ack)
        if not emitter.has_listeners():
            if event in self._emitters:
                del self._emitters[event]

get_subscriptions

get_subscriptions() -> list[str]

Get list of currently subscribed events.

Returns:

Type	Description
list[str]	List of event names

Source code in vscode_sockpuppet/client.py

def get_subscriptions(self) -> list[str]:
    """
    Get list of currently subscribed events.

    Returns:
        List of event names
    """
    return self._send_request("events.listSubscriptions")

get_emitter

get_emitter(event: str) -> EventEmitter

Return the EventEmitter for a named event, creating it if necessary.

This allows callers to access the per-event emitter when they need to inspect or interact with listeners directly. It preserves the same on_first_add/on_no_listeners hook behavior as before.

Source code in vscode_sockpuppet/client.py

def get_emitter(self, event: str) -> EventEmitter:
    """Return the EventEmitter for a named event, creating it if necessary.

    This allows callers to access the per-event emitter when they need to
    inspect or interact with listeners directly. It preserves the same
    on_first_add/on_no_listeners hook behavior as before.
    """
    if event not in self._emitters:

        def _on_first_add() -> None:
            try:
                self._send_request("events.subscribe", {"event": event})
            finally:
                self._notify_session_listeners("subscription-ack", {"event": event})

        def _on_no_listeners() -> None:
            try:
                self._send_request("events.unsubscribe", {"event": event})
            finally:
                self._notify_session_listeners("unsubscription-ack", {"event": event})

        self._emitters[event] = EventEmitter(
            on_first_add=_on_first_add, on_no_listeners=_on_no_listeners
        )

    return self._emitters[event]

__enter__

__enter__()

Context manager entry.

Source code in vscode_sockpuppet/client.py

def __enter__(self):
    """Context manager entry."""
    self.connect()
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in vscode_sockpuppet/client.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""
    self._running = False
    if self._event_thread:
        self._event_thread.join(timeout=1.0)
    self.disconnect()

add_session_listener

add_session_listener(listener: Callable[[str, Dict[str, Any]], None]) -> Callable[[], None]

Register a listener for session state notifications.

Source code in vscode_sockpuppet/client.py

def add_session_listener(
    self, listener: Callable[[str, Dict[str, Any]], None]
) -> Callable[[], None]:
    """Register a listener for session state notifications."""

    with self._session_listener_lock:
        self._session_listeners.append(listener)

    def _remove() -> None:
        self.remove_session_listener(listener)

    return _remove

remove_session_listener

remove_session_listener(listener: Callable[[str, Dict[str, Any]], None]) -> None

Remove a previously registered session listener.

Source code in vscode_sockpuppet/client.py

def remove_session_listener(self, listener: Callable[[str, Dict[str, Any]], None]) -> None:
    """Remove a previously registered session listener."""

    with self._session_listener_lock:
        if listener in self._session_listeners:
            self._session_listeners.remove(listener)

session_listener

session_listener(listener: Callable[[str, Dict[str, Any]], None]) -> Any

Context manager to automatically add and remove a session listener.

Source code in vscode_sockpuppet/client.py

@contextmanager
def session_listener(self, listener: Callable[[str, Dict[str, Any]], None]) -> Any:
    """Context manager to automatically add and remove a session listener."""

    remove = self.add_session_listener(listener)
    try:
        yield listener
    finally:
        remove()

subscription

subscription(event: str, handler: Callable[[Any], None]) -> Any

Context manager that registers a subscription and ensures it's removed.

Example:

with client.subscription('webview.onDidReceiveMessage', handler):
    # handler will receive events here
    ...

Source code in vscode_sockpuppet/client.py

@contextmanager
def subscription(self, event: str, handler: Callable[[Any], None]) -> Any:
    """Context manager that registers a subscription and ensures it's removed.

    Example:

        with client.subscription('webview.onDidReceiveMessage', handler):
            # handler will receive events here
            ...
    """

    # Register the handler and ensure we remove it when the context exits.
    unsubscribe = self.add_event_listener(event, handler)
    try:
        yield handler
    finally:
        try:
            unsubscribe()
        except Exception:
            pass