diff --git a/src/mcp_agent/cli/cloud/main.py b/src/mcp_agent/cli/cloud/main.py index 32c469f39..4dc218ddf 100644 --- a/src/mcp_agent/cli/cloud/main.py +++ b/src/mcp_agent/cli/cloud/main.py @@ -2,16 +2,12 @@ import logging import os +from importlib.metadata import version as metadata_version from logging.handlers import RotatingFileHandler from pathlib import Path from typing import Optional -from importlib.metadata import version as metadata_version -import click import typer -from rich.console import Console -from rich.panel import Panel -from typer.core import TyperGroup from mcp_agent.cli.cloud.commands import ( configure_app, @@ -20,27 +16,27 @@ logout, whoami, ) -from mcp_agent.cli.cloud.commands.logger import tail_logs from mcp_agent.cli.cloud.commands.app import ( delete_app, get_app_status, list_app_workflows, ) from mcp_agent.cli.cloud.commands.apps import list_apps +from mcp_agent.cli.cloud.commands.logger import tail_logs +from mcp_agent.cli.cloud.commands.servers import ( + delete_server, + describe_server, + list_servers, +) from mcp_agent.cli.cloud.commands.workflows import ( + cancel_workflow, describe_workflow, + list_workflow_runs, + list_workflows, resume_workflow, suspend_workflow, - cancel_workflow, - list_workflows, - list_workflow_runs, ) -from mcp_agent.cli.cloud.commands.servers import ( - list_servers, - describe_server, - delete_server, -) -from mcp_agent.cli.exceptions import CLIError +from mcp_agent.cli.utils.typer_utils import HelpfulTyperGroup from mcp_agent.cli.utils.ux import print_error # Setup file logging @@ -63,36 +59,6 @@ logging.basicConfig(level=logging.INFO, handlers=[file_handler]) -class HelpfulTyperGroup(TyperGroup): - """Typer group that shows help before usage errors for better UX.""" - - def resolve_command(self, ctx, args): - try: - return super().resolve_command(ctx, args) - except click.UsageError as e: - click.echo(ctx.get_help()) - - console = Console(stderr=True) - error_panel = Panel( - str(e), - title="Error", - title_align="left", - border_style="red", - expand=True, - ) - console.print(error_panel) - ctx.exit(2) - - def invoke(self, ctx): - try: - return super().invoke(ctx) - except CLIError as e: - # Handle CLIError cleanly - show error message and exit - logging.error(f"CLI error: {str(e)}") - print_error(str(e)) - ctx.exit(e.exit_code) - - # Root typer for `mcp-agent` CLI commands app = typer.Typer( help="MCP Agent Cloud CLI for deployment and management", diff --git a/src/mcp_agent/cli/main.py b/src/mcp_agent/cli/main.py index 67525c3b4..fe14a50f8 100644 --- a/src/mcp_agent/cli/main.py +++ b/src/mcp_agent/cli/main.py @@ -9,10 +9,13 @@ from __future__ import annotations +import logging import typer from rich.console import Console +from mcp_agent.cli.utils.ux import print_error + # Mount existing cloud CLI try: from mcp_agent.cli.cloud.main import app as cloud_app # type: ignore @@ -21,23 +24,36 @@ # Local command groups (scaffolded) +from mcp_agent.cli.cloud.commands import deploy_config, login, logout, whoami +from mcp_agent.cli.commands import ( + check as check_cmd, +) from mcp_agent.cli.commands import ( - init as init_cmd, - quickstart as quickstart_cmd, config as config_cmd, +) +from mcp_agent.cli.commands import ( + go as go_cmd, +) +from mcp_agent.cli.commands import ( + init as init_cmd, +) +from mcp_agent.cli.commands import ( keys as keys_cmd, +) +from mcp_agent.cli.commands import ( models as models_cmd, - go as go_cmd, - check as check_cmd, ) - -from mcp_agent.cli.cloud.commands import deploy_config, login, logout, whoami +from mcp_agent.cli.commands import ( + quickstart as quickstart_cmd, +) +from mcp_agent.cli.utils.typer_utils import HelpfulTyperGroup app = typer.Typer( help="mcp-agent CLI", add_completion=False, no_args_is_help=False, context_settings={"help_option_names": ["-h", "--help"]}, + cls=HelpfulTyperGroup, ) @@ -139,8 +155,14 @@ def main( def run() -> None: - """Entry for __main__.""" - app() + """Run the CLI application.""" + try: + app() + except Exception as e: + # Unexpected errors - log full exception and show clean error to user + logging.exception("Unhandled exception in CLI") + print_error(f"An unexpected error occurred: {str(e)}") + raise typer.Exit(1) from e if __name__ == "__main__": diff --git a/src/mcp_agent/cli/utils/typer_utils.py b/src/mcp_agent/cli/utils/typer_utils.py new file mode 100644 index 000000000..f7b3521c6 --- /dev/null +++ b/src/mcp_agent/cli/utils/typer_utils.py @@ -0,0 +1,40 @@ +"""Shared Typer utilities for MCP Agent CLI.""" + +import logging +import click +from rich.console import Console +from rich.panel import Panel +from typer.core import TyperGroup + +from mcp_agent.cli.exceptions import CLIError +from mcp_agent.cli.utils.ux import print_error + + +class HelpfulTyperGroup(TyperGroup): + """Typer group that shows help before usage errors for better UX.""" + + def resolve_command(self, ctx, args): + try: + return super().resolve_command(ctx, args) + except click.UsageError as e: + click.echo(ctx.get_help()) + + console = Console(stderr=True) + error_panel = Panel( + str(e), + title="Error", + title_align="left", + border_style="red", + expand=True, + ) + console.print(error_panel) + ctx.exit(2) + + def invoke(self, ctx): + try: + return super().invoke(ctx) + except CLIError as e: + # Handle CLIError cleanly - show error message and exit + logging.error(f"CLI error: {str(e)}") + print_error(str(e)) + ctx.exit(e.exit_code) diff --git a/src/mcp_agent/server/app_server.py b/src/mcp_agent/server/app_server.py index 1acaa03b9..6dbdbd376 100644 --- a/src/mcp_agent/server/app_server.py +++ b/src/mcp_agent/server/app_server.py @@ -338,7 +338,9 @@ async def _relay_notify(request: Request): if gw_token: bearer = request.headers.get("Authorization", "") bearer_token = ( - bearer.split(" ", 1)[1] if bearer.lower().startswith("bearer ") else "" + bearer.split(" ", 1)[1] + if bearer.lower().startswith("bearer ") + else "" ) header_tok = request.headers.get("X-MCP-Gateway-Token", "") if not ( @@ -512,7 +514,9 @@ async def _internal_workflows_log(request: Request): if gw_token: bearer = request.headers.get("Authorization", "") bearer_token = ( - bearer.split(" ", 1)[1] if bearer.lower().startswith("bearer ") else "" + bearer.split(" ", 1)[1] + if bearer.lower().startswith("bearer ") + else "" ) header_tok = request.headers.get("X-MCP-Gateway-Token", "") if not ( @@ -558,7 +562,9 @@ async def _internal_human_prompts(request: Request): if gw_token: bearer = request.headers.get("Authorization", "") bearer_token = ( - bearer.split(" ", 1)[1] if bearer.lower().startswith("bearer ") else "" + bearer.split(" ", 1)[1] + if bearer.lower().startswith("bearer ") + else "" ) header_tok = request.headers.get("X-MCP-Gateway-Token", "") if not (