cli-mcp-server

Instructions

Implementation Reference

• src/cli_mcp_server/server.py:475-516 (handler)

  Handler for the 'run_command' tool call. Validates arguments, executes the command via CommandExecutor, and returns stdout, stderr, and return code in MCP TextContent format.

  if name == "run_command":
      if not arguments or "command" not in arguments:
          return [
              types.TextContent(type="text", text="No command provided", error=True)
          ]
  
      try:
          result = executor.execute(arguments["command"])
  
          response = []
          if result.stdout:
              response.append(types.TextContent(type="text", text=result.stdout))
          if result.stderr:
              response.append(
                  types.TextContent(type="text", text=result.stderr, error=True)
              )
  
          response.append(
              types.TextContent(
                  type="text",
                  text=f"\nCommand completed with return code: {result.returncode}",
              )
          )
  
          return response
  
      except CommandSecurityError as e:
          return [
              types.TextContent(
                  type="text", text=f"Security violation: {str(e)}", error=True
              )
          ]
      except subprocess.TimeoutExpired:
          return [
              types.TextContent(
                  type="text",
                  text=f"Command timed out after {executor.security_config.command_timeout} seconds",
                  error=True,
              )
          ]
      except Exception as e:
          return [types.TextContent(type="text", text=f"Error: {str(e)}", error=True)]

• src/cli_mcp_server/server.py:447-456 (schema)

  JSON schema defining the input for the 'run_command' tool: an object with a required 'command' string property.

  inputSchema={
      "type": "object",
      "properties": {
          "command": {
              "type": "string",
              "description": "Single command to execute (example: 'ls -l' or 'cat file.txt')",
          }
      },
      "required": ["command"],
  },

• src/cli_mcp_server/server.py:439-457 (registration)

  Registration of the 'run_command' tool in the list_tools() handler, including name, security-aware description, and input schema.

  types.Tool(
      name="run_command",
      description=(
          f"Allows command (CLI) execution in the directory: {executor.allowed_dir}\n\n"
          f"Available commands: {commands_desc}\n"
          f"Available flags: {flags_desc}\n\n"
          f"Shell operators (&&, ||, |, >, >>, <, <<, ;) are {'supported' if executor.security_config.allow_shell_operators else 'not supported'}. Set ALLOW_SHELL_OPERATORS=true to enable."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "command": {
                  "type": "string",
                  "description": "Single command to execute (example: 'ls -l' or 'cat file.txt')",
              }
          },
          "required": ["command"],
      },
  ),

• src/cli_mcp_server/server.py:297-370 (helper)

  The core helper method in CommandExecutor that validates the command (checking length, security rules, shell operators) and executes it using subprocess.run, with appropriate shell=True/False, timeout, and cwd restrictions.

  def execute(self, command_string: str) -> subprocess.CompletedProcess:
      """
      Executes a command string in a secure, controlled environment.
  
      Runs the command after validating it against security constraints including length limits
      and shell operator restrictions. Executes with controlled parameters for safety.
  
      Args:
          command_string (str): The command string to execute.
  
      Returns:
          subprocess.CompletedProcess: The result of the command execution containing
              stdout, stderr, and return code.
  
      Raises:
          CommandSecurityError: If the command:
              - Exceeds maximum length
              - Fails security validation
              - Fails during execution
  
      Notes:
          - Uses shell=True for commands with shell operators, shell=False otherwise
          - Uses timeout and working directory constraints
          - Captures both stdout and stderr
      """
      if len(command_string) > self.security_config.max_command_length:
          raise CommandSecurityError(
              f"Command exceeds maximum length of {self.security_config.max_command_length}"
          )
  
      try:
          command, args = self.validate_command(command_string)
  
          # Check if this is a command with shell operators
          shell_operators = ["&&", "||", "|", ">", ">>", "<", "<<", ";"]
          use_shell = any(operator in command_string for operator in shell_operators)
  
          # Double-check that shell operators are allowed if they are present
          if use_shell and not self.security_config.allow_shell_operators:
              for operator in shell_operators:
                  if operator in command_string:
                      raise CommandSecurityError(
                          f"Shell operator '{operator}' is not supported. Set ALLOW_SHELL_OPERATORS=true to enable."
                      )
  
          if use_shell:
              # For commands with shell operators, execute with shell=True
              return subprocess.run(
                  command,  # command is the full command string in this case
                  shell=True,
                  text=True,
                  capture_output=True,
                  timeout=self.security_config.command_timeout,
                  cwd=self.allowed_dir,
              )
          else:
              # For regular commands, execute with shell=False
              return subprocess.run(
                  [command] + args,
                  shell=False,
                  text=True,
                  capture_output=True,
                  timeout=self.security_config.command_timeout,
                  cwd=self.allowed_dir,
              )
      except subprocess.TimeoutExpired:
          raise CommandTimeoutError(
              f"Command timed out after {self.security_config.command_timeout} seconds"
          )
      except CommandError:
          raise
      except Exception as e:
          raise CommandExecutionError(f"Command execution failed: {str(e)}")