"""Entry point for domin8.
This module provides a small CLI to run the MCP server locally. By default
it runs the server in stdio mode (suitable for local MCP clients). If you
pass the `--tcp-port` option, a simple TCP server will be started and will
accept MCP connections on the local loopback interface only (127.0.0.1).
"""
from __future__ import annotations
import argparse
import asyncio
import logging
from typing import Optional
from domin8 import server as domin8_server
logger = logging.getLogger("domin8.main")
async def _run_stdio_mode() -> None:
"""Run the MCP server in stdio mode (default).
This is the same behavior as running the module directly (python -m domin8.server)
and is safe for local use because it communicates over the process stdio.
"""
await domin8_server.main()
async def _run_tcp_mode(host: str, port: int) -> None:
"""Run a simple TCP server bound to `host` and `port`.
Each incoming connection is handled by calling the domin8 MCP app's
asynchronous `run` method with the connection's `StreamReader` and
`StreamWriter`. For safety, this server is intended to be bound to the
loopback interface (127.0.0.1) so it's only accessible locally.
"""
async def _handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
peer = writer.get_extra_info("peername")
logger.info("Accepted connection from %s", peer)
try:
# Run the domin8 MCP server protocol on the connection streams.
await domin8_server.app.run(
reader,
writer,
domin8_server.app.create_initialization_options(),
)
except Exception:
logger.exception("Error handling MCP client %s", peer)
finally:
try:
writer.close()
await writer.wait_closed()
except Exception:
pass
server = await asyncio.start_server(_handle_client, host=host, port=port)
addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets or [])
logger.info("Domin8 MCP TCP server listening on %s", addrs)
async with server:
await server.serve_forever()
def _parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run the domin8 MCP server locally")
parser.add_argument(
"--tcp-port",
type=int,
default=None,
help="If set, start a TCP MCP server bound to --host (default: 127.0.0.1)."
)
parser.add_argument(
"--host",
type=str,
default="127.0.0.1",
help="Host/interface to bind the TCP server to (default: 127.0.0.1)."
)
parser.add_argument(
"--debug",
action="store_true",
help="Enable debug logging."
)
return parser.parse_args(argv)
def main(argv: Optional[list[str]] = None) -> None:
"""Console entrypoint.
By default this runs the MCP server in stdio mode. To run a TCP server on
the local loopback, pass `--tcp-port <port>`.
"""
args = _parse_args(argv)
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
if args.tcp_port is None:
# StdIO mode: delegate to the server module's main coroutine.
asyncio.run(_run_stdio_mode())
else:
# Enforce loopback-only binding for safety.
if args.host not in ("127.0.0.1", "::1", "localhost"):
raise SystemExit("Refusing to bind TCP server to non-loopback interface")
asyncio.run(_run_tcp_mode(args.host, args.tcp_port))
if __name__ == "__main__":
main()
