139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""Command line entry point (``aegis-gitea-mcp``).
|
|
|
|
Subcommands: ``serve`` (default), ``setup`` and ``version``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from collections.abc import Sequence
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from aegis_gitea_mcp import __version__, setup_cmd
|
|
from aegis_gitea_mcp.config import (
|
|
ENV_FILE_VARIABLE,
|
|
EnvFileError,
|
|
Settings,
|
|
get_settings,
|
|
reset_settings,
|
|
set_env_file,
|
|
)
|
|
from aegis_gitea_mcp.logging_utils import configure_logging
|
|
|
|
|
|
def _port(value: str) -> int:
|
|
"""Parse ``--port`` (argparse ``type``): an integer from 1 to 65535."""
|
|
try:
|
|
port = int(value)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError("must be an integer from 1 to 65535") from None
|
|
if not 1 <= port <= 65535:
|
|
raise argparse.ArgumentTypeError("must be between 1 and 65535")
|
|
return port
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
"""Build the argument parser."""
|
|
parser = argparse.ArgumentParser(
|
|
prog="aegis-gitea-mcp",
|
|
description="MCP server exposing the Gitea API over stdio or streamable HTTP.",
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
|
|
serve = subparsers.add_parser("serve", help="run the MCP server (default)")
|
|
serve.add_argument(
|
|
"--transport",
|
|
choices=("stdio", "http"),
|
|
default=None,
|
|
help="transport to serve (default: MCP_TRANSPORT, else stdio)",
|
|
)
|
|
serve.add_argument("--host", default=None, help="HTTP bind address (default: MCP_HOST)")
|
|
serve.add_argument(
|
|
"--port",
|
|
type=_port,
|
|
default=None,
|
|
metavar="PORT",
|
|
help="HTTP port (default: MCP_PORT)",
|
|
)
|
|
serve.add_argument(
|
|
"--env-file",
|
|
default=None,
|
|
metavar="PATH",
|
|
help=(
|
|
f"env file to load (default: {ENV_FILE_VARIABLE}, else ./.env when it is a regular "
|
|
"file owned by you with mode 0600; real environment variables always win)"
|
|
),
|
|
)
|
|
|
|
setup = subparsers.add_parser("setup", help="verify a Gitea instance and write the env file")
|
|
setup_cmd.add_arguments(setup)
|
|
subparsers.add_parser("version", help="print the version and exit")
|
|
return parser
|
|
|
|
|
|
def _load_settings(args: argparse.Namespace) -> Settings | None:
|
|
"""Load settings with the ``serve`` flags applied; print problems and return None."""
|
|
port = getattr(args, "port", None)
|
|
overrides = {
|
|
"MCP_TRANSPORT": getattr(args, "transport", None),
|
|
"MCP_HOST": getattr(args, "host", None),
|
|
"MCP_PORT": None if port is None else str(port),
|
|
}
|
|
for name, value in overrides.items():
|
|
if value is not None:
|
|
os.environ[name] = value
|
|
set_env_file(getattr(args, "env_file", None))
|
|
reset_settings()
|
|
try:
|
|
return get_settings()
|
|
except EnvFileError as exc:
|
|
print(f"aegis-gitea-mcp: {exc}", file=sys.stderr)
|
|
return None
|
|
except ValidationError as exc:
|
|
# Print field names and messages only: str(exc) would echo the offending values.
|
|
print("aegis-gitea-mcp: invalid configuration:", file=sys.stderr)
|
|
for error in exc.errors(include_input=False, include_url=False):
|
|
field = ".".join(str(part) for part in error["loc"]).upper() or "settings"
|
|
print(f" {field}: {error['msg']}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def run_serve(args: argparse.Namespace) -> int:
|
|
"""Run the MCP server over the selected transport and return the exit code."""
|
|
from aegis_gitea_mcp import transports
|
|
|
|
settings = _load_settings(args)
|
|
if settings is None:
|
|
return 2
|
|
configure_logging(settings.log_level)
|
|
if settings.insecure_gitea_url():
|
|
print(
|
|
"aegis-gitea-mcp: warning: GITEA_URL uses plain http to a host that is not loopback: "
|
|
"the Gitea token, the OAuth client secret and every user's token travel unencrypted. "
|
|
"Use https",
|
|
file=sys.stderr,
|
|
)
|
|
if settings.mcp_transport == "http":
|
|
return transports.run_http(settings)
|
|
return transports.run_stdio(settings)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
"""Run the CLI and return the process exit code."""
|
|
args = build_parser().parse_args(argv)
|
|
command = args.command or "serve"
|
|
|
|
if command == "version":
|
|
print(__version__)
|
|
return 0
|
|
if command == "setup":
|
|
return setup_cmd.run_setup(args)
|
|
return run_serve(args)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
raise SystemExit(main())
|