from __future__ import annotations
import json
import os
import subprocess
import sys
from dataclasses import dataclass
from typing import Any, Callable, Dict, List
JSON = Dict[str, Any]
@dataclass(frozen=True)
class ToolDef:
name: str
description: str
input_schema: JSON
build_args: Callable[[JSON], List[str]]
def _optional_bool_flag(value: Any, flag: str, out: List[str]) -> None:
if value is True:
out.append(flag)
def _optional_str_flag(value: Any, flag: str, out: List[str]) -> None:
if isinstance(value, str) and value.strip():
out.extend([flag, value.strip()])
def _run_xbp(args: List[str], arguments: JSON) -> JSON:
command = ["xbp", *args]
timeout_seconds = arguments.get("timeout_seconds", 300)
cwd = arguments.get("cwd")
if isinstance(cwd, str) and cwd.strip():
run_cwd = cwd.strip()
else:
run_cwd = os.getcwd()
env = os.environ.copy()
if arguments.get("debug") is True:
command.insert(1, "--debug")
try:
completed = subprocess.run(
command,
cwd=run_cwd,
env=env,
capture_output=True,
text=True,
timeout=timeout_seconds,
check=False,
)
return {
"ok": completed.returncode == 0,
"command": command,
"cwd": run_cwd,
"exit_code": completed.returncode,
"stdout": completed.stdout,
"stderr": completed.stderr,
}
except FileNotFoundError:
return {
"ok": False,
"command": command,
"cwd": run_cwd,
"exit_code": 127,
"stdout": "",
"stderr": "xbp executable was not found on PATH.",
}
except subprocess.TimeoutExpired as exc:
return {
"ok": False,
"command": command,
"cwd": run_cwd,
"exit_code": 124,
"stdout": exc.stdout or "",
"stderr": (exc.stderr or "") + f"\nCommand timed out after {timeout_seconds}s.",
}
def _args_ports(a: JSON) -> List[str]:
out = ["ports"]
if isinstance(a.get("port"), int):
out.extend(["--port", str(a["port"])])
_optional_bool_flag(a.get("kill"), "--kill", out)
_optional_bool_flag(a.get("nginx"), "--nginx", out)
_optional_bool_flag(a.get("full"), "--full", out)
_optional_bool_flag(a.get("no_local"), "--no-local", out)
return out
def _args_setup(_: JSON) -> List[str]:
return ["setup"]
def _args_redeploy(a: JSON) -> List[str]:
out = ["redeploy"]
if isinstance(a.get("service_name"), str) and a["service_name"].strip():
out.append(a["service_name"].strip())
return out
def _args_redeploy_v2(a: JSON) -> List[str]:
out = ["redeploy-v2"]
_optional_str_flag(a.get("password"), "--password", out)
_optional_str_flag(a.get("username"), "--username", out)
_optional_str_flag(a.get("host"), "--host", out)
_optional_str_flag(a.get("project_dir"), "--project-dir", out)
return out
def _args_config(a: JSON) -> List[str]:
out = ["config"]
_optional_bool_flag(a.get("project"), "--project", out)
_optional_bool_flag(a.get("no_open"), "--no-open", out)
return out
def _args_install(a: JSON) -> List[str]:
return ["install", str(a["package"])]
def _args_logs(a: JSON) -> List[str]:
out = ["logs"]
if isinstance(a.get("project"), str) and a["project"].strip():
out.append(a["project"].strip())
_optional_str_flag(a.get("ssh_host"), "--ssh-host", out)
_optional_str_flag(a.get("ssh_username"), "--ssh-username", out)
_optional_str_flag(a.get("ssh_password"), "--ssh-password", out)
return out
def _args_list(_: JSON) -> List[str]:
return ["list"]
def _args_curl(a: JSON) -> List[str]:
out = ["curl"]
if isinstance(a.get("url"), str) and a["url"].strip():
out.append(a["url"].strip())
_optional_bool_flag(a.get("no_timeout"), "--no-timeout", out)
return out
def _args_services(_: JSON) -> List[str]:
return ["services"]
def _args_service(a: JSON) -> List[str]:
out = ["service"]
if isinstance(a.get("command"), str) and a["command"].strip():
out.append(a["command"].strip())
if isinstance(a.get("service_name"), str) and a["service_name"].strip():
out.append(a["service_name"].strip())
return out
def _args_nginx_setup(a: JSON) -> List[str]:
return ["nginx", "setup", "--domain", str(a["domain"]), "--port", str(a["port"])]
def _args_nginx_list(_: JSON) -> List[str]:
return ["nginx", "list"]
def _args_nginx_show(a: JSON) -> List[str]:
out = ["nginx", "show"]
if isinstance(a.get("domain"), str) and a["domain"].strip():
out.append(a["domain"].strip())
return out
def _args_nginx_edit(a: JSON) -> List[str]:
return ["nginx", "edit", str(a["domain"])]
def _args_nginx_update(a: JSON) -> List[str]:
return ["nginx", "update", "--domain", str(a["domain"]), "--port", str(a["port"])]
def _args_diag(a: JSON) -> List[str]:
out = ["diag"]
_optional_bool_flag(a.get("nginx"), "--nginx", out)
_optional_str_flag(a.get("ports"), "--ports", out)
_optional_bool_flag(a.get("no_speed_test"), "--no-speed-test", out)
return out
def _args_monitor_check(_: JSON) -> List[str]:
return ["monitor", "check"]
def _args_monitor_start(_: JSON) -> List[str]:
return ["monitor", "start"]
def _args_snapshot(_: JSON) -> List[str]:
return ["snapshot"]
def _args_resurrect(_: JSON) -> List[str]:
return ["resurrect"]
def _args_stop(a: JSON) -> List[str]:
out = ["stop"]
if isinstance(a.get("target"), str) and a["target"].strip():
out.append(a["target"].strip())
return out
def _args_flush(a: JSON) -> List[str]:
out = ["flush"]
if isinstance(a.get("target"), str) and a["target"].strip():
out.append(a["target"].strip())
return out
def _args_login(_: JSON) -> List[str]:
return ["login"]
def _args_version(a: JSON) -> List[str]:
out = ["version"]
if isinstance(a.get("target"), str) and a["target"].strip():
out.append(a["target"].strip())
_optional_bool_flag(a.get("git"), "--git", out)
return out
def _args_env(a: JSON) -> List[str]:
return ["env", str(a["target"])]
def _args_tail(a: JSON) -> List[str]:
out = ["tail"]
_optional_bool_flag(a.get("kafka"), "--kafka", out)
_optional_bool_flag(a.get("ship"), "--ship", out)
return out
def _args_start(a: JSON) -> List[str]:
out = ["start"]
cmd_args = a.get("args")
if isinstance(cmd_args, list):
out.extend(str(v) for v in cmd_args)
return out
def _args_generate_systemd(a: JSON) -> List[str]:
out = ["generate", "systemd"]
_optional_str_flag(a.get("output_dir"), "--output-dir", out)
_optional_str_flag(a.get("service"), "--service", out)
return out
def _secrets_base(a: JSON) -> List[str]:
out = ["secrets"]
_optional_str_flag(a.get("repo"), "--repo", out)
return out
def _args_secrets_list(a: JSON) -> List[str]:
return [*_secrets_base(a), "list"]
def _args_secrets_push(a: JSON) -> List[str]:
out = [*_secrets_base(a), "push"]
_optional_str_flag(a.get("file"), "--file", out)
_optional_bool_flag(a.get("force"), "--force", out)
return out
def _args_secrets_pull(a: JSON) -> List[str]:
out = [*_secrets_base(a), "pull"]
_optional_str_flag(a.get("output"), "--output", out)
return out
def _args_secrets_generate_default(a: JSON) -> List[str]:
out = [*_secrets_base(a), "generate-default"]
_optional_str_flag(a.get("output"), "--output", out)
return out
def _args_secrets_verify(a: JSON) -> List[str]:
return [*_secrets_base(a), "verify"]
def _args_secrets_help(a: JSON) -> List[str]:
return [*_secrets_base(a), "help"]
def _args_monitoring_serve(a: JSON) -> List[str]:
out = ["monitoring", "serve"]
_optional_str_flag(a.get("file"), "--file", out)
return out
def _args_monitoring_run_once(a: JSON) -> List[str]:
out = ["monitoring", "run-once"]
_optional_str_flag(a.get("file"), "--file", out)
_optional_bool_flag(a.get("probes_only"), "--probes-only", out)
_optional_bool_flag(a.get("stories_only"), "--stories-only", out)
return out
def _args_monitoring_list(a: JSON) -> List[str]:
out = ["monitoring", "list"]
_optional_str_flag(a.get("file"), "--file", out)
return out
def _args_raw(a: JSON) -> List[str]:
values = a.get("args")
if not isinstance(values, list) or any(not isinstance(v, str) for v in values):
raise ValueError("`args` must be an array of strings.")
return values
TOOLS: List[ToolDef] = [
ToolDef("xbp_ports", "Run `xbp ports`.", {
"type": "object",
"properties": {
"port": {"type": "integer"},
"kill": {"type": "boolean"},
"nginx": {"type": "boolean"},
"full": {"type": "boolean"},
"debug": {"type": "boolean"},
"cwd": {"type": "string"},
"timeout_seconds": {"type": "integer", "minimum": 1}
}
}, _args_ports),
ToolDef("xbp_setup", "Run `xbp setup`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_setup),
ToolDef("xbp_redeploy", "Run `xbp redeploy [service_name]`.", {"type": "object", "properties": {"service_name": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_redeploy),
ToolDef("xbp_redeploy_v2", "Run `xbp redeploy-v2`.", {"type": "object", "properties": {"password": {"type": "string"}, "username": {"type": "string"}, "host": {"type": "string"}, "project_dir": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_redeploy_v2),
ToolDef("xbp_config", "Run `xbp config`.", {"type": "object", "properties": {"project": {"type": "boolean"}, "no_open": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_config),
ToolDef("xbp_install", "Run `xbp install <package>`.", {"type": "object", "required": ["package"], "properties": {"package": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_install),
ToolDef("xbp_logs", "Run `xbp logs`.", {"type": "object", "properties": {"project": {"type": "string"}, "ssh_host": {"type": "string"}, "ssh_username": {"type": "string"}, "ssh_password": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_logs),
ToolDef("xbp_list", "Run `xbp list`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_list),
ToolDef("xbp_curl", "Run `xbp curl [url]`.", {"type": "object", "properties": {"url": {"type": "string"}, "no_timeout": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_curl),
ToolDef("xbp_services", "Run `xbp services`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_services),
ToolDef("xbp_service", "Run `xbp service [command] [service_name]`.", {"type": "object", "properties": {"command": {"type": "string"}, "service_name": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_service),
ToolDef("xbp_nginx_setup", "Run `xbp nginx setup`.", {"type": "object", "required": ["domain", "port"], "properties": {"domain": {"type": "string"}, "port": {"type": "integer"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_nginx_setup),
ToolDef("xbp_nginx_list", "Run `xbp nginx list`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_nginx_list),
ToolDef("xbp_nginx_show", "Run `xbp nginx show [domain]`.", {"type": "object", "properties": {"domain": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_nginx_show),
ToolDef("xbp_nginx_edit", "Run `xbp nginx edit <domain>`.", {"type": "object", "required": ["domain"], "properties": {"domain": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_nginx_edit),
ToolDef("xbp_nginx_update", "Run `xbp nginx update`.", {"type": "object", "required": ["domain", "port"], "properties": {"domain": {"type": "string"}, "port": {"type": "integer"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_nginx_update),
ToolDef("xbp_diag", "Run `xbp diag`.", {"type": "object", "properties": {"nginx": {"type": "boolean"}, "ports": {"type": "string"}, "no_speed_test": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_diag),
ToolDef("xbp_monitor_check", "Run `xbp monitor check`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_monitor_check),
ToolDef("xbp_monitor_start", "Run `xbp monitor start`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_monitor_start),
ToolDef("xbp_snapshot", "Run `xbp snapshot`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_snapshot),
ToolDef("xbp_resurrect", "Run `xbp resurrect`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_resurrect),
ToolDef("xbp_stop", "Run `xbp stop [target]`.", {"type": "object", "properties": {"target": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_stop),
ToolDef("xbp_flush", "Run `xbp flush [target]`.", {"type": "object", "properties": {"target": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_flush),
ToolDef("xbp_login", "Run `xbp login`.", {"type": "object", "properties": {"debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_login),
ToolDef("xbp_version", "Run `xbp version [target] [--git]`.", {"type": "object", "properties": {"target": {"type": "string"}, "git": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_version),
ToolDef("xbp_env", "Run `xbp env <target>`.", {"type": "object", "required": ["target"], "properties": {"target": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_env),
ToolDef("xbp_tail", "Run `xbp tail`.", {"type": "object", "properties": {"kafka": {"type": "boolean"}, "ship": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_tail),
ToolDef("xbp_start", "Run `xbp start ...`.", {"type": "object", "properties": {"args": {"type": "array", "items": {"type": "string"}}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_start),
ToolDef("xbp_generate_systemd", "Run `xbp generate systemd`.", {"type": "object", "properties": {"output_dir": {"type": "string"}, "service": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_generate_systemd),
ToolDef("xbp_secrets_list", "Run `xbp secrets list`.", {"type": "object", "properties": {"repo": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_list),
ToolDef("xbp_secrets_push", "Run `xbp secrets push`.", {"type": "object", "properties": {"repo": {"type": "string"}, "file": {"type": "string"}, "force": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_push),
ToolDef("xbp_secrets_pull", "Run `xbp secrets pull`.", {"type": "object", "properties": {"repo": {"type": "string"}, "output": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_pull),
ToolDef("xbp_secrets_generate_default", "Run `xbp secrets generate-default`.", {"type": "object", "properties": {"repo": {"type": "string"}, "output": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_generate_default),
ToolDef("xbp_secrets_verify", "Run `xbp secrets verify`.", {"type": "object", "properties": {"repo": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_verify),
ToolDef("xbp_secrets_help", "Run `xbp secrets help`.", {"type": "object", "properties": {"repo": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_secrets_help),
ToolDef("xbp_monitoring_serve", "Run `xbp monitoring serve` (feature-gated command).", {"type": "object", "properties": {"file": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_monitoring_serve),
ToolDef("xbp_monitoring_run_once", "Run `xbp monitoring run-once` (feature-gated command).", {"type": "object", "properties": {"file": {"type": "string"}, "probes_only": {"type": "boolean"}, "stories_only": {"type": "boolean"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_monitoring_run_once),
ToolDef("xbp_monitoring_list", "Run `xbp monitoring list` (feature-gated command).", {"type": "object", "properties": {"file": {"type": "string"}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_monitoring_list),
ToolDef("xbp_raw", "Run any `xbp` command by passing raw argument array.", {"type": "object", "required": ["args"], "properties": {"args": {"type": "array", "items": {"type": "string"}}, "debug": {"type": "boolean"}, "cwd": {"type": "string"}, "timeout_seconds": {"type": "integer", "minimum": 1}}}, _args_raw),
]
TOOL_MAP = {tool.name: tool for tool in TOOLS}
class McpServer:
def __init__(self) -> None:
self.stdin = sys.stdin.buffer
self.stdout = sys.stdout.buffer
def _read_message(self) -> JSON | None:
headers: Dict[str, str] = {}
while True:
line = self.stdin.readline()
if not line:
return None
if line in (b"\r\n", b"\n"):
break
key, _, value = line.decode("utf-8").partition(":")
headers[key.strip().lower()] = value.strip()
if "content-length" not in headers:
return None
length = int(headers["content-length"])
payload = self.stdin.read(length)
if not payload:
return None
return json.loads(payload.decode("utf-8"))
def _send(self, message: JSON) -> None:
encoded = json.dumps(message, separators=(",", ":")).encode("utf-8")
header = f"Content-Length: {len(encoded)}\r\n\r\n".encode("utf-8")
self.stdout.write(header)
self.stdout.write(encoded)
self.stdout.flush()
def _send_result(self, request_id: Any, result: JSON) -> None:
self._send({"jsonrpc": "2.0", "id": request_id, "result": result})
def _send_error(self, request_id: Any, code: int, message: str) -> None:
self._send(
{
"jsonrpc": "2.0",
"id": request_id,
"error": {"code": code, "message": message},
}
)
def _handle_initialize(self, request_id: Any) -> None:
self._send_result(
request_id,
{
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "xbp-mcp", "version": "0.1.0"},
},
)
def _handle_tools_list(self, request_id: Any) -> None:
self._send_result(
request_id,
{
"tools": [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.input_schema,
}
for tool in TOOLS
]
},
)
def _handle_tools_call(self, request_id: Any, params: JSON) -> None:
name = params.get("name")
arguments = params.get("arguments") or {}
if not isinstance(name, str) or name not in TOOL_MAP:
self._send_error(request_id, -32602, f"Unknown tool: {name}")
return
if not isinstance(arguments, dict):
self._send_error(request_id, -32602, "Tool arguments must be an object.")
return
tool = TOOL_MAP[name]
try:
args = tool.build_args(arguments)
except Exception as exc:
self._send_result(
request_id,
{
"content": [{"type": "text", "text": f"Invalid arguments: {exc}"}],
"isError": True,
},
)
return
result = _run_xbp(args, arguments)
text_output = (
f"$ {' '.join(result['command'])}\n"
f"cwd: {result['cwd']}\n"
f"exit_code: {result['exit_code']}\n\n"
f"stdout:\n{result['stdout']}\n"
f"stderr:\n{result['stderr']}"
)
self._send_result(
request_id,
{
"content": [{"type": "text", "text": text_output}],
"structuredContent": result,
"isError": not result["ok"],
},
)
def serve_forever(self) -> int:
while True:
request = self._read_message()
if request is None:
return 0
method = request.get("method")
request_id = request.get("id")
if not isinstance(method, str):
if request_id is not None:
self._send_error(request_id, -32600, "Invalid request: missing method.")
continue
if method == "initialize":
self._handle_initialize(request_id)
elif method == "tools/list":
self._handle_tools_list(request_id)
elif method == "tools/call":
self._handle_tools_call(request_id, request.get("params") or {})
elif method == "notifications/initialized":
continue
else:
if request_id is not None:
self._send_error(request_id, -32601, f"Method not found: {method}")
def main() -> int:
server = McpServer()
return server.serve_forever()
if __name__ == "__main__":
raise SystemExit(main())