marcus
6ba4c7ca92
Mechanical, no behavior change. `ruff check src/ tests/` now passes with zero findings. - cli.py:147 (SIM105) — replace `try/except SynologyError/pass` around the cleanup logout with `contextlib.suppress(SynologyError)`. - compose.py:271 (B007) — drop the unused `i` from the env_list preview-detection loop (the apply loop below still uses enumerate). - compose.py:329 (E501) — extract `verb = "Updated" if … else "Added"` into a local before the return so the f-string fits in 100 cols. - images.py:237 (E501) — extract `stopped_name = in_use_stopped[0]` before the return and split the message across two f-strings. - test_auth.py:38, 127, 140 (SIM117) — combine nested `with patch(…):` / `with pytest.raises(…):` into single parenthesised with-statements. 236 tests pass. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""CLI entry point: setup, check, serve commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import sys
|
|
from typing import Any
|
|
|
|
import click
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _configure_logging(level: str = "warning") -> None:
|
|
"""Configure stderr logging."""
|
|
numeric = getattr(logging, level.upper(), logging.WARNING)
|
|
logging.basicConfig(
|
|
level=numeric,
|
|
format="%(levelname)s %(name)s: %(message)s",
|
|
stream=sys.stderr,
|
|
)
|
|
|
|
|
|
@click.group()
|
|
def main() -> None:
|
|
"""MCP server for Synology Container Manager."""
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# setup
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@main.command()
|
|
@click.option("--verbose", "-v", is_flag=True, help="Enable debug logging")
|
|
def setup(verbose: bool) -> None:
|
|
"""Interactive setup wizard: configure connection and store credentials."""
|
|
_configure_logging("debug" if verbose else "warning")
|
|
asyncio.run(_run_setup())
|
|
|
|
|
|
async def _run_setup() -> None:
|
|
"""Interactive setup flow."""
|
|
from mcp_synology_container.auth import AuthManager
|
|
from mcp_synology_container.config import CONFIG_PATH, AppConfig, ConnectionConfig, save_config
|
|
from mcp_synology_container.dsm_client import DsmClient, SynologyError
|
|
|
|
click.echo("=== mcp-synology-container setup ===\n")
|
|
|
|
# Connection details
|
|
host = click.prompt("NAS hostname or IP")
|
|
use_https = click.confirm("Use HTTPS?", default=True)
|
|
default_port = 443 if use_https else 5000
|
|
port = click.prompt("Port", default=default_port, type=int)
|
|
|
|
verify_ssl = True
|
|
if use_https:
|
|
verify_ssl = click.confirm("Verify SSL certificate?", default=True)
|
|
|
|
compose_base = click.prompt(
|
|
"Base path for compose projects on NAS",
|
|
default="/volume1/docker",
|
|
)
|
|
alias_input = click.prompt("Alias (friendly name, optional)", default="", show_default=False)
|
|
alias: str | None = alias_input.strip() or None
|
|
|
|
config = AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(
|
|
host=host,
|
|
port=port,
|
|
https=use_https,
|
|
verify_ssl=verify_ssl,
|
|
),
|
|
compose_base_path=compose_base,
|
|
alias=alias,
|
|
)
|
|
|
|
click.echo()
|
|
username = click.prompt("DSM username")
|
|
password = click.prompt("DSM password", hide_input=True)
|
|
|
|
# Store in keyring
|
|
auth = AuthManager(config)
|
|
keyring_ok = auth.store_credentials(username, password)
|
|
if keyring_ok:
|
|
click.echo("Credentials stored in OS keyring.")
|
|
else:
|
|
click.echo(
|
|
click.style("Warning: keyring unavailable.", fg="yellow")
|
|
+ " Set SYNOLOGY_USERNAME and SYNOLOGY_PASSWORD environment variables instead."
|
|
)
|
|
|
|
# Test connection
|
|
click.echo("\nTesting connection...")
|
|
base_url = config.base_url
|
|
|
|
try:
|
|
async with DsmClient(base_url, verify_ssl=verify_ssl) as client:
|
|
await client.query_api_info()
|
|
|
|
# Attempt login with possible 2FA handling
|
|
params: dict[str, Any] = {
|
|
"account": username,
|
|
"passwd": password,
|
|
"format": "sid",
|
|
}
|
|
try:
|
|
data = await client.request("SYNO.API.Auth", "login", version=6, params=params)
|
|
sid = data.get("sid")
|
|
except SynologyError as e:
|
|
if e.code == 403:
|
|
# 2FA required
|
|
click.echo(
|
|
click.style("2FA is enabled.", fg="yellow")
|
|
+ " Enter the OTP code from your authenticator app."
|
|
)
|
|
otp_code = click.prompt("OTP code")
|
|
data = await client.request(
|
|
"SYNO.API.Auth",
|
|
"login",
|
|
version=6,
|
|
params={
|
|
"account": username,
|
|
"passwd": password,
|
|
"otp_code": otp_code,
|
|
"enable_device_token": "yes",
|
|
"device_name": "MCPSynologyContainer",
|
|
"format": "sid",
|
|
},
|
|
)
|
|
sid = data.get("sid")
|
|
device_token = data.get("did", "")
|
|
if device_token and keyring_ok:
|
|
auth.store_device_token(device_token)
|
|
click.echo(click.style("2FA device token stored in keyring.", fg="green"))
|
|
else:
|
|
click.echo(click.style(f"Login failed: {e}", fg="red"), err=True)
|
|
sys.exit(1)
|
|
|
|
if sid:
|
|
client.sid = sid
|
|
click.echo(click.style("Login successful!", fg="green"))
|
|
# Logout cleanly
|
|
with contextlib.suppress(SynologyError):
|
|
await client.request("SYNO.API.Auth", "logout", version=6, params={})
|
|
else:
|
|
click.echo(click.style("Login failed: no session ID returned.", fg="red"), err=True)
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
click.echo(click.style(f"Connection failed: {e}", fg="red"), err=True)
|
|
sys.exit(1)
|
|
|
|
# Save config
|
|
save_config(config)
|
|
click.echo(f"\nConfig saved to {CONFIG_PATH}")
|
|
|
|
# Emit Claude Desktop snippet
|
|
_emit_desktop_snippet()
|
|
|
|
|
|
def _emit_desktop_snippet() -> None:
|
|
"""Print the Claude Desktop configuration snippet."""
|
|
import shutil
|
|
|
|
cmd = shutil.which("mcp-synology-container") or "mcp-synology-container"
|
|
snippet = {
|
|
"mcpServers": {
|
|
"synology-container": {
|
|
"command": cmd,
|
|
"args": ["serve"],
|
|
}
|
|
}
|
|
}
|
|
click.echo("\nAdd this to your Claude Desktop config (claude_desktop_config.json):\n")
|
|
click.echo(json.dumps(snippet, indent=2))
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# check
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@main.command()
|
|
@click.option("--config", "-c", "config_path", type=click.Path(), help="Config file path")
|
|
@click.option("--verbose", "-v", is_flag=True, help="Enable debug logging")
|
|
def check(config_path: str | None, verbose: bool) -> None:
|
|
"""Test connection to DSM and print status."""
|
|
_configure_logging("debug" if verbose else "warning")
|
|
ok = asyncio.run(_run_check(config_path))
|
|
sys.exit(0 if ok else 1)
|
|
|
|
|
|
async def _run_check(config_path: str | None) -> bool:
|
|
"""Run connectivity check. Returns True on success."""
|
|
from mcp_synology_container.auth import AuthenticationError, AuthManager
|
|
from mcp_synology_container.config import load_config
|
|
from mcp_synology_container.dsm_client import DsmClient, SynologyError
|
|
|
|
try:
|
|
config = load_config(config_path)
|
|
except (FileNotFoundError, ValueError) as e:
|
|
click.echo(click.style(f"Config error: {e}", fg="red"), err=True)
|
|
return False
|
|
|
|
host = config.connection.host
|
|
port = config.connection.port
|
|
click.echo(f"Host: {host}:{port}")
|
|
click.echo(f"HTTPS: {config.connection.https}")
|
|
click.echo(f"Verify SSL: {config.connection.verify_ssl}")
|
|
click.echo(f"Compose: {config.compose_base_path}")
|
|
if config.alias:
|
|
click.echo(f"Alias: {config.alias}")
|
|
click.echo()
|
|
|
|
auth = AuthManager(config)
|
|
try:
|
|
username, _, device_token = auth.resolve_credentials()
|
|
click.echo(f"Credentials: found (user={username}, 2FA={'yes' if device_token else 'no'})")
|
|
except AuthenticationError as e:
|
|
click.echo(click.style(f"Credentials: {e}", fg="red"), err=True)
|
|
return False
|
|
|
|
try:
|
|
async with DsmClient(config.base_url, config.connection.verify_ssl) as client:
|
|
await client.query_api_info()
|
|
click.echo("API info: fetched successfully")
|
|
|
|
client.set_auth_manager(auth)
|
|
sid = await auth.login(client)
|
|
client.sid = sid
|
|
click.echo(click.style("Login: successful", fg="green"))
|
|
|
|
# Check Docker API availability
|
|
docker_apis = [
|
|
"SYNO.Docker.Project",
|
|
"SYNO.Docker.Container",
|
|
"SYNO.Docker.Container.Log",
|
|
"SYNO.Docker.Image",
|
|
"SYNO.FileStation.Download",
|
|
"SYNO.FileStation.Upload",
|
|
]
|
|
click.echo("\nRequired APIs:")
|
|
all_ok = True
|
|
for api_name in docker_apis:
|
|
if api_name in client._api_cache:
|
|
info = client._api_cache[api_name]
|
|
click.echo(f" {api_name}: v{info['minVersion']}-v{info['maxVersion']} ✓")
|
|
else:
|
|
click.echo(click.style(f" {api_name}: NOT FOUND", fg="red"))
|
|
all_ok = False
|
|
|
|
await auth.logout(client)
|
|
|
|
if all_ok:
|
|
click.echo(click.style("\nAll checks passed.", fg="green"))
|
|
else:
|
|
click.echo(
|
|
click.style(
|
|
"\nSome APIs not found. Ensure Container Manager is installed.", fg="yellow"
|
|
)
|
|
)
|
|
return all_ok
|
|
|
|
except SynologyError as e:
|
|
click.echo(click.style(f"DSM error: {e}", fg="red"), err=True)
|
|
return False
|
|
except Exception as e:
|
|
click.echo(click.style(f"Connection error: {e}", fg="red"), err=True)
|
|
return False
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# serve
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@main.command()
|
|
@click.option("--config", "-c", "config_path", type=click.Path(), help="Config file path")
|
|
def serve(config_path: str | None) -> None:
|
|
"""Start the MCP server in stdio mode."""
|
|
sys.stderr.write("SERVER STARTING\n")
|
|
sys.stderr.flush()
|
|
_configure_logging("warning")
|
|
# Use anyio.run() — FastMCP uses anyio internally (anyio.create_task_group),
|
|
# and anyio.run() ensures the correct backend context is set up.
|
|
# asyncio.run() can cause issues on Windows (ProactorEventLoop + anyio).
|
|
import anyio
|
|
|
|
anyio.run(_run_serve, config_path)
|
|
|
|
|
|
async def _run_serve(config_path: str | None) -> None:
|
|
"""Initialize and run the MCP server."""
|
|
# Eagerly import all modules so any ImportError surfaces immediately on stderr
|
|
# instead of silently killing the process.
|
|
from mcp_synology_container.auth import AuthManager
|
|
from mcp_synology_container.config import load_config
|
|
from mcp_synology_container.dsm_client import DsmClient
|
|
from mcp_synology_container.server import create_server
|
|
|
|
logger.debug("Loading config from: %s", config_path or "default path")
|
|
try:
|
|
config = load_config(config_path)
|
|
except (FileNotFoundError, ValueError) as e:
|
|
sys.stderr.write(f"Config error: {e}\n")
|
|
sys.stderr.flush()
|
|
return
|
|
|
|
# Open the HTTP client and register auth — but do NOT connect to the NAS yet.
|
|
# Lazy init (_ensure_initialized) runs on the first tool call, so Claude Desktop
|
|
# sees the server as ready immediately without waiting for NAS connectivity.
|
|
async with DsmClient(config.base_url, config.connection.verify_ssl) as client:
|
|
auth = AuthManager(config)
|
|
client.set_auth_manager(auth)
|
|
|
|
mcp_server = create_server(config, client)
|
|
sys.stderr.write("MCP server ready\n")
|
|
sys.stderr.flush()
|
|
try:
|
|
await mcp_server.run_stdio_async()
|
|
except Exception as e:
|
|
sys.stderr.write(f"MCP server error: {e}\n")
|
|
sys.stderr.flush()
|
|
raise
|