Most projects interact with the IPC server through CmdMox, but advanced
scenarios can instantiate cmd_mox.ipc.IPCServer themselves. The server
accepts optional callbacks so invocation handling can be customised without
subclassing:
from cmd_mox.ipc import IPCHandlers, IPCServer, Response
def handle(invocation):
return Response(stdout="custom output")
handlers = IPCHandlers(handler=handle)
with IPCServer(socket_path, handlers=handlers):
...
Providing passthrough_handler= to IPCHandlers intercepts passthrough
completions in the same fashion. When no callbacks are supplied the server
keeps its default echo behaviour, so existing code continues to work unchanged.
On Windows the transport can be forced explicitly by swapping IPCServer for
:class:NamedPipeServer; CmdMox selects it automatically based on
os.name.
Projects that rely on :class:CallbackIPCServer can still customise startup
and accept timeouts by passing a :class:TimeoutConfig dataclass:
import os
from cmd_mox.ipc import (
CallbackIPCServer,
CallbackNamedPipeServer,
IPCHandlers,
TimeoutConfig,
Response,
)
Server = CallbackNamedPipeServer if os.name == "nt" else CallbackIPCServer
def handle_passthrough(result):
return Response(stdout=result.stdout, stderr=result.stderr, exit_code=result.exit_code)
server = Server(
socket_path,
handler=handle,
passthrough_handler=handle_passthrough,
timeouts=TimeoutConfig(timeout=1.5, accept_timeout=0.05),
)